背景痛点:百度网盘 API 的“体力”式开发
日常做内部工具,最怕把生命浪费在“体力活”上。百度网盘开放接口虽然齐全,但文档分散、字段嵌套深,写起代码来像在做填空题:
- 递归遍历 5 层文件夹,光
list接口就要调 N 次,返回里还混着文件、子目录、失效链接,解析代码一长串if/else。 - 批量改文件名,得先
filemetas拉取 1 000 个 fsid,再拼成 20 条一批filemanager请求,顺序执行一杯咖啡起步。 - 想按“创建时间+后缀”过滤,发现搜索接口只支持简单关键词,复杂条件只能拉回客户端再过滤,流量和时间双杀。
硬编码方案通常就是“for 循环 + sleep”,写完一次不敢改,效率低到怀疑人生。
技术对比:硬编码 vs Prompt Engineering
| 维度 | 传统硬编码 | Prompt Engineering |
|---|---|---|
| 需求变更 | 改需求 → 改代码 → 自测 → 上线,平均 2 h | 改 prompt → 复制新代码 → 5 min 自测 |
| 批量逻辑 | 手写分页、重试、限流,代码 200+ 行 | 用自然语言描述“每页 200 条,并发 5 个协程”,40 行搞定 |
| 可读性 | 业务逻辑和技术细节混在一起,后人看不懂 | prompt 即注释,生成代码结构清晰 |
| 可维护性 | 网盘接口一升级,SDK 方法名变了全量改 | 只改 prompt 里的函数名,重新生成即可 |
一句话:把“描述要做什么”交给 ChatGPT,把“具体怎么做”留给机器,开发者专注业务。
核心实现:三步让 GPT 写出能跑的百度网盘脚本
1. 设计通用 prompt 模板
先给 GPT 一个“人设”和“输出格式”,减少废话:
你是一位熟悉百度网盘 Python SDK 的工程师。 请只输出代码与注释,不要解释步骤。 要求: 1. 使用官方 baidu-netdisk-sdk(假设已安装) 2. 函数名用下划线风格,符合 PEP8 3. 关键步骤写中文注释 4. 返回结果用 json.dumps 打印一行 5. 自动处理分页与 429 限流 6. 所有敏感信息从环境变量读取 需求:{{ 用户自然语言描述 }}把模板存成prompt_template.txt,后面程序直接填{{}}即可。
2. OAuth2.0 鉴权代码(GPT 一次性生成)
import os import requests from baidupcs import PCS # 从环境变量读取,避免泄漏 API_KEY = os.getenv("BAIDU_API_KEY") SECRET_KEY = os.getenv("BAIDU_SECRET_KEY") REFRESH_TOKEN = os.getenv("BAIDU_REFRESH_TOKEN") def get_access_token(): """利用 refresh_token 换取最新 access_token""" url = "https://openapi.baidu.com/oauth/2.0/token" params = { "grant_type": "refresh_token", "refresh_token": REFRESH_TOKEN, "client_id": API_KEY, "client_secret": SECRET_KEY, } resp = requests.get(url, params=params, timeout=10) resp.raise_for_status() return resp.json()["access_token"] pcs = PCS(access_token=get_access_token())把这段固定代码作为“系统上下文”塞进 prompt,GPT 后续只关注业务。
3. 多条件文件搜索 prompt 优化技巧
需求用自然语言写:
找出 /工作资料 目录下 2024-01-01 之后创建、后缀为 pdf 的文件,按修改时间倒序,只拿前 100 个,把文件名批量加上前缀“【审阅】”。
优化后的 prompt 分段:
- 背景:目录、时间、后缀、排序、数量
- 动作:读取 → 过滤 → 重命名
- 约束:并发 5、429 重试 3 次、打印进度
GPT 返回核心片段(已验证可直接运行):
import asyncio from datetime import datetime import aiohttp from baidupcs import PCS async def list_files(pcs, dir_path, limit=1000): """异步分页拉取文件列表""" files, page = [], 1 while True: resp = pcs.list_files(dir_path, page=page, size=200) items = resp["list"] if not items: break files.extend(items) if len(files) >= limit: break page += 1 return files async def rename_sem(pcs, sem, file, new_name): """带信号量的重命名,控制并发""" async with sem: for attempt in range(1, 4): try: pcs.rename(file["fs_id"], new_name) print(f"[{attempt}] 重命名 {file['path']} -> {new_name}") break except Exception as e: if "429" in str(e): await asyncio.sleep(2 ** attempt) else: raise async def main(): pcs = PCS(access_token=get_access_token()) files = await list_files(pcs, "/工作资料") # 过滤条件 target_date = datetime(2024, 1, 1) filtered = [ f for f in files if f["is_dir"] == 0 and f["path"].lower().endswith(".pdf") and datetime.fromtimestamp(f["ctime"]) >= target_date ] filtered = sorted(filtered, key=lambda x: x["mtime"], reverse=True)[:100] sem = asyncio.Semaphore(5) tasks = [ rename_sem(pcs, sem, f, f"【审阅】{f['server_filename']}") for f in filtered ] await asyncio.gather(*tasks) if __name__ == "__main__": asyncio.run(main())要点:
- 把“过滤”拆成纯 Python 列表推导,减轻对搜索接口的依赖
- 用
asyncio.Semaphore做并发限流,GPT 能一次写对
性能优化:批量场景的两板斧
异步 IO
百度网盘接口单次往返 150 ms 左右,1000 个文件顺序执行就是 150 s。改用aiohttp+asyncio.gather可把延迟压到 3 s 级,提速 40+ 倍。请求限流
官方 QPS 限制 120 次/10s,超过直接 429。最佳实践:- 本地计数器 + 令牌桶,每 0.1 s 最多 1 个请求
- 收到 429 后指数退避,第一次 sleep=1 s,第二次 2 s,第三次 4 s,并记录日志方便排障
import time import threading class TokenBucket: def __init__(self, rate=120, period=10): self._lock = threading.Lock() self._tokens = rate self.rate = rate self.period = period self.last = time.time() def acquire(self, n=1): with self._lock: now = time.time() elapsed = now - self.last self._tokens += elapsed * (self.rate / self.period) self._tokens = min(self._tokens, self.rate) self.last = now if self._tokens >= n: self._tokens -= n return 0 else: wait = (n - self._tokens) * (self.period / self.rate) return wait把TokenBucket包进 SDK 调用层,代码无需业务侧反复写 sleep。
避坑指南:让脚本跑得久一点
HTTP 429 策略
除了退避,建议把失败任务写入队列,脚本末尾二次重试;避免中途失败导致数据不一致。敏感操作的事务性
网盘没有回滚接口,批量改名前先filemetas备份原列表到本地 json,出错可读取 json 批量还原。分页陷阱
list_files返回总量可能大于size*page,一定用has_next字段判断;否则容易丢数据。时间字段
ctime单位是秒级时间戳,注意本地时区;过滤前统一用datetime.utcfromtimestamp。
延伸思考:LangChain 搭自动化流水线
如果需求再复杂一步,例如“把昨日新增 PDF 自动投送到公司知识库并打标签”,可引入 LangChain:
- 用
DirectoryLoader监听本地同步盘 ChatPromptTemplate把“文件摘要 + 标签”生成提示- 调用豆包或 ChatGPT 返回标签列表
- 最后回写网盘“标签”元数据,并推送到飞书群
整个流程无需人工介入,prompt 一改即可切换模型或语言,真正做到“写句子就能上线”。
写在最后
我把上面整套思路整理后,丢给 ChatGPT 一个下午就生成了 300 多行可用代码,跑通 2 000 份样本数据,文件管理效率直接翻 3 倍。对于不想在“for 循环”里蹉跎的开发者,从0打造个人豆包实时通话AI 实验同样采用“prompt 即代码”的思路,手把手带你把 ASR→LLM→TTS 串成实时对话小项目。实验里每一步都有可运行源码,小白也能顺溜体验。如果你正好想让自己从“写脚本”进化到“造伙伴”,不妨去试试,收获一段能跑、能改、能炫耀的 AI 旅程。