简介:这是一款面向百度网盘批量转存的浏览器扩展源码,适合计算机、数学、电子信息等专业学生作为课程设计、期末大作业或毕业设计参考资料,可有效解决多文件手动逐条转存效率低、重复操作多的问题,同时也能帮助前端开发者快速了解浏览器扩展的完整开发流程。压缩包共12个文件,以JavaScript逻辑脚本为主,配合HTML弹窗页面、CSS进度样式、PNG/JPG图标资源和manifest配置文件,整体仅31KB,代码精简、结构清晰,便于按模块研读与调试。目前已有663人学习下载,具备一定参考热度。源码完整可运行,涵盖弹出窗口、进度展示、批量处理与配置声明等完整实现,读者既能借鉴其界面交互和数据请求写法,也可在此基础上升级为支持更多网盘或自定义转存任务,是练手与二次开发的不错起点。
1. 批量转存不是下载:先看百度网盘转存的接口本质
网上流传的“百度网盘批量转存工具源码.zip”,下载下来后大概率是一堆.py文件加一份写得不完整的 README。真正跑不起来的原因并不是代码本身,而是它默认你已经知道:转存不是下载,而是把别人分享的文件“保存到你自己网盘”的一次接口调用。手动操作时你只需要打开分享链接、点“保存到网盘”,批量化之后就要自己处理登录态、分享信息解析、签名参数、并发和重试。这篇文章把这套链路拆开讲清楚:先理解接口,再写最小批量脚本,最后补上工程化兜底。与其在“免费 python 源码大全”里碰运气,不如自己把这条转存链路完整搭一遍,跑通了它就是个能长期改的框架,而不是一个解不开的谜。
2. 从分享链接到转存参数:解析分享页与构造 save 请求
2.1 手动点“保存到网盘”时,浏览器实际做了什么
打开一个带提取码的分享链接,手动保存到网盘,背后至少发生了三次请求:第一次打开分享详情页,拿到分享者 UID(uk)、分享 ID(shareid)、文件列表(fid_list);第二次提交提取码,换回一个用于转存的凭证;第三次才是真正执行保存动作,也就是share/save接口。
常见做法是直接看浏览器开发者工具里 Network 面板的请求记录,把“保存到网盘”前后那几条 XHR 请求的 URL、参数和返回结构抄下来。这个工具的“源码”价值,其实就是对这三步请求的封装。拿到手之后,你要先确认它请求的 URL 和参数跟当前网页版是否一致,因为百度网盘前端接口的参数名变动并不少见。
请求链路里有一个关键点:转存动作本身是服务端到服务端的文件复制,不经过本地流量。也就是说,批量转存的耗时取决于接口响应速度,而不是你的带宽。这决定了后面做并发设计时可以走线程池而不是协程池。
2.2 登录态:BDUSS 是唯一鉴权依据
所有转存请求都依赖登录态。常见做法是从浏览器 Cookie 里取BDUSS,它是账号凭证的核心;STOKEN一类字段在某些接口里也会用到。
import requests session = requests.Session() session.headers.update({ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Referer": "https://pan.baidu.com/" }) session.cookies.set("BDUSS", "你的BDUSS值")逻辑说明:session.cookies.set把BDUSS注入到后续所有请求的 Cookie 头里。User-Agent和Referer也要带上,否则部分接口直接拒绝。写工具时应把BDUSS放到独立配置文件里,而不是写死在源码中,这样源码 zip 给别人看时不会泄号。
注意:BDUSS 等同于网盘账号的登录凭证,泄漏后别人可以直接操作你的网盘。做分享或开源时务必从代码里剥离。
2.3 解析分享详情:拿 shareid、uk、fid_list
这一步的行为是:先请求分享链接对应的详情页,从返回内容里解析出分享信息;如果链接带提取码,还需要先提交提取码。
import re import requests def get_share_detail(session, share_url, pwd=None): # 提取分享链接中的 surl 标识 surl = share_url.split("/s/")[-1].split("?")[0] url = "https://pan.baidu.com/s/" + surl resp = session.get(url, allow_redirects=True) text = resp.text # 部分页面会把 shareid/uk 直接写在 HTML 的 data 属性里 shareid_match = re.search(r'"shareid":\s*(\d+)', text) uk_match = re.search(r'"uk":\s*(\d+)', text) # 较新的网页端是 SPA,分享页 HTML 里可能拿不到完整参数 # 常见做法是改请求 web/share/list 接口拿 JSON if "(list)" in text or shareid_match is None: api_url = "https://pan.baidu.com/share/list" params = { "uk": uk_match.group(1) if uk_match else "", "shareid": shareid_match.group(1) if shareid_match else "", "order": "time", "desc": "1", "showempty": "0", "web": "1", "page": "1", "num": "1000", } if pwd: params["pwd"] = pwd api_resp = session.get(api_url, params=params) return api_resp.json() return text参数说明:
surl是链接里的短标识,从/s/后面取一段,去掉查询参数。- 先直接 GET 分享页,部分老页面会把
shareid和uk直接塞在 HTML 里,正则能拿到。 - 如果拿不到,就请求
share/list接口,num=1000表示单页返回 1000 个文件条目,超大批量时要翻页。 - 带提取码时把
pwd加进查询参数,有些接口还要额外 POST 一次提取码校验。
手头拿到别人的工具源码时,优先看这个函数:如果它还在解析旧版 HTML 结构,基本可以判断已经失效。新版接口返回的是 JSON,字段更稳定。
2.4 构造转存请求:save 接口与签名参数
执行转存的核心请求是share/save,常见参数如下:
| 参数 | 含义 | 来源 |
|---|---|---|
shareid | 分享 ID | 分享详情返回 |
from | 分享者标识,填uk值 | 分享详情返回 |
sekey | 转存凭证,提取码校验后返回 | 提取码校验/randsk |
fid_list | 要转存的文件 ID 列表 | file_list返回 |
path | 你网盘里的目标目录 | 自己指定 |
sign | 动态签名 | 网页端 JS 生成 |
timestamp | 时间戳 | 网页端 JS 生成 |
sign和timestamp是从网盘首页加载的 JS 文件里生成出来的。常见做法是下载首页引用的 JS,用正则抓取生成逻辑,然后在工具里复刻一遍。
import re, time, hashlib def get_sign(session): home = session.get("https://pan.baidu.com/") js_urls = re.findall(r'<script[^>]+src="([^"]+\.js[^"]*)"', home.text) for js_url in js_urls[:10]: js_text = session.get(js_url if js_url.startswith("http") else "https:" + js_url).text sign_match = re.search(r'"sign"\s*:\s*function[^}]+', js_text) if sign_match: # 实际生产环境是从匹配到的逻辑里计算而成,这里只做占位 break timestamp = int(time.time() * 1000) return timestamp, timestamp # 返回占位,需按抓到的逻辑替换逻辑说明:这个函数的核心思路是“动态获取签名逻辑”,因为 JS 文件名带版本号,写死 URL 会随版本更新失效。代码里sign_match找到的是一段函数体,需要把它转换成 Python 实现,比如用exec、py_mini_racer或直接人工翻译。
注意:签名生成逻辑是接口能否调通的关键。如果源码里
sign是写死的常量,说明它很快就会过期;如果根本没有sign参数,说明它对应的可能是旧版接口,能不能用取决于当前服务端是否还兼容。
3. 批量与并发:用 Python 把转存循环改成可控任务流
3.1 输入清单与任务组织
批量转存的输入通常是一个文本文件,每一行一个分享链接,可选的提取码用空格分隔。这个格式足够简洁,也方便和表格工具互导。
https://pan.baidu.com/s/1ABC123… abcd https://pan.baidu.com/s/1DEF456…对应的数据结构用dataclass定义,比裸字典清晰:
from dataclasses import dataclass @dataclass class ShareTask: url: str pwd: str | None = None target_dir: str = "/来自批量转存" def to_skip_key(self) -> str: # 用于后续增量记录 return self.url.split("/s/")[-1] + (self.pwd or "")to_skip_key的用途是关键:批量转存最怕任务跑到一半中断,下次全量重跑。用链接标识加提取码拼一个唯一键,结合断点记录就能跳过已完成的条目。
3.2 串行版批量转存:先把链路跑通
并发之前,先写一份串行版本,保证单个任务能从链接到转存完整走通。这份代码也是后续排错的基准。
import json import time import requests from pathlib import Path def load_tasks(file_path: str) -> list[ShareTask]: tasks = [] for line in Path(file_path).read_text(encoding="utf-8").splitlines(): line = line.strip() if not line or line.startswith("#"): continue parts = line.split(maxsplit=1) tasks.append(ShareTask(url=parts[0], pwd=parts[1] if len(parts) > 1 else None)) return tasks def process_one(session, task: ShareTask) -> dict: info = get_share_detail(session, task.url, task.pwd) fid_list = [item["fs_id"] for item in info.get("list", [])] if not fid_list: return {"task": task.url, "status": "skipped", "reason": "empty_fid"} params = { "shareid": info["shareid"], "from": info["uk"], "sekey": info.get("sekey", ""), "fid_list": json.dumps(fid_list), "path": task.target_dir, "sign": get_sign(session)[0], "timestamp": get_sign(session)[1], } resp = session.post("https://pan.baidu.com/share/save", params=params) return {"task": task.url, "status": resp.json().get("errno"), "detail": resp.json()} tasks = load_tasks("links.txt") session = requests.Session() session.cookies.set("BDUSS", "你的BDUSS值") for task in tasks: result = process_one(session, task) print(result) time.sleep(1) # 避免请求过快逻辑说明:fid_list必须是 JSON 数组字符串,不能直接传 Python 列表,requests会把列表参数编码成多个同名键。path指向你网盘里的目标目录,如果目录不存在需要先调create接口,或手动在网盘里建好。循环末尾的sleep(1)是最简单的限速手段。
3.3 并发提速:为什么用线程池而不是 asyncio
接口调用是网络 IO 密集型操作,单个任务的大头时间花在等待响应上。常见做法是用ThreadPoolExecutor,因为requests是同步阻塞调用,起多线程最简单;asyncio配httpx也可以,但改造成本高,出错时不好定位。
from concurrent.futures import ThreadPoolExecutor, as_completed MAX_WORKERS = 5 def run_batch(tasks, session_factory): with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool: futures = { pool.submit(process_one, session_factory(), task): task for task in tasks } for future in as_completed(futures): task = futures[future] try: result = future.result() print(task.url, result["status"]) except Exception as e: print(task.url, "failed", repr(e))session_factory是每次提交任务时创建一个新会话,不要多个线程共用一个requests.Session。原因有两个:一是Session内部维护的 Cookie 和连接池不是完全线程安全的;二是某个线程请求异常导致连接池污染时,会连累其他任务。
3.4 并发上限与限速参数怎么定
MAX_WORKERS不是越大越好。网盘服务端对单账号的请求频率有限制,超过阈值会返回限流错误码。我一般从 3 开始调,观察两个指标:成功率是否下降、单个请求响应时间是否显著变长。如果errno里频繁出现限流类错误,就把并发降回去。
另外一个实用做法是给每个任务加随机间隔:
import random # 在 process_one 开头执行 time.sleep(random.uniform(0.3, 0.8))随机间隔的好处是避免所有线程在同一时刻集中发请求,形成明显的请求峰。固定间隔在并发场景下反而容易踩限频阈值。
4. 工程化兜底:重试、限速、增量记录与结果校验
4.1 按返回码决定是否重试
批量转存跑几百条链接时,网络抖动和临时限流是常态。重试策略不能所有错误一视同仁,否则会把永久性失败的任务反复跑好几遍。
RETRYABLE_ERRNO = {-9, -10, 0, 1} MAX_RETRY = 3 def process_with_retry(session, task): for attempt in range(MAX_RETRY): result = process_one(session, task) errno = result["status"] if errno == 0: return result if errno in RETRYABLE_ERRNO and attempt < MAX_RETRY - 1: wait = (2 ** attempt) + random.uniform(0, 1) time.sleep(wait) continue return result return result参数说明:MAX_RETRY是重试次数,wait = 2 ** attempt是指数退避;错误码0表示接口返回成功但可能转存被异步执行,需要二次校验确认,所以也放进可重试集合。
注意:永久性错误不要重试。常见的永久性失败包括分享链接已失效、文件已被删除、提取码错误,重试只会浪费请求额度,应该在日志里直接标记为
SKIP。
4.2 增量记录:中断后不用重头跑
大批量操作不能没有断点记录。常见做法是维护一个done.txt,每成功一条就追加一行,下次启动时先读取这份记录,跳过已完成的条目。
def load_done(skip_path: str) -> set: if not Path(skip_path).exists(): return set() return set(Path(skip_path).read_text(encoding="utf-8").splitlines()) def append_done(skip_path: str, key: str): with open(skip_path, "a", encoding="utf-8") as f: f.write(key + "\n") # 主流程里 done_set = load_done("done.txt") for task in tasks: key = task.to_skip_key() if key in done_set: print(task.url, "already done") continue result = process_with_retry(session, task) if result["status"] == 0 and verify_target(task): append_done("done.txt", key)verify_target是结果校验函数。只记录“确认转存成功”的条目,而不是“接口返回成功”的条目,这样断点记录才有意义。
4.3 转存后的二次校验
share/save返回errno=0只说明请求已受理,文件可能还在后台拷贝,尤其是超大目录。所以我一般会在转存后延迟几秒,再调file/list检查目标目录里的文件数是否增加了。
def verify_target(session, task) -> bool: time.sleep(5) url = "https://pan.baidu.com/api/list" params = {"dir": task.target_dir, "order": "time", "desc": "1"} resp = session.get(url, params=params) data = resp.json() return data.get("errno") == 0 and len(data.get("list", [])) > 0逻辑说明:这里用list接口读目标目录,只验证“目录里有没有东西”,不精确比对文件数量。要精确比对,就需要把get_share_detail里拿到的fid_list数量与该接口返回条目数做对比,数量一致才算完整转存。
4.4 常见失败场景与排查方向
| 现象 | 可能原因 | 排查手段 |
|---|---|---|
errno=-6或类似鉴权错误 | BDUSS 过期或被风控 | 重新登录网页版,检查 Cookie 是否失效 |
sign相关报错 | 签名算法过期 | 重新抓取首页 JS,更新签名生成逻辑 |
errno=0但目录为空 | 异步拷贝未完成 | 加大verify_target的等待时间 |
| 转存成功但文件缺失 | 部分文件被分享者删除 | 比对fid_list与目录实际条目 |
| 请求全部超时 | 请求频率过高被临时限制 | 降低并发,加大随机 sleep 间隔 |
这四个场景覆盖了批量转存工具 90% 以上的问题。任何一行报错,先对照这个表定位是“登录态问题”“接口版本问题”还是“频率限制问题”,再动手改代码。
5. 验证与排错:用 errno 和文件清单确认每次转存成功
批量工具上线前,先拿 3 条测试链接跑串行版。三条测试链接分别覆盖三种情况:无提取码链接、带提取码链接、包含多层级目录的分享链接。跑完去网页端核对路径,确认目录结构和文件数量与预期一致。这一步通过后再把MAX_WORKERS调到 5 跑全量。
日志是唯一可靠的排错依据。每一行输出建议固定格式:
2025-01-12 10:23:45 | OK | /s/ABC123 | 20 files | 12.3s 2025-01-12 10:24:01 | SKIP | /s/DEF456 | already done 2025-01-12 10:24:30 | FAIL | /s/GHI789 | errno=-18 link expired这个格式包含时间、状态、任务标识、文件数和耗时。跑完批量任务后,用grep "FAIL"看失败列表,用grep "OK" | wc -l统计成功数。耗时列还能帮你判断限速情况:如果单个任务平均耗时从 5 秒涨到 15 秒,说明请求已经被服务端降速处理。
最后一个小技巧:转存后校验不要只看接口返回码。在verify_target里对比fid_list的数量与list接口返回的条目数,不一致时主动在日志里标PARTIAL,并把任务写进单独的need_check.txt,而不是直接标记失败。这样处理的好处是能区分“转存完全失败”和“部分文件丢失”,后者往往是因为分享者在转存过程中删除了文件,重试也解决不了,人工核对更合适。
本文还有配套的精品资源,点击获取