百度网盘开放平台 API v3 实战：Python 脚本批量获取文件列表与生成分享链接

百度网盘开放平台 API v3 实战Python 脚本批量获取文件列表与生成分享链接在当今数字化办公环境中云存储服务已成为企业和个人日常工作中不可或缺的一部分。百度网盘作为国内领先的云存储平台其开放平台提供的 API 接口为开发者提供了强大的自动化能力。本文将深入探讨如何利用 Python 语言结合百度网盘开放平台 API v3 实现高效的文件管理自动化包括 OAuth2.0 授权、递归获取文件列表、批量创建分享链接等核心功能。1. 环境准备与授权配置在开始编写自动化脚本之前我们需要完成必要的准备工作。首先确保您已在百度开放平台注册开发者账号并创建应用。这一步骤至关重要因为只有获得合法的 API Key 和 Secret Key 才能调用百度网盘 API。1.1 注册开发者账号与创建应用访问百度开放平台官网并登录您的百度账号进入开发者中心选择创建应用填写应用基本信息特别注意选择网盘作为接入范围创建完成后记录下系统分配的 API Key 和 Secret Key1.2 安装必要的 Python 库我们将使用以下几个核心 Python 库来实现功能pip install requests requests-oauthlib python-dotenv这些库分别用于requests处理 HTTP 请求requests-oauthlib简化 OAuth2.0 授权流程python-dotenv管理环境变量1.3 配置环境变量创建一个.env文件来存储敏感信息# .env 文件内容 BAIDU_API_KEYyour_api_key BAIDU_SECRET_KEYyour_secret_key REDIRECT_URIhttps://your-domain.com/callback2. OAuth2.0 授权流程实现百度网盘 API 使用 OAuth2.0 协议进行授权这是现代 API 授权的标准方式。下面我们将详细实现完整的授权流程。2.1 初始化 OAuth2Sessionimport os from dotenv import load_dotenv from requests_oauthlib import OAuth2Session # 加载环境变量 load_dotenv() # 配置 OAuth2 参数 oauth_params { client_id: os.getenv(BAIDU_API_KEY), redirect_uri: os.getenv(REDIRECT_URI), scope: basic,netdisk, response_type: code } # 创建 OAuth2Session 实例 baidu OAuth2Session( oauth_params[client_id], redirect_urioauth_params[redirect_uri], scopeoauth_params[scope].split(,) )2.2 生成授权 URL 并获取授权码# 生成授权 URL authorization_url https://openapi.baidu.com/oauth/2.0/authorize auth_url, state baidu.authorization_url(authorization_url) print(f请访问以下 URL 完成授权:\n{auth_url}) # 用户访问授权 URL 并授权后会重定向到 REDIRECT_URI # 从重定向 URL 中提取授权码 redirect_response input(请输入完整的重定向 URL: ) code baidu.parse_authorization_response(redirect_response)[code]2.3 获取访问令牌# 获取访问令牌的端点 token_url https://openapi.baidu.com/oauth/2.0/token # 准备令牌请求参数 token_params { client_id: os.getenv(BAIDU_API_KEY), client_secret: os.getenv(BAIDU_SECRET_KEY), code: code, redirect_uri: os.getenv(REDIRECT_URI), grant_type: authorization_code } # 获取访问令牌 token baidu.fetch_token(token_url, **token_params) print(f获取到的访问令牌:\n{token})3. 文件列表获取与处理获得有效的访问令牌后我们就可以开始与百度网盘 API 进行交互了。首先实现获取文件列表的功能。3.1 获取指定路径下的文件列表def get_file_list(access_token, path/, recursion0): 获取指定路径下的文件列表 :param access_token: OAuth2 访问令牌 :param path: 要查询的路径默认为根目录 :param recursion: 是否递归获取子目录0-不递归1-递归 :return: 文件列表 url https://pan.baidu.com/rest/2.0/xpan/file headers { User-Agent: pan.baidu.com, Authorization: fBearer {access_token} } params { method: list, dir: path, recursion: recursion, web: 0 } response requests.get(url, headersheaders, paramsparams) response.raise_for_status() return response.json()[list]3.2 递归获取所有文件对于需要批量处理大量文件的情况递归获取整个目录结构非常有用def get_all_files(access_token, root_path/): 递归获取指定路径下的所有文件 :param access_token: OAuth2 访问令牌 :param root_path: 起始路径 :return: 所有文件的列表 all_files [] def _recursive_get(path): items get_file_list(access_token, path, recursion0) for item in items: if item[isdir] 1: # 如果是目录 _recursive_get(item[path]) # 递归获取子目录 else: all_files.append(item) _recursive_get(root_path) return all_files3.3 文件列表过滤与处理获取到文件列表后我们可能需要对文件进行筛选和处理def filter_files_by_extension(files, extensions): 按文件扩展名过滤文件 :param files: 文件列表 :param extensions: 允许的扩展名列表如 [.pdf, .docx] :return: 过滤后的文件列表 return [f for f in files if os.path.splitext(f[server_filename])[1].lower() in extensions] def sort_files_by_size(files, reverseFalse): 按文件大小排序 :param files: 文件列表 :param reverse: 是否降序排列 :return: 排序后的文件列表 return sorted(files, keylambda x: x[size], reversereverse)4. 批量创建分享链接获取文件列表后我们可以为这些文件批量创建分享链接。百度网盘 API 提供了灵活的分享功能可以设置有效期、密码等参数。4.1 创建单个文件分享链接def create_share_link(access_token, fs_ids, passwordNone, period7): 创建文件分享链接 :param access_token: OAuth2 访问令牌 :param fs_ids: 文件ID列表 :param password: 分享密码None表示无密码 :param period: 有效期(天)最大为7 :return: 分享信息 url https://pan.baidu.com/rest/2.0/xpan/share headers { User-Agent: pan.baidu.com, Authorization: fBearer {access_token} } data { method: set, fs_ids: json.dumps(fs_ids), period: min(period, 7) } if password: data[pwd] str(password) response requests.post(url, headersheaders, datadata) response.raise_for_status() return response.json()4.2 批量创建分享链接为了提高效率我们可以批量处理多个文件的分享def batch_create_share_links(access_token, files, batch_size10): 批量创建分享链接 :param access_token: OAuth2 访问令牌 :param files: 文件列表 :param batch_size: 每批处理的文件数量 :return: 所有分享链接的列表 all_share_links [] for i in range(0, len(files), batch_size): batch files[i:ibatch_size] fs_ids [f[fs_id] for f in batch] try: result create_share_link(access_token, fs_ids) if result[errno] 0: share_links [{ filename: f[server_filename], link: fhttps://pan.baidu.com/s/{result[shareid]}?pwd{result[pwd]} } for f in batch] all_share_links.extend(share_links) except Exception as e: print(f批量分享失败: {str(e)}) return all_share_links4.3 分享链接管理创建分享链接后我们可能还需要管理这些链接def list_share_links(access_token): 列出所有分享链接 :param access_token: OAuth2 访问令牌 :return: 分享链接列表 url https://pan.baidu.com/rest/2.0/xpan/share headers { User-Agent: pan.baidu.com, Authorization: fBearer {access_token} } params { method: list } response requests.get(url, headersheaders, paramsparams) response.raise_for_status() return response.json()[list] def cancel_share_links(access_token, share_ids): 取消分享 :param access_token: OAuth2 访问令牌 :param share_ids: 要取消的分享ID列表 :return: 操作结果 url https://pan.baidu.com/rest/2.0/xpan/share headers { User-Agent: pan.baidu.com, Authorization: fBearer {access_token} } data { method: cancel, shareid_list: json.dumps(share_ids) } response requests.post(url, headersheaders, datadata) response.raise_for_status() return response.json()5. 完整脚本与错误处理将上述功能整合成一个完整的脚本并添加适当的错误处理和日志记录功能。5.1 完整脚本示例import os import json import logging from dotenv import load_dotenv from requests_oauthlib import OAuth2Session import requests # 配置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s ) logger logging.getLogger(__name__) class BaiduPanManager: def __init__(self): load_dotenv() self.client_id os.getenv(BAIDU_API_KEY) self.client_secret os.getenv(BAIDU_SECRET_KEY) self.redirect_uri os.getenv(REDIRECT_URI) self.access_token None self.refresh_token None self.session None def authorize(self): 完整的授权流程 try: # 初始化 OAuth2Session self.session OAuth2Session( self.client_id, redirect_uriself.redirect_uri, scope[basic, netdisk] ) # 生成授权 URL authorization_url https://openapi.baidu.com/oauth/2.0/authorize auth_url, _ self.session.authorization_url(authorization_url) logger.info(f请访问以下 URL 完成授权:\n{auth_url}) # 获取授权码 redirect_response input(请输入完整的重定向 URL: ) code self.session.parse_authorization_response(redirect_response)[code] # 获取访问令牌 token_url https://openapi.baidu.com/oauth/2.0/token token self.session.fetch_token( token_url, client_secretself.client_secret, codecode ) self.access_token token[access_token] self.refresh_token token.get(refresh_token) logger.info(授权成功已获取访问令牌) except Exception as e: logger.error(f授权过程中发生错误: {str(e)}) raise def refresh_access_token(self): 刷新访问令牌 if not self.refresh_token: raise ValueError(没有可用的刷新令牌) try: token_url https://openapi.baidu.com/oauth/2.0/token token self.session.refresh_token( token_url, client_idself.client_id, client_secretself.client_secret, refresh_tokenself.refresh_token ) self.access_token token[access_token] self.refresh_token token.get(refresh_token, self.refresh_token) logger.info(访问令牌已刷新) except Exception as e: logger.error(f刷新令牌时发生错误: {str(e)}) raise def get_file_list(self, path/, recursion0): 获取文件列表 try: url https://pan.baidu.com/rest/2.0/xpan/file headers { User-Agent: pan.baidu.com, Authorization: fBearer {self.access_token} } params { method: list, dir: path, recursion: recursion, web: 0 } response requests.get(url, headersheaders, paramsparams) response.raise_for_status() return response.json()[list] except requests.exceptions.HTTPError as e: if e.response.status_code 401: # 未授权 logger.warning(访问令牌可能已过期尝试刷新...) self.refresh_access_token() return self.get_file_list(path, recursion) logger.error(f获取文件列表失败: {str(e)}) raise def create_share_links(self, fs_ids, passwordNone, period7): 创建分享链接 try: url https://pan.baidu.com/rest/2.0/xpan/share headers { User-Agent: pan.baidu.com, Authorization: fBearer {self.access_token} } data { method: set, fs_ids: json.dumps(fs_ids), period: min(period, 7) } if password: data[pwd] str(password) response requests.post(url, headersheaders, datadata) response.raise_for_status() result response.json() if result[errno] ! 0: raise ValueError(fAPI 返回错误: {result.get(errmsg, 未知错误)}) return result except requests.exceptions.HTTPError as e: if e.response.status_code 401: # 未授权 logger.warning(访问令牌可能已过期尝试刷新...) self.refresh_access_token() return self.create_share_links(fs_ids, password, period) logger.error(f创建分享链接失败: {str(e)}) raise def export_links_to_csv(self, links, filenameshare_links.csv): 将分享链接导出为 CSV 文件 import csv try: with open(filename, w, newline, encodingutf-8) as f: writer csv.writer(f) writer.writerow([文件名, 分享链接]) for link in links: writer.writerow([link[filename], link[link]]) logger.info(f分享链接已成功导出到 {filename}) except Exception as e: logger.error(f导出 CSV 文件失败: {str(e)}) raise if __name__ __main__: try: manager BaiduPanManager() manager.authorize() # 获取所有文件 files manager.get_file_list(recursion1) logger.info(f共获取到 {len(files)} 个文件) # 过滤出 PDF 文件 pdf_files [f for f in files if f[server_filename].lower().endswith(.pdf)] logger.info(f找到 {len(pdf_files)} 个 PDF 文件) # 批量创建分享链接 fs_ids [f[fs_id] for f in pdf_files[:10]] # 限制前10个文件 share_result manager.create_share_links(fs_ids, password1234) # 构建分享链接列表 share_links [{ filename: f[server_filename], link: fhttps://pan.baidu.com/s/{share_result[shareid]}?pwd{share_result[pwd]} } for f in pdf_files[:10]] # 导出到 CSV manager.export_links_to_csv(share_links) except Exception as e: logger.error(f程序运行出错: {str(e)})5.2 错误处理最佳实践在实际应用中我们需要考虑各种可能的错误情况网络请求错误添加重试机制和超时设置API 限流实现请求速率限制和退避策略令牌过期自动刷新访问令牌文件不存在检查文件状态后再操作权限不足明确提示用户需要哪些权限def safe_api_call(self, func, *args, max_retries3, **kwargs): 带错误处理和重试的 API 调用封装 last_exception None for attempt in range(max_retries): try: return func(*args, **kwargs) except requests.exceptions.HTTPError as e: last_exception e if e.response.status_code 401: # 未授权 self.refresh_access_token() kwargs[headers] kwargs.get(headers, {}) kwargs[headers][Authorization] fBearer {self.access_token} elif e.response.status_code 429: # 请求过多 retry_after int(e.response.headers.get(Retry-After, 5)) logger.warning(fAPI 限流等待 {retry_after} 秒后重试...) time.sleep(retry_after) else: break except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e: last_exception e wait_time (attempt 1) * 2 # 指数退避 logger.warning(f网络错误等待 {wait_time} 秒后重试...) time.sleep(wait_time) logger.error(fAPI 调用失败已达到最大重试次数 {max_retries}) raise last_exception if last_exception else Exception(未知错误)6. 高级功能与性能优化对于需要处理大量文件或要求高性能的场景我们可以进一步优化脚本。6.1 多线程处理使用线程池加速批量操作from concurrent.futures import ThreadPoolExecutor def batch_process_files(self, files, process_func, batch_size10, max_workers4): 多线程批量处理文件 results [] def process_batch(batch): try: fs_ids [f[fs_id] for f in batch] return process_func(fs_ids) except Exception as e: logger.error(f处理批次时出错: {str(e)}) return None with ThreadPoolExecutor(max_workersmax_workers) as executor: futures [] for i in range(0, len(files), batch_size): batch files[i:ibatch_size] futures.append(executor.submit(process_batch, batch)) for future in futures: result future.result() if result: results.append(result) return results6.2 缓存机制减少重复 API 调用提高性能import pickle from functools import lru_cache class BaiduPanManager: def __init__(self, cache_filebaidu_cache.pkl): self.cache_file cache_file self._load_cache() def _load_cache(self): 加载缓存 try: with open(self.cache_file, rb) as f: self.cache pickle.load(f) except (FileNotFoundError, EOFError): self.cache {file_list: {}, share_links: {}} def _save_cache(self): 保存缓存 with open(self.cache_file, wb) as f: pickle.dump(self.cache, f) lru_cache(maxsize100) def get_file_list_cached(self, path/, recursion0): 带缓存的文件列表获取 cache_key f{path}:{recursion} if cache_key in self.cache[file_list]: return self.cache[file_list][cache_key] files self.get_file_list(path, recursion) self.cache[file_list][cache_key] files self._save_cache() return files6.3 增量同步功能实现本地与云端文件的增量同步def sync_local_with_cloud(self, local_dir, cloud_dir/, excludeNone): 同步本地目录与云端目录 :param local_dir: 本地目录路径 :param cloud_dir: 云端目录路径 :param exclude: 要排除的文件模式列表 if exclude is None: exclude [] # 获取云端文件列表 cloud_files self.get_file_list(cloud_dir) cloud_files {f[server_filename]: f for f in cloud_files if not any( fnmatch.fnmatch(f[server_filename], pattern) for pattern in exclude )} # 获取本地文件列表 local_files {} for root, _, files in os.walk(local_dir): for filename in files: if any(fnmatch.fnmatch(filename, pattern) for pattern in exclude): continue filepath os.path.join(root, filename) relpath os.path.relpath(filepath, local_dir) local_files[relpath] { path: filepath, mtime: os.path.getmtime(filepath), size: os.path.getsize(filepath) } # 比较并执行同步操作 for relpath, local_info in local_files.items(): cloud_file cloud_files.get(relpath) if not cloud_file: # 本地有但云端没有 self.upload_file(local_info[path], os.path.join(cloud_dir, relpath)) elif (local_info[size] ! cloud_file[size] or local_info[mtime] cloud_file[server_mtime]): # 文件有变化 self.upload_file(local_info[path], os.path.join(cloud_dir, relpath)) for filename, cloud_file in cloud_files.items(): if filename not in local_files: # 云端有但本地没有 local_path os.path.join(local_dir, filename) os.makedirs(os.path.dirname(local_path), exist_okTrue) self.download_file(cloud_file[fs_id], local_path)7. 安全注意事项与最佳实践在使用百度网盘 API 进行开发时安全性是不可忽视的重要方面。7.1 敏感信息保护不要硬编码密钥始终使用环境变量或配置文件使用最小权限原则只申请必要的 API 权限定期轮换密钥定期更新 API Key 和 Secret Key保护访问令牌不要将令牌记录在日志或版本控制系统中7.2 API 使用限制百度网盘 API 有以下限制需要注意请求频率限制通常为每分钟 100-200 次请求分享链接有效期最长 7 天单次请求最大文件数通常为 100 个每日分享次数限制根据账号等级不同7.3 合规使用建议遵守百度网盘用户协议确保您的使用方式符合平台规定避免滥用 API合理控制请求频率避免影响服务稳定性用户隐私保护如果处理用户数据确保符合相关法律法规错误处理优雅地处理各种错误情况避免无限重试def check_api_limits(self): 检查 API 使用限制 url https://pan.baidu.com/rest/2.0/xpan/multimedia headers { User-Agent: pan.baidu.com, Authorization: fBearer {self.access_token} } params { method: quota } response requests.get(url, headersheaders, paramsparams) response.raise_for_status() quota_info response.json() logger.info(fAPI 配额信息: 总空间 {quota_info[total]/1024**3:.2f}GB, f已用 {quota_info[used]/1024**3:.2f}GB, f剩余 {quota_info[free]/1024**3:.2f}GB) return quota_info