# 使用Python调用FileBrowser REST API实现文件上传的完整指南 FileBrowser是一个轻量级的文件管理器，提供了丰富的REST API接口，可以通过Python实现自动化文件操作。下面将详细介绍如何使用Python调用FileBrowser的REST API实现文件上传功能。 ## 1. FileBrowser REST API基础配置 ### 1.1 认证配置 FileBrowser API使用基于令牌的身份验证，需要在请求头中添加认证信息： ```python import requests import json # FileBrowser基础配置 FILEBROWSER_BASE_URL = "http://your-filebrowser-domain:port" USERNAME = "your_username" PASSWORD = "your_password" ``` ### 1.2 获取认证令牌 首先需要获取访问令牌才能调用其他API接口： ```python def get_auth_token(base_url, username, password): """ 获取FileBrowser认证令牌 """ auth_url = f"{base_url}/api/login" auth_data = { "username": username, "password": password } try: response = requests.post(auth_url, json=auth_data) response.raise_for_status() token = response.json().get("token") return token except requests.exceptions.RequestException as e: print(f"认证失败: {e}") return None # 使用示例 auth_token = get_auth_token(FILEBROWSER_BASE_URL, USERNAME, PASSWORD) print(f"获取到的令牌: {auth_token}") ``` ## 2. 文件上传实现方法 ### 2.1 基础文件上传 FileBrowser提供了多种文件上传方式，以下是使用multipart/form-data格式上传单个文件的方法： ```python def upload_file_single(base_url, token, file_path, target_path="/"): """ 上传单个文件到FileBrowser :param base_url: FileBrowser基础URL :param token: 认证令牌 :param file_path: 本地文件路径 :param target_path: 目标目录路径 """ upload_url = f"{base_url}/api/resources{target_path}" headers = { "X-Auth": token } try: with open(file_path, 'rb') as file: files = { 'file': (os.path.basename(file_path), file, 'application/octet-stream') } response = requests.post(upload_url, headers=headers, files=files) response.raise_for_status() print(f"文件上传成功: {file_path} -> {target_path}") return response.json() except FileNotFoundError: print(f"文件不存在: {file_path}") return None except requests.exceptions.RequestException as e: print(f"上传失败: {e}") return None # 使用示例 result = upload_file_single( FILEBROWSER_BASE_URL, auth_token, "/path/to/local/file.txt", "/uploads/" ) ``` ### 2.2 批量文件上传 对于需要上传多个文件的情况，可以实现批量上传功能： ```python import os from pathlib import Path def upload_files_batch(base_url, token, local_directory, target_directory="/"): """ 批量上传目录中的所有文件 :param local_directory: 本地目录路径 :param target_directory: 目标目录路径 """ success_count = 0 fail_count = 0 for file_path in Path(local_directory).rglob('*'): if file_path.is_file(): relative_path = file_path.relative_to(local_directory) target_path = f"{target_directory}/{relative_path}" result = upload_file_single(base_url, token, str(file_path), target_path) if result: success_count += 1 print(f"✓ 成功上传: {file_path.name}") else: fail_count += 1 print(f"✗ 上传失败: {file_path.name}") print(f"\n上传完成: 成功 {success_count} 个, 失败 {fail_count} 个") return success_count, fail_count # 使用示例 success, fails = upload_files_batch( FILEBROWSER_BASE_URL, auth_token, "/local/photos", "/backup/photos" ) ``` ## 3. 高级文件操作功能 ### 3.1 带进度显示的文件上传 对于大文件上传，可以添加进度显示功能： ```python def upload_file_with_progress(base_url, token, file_path, target_path="/"): """ 带进度显示的文件上传 """ upload_url = f"{base_url}/api/resources{target_path}" headers = { "X-Auth": token } file_size = os.path.getsize(file_path) def read_in_chunks(file_object, chunk_size=8192): """生成器函数，分块读取文件""" while True: data = file_object.read(chunk_size) if not data: break yield data try: with open(file_path, 'rb') as file: uploaded_size = 0 # 创建自定义文件对象用于跟踪进度 class ProgressFile: def __init__(self, file_obj): self.file_obj = file_obj self.uploaded = 0 def read(self, size=-1): data = self.file_obj.read(size) self.uploaded += len(data) progress = (self.uploaded / file_size) * 100 print(f"\r上传进度: {progress:.1f}% ({self.uploaded}/{file_size} bytes)", end="") return data progress_file = ProgressFile(file) files = {'file': (os.path.basename(file_path), progress_file)} response = requests.post(upload_url, headers=headers, files=files) response.raise_for_status() print("\n✓ 文件上传完成!") return response.json() except Exception as e: print(f"\n✗ 上传失败: {e}") return None ``` ### 3.2 文件上传前的目录检查与创建 确保目标目录存在，如果不存在则自动创建： ```python def ensure_directory_exists(base_url, token, directory_path): """ 检查目录是否存在，如果不存在则创建 """ check_url = f"{base_url}/api/resources{directory_path}" create_url = f"{base_url}/api/resources{directory_path}" headers = { "X-Auth": token } # 检查目录是否存在 response = requests.get(check_url, headers=headers) if response.status_code == 200: return True # 创建目录 create_data = { "name": os.path.basename(directory_path), "type": "directory" } try: response = requests.post(create_url, headers=headers, json=create_data) response.raise_for_status() print(f"目录创建成功: {directory_path}") return True except requests.exceptions.RequestException as e: print(f"目录创建失败: {e}") return False def upload_file_safe(base_url, token, file_path, target_path="/"): """ 安全上传文件：确保目标目录存在 """ target_dir = os.path.dirname(target_path) if ensure_directory_exists(base_url, token, target_dir): return upload_file_single(base_url, token, file_path, target_path) else: print("目录准备失败，无法上传文件") return None ``` ## 4. 错误处理与重试机制 ### 4.1 完善的错误处理 ```python class FileBrowserClient: def __init__(self, base_url, username, password): self.base_url = base_url self.username = username self.password = password self.token = None self.session = requests.Session() def authenticate(self, max_retries=3): """带重试的认证方法""" for attempt in range(max_retries): try: self.token = get_auth_token(self.base_url, self.username, self.password) if self.token: self.session.headers.update({"X-Auth": self.token}) return True except Exception as e: print(f"认证尝试 {attempt + 1} 失败: {e}") if attempt == max_retries - 1: raise Exception("认证失败，已达到最大重试次数") return False def upload_with_retry(self, file_path, target_path, max_retries=3): """带重试的文件上传""" for attempt in range(max_retries): try: result = upload_file_single( self.base_url, self.token, file_path, target_path ) if result: return result except Exception as e: print(f"上传尝试 {attempt + 1} 失败: {e}") if attempt == max_retries - 1: raise Exception("文件上传失败，已达到最大重试次数") return None # 使用完整的客户端 client = FileBrowserClient(FILEBROWSER_BASE_URL, USERNAME, PASSWORD) if client.authenticate(): result = client.upload_with_retry( "/path/to/important/file.pdf", "/documents/backup/" ) ``` ## 5. 实际应用场景示例 ### 5.1 自动化备份脚本 ```python import schedule import time from datetime import datetime def automated_backup(): """自动化备份重要文件到FileBrowser""" print(f"开始自动备份: {datetime.now()}") client = FileBrowserClient(FILEBROWSER_BASE_URL, USERNAME, PASSWORD) if client.authenticate(): # 备份重要配置文件 important_files = [ "/etc/nginx/nginx.conf", "/home/user/documents/*.pdf", "/var/log/important.log" ] backup_dir = f"/backups/{datetime.now().strftime('%Y%m%d_%H%M%S')}" for file_pattern in important_files: for file_path in Path().glob(file_pattern): if file_path.exists(): target_path = f"{backup_dir}/{file_path.name}" client.upload_with_retry(str(file_path), target_path) print("自动备份完成") # 设置定时任务：每天凌晨2点执行备份 schedule.every().day.at("02:00").do(automated_backup) while True: schedule.run_pending() time.sleep(60) ``` ### 5.2 文件同步监控 ```python import hashlib def calculate_file_hash(file_path): """计算文件哈希值用于验证""" hash_md5 = hashlib.md5() with open(file_path, "rb") as f: for chunk in iter(lambda: f.read(4096), b""): hash_md5.update(chunk) return hash_md5.hexdigest() def sync_directory_with_verification(local_dir, remote_dir): """带验证的目录同步""" client = FileBrowserClient(FILEBROWSER_BASE_URL, USERNAME, PASSWORD) client.authenticate() for local_file in Path(local_dir).rglob('*'): if local_file.is_file(): remote_path = f"{remote_dir}/{local_file.relative_to(local_dir)}" local_hash = calculate_file_hash(local_file) # 上传文件 result = client.upload_with_retry(str(local_file), remote_path) if result: print(f"✓ 同步成功: {local_file.name}") else: print(f"✗ 同步失败: {local_file.name}") ``` 通过上述完整的Python实现，您可以轻松地集成FileBrowser REST API到您的自动化工作流中，实现高效、可靠的文件上传操作。这些代码示例涵盖了从基础认证到高级功能的各种场景，可以根据实际需求进行调整和扩展[ref_1]。