Skill-123yunpan/references/bulk-operations.md

123云盘批量操作指南

批量创建分享链接

为大量文件创建分享链接时，需要注意速率限制和断点续传。

⚠️ 关键：fileIDList 必须是字符串

API 要求 fileIDList 为逗号分隔的字符串，不是数组：

# ✅ 正确
body = {"fileIDList": "12345,67890", "shareName": "xxx", "shareExpire": 0}

# ❌ 错误 — 会返回 "文件列表不能为空"
body = {"fileIDList": [12345, 67890], "shareName": "xxx", "shareExpire": 0}

关键参数

• API 端点: POST /api/v1/share/create
• 单次最多 100 个 fileID
• shareExpire: 0=永久, 1/7/30 天
• 建议请求间隔: 0.5 秒（每秒不超过 2 个请求）

⚠️ 分享链接构建

API 返回 shareID + shareKey，不会返回完整 URL，需要自行拼接。支持两种格式：

# Short 格式 (默认)
share_url = f"https://www.123pan.com/s/{share_key}"

# UID 格式 — 需要 user_id（可从 JWT token 解析，无需额外 API）
share_url = f"https://{user_id}.share.123pan.cn/123pan/{share_key}"

响应示例：

{"code": 0, "message": "ok", "data": {"shareID": 5193007, "shareKey": "JyAKVv-o6bI"}}

→ Short: https://www.123pan.com/s/JyAKVv-o6bI → UID: https://666666666.share.123pan.cn/123pan/JyAKVv-o6bI

完整示例：批量为文件创建分享并导出 Excel

import json, time, urllib.request, urllib.parse
from collections import deque

BASE = "https://open-api.123pan.com"

def get_token():
    with open(os.path.expanduser("~/.123yunpan_config.json")) as f:
        return json.load(f)["access_token"]

def api_get(path, params=None):
    url = BASE + path
    if params:
        url += "?" + urllib.parse.urlencode(params)
    req = urllib.request.Request(url, headers={
        "Authorization": f"Bearer {get_token()}", "Platform": "open_platform"})
    return json.loads(urllib.request.urlopen(req, timeout=30).read())

def api_post(path, body):
    url = BASE + path
    data = json.dumps(body).encode()
    req = urllib.request.Request(url, data=data, method="POST", headers={
        "Authorization": f"Bearer {get_token()}", "Platform": "open_platform",
        "Content-Type": "application/json"})
    return json.loads(urllib.request.urlopen(req, timeout=30).read())

def list_folder(parent_id):
    items, last_id = [], 0
    while True:
        params = {"parentFileId": parent_id, "limit": 100}
        if last_id:
            params["lastFileId"] = last_id
        resp = api_get("/api/v2/file/list", params)
        data = resp.get("data", {})
        items.extend(data.get("fileList", []))
        last_id = data.get("lastFileId", -1)
        if last_id == -1:
            break
        time.sleep(0.3)
    return items

def find_all_files(root_id):
    queue, all_files = deque([(root_id, [])]), []
    while queue:
        pid, path = queue.popleft()
        for item in list_folder(pid):
            if item.get("trashed"):
                continue
            if item["type"] == 1:
                queue.append((item["fileId"], path + [item["filename"]]))
            else:
                all_files.append({"fileId": item["fileId"],
                    "filename": item["filename"], "path": "/".join(path)})
    return all_files

def create_share(file_id, name, user_id=None):
    """创建分享并返回 URL。user_id 非空时使用 UID 格式。"""
    resp = api_post("/api/v1/share/create", {
        "fileIDList": str(file_id),  # ← 字符串，不是数组！
        "shareName": name, "shareExpire": 0})
    data = resp.get("data", {})
    if resp.get("code") == 0:
        key = data["shareKey"]
        if user_id:
            return f"https://{user_id}.share.123pan.cn/123pan/{key}"
        return f"https://www.123pan.com/s/{key}"
    return None

# --- Usage ---
all_files = find_all_files(root_folder_id)
results = []
for i, f in enumerate(all_files):
    url = create_share(f["fileId"], f["filename"])
    results.append({"name": f["filename"], "url": url})
    time.sleep(0.6)
    if (i + 1) % 10 == 0:
        json.dump(results, open("progress.json", "w"), ensure_ascii=False)

断点续传模式

# 保存进度
results_file = "/tmp/shares_progress.json"
if os.path.exists(results_file):
    results = json.load(open(results_file))
    done_ids = {r["fileId"] for r in results if r.get("share_url")}

# 跳过已完成的
for item in all_files:
    if item["fileId"] in done_ids:
        continue
    # ... 创建分享 ...

    # 每 N 条保存一次
    if done_count % 10 == 0:
        json.dump(results, open(results_file, "w"))

速率控制

import time
time.sleep(0.5)  # 每个请求后暂停
# 每 5 个请求可额外暂停
if (i + 1) % 5 == 0:
    time.sleep(1)

短直链批量生成

为大量文件生成短直链（不暴露文件名），利用 JWT 解析 user_id，无需额外 API 调用：

import base64, json, time

def get_user_id_from_token(token):
    """从 JWT token 解析 user_id"""
    parts = token.split(".")
    payload = parts[1] + "=" * (4 - len(parts[1]) % 4)
    return json.loads(base64.b64decode(payload)).get("id", 0)

# 一次性获取 user_id
user_id = get_user_id_from_token(access_token)

# 批量生成短直链
for file_id in file_ids:
    short_url = f"https://{user_id}.v.123pan.cn/{user_id}/{file_id}"
    print(f"File {file_id}: {short_url}")

⚠️ 短直链要求文件在已开启直链功能的文件夹中。

深层目录遍历

123云盘文件夹可达 4-5 层深（学段/科目/版本/年级/文件）。

BFS 遍历 + 缓存（推荐）

遍历大量文件夹时，API 调用次数 = 文件夹数量。100+ 文件夹需要 5-10 分钟。 关键优化：将遍历结果缓存到 JSON，5 分钟内复用，跳过所有 API 调用。

实测效果：357 个文件遍历从 57 秒 → 0.15 秒（380 倍加速）。

import os, json, time
from collections import deque

CACHE_DIR = os.path.expanduser("~/.123yunpan_cache")

def bfs_with_cache(client, root_id, max_age=300):
    """BFS 遍历 + JSON 缓存"""
    cache_path = os.path.join(CACHE_DIR, f"files_{root_id}.json")
    os.makedirs(CACHE_DIR, exist_ok=True)

    # 尝试加载缓存（默认 5 分钟有效）
    if os.path.exists(cache_path):
        age = time.time() - os.path.getmtime(cache_path)
        if age < max_age:
            with open(cache_path) as f:
                data = json.load(f)
            print(f"使用缓存 ({int(age)}秒前, {len(data['files'])} 个文件)")
            return data["files"]

    # BFS 遍历
    queue = deque([(root_id, [])])
    all_files = []
    while queue:
        pid, path = queue.popleft()
        items = list_folder(pid)  # 自动翻页
        for item in items:
            if item.get("trashed"):
                continue
            if item["type"] == 1:
                queue.append((item["fileId"], path + [item["filename"]]))
            else:
                all_files.append({
                    "fileId": item["fileId"],
                    "filename": item["filename"],
                    "path": "/".join(path),
                    "size": item.get("size", 0),
                })

    # 保存缓存
    with open(cache_path, "w") as f:
        json.dump({"files": all_files, "ts": time.time()}, f, ensure_ascii=False)

    return all_files

缓存路径: ~/.123yunpan_cache/files_{root_id}.json 强制刷新: 删除缓存文件或传 max_age=0

分层遍历（已知结构时更快）

# 已知 学段/科目/版本/年级/文件 结构
for stage in list_folder(root_id):
    for subject in list_folder(stage["fileId"]):
        for version in list_folder(subject["fileId"]):
            for grade in list_folder(version["fileId"]):
                files = list_folder(grade["fileId"])

翻页处理

def list_folder(parent_id):
    items = []
    last_id = 0
    while True:
        params = {"parentFileId": parent_id, "limit": 100}
        if last_id:
            params["lastFileId"] = last_id
        resp = api_get("/api/v2/file/list", params)
        data = _extract(resp)  # v1.2.0: always use _extract() for {code,data} wrappers
        items.extend(data.get("fileList", []))
        last_id = data.get("lastFileId", -1)
        if last_id == -1:
            break
    return items

后台长时间运行

批量操作（如 2500+ 文件）需要 20+ 分钟，应在后台运行：

python3 -u script.py 2>&1 | tee /tmp/progress.log

监控进度：

tail -5 /tmp/progress.log

环境限制：pip 不可用时安装 Python 包

系统 Python 受限（externally managed），用临时 venv：

uv venv /tmp/xlsx_env
uv pip install openpyxl --python /tmp/xlsx_env/bin/python
/tmp/xlsx_env/bin/python your_script.py