#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 123云盘 (123 Cloud Drive) 开放平台 API 命令行工具 v1.2.1 支持全部 70+ API 端点：文件管理、上传、分享、离线下载、直链、图床、视频转码、保险箱等 仅使用 Python 标准库，无需安装第三方依赖。 """ import os import sys import json import hashlib import math import argparse import mimetypes import time import re import base64 import zipfile from urllib.request import Request, urlopen, build_opener, HTTPCookieProcessor from urllib.parse import urlencode, quote from urllib.error import HTTPError, URLError from http.cookiejar import CookieJar # ─── 常量 ─────────────────────────────────────────────────────────────────── VERSION = "1.3.0" BASE_URL = "https://open-api.123pan.com" CONFIG_PATH = os.path.expanduser("~/.123yunpan_config.json") CACHE_DIR = os.path.expanduser("~/.123yunpan_cache") DEFAULT_LIMIT = 100 # API 硬性上限：limit 最大 100 MAX_LIMIT = 100 SLICE_SIZE = 10 * 1024 * 1024 # 10MB 分片 # 超过该数量的分享链接不再逐个打印，改为生成 Excel SHARE_EXCEL_THRESHOLD = 10 # 付费分享名称长度限制（普通分享同样受此约束，主动截断避免 400 错误） SHARE_NAME_MAX_LEN = 35 # 单次分享最多拼接的文件 ID 数 SHARE_MAX_FILE_IDS = 100 # 每次请求最小间隔（秒）。QPS 上限对应最小间隔 = 1/QPS，这里留出安全余量。 MIN_REQUEST_INTERVAL = 0.72 # 约 1.4 次/秒，低于 v2/file/list 的 1.5 QPS 限流 NO_COLOR = os.environ.get("NO_COLOR") is not None # https://no-color.org/ JSON_OUTPUT = False # --json 模式 # ─── QPS 限流表（来源：官方「开发须知」） ───────────────────────────────────── # 未列出的端点使用上面的通用最小间隔。 QPS_RULES = { "/api/v2/file/list": 1.5, # 最严格，BFS 遍历时的瓶颈 "/api/v1/file/list": 10, "/api/v1/user/info": 10, "/api/v1/file/move": 2, "/api/v1/file/trash": 10, "/api/v1/access_token": 10, "/api/v1/share/create": 3, "/api/v1/share/list": 3, "/api/v1/share/delete": 3, "/upload/v1/file/mkdir": 20, "/api/v1/transcode/folder/info": 20, "/api/v1/transcode/upload/from_cloud_disk": 1, "/api/v1/transcode/delete": 10, "/api/v1/transcode/video/resolutions": 1, "/api/v1/transcode/video": 3, "/api/v1/transcode/video/record": 20, "/api/v1/transcode/video/result": 20, "/api/v1/transcode/file/download": 10, "/api/v1/transcode/m3u8_ts/download": 20, "/api/v1/transcode/file/download/all": 1, } # ─── 错误码中文说明 ────────────────────────────────────────────────────────── # key: (message, 处理建议) —— 未命中时回退到原始 message ERROR_HINTS = { 401: ("access_token 无效或已过期", "重新执行 setup 获取新 token"), 429: ("请求太频繁，触发限流", "降低调用频率后重试，或稍等几秒"), 5113: ("自用下载流量不足", "升级 VIP 可获得无限流量下载"), 5066: ("文件不存在", "检查传入的 fileId 是否正确"), 1: ("参数错误", "检查请求参数格式，详见接口文档"), 2: ("分享ID非法", "文件不存在或没有分享权限"), 10: ("文件名已存在", "换一个文件名，或用 --duplicate 2 覆盖"), 16: ("文件不存在", "file_id 错误或文件已被彻底删除"), 20: ("分享文件涉嫌违规，禁止分享", "文件内容被平台策略拦截，换一个文件分享"), } def error_hint(code, message=""): """把 API 错误码翻译成人话，附加处理建议。""" if code in ERROR_HINTS: msg, hint = ERROR_HINTS[code] return f"{message or msg}（{hint}）" return message or f"请求失败（code={code}）" # ─── 颜色控制 ──────────────────────────────────────────────────────────────── class _AnsiStripper: """NO_COLOR 时自动剥离 ANSI 转义序列的输出流包装""" def __init__(self, stream): self._stream = stream def write(self, text): if NO_COLOR and isinstance(text, str): text = re.sub("\x1b\\[[0-9;]*m", "", text) self._stream.write(text) def flush(self): self._stream.flush() def __getattr__(self, name): return getattr(self._stream, name) if NO_COLOR: sys.stdout = _AnsiStripper(sys.stdout) sys.stderr = _AnsiStripper(sys.stderr) # ─── 工具函数 ─────────────────────────────────────────────────────────────── def parse_int_list(value): """Parse comma-separated or space-separated integers. Supports: --file-ids 1,2,3 | --file-ids "1,2,3" | --file-ids 1 2 3 """ if isinstance(value, list): result = [] for v in value: result.extend(parse_int_list(v)) return result return [int(x.strip()) for x in str(value).replace(" ", ",").split(",") if x.strip()] def human_size(nbytes): """将字节数转换为人类可读的文件大小""" if nbytes is None: return "未知" try: nbytes = int(nbytes) except (ValueError, TypeError): return str(nbytes) if nbytes == 0: return "0 B" units = ["B", "KB", "MB", "GB", "TB", "PB"] i = int(math.floor(math.log(nbytes, 1024))) if nbytes > 0 else 0 if i >= len(units): i = len(units) - 1 val = nbytes / (1024 ** i) return f"{val:.2f} {units[i]}" def _colorize(text, code): """根据 NO_COLOR 设置决定是否着色""" if NO_COLOR: return text return f"\033[{code}m{text}\033[0m" def print_json(data, extra_info=None): """格式化打印 JSON 数据，可附带额外信息""" if JSON_OUTPUT: print(json.dumps(data, ensure_ascii=False)) return if extra_info: print(_colorize(extra_info, "36")) print(json.dumps(data, ensure_ascii=False, indent=2)) def print_error(msg): """打印错误信息""" print(f"{_colorize('❌ 错误：', '31')}{msg}", file=sys.stderr) def print_success(msg): """打印成功信息""" print(f"{_colorize('✅ ', '32')}{msg}") def print_warn(msg): """打印警告信息""" print(f"{_colorize('⚠️ ', '33')}{msg}") # status > 100 表示文件审核驳回（官方字段说明），这类文件无法分享 def _is_rejected(item): return int(item.get("status", 0) or 0) > 100 def md5_file(path): """计算文件的 MD5""" h = hashlib.md5() with open(path, "rb") as f: while True: chunk = f.read(8192) if not chunk: break h.update(chunk) return h.hexdigest() # ─── BFS 缓存 ──────────────────────────────────────────────────────────────── def _cache_path(root_id): """获取缓存文件路径""" os.makedirs(CACHE_DIR, exist_ok=True) return os.path.join(CACHE_DIR, f"files_{root_id}.json") def _cache_load(root_id, max_age=300): """加载缓存（默认 5 分钟有效）""" path = _cache_path(root_id) if not os.path.exists(path): return None age = time.time() - os.path.getmtime(path) if age > max_age: return None try: with open(path, "r", encoding="utf-8") as f: data = json.load(f) print(_colorize(f"📂 使用缓存 ({int(age)}秒前, {len(data['files'])} 个文件)", "36")) return data except Exception: return None def _cache_save(root_id, files, folders): """保存缓存""" path = _cache_path(root_id) with open(path, "w", encoding="utf-8") as f: json.dump({"files": files, "folders": folders, "ts": time.time()}, f, ensure_ascii=False) def bfs_list_files(client, root_id, use_cache=True, max_age=300, show_progress=True, max_depth=None): """BFS 遍历文件夹，带缓存支持。max_depth=None 表示不限深度。""" from collections import deque # 尝试加载缓存 if use_cache: cached = _cache_load(root_id, max_age) if cached: return cached["files"], cached["folders"] if show_progress: depth_info = f" (最大深度: {max_depth})" if max_depth else "" print(_colorize(f"📂 正在遍历文件夹 {root_id}{depth_info}...", "36")) queue = deque([(root_id, [], 0)]) # (file_id, path, depth) all_files = [] all_folders = [] folder_count = 0 while queue: pid, path, depth = queue.popleft() # 深度限制：如果当前深度 >= max_depth，不再进入子文件夹 if max_depth is not None and depth >= max_depth: continue last_id = 0 while True: params = {"parentFileId": pid, "limit": 100} if last_id: params["lastFileId"] = last_id resp = client._get("/api/v2/file/list", params=params) data = client._extract(resp) if not data: break for item in data.get("fileList", []): if item.get("trashed"): continue if item.get("type") == 1: # 文件夹 queue.append((item["fileId"], path + [item["filename"]], depth + 1)) all_folders.append({ "fileId": item["fileId"], "filename": item["filename"], "path": "/".join(path), }) folder_count += 1 else: # 文件 all_files.append({ "fileId": item["fileId"], "filename": item.get("filename", ""), "path": "/".join(path), "size": item.get("size", 0), "status": item.get("status", 0), }) last_id = data.get("lastFileId", -1) if last_id == -1 or not last_id: break if show_progress and folder_count % 50 == 0 and folder_count > 0: print(f" 已扫描 {folder_count} 个文件夹，找到 {len(all_files)} 个文件...", flush=True) if show_progress: print(f"\033[36m📁 扫描完成: {folder_count} 个文件夹, {len(all_files)} 个文件\033[0m") # 保存缓存 _cache_save(root_id, all_files, all_folders) return all_files, all_folders def progress_bar(current, total, prefix="", suffix="", width=40, start_time=None): """显示下载/上传进度条，支持ETA估算""" if total <= 0: return pct = current / total filled = int(width * pct) bar = "█" * filled + "░" * (width - filled) # ETA计算 eta_str = "" if start_time and current > 0: elapsed = time.time() - start_time if elapsed > 0 and pct > 0: remaining = elapsed * (1 - pct) / pct if remaining >= 3600: eta_str = f" ETA:{int(remaining // 3600)}h{int((remaining % 3600) // 60)}m" elif remaining >= 60: eta_str = f" ETA:{int(remaining // 60)}m{int(remaining % 60)}s" else: eta_str = f" ETA:{int(remaining)}s" print(f"\r{prefix} [{bar}] {pct*100:.1f}% {suffix}{eta_str}", end="", flush=True) if current >= total: print() # ─── 零依赖 Excel (xlsx) 写入器 ─────────────────────────────────────────────── # xlsx 本质是一个 zip 包 + XML，用标准库即可生成，无需 openpyxl。 # 这样在任何没有安装第三方库的环境里都能稳定导出 Excel。 _XML_ILLEGAL = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f]") def _xml_escape(text): """转义为合法 XML 文本，并剔除 XML 1.0 不允许的控制字符。""" text = _XML_ILLEGAL.sub("", str(text)) return (text.replace("&", "&") .replace("<", "<") .replace(">", ">") .replace('"', """)) def _xml_attr(text): """属性值版本（额外转义引号，XML 属性用双引号包裹）。""" return _xml_escape(text).replace("'", "'") def _col_letter(idx): """1 -> A, 27 -> AA""" letters = "" while idx > 0: idx, rem = divmod(idx - 1, 26) letters = chr(65 + rem) + letters return letters def write_xlsx(path, sheet_name, headers, rows, widths=None, hyperlink_cols=()): """写出一个最小可用的 .xlsx 文件。 headers: 表头列表 rows: 二维列表，每行长度与 headers 一致 widths: 列宽列表（可选） hyperlink_cols: 需要做成可点击超链接的列下标（从 0 开始） 返回写入的行数（不含表头） """ n_cols = len(headers) n_rows = len(rows) sheet_name = _xml_attr(sheet_name[:31]) or "Sheet1" # ── 单元格样式索引 ────────────────────────────────────────────────────── S_DEFAULT, S_HEADER, S_ZEBRA, S_LINK, S_CELL = 0, 1, 2, 3, 4 parts = [] # [Content_Types].xml parts.append('' '' '' '' '' '' '' '') # _rels/.rels parts.append('' '' '' '') # xl/workbook.xml parts.append('' '' f'' '') # xl/_rels/workbook.xml.rels parts.append('' '' '' '' '') # ── 样式表 ────────────────────────────────────────────────────────────── style = ( '' '' '' '' '' '' '' '' '' '' '' '' '' '' '' '' '' '' '' '' '' '' '' '' '' f'' '' f'' '' f'' '' f'' '' f'' '' '' '' '' '' ) parts.append(style) # ── 工作表 ────────────────────────────────────────────────────────────── last_col = _col_letter(max(n_cols, 1)) sheet = ['', '', f'', '', ''] if widths: sheet.append("") for i, w in enumerate(widths[:n_cols], 1): sheet.append(f'') sheet.append("") sheet.append("") # 表头行 sheet.append(f'') for c, h in enumerate(headers, 1): ref = f"{_col_letter(c)}1" sheet.append(f'' f'{_xml_escape(h)}') sheet.append("") # 数据行 for r_i, row in enumerate(rows, start=2): zebra = "1" if r_i % 2 == 0 else "0" sheet.append(f'') for c_i, value in enumerate(row[:n_cols]): if value is None or value == "": sheet.append(f'') continue is_hyperlink = c_i in hyperlink_cols and isinstance(value, str) and value.startswith("http") if is_hyperlink: style_idx = S_LINK else: style_idx = S_ZEBRA if zebra == "1" else S_CELL ref = f"{_col_letter(c_i + 1)}{r_i}" if is_hyperlink: # 单元格里放可显示的文本，点击跳转由 关系决定 sheet.append(f'' f'{_xml_escape(value)}') elif isinstance(value, (int, float)) and not isinstance(value, bool): sheet.append(f'{value}') else: sheet.append(f'' f'{_xml_escape(value)}') sheet.append("") sheet.append("") # 超链接关系 links = [(c_i, r_i, row[c_i]) for r_i, row in enumerate(rows, start=2) for c_i in hyperlink_cols if isinstance(row[c_i], str) and row[c_i].startswith("http")] if links: sheet.append("") for rel_i, (c_i, r_i, url) in enumerate(links, start=1): col = _col_letter(c_i + 1) rid = f"rId{rel_i}" sheet.append(f'') sheet.append("") sheet.append("") parts.append("".join(sheet)) # 超链接的关系文件（targetMode=External 不可省略） rel_items = "".join( f'' for i, (_, _, url) in enumerate(links, start=1)) parts.append('' '' f'{rel_items}') # 打包 tmp_path = path + ".tmp" with zipfile.ZipFile(tmp_path, "w", zipfile.ZIP_DEFLATED) as z: z.writestr("[Content_Types].xml", parts[0]) z.writestr("_rels/.rels", parts[1]) z.writestr("xl/workbook.xml", parts[2]) z.writestr("xl/_rels/workbook.xml.rels", parts[3]) z.writestr("xl/styles.xml", parts[4]) z.writestr("xl/worksheets/sheet1.xml", parts[5]) if links: z.writestr("xl/worksheets/_rels/sheet1.xml.rels", parts[6]) os.replace(tmp_path, path) return n_rows # ─── 分享链接表格数据整理 ──────────────────────────────────────────────────── def split_name_and_path(file_item): """把一个文件条目拆成「文件名称」与「路径」。 文件名称：去掉路径前缀，再去掉扩展名 "Cinema 4D/Cinema 4D R20.zip" -> "Cinema 4D R20" 路径：云盘内的完整相对路径，含文件名与扩展名 "软件/office/离线安装包/2021/ProPlus2021Retail.img" """ filename = str(file_item.get("filename", "") or "") parent_path = str(file_item.get("path", "") or "").strip("/") # 去掉扩展名（最后一段里最后一个点之后非空） stem = filename dot = stem.rfind(".") if dot > 0 and dot < len(stem) - 1: stem = stem[:dot] # 去掉「分类前缀」：若文件名里含有父目录名，剔除该重复前缀部分 if parent_path: parent_last = parent_path.split("/")[-1] if parent_last and stem.startswith(parent_last): rest = stem[len(parent_last):].lstrip(" -_") stem = (parent_last + " " + rest).strip() if rest else parent_last return stem or filename, (f"{parent_path}/{filename}" if parent_path else filename) def build_share_rows(results): """把批量分享结果整理成 Excel 三列数据。""" rows = [] for r in results: if not r.get("url"): continue file_name, full_path = split_name_and_path(r) rows.append([file_name, r["url"], full_path]) return rows def save_share_table(results, output_file, headers=None, widths=None): """按输出扩展名保存分享结果。xlsx 走零依赖写入器，其它格式回退 JSON。""" if output_file.endswith(".json"): with open(output_file, "w", encoding="utf-8") as f: json.dump(results, f, ensure_ascii=False, indent=2) return "JSON", len(results) if not output_file.endswith((".xlsx", ".xls")): output_file += ".xlsx" # 未指定扩展名时默认导出 Excel rows = build_share_rows(results) headers = headers or ["文件名称", "链接", "路径"] widths = widths or [38, 52, 60] write_xlsx(output_file, "分享链接", headers, rows, widths, hyperlink_cols=(1,)) return "Excel", len(results) # ─── 配置管理 ─────────────────────────────────────────────────────────────── class Config: """管理 API 配置（client_id, client_secret, access_token）""" def __init__(self): self.client_id = "" self.client_secret = "" self.access_token = "" self.expired_at = "" self.load() def load(self): """从配置文件或环境变量加载""" # 优先级: 环境变量 > ~/.123yunpan_config.json > skill 目录 config.json # 1. 尝试从 skill 目录的 config.json 加载（最低优先级） skill_config = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), "config.json") if os.path.exists(skill_config): try: with open(skill_config, "r", encoding="utf-8") as f: data = json.load(f) self.client_id = data.get("client_id", "") self.client_secret = data.get("client_secret", "") self.access_token = data.get("access_token", "") self.expired_at = data.get("expired_at", "") except (json.JSONDecodeError, IOError): pass # 2. 从 ~/.123yunpan_config.json 加载（覆盖 skill 目录配置） if os.path.exists(CONFIG_PATH): try: with open(CONFIG_PATH, "r", encoding="utf-8") as f: data = json.load(f) self.client_id = data.get("client_id", "") or self.client_id self.client_secret = data.get("client_secret", "") or self.client_secret self.access_token = data.get("access_token", "") or self.access_token self.expired_at = data.get("expired_at", "") or self.expired_at except (json.JSONDecodeError, IOError): pass # 3. 环境变量覆盖（最高优先级） if os.environ.get("TWOPAN_CLIENT_ID"): self.client_id = os.environ["TWOPAN_CLIENT_ID"] if os.environ.get("TWOPAN_CLIENT_SECRET"): self.client_secret = os.environ["TWOPAN_CLIENT_SECRET"] if os.environ.get("TWOPAN_ACCESS_TOKEN"): self.access_token = os.environ["TWOPAN_ACCESS_TOKEN"] def save(self): """保存配置到文件""" data = { "client_id": self.client_id, "client_secret": self.client_secret, "access_token": self.access_token, "expired_at": self.expired_at, } os.makedirs(os.path.dirname(CONFIG_PATH), exist_ok=True) with open(CONFIG_PATH, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) def check(self): """检查配置是否完整""" missing = [] if not self.client_id: missing.append("TWOPAN_CLIENT_ID (client_id)") if not self.client_secret: missing.append("TWOPAN_CLIENT_SECRET (client_secret)") if not self.access_token: missing.append("TWOPAN_ACCESS_TOKEN (access_token)") return missing def check_warning(self): """检查配置，如果不完整则打印中文提醒""" missing = self.check() if missing: print_warn("⚠️ 配置不完整！请运行以下命令进行配置：") print() print(" 方法一：使用 setup 子命令自动配置：") print(f" python3 {sys.argv[0]} setup") print() print(" 方法二：手动创建配置文件 ~/.123yunpan_config.json：") print(' {') print(' "client_id": "你的ClientID",') print(' "client_secret": "你的ClientSecret",') print(' "access_token": ""') print(" }") print() print(" 方法三：设置环境变量：") print(" export TWOPAN_CLIENT_ID='你的ClientID'") print(" export TWOPAN_CLIENT_SECRET='你的ClientSecret'") print(" export TWOPAN_ACCESS_TOKEN='你的AccessToken'") print() print(f" 缺少的配置项：{', '.join(missing)}") print() return False return True # ─── API 客户端 ────────────────────────────────────────────────────────────── class ApiClient: """123云盘 API 客户端 内置三层稳定性保障： 1. QPS 限流 —— 按官方「开发须知」的限流表控制请求节奏 2. 429 / 5xx 指数退避重试 —— 被限流或服务端抖动时自动重试 3. Token 自动刷新 —— 收到 401 时刷新后重试一次 """ # 限流状态：{path: 上次请求时间} _last_call = {} _lock_guard = None @classmethod def _wait_for_slot(cls, path): """按端点 QPS 限制等待，返回该端点的每秒配额""" qps = QPS_RULES.get(path, 1 / MIN_REQUEST_INTERVAL) min_interval = 1.0 / qps if qps > 0 else MIN_REQUEST_INTERVAL last = cls._last_call.get(path) if last is not None: wait = min_interval - (time.time() - last) if wait > 0: time.sleep(wait) cls._last_call[path] = time.time() return qps def __init__(self, config): self.config = config self.base = BASE_URL self._refreshing = False # 防止递归刷新 def _refresh_token(self): """自动刷新 access_token""" if not self.config.client_id or not self.config.client_secret: return False try: url = self.base + "/api/v1/access_token" body = json.dumps({ "clientID": self.config.client_id, "clientSecret": self.config.client_secret, }).encode("utf-8") req = Request(url, data=body, headers={ "Content-Type": "application/json", "Platform": "open_platform", }, method="POST") resp = urlopen(req, timeout=30) result = json.loads(resp.read().decode("utf-8")) if result.get("code") == 0 and "data" in result: self.config.access_token = result["data"].get("accessToken", "") self.config.expired_at = result["data"].get("expiredAt", "") self.config.save() print_warn("Token 已自动刷新") return True except Exception: pass return False # ── 单次 HTTP 尝试 ────────────────────────────────────────────────────── def _send(self, url, method, data, req_headers, raw, timeout=60): """发起一次请求，返回 (结果 dict, http_code)。成功时 http_code=200。""" req = Request(url, data=data, headers=req_headers, method=method) try: resp = urlopen(req, timeout=timeout) if raw: return resp.read(), 200 resp_data = resp.read().decode("utf-8") if resp_data: return json.loads(resp_data), 200 return {}, 200 except HTTPError as e: body = e.read().decode("utf-8", errors="replace") try: parsed = json.loads(body) except json.JSONDecodeError: parsed = {"error": body} # 把 HTTP 状态码塞进结果，方便上层判断重试 if isinstance(parsed, dict): parsed["http_code"] = e.code return parsed, e.code except URLError as e: return {"error": str(e.reason), "http_code": 0}, 0 except Exception as e: # 网络层之外的意外（超时、JSON 解析等） return {"error": str(e), "http_code": 0}, 0 def _request(self, method, path, body=None, params=None, headers=None, raw=False, base_url=None): """发送 HTTP 请求，带限流与自动重试。base_url 可覆盖默认域名（如上传域名）。""" url = (base_url or self.base) + path if params: url += "?" + urlencode(params, doseq=True) req_headers = {"Platform": "open_platform"} if self.config.access_token: req_headers["Authorization"] = f"Bearer {self.config.access_token}" if headers: req_headers.update(headers) data = None if body is not None: data = json.dumps(body).encode("utf-8") req_headers["Content-Type"] = "application/json" ApiClient._wait_for_slot(path) max_attempts = 4 # 含首次，共 3 次退避重试 for attempt in range(1, max_attempts + 1): resp, http_code = self._send(url, method, data, req_headers, raw) # 网络层完全失败（无响应 body）—— 属于可重试错误 if http_code == 0: if attempt < max_attempts: time.sleep(2 ** attempt) continue return resp # 401：Token 失效 → 刷新后重试（只刷新一次，避免无限循环） if http_code == 401 and not self._refreshing and not raw: self._refreshing = True try: if self._refresh_token(): req_headers["Authorization"] = f"Bearer {self.config.access_token}" ApiClient._wait_for_slot(path) resp, http_code = self._send(url, method, data, req_headers, raw) if http_code != 0: return resp finally: self._refreshing = False if http_code != 0: continue # 刷新后仍失败，走下面的退避/返回逻辑 # 429 / 5xx：限流或服务端抖动 → 指数退避后重试 if http_code == 429 or http_code >= 500: if attempt < max_attempts: delay = 2 ** attempt + (0.5 if http_code == 429 else 0) if not JSON_OUTPUT: print(_colorize( f" ⟳ HTTP {http_code}，{delay:.1f}s 后重试 ({attempt}/{max_attempts - 1})", "33"), file=sys.stderr) time.sleep(delay) continue return resp return resp def _get(self, path, params=None, headers=None, base_url=None): return self._request("GET", path, params=params, headers=headers, base_url=base_url) def _post(self, path, body=None, params=None, headers=None, base_url=None): return self._request("POST", path, body=body, params=params, headers=headers, base_url=base_url) def _put(self, path, body=None, params=None, headers=None, base_url=None): return self._request("PUT", path, body=body, params=params, headers=headers, base_url=base_url) def _delete(self, path, body=None, params=None, headers=None, base_url=None): return self._request("DELETE", path, body=body, params=params, headers=headers, base_url=base_url) def get_user_id_from_token(self): """从 JWT token 中解析 user_id，无需额外 API 调用""" token = self.config.access_token if not token: return 0 try: parts = token.split(".") if len(parts) >= 2: payload = parts[1] # 补齐 base64 padding padding = 4 - len(payload) % 4 if padding != 4: payload += "=" * padding data = json.loads(base64.b64decode(payload)) return data.get("id", 0) except Exception: pass return 0 @staticmethod def _extract(resp): """从 API 响应中提取数据。123云盘返回 {code, message, data} 或 {result}""" if resp.get("code") == 0 and "data" in resp: return resp["data"] if "result" in resp: return resp["result"] return resp def _upload_file(self, url, file_path, headers=None, content_type=None): """上传文件到指定 URL""" if not content_type: content_type, _ = mimetypes.guess_type(file_path) if not content_type: content_type = "application/octet-stream" with open(file_path, "rb") as f: file_data = f.read() req_headers = headers or {} req_headers["Content-Type"] = content_type req_headers["Content-Length"] = str(len(file_data)) req = Request(url, data=file_data, headers=req_headers, method="PUT") try: resp = urlopen(req, timeout=300) resp_data = resp.read().decode("utf-8") if resp_data: return json.loads(resp_data) return {} except HTTPError as e: resp_body = e.read().decode("utf-8", errors="replace") try: return json.loads(resp_body) except json.JSONDecodeError: return {"error": resp_body, "http_code": e.code} def _upload_bytes(self, url, data, headers=None, content_type="application/octet-stream"): """上传字节数据到指定 URL（用于分片上传）""" req_headers = headers or {} req_headers["Content-Type"] = content_type req_headers["Content-Length"] = str(len(data)) req = Request(url, data=data, headers=req_headers, method="PUT") try: resp = urlopen(req, timeout=300) resp_data = resp.read().decode("utf-8") if resp_data: return json.loads(resp_data) return {} except HTTPError as e: resp_body = e.read().decode("utf-8", errors="replace") try: return json.loads(resp_body) except json.JSONDecodeError: return {"error": resp_body, "http_code": e.code} def _multipart_upload(self, url, fields, file_path, file_field="file", headers=None): """multipart/form-data 上传（用于 single/create 等端点）""" boundary = f"----HermesBoundary{int(time.time()*1000)}" body = b"" for name, value in fields.items(): body += f"--{boundary}\r\nContent-Disposition: form-data; name=\"{name}\"\r\n\r\n{value}\r\n".encode() filename = os.path.basename(file_path) with open(file_path, "rb") as f: file_data = f.read() body += f"--{boundary}\r\nContent-Disposition: form-data; name=\"{file_field}\"; filename=\"{filename}\"\r\nContent-Type: application/octet-stream\r\n\r\n".encode() body += file_data body += f"\r\n--{boundary}--\r\n".encode() req_headers = headers or {} req_headers["Content-Type"] = f"multipart/form-data; boundary={boundary}" if self.config.access_token: req_headers["Authorization"] = f"Bearer {self.config.access_token}" req_headers["Platform"] = "open_platform" req = Request(url, data=body, headers=req_headers, method="POST") try: resp = urlopen(req, timeout=600) resp_data = resp.read().decode("utf-8") if resp_data: return json.loads(resp_data) return {} except HTTPError as e: resp_body = e.read().decode("utf-8", errors="replace") try: return json.loads(resp_body) except json.JSONDecodeError: return {"error": resp_body, "http_code": e.code} def _download_file(self, url, save_path, file_size=None): """下载文件并显示进度""" req = Request(url, method="GET") req.add_header("User-Agent", "123yunpan-cli/1.0") resp = urlopen(req, timeout=300) content_length = int(resp.headers.get("Content-Length", 0)) if file_size is None: file_size = content_length os.makedirs(os.path.dirname(os.path.abspath(save_path)), exist_ok=True) downloaded = 0 chunk_size = 65536 with open(save_path, "wb") as f: while True: chunk = resp.read(chunk_size) if not chunk: break f.write(chunk) downloaded += len(chunk) if file_size > 0: progress_bar(downloaded, file_size, prefix="下载进度", suffix=f"{human_size(downloaded)}/{human_size(file_size)}") print_success(f"文件已保存到: {save_path}") # ─── 子命令处理函数 ────────────────────────────────────────────────────────── def cmd_setup(client, args): """设置配置并获取 access_token""" if args.client_id: client.config.client_id = args.client_id if args.client_secret: client.config.client_secret = args.client_secret # 未提供凭证时，交互式提示输入 if not client.config.client_id: print("⚠️ 未检测到 client_id") print(" 获取方式：访问 https://www.123pan.com/member 购买开发者权益包") print(" 购买后通过站内信获取 client_id 和 client_secret") print() cid = input("请输入 client_id: ").strip() if not cid: print_error("client_id 不能为空，退出。") return client.config.client_id = cid if not client.config.client_secret: print("⚠️ 未检测到 client_secret") csec = input("请输入 client_secret: ").strip() if not csec: print_error("client_secret 不能为空，退出。") return client.config.client_secret = csec print("正在获取 access_token...") resp = client._post("/api/v1/access_token", body={ "clientID": client.config.client_id, "clientSecret": client.config.client_secret, }) # 123云盘 API 返回 {code: 0, message: "ok", data: {accessToken, expiredAt}} if resp.get("code") == 0 and "data" in resp: result = resp["data"] client.config.access_token = result.get("accessToken", "") client.config.expired_at = result.get("expiredAt", "") client.config.save() print_success("配置已保存！") print(f" Access Token: {client.config.access_token[:30]}...") print(f" 过期时间: {client.config.expired_at}") elif "result" in resp: result = resp["result"] client.config.access_token = result.get("accessToken", "") client.config.expired_at = result.get("expiredAt", "") client.config.save() print_success("配置已保存！") print_json(resp) else: print_error("获取 access_token 失败") print_json(resp) # ─── 文件管理 ──────────────────────────────────────────────────────────────── def cmd_file_list(client, args): """列出文件""" params = {"parentFileId": args.parent_id, "limit": args.limit} if args.last_file_id: params["lastFileId"] = args.last_file_id if args.search: params["searchData"] = args.search params["searchMode"] = args.search_mode # --trashed 需要全量拉取才能过滤 fetch_all = args.all_pages or args.trashed or args.show_trashed if fetch_all: # 自动分页获取所有文件 all_files = [] last_id = args.last_file_id or "" while True: if last_id: params["lastFileId"] = last_id resp = client._get("/api/v2/file/list", params=params) data = client._extract(resp) if data and "fileList" in data: files = data["fileList"] all_files.extend(files) last_id = data.get("lastFileId", "") if len(files) < args.limit or not last_id: break else: print_json(resp) return # 过滤回收站文件 if args.trashed: all_files = [f for f in all_files if f.get("trashed")] elif not args.show_trashed: all_files = [f for f in all_files if not f.get("trashed")] # 显示结果 print(f"\033[36m📁 共 {len(all_files)} 个文件\033[0m") for f in all_files: ftype = "📁" if f.get("type") == 1 else "📄" trash_mark = " 🗑️" if f.get("trashed") else "" size = human_size(f.get("size", 0)) print(f" {ftype} [{f.get('fileId', '')}] {f.get('filename', '')} ({size}){trash_mark}") return resp = client._get("/api/v2/file/list", params=params) print_json(resp) def cmd_file_detail(client, args): """获取文件详情""" resp = client._get("/api/v1/file/detail", params={"fileID": args.file_id}) print_json(resp) def cmd_file_search(client, args): """跨文件夹递归搜索文件""" keyword = args.keyword.lower() root_id = args.parent_id max_depth = args.max_depth print(f"\033[36m🔍 正在搜索 '{args.keyword}' (最大深度: {max_depth})...\033[0m") # 使用缓存的 BFS 获取所有文件（传入 max_depth 限制遍历深度） use_cache = not args.no_cache all_files, _ = bfs_list_files(client, root_id, use_cache=use_cache, show_progress=True, max_depth=max_depth) # 在结果中过滤 results = [f for f in all_files if keyword in f.get("filename", "").lower()] if results: print(f"\n\033[36m🎯 找到 {len(results)} 个匹配文件:\033[0m") for r in results: full_path = r["path"].split("/")[-1] + "/" + r["filename"] if r["path"] else r["filename"] print(f" 📄 [{r['fileId']}] {full_path} ({human_size(r['size'])})") else: print_warn(f"未找到包含 '{args.keyword}' 的文件") def cmd_file_details_multi(client, args): """批量获取文件详情""" body = {"fileIds": args.file_ids} resp = client._post("/api/v1/file/infos", body=body) print_json(resp) def cmd_mkdir(client, args): """创建文件夹""" body = {"name": args.name, "parentID": args.parent_id} resp = client._post("/upload/v1/file/mkdir", body=body) print_json(resp) data = client._extract(resp) if data and "dirID" in data: print_success(f"文件夹 '{args.name}' 创建成功，ID: {data.get('dirID', '')}") def cmd_rename(client, args): """重命名文件""" body = {"fileId": args.file_id, "name": args.new_name} resp = client._put("/api/v1/file/name", body=body) print_json(resp) def cmd_rename_batch(client, args): """批量重命名文件""" body = {"fileIdList": args.file_ids, "nameList": args.names} resp = client._post("/api/v1/file/rename", body=body) print_json(resp) def cmd_move(client, args): """移动文件""" body = {"fileIDs": args.file_ids, "toParentFileID": args.target_parent_id} resp = client._post("/api/v1/file/move", body=body) print_json(resp) def cmd_copy(client, args): """复制文件""" body = {"fileID": args.file_id, "targetParentFileID": args.target_parent_id} resp = client._post("/api/v1/file/copy", body=body) print_json(resp) def cmd_copy_batch(client, args): """批量复制文件""" body = {"fileIDs": args.file_ids, "targetParentFileID": args.target_parent_id} resp = client._post("/api/v1/file/async/copy", body=body) print_json(resp) def cmd_copy_progress(client, args): """查询批量复制进度""" resp = client._get("/api/v1/file/async/copy/process", params={"taskID": args.task_id}) print_json(resp) def cmd_trash(client, args): """移入回收站""" if args.dry_run: # 预览模式：获取文件信息但不执行删除 print(f"\033[36m🔍 预览模式：以下 {len(args.file_ids)} 个文件将被移入回收站\033[0m") for fid in args.file_ids: resp = client._get("/api/v1/file/detail", params={"fileID": fid}) data = client._extract(resp) if data: name = data.get("filename", "未知") size = human_size(data.get("size", 0)) print(f" 📄 [{fid}] {name} ({size})") else: print(f" ❓ [{fid}] (无法获取信息)") print(f"\n\033[33m⚠️ 这是预览模式，未执行任何操作。去掉 --dry-run 才会真正删除。\033[0m") return body = {"fileIDs": args.file_ids} resp = client._post("/api/v1/file/trash", body=body) print_json(resp) def cmd_recover(client, args): """从回收站恢复文件""" body = {"fileIDs": args.file_ids} resp = client._post("/api/v1/file/recover", body=body) print_json(resp) def cmd_recover_by_path(client, args): """按路径恢复文件""" body = { "fileID": args.file_id, "targetParentFileID": args.target_parent_id, "filename": args.filename, } resp = client._post("/api/v1/file/recover/by_path", body=body) print_json(resp) def get_download_info(client, file_id): """获取下载地址。 download_info 的参数名在文档里写作 fileId，而 file/detail 写作 fileID， 这里两种拼写都试一遍，保证不同版本的实现都能拿到结果。 """ for key in ("fileId", "fileID"): resp = client._get("/api/v1/file/download_info", params={key: file_id}) data = client._extract(resp) if isinstance(data, dict) and "downloadUrl" in data: return data return None def cmd_download(client, args): """下载文件""" data = get_download_info(client, args.file_id) if data is None: print_error("获取下载地址失败：文件不存在（5066）或自用下载流量不足（5113）") return url = data["downloadUrl"] filename = args.output or data.get("fileName") or data.get("filename") or f"download_{args.file_id}" print(f"开始下载: {filename}") try: client._download_file(url, filename) except (HTTPError, URLError) as e: reason = f"HTTP {e.code}" if isinstance(e, HTTPError) else str(e.reason) print_error(f"下载失败: {reason}") if isinstance(e, HTTPError) and e.code == 403: print_warn("可能是自用下载流量不足(5113)，升级 VIP 可获得无限流量") # ─── 上传 (V2) ───────────────────────────────────────────────────────────── def cmd_upload(client, args): """上传文件（自动选择单次上传或分片上传）""" file_path = args.file if not os.path.exists(file_path): print_error(f"文件不存在: {file_path}") return filename = args.filename or os.path.basename(file_path) file_size = os.path.getsize(file_path) file_md5 = md5_file(file_path) parent_id = args.parent_id or "0" duplicate = args.duplicate # 0=跳过, 1=重命名, 2=覆盖 # 上传前空间预检：专业空间/标准空间超额时，开放平台上传入口会被直接阻断 try: _resp, _err = _get_user_info(client) if not _err: warn_if_professional_space_low(format_space_report( client._extract(_resp) or {})) except Exception: pass # 获取上传域名 print("获取上传域名...") domain_resp = client._get("/upload/v2/file/domain") domain_data = client._extract(domain_resp) if not domain_data: print_error("获取上传域名失败") print_json(domain_resp) return # API 返回列表 ["https://..."] 或对象 {"host": "https://..."} if isinstance(domain_data, list) and domain_data: upload_domain = domain_data[0] elif isinstance(domain_data, dict): upload_domain = domain_data.get("host", "") else: upload_domain = "" if not upload_domain: print_error("上传域名为空") print_json(domain_resp) return print(f"上传域名: {upload_domain}") # 判断使用单次上传还是分片上传 SINGLE_FILE_MAX = 100 * 1024 * 1024 # 100MB if file_size <= SINGLE_FILE_MAX: # 单次上传（multipart/form-data） print(f"使用单次上传模式 ({human_size(file_size)})") fields = { "parentFileID": str(int(parent_id)), "filename": filename, "etag": file_md5, "size": str(file_size), } # duplicate: 1=重命名, 2=覆盖（不传=覆盖，默认行为） if duplicate is not None: fields["duplicate"] = str(duplicate) resp = client._multipart_upload( f"{upload_domain}/upload/v2/file/single/create", fields, file_path ) if resp.get("code") != 0: print_error(f"上传失败: {resp.get('message', '未知错误')}") print_json(resp) return result = client._extract(resp) if not result: print_error("创建上传任务失败") print_json(resp) return # 检查秒传（SHA1 复用） if result.get("reuse"): file_id = result.get("fileID", "") print_success(f"⚡ 秒传成功！文件已存在，跳过上传 (fileID: {file_id})") print_json(result) return file_id = result.get("fileID", "") print_success(f"✅ 文件上传完成！(fileID: {file_id})") print_json(result) else: # 分片上传 print(f"使用分片上传模式 ({human_size(file_size)})") body = { "parentFileID": int(parent_id), "filename": filename, "etag": file_md5, "size": file_size, } # duplicate: 1=重命名, 2=覆盖（不传=覆盖，默认行为） if duplicate is not None: body["duplicate"] = duplicate # create 用主 API 域名 resp = client._post("/upload/v2/file/create", body=body) if resp.get("code") != 0: print_error(f"创建分片上传任务失败: {resp.get('message', '未知错误')}") print_json(resp) return result = client._extract(resp) if not result: print_error("创建分片上传任务失败") print_json(resp) return # 检查秒传（SHA1 复用） if result.get("reuse"): file_id = result.get("fileID", "") print_success(f"⚡ 秒传成功！文件已存在，跳过上传 (fileID: {file_id})") print_json(result) return slice_size = result.get("sliceSize", SLICE_SIZE) preupload_id = result.get("preuploadID", "") total_slices = math.ceil(file_size / slice_size) print(f"分片数: {total_slices}, 分片大小: {human_size(slice_size)}") with open(file_path, "rb") as f: for i in range(1, total_slices + 1): chunk = f.read(slice_size) chunk_md5 = hashlib.md5(chunk).hexdigest() slice_body = { "preuploadID": preupload_id, "sliceNo": i, "partMd5": chunk_md5, } # 需要先获取上传URL slice_url_resp = client._post(f"/upload/v2/file/slice", body=slice_body, base_url=upload_domain) slice_data = client._extract(slice_url_resp) if not slice_data: print_error(f"分片 {i}/{total_slices} 获取上传URL失败") print_json(slice_url_resp) return upload_url = slice_data.get("uploadUrl", "") if not upload_url: print_error(f"分片 {i}/{total_slices} 上传URL为空") print_json(slice_url_resp) return # 上传分片（使用 chunk 字节数据，而非整个文件） print(f"上传分片 {i}/{total_slices}...", end="", flush=True) resp_upload = client._upload_bytes(upload_url, chunk, content_type="application/octet-stream") print(f" ✓") # 完成上传（带轮询校验） print("等待服务器校验分片...") max_retries = max(60, min(300, total_slices * 2)) # 根据分片数自适应 for attempt in range(max_retries): complete_resp = client._post("/upload/v2/file/upload_complete", body={"preuploadID": preupload_id}, base_url=upload_domain) complete_data = client._extract(complete_resp) if complete_data and complete_data.get("completed"): print(f" 校验完成（第 {attempt + 1} 次轮询）") print_json(complete_resp) print_success("文件上传完成！") return # 检查是否有需要等待的提示 msg = complete_resp.get("message", "") if any(kw in msg for kw in ["校验", "verify", "check", "间隔"]): if attempt % 10 == 0: print(f" 服务器校验中... 已等待 {attempt}秒") elif attempt >= 5 and complete_resp.get("code") != 0: # 非等待类错误，提前退出 print_error(f"上传完成确认失败: {msg}") print_json(complete_resp) return time.sleep(1) print_warn(f"轮询超时（{max_retries}秒），文件可能仍在校验中") print_json(complete_resp) # ─── 分享管理 ──────────────────────────────────────────────────────────────── def cmd_share_create(client, args): """创建分享""" body = { "shareName": args.name, "shareExpire": args.expire, "fileIDList": ",".join(str(x) for x in args.file_ids), } if args.pwd: body["sharePwd"] = args.pwd if args.traffic_switch is not None: body["trafficSwitch"] = args.traffic_switch if args.traffic_limit_switch is not None: body["trafficLimitSwitch"] = args.traffic_limit_switch if args.traffic_limit is not None: body["trafficLimit"] = args.traffic_limit # 文件过多时主动分段，避免超过 API 的 100 个上限 file_ids = args.file_ids if len(file_ids) > SHARE_MAX_FILE_IDS: print_warn(f"文件数 {len(file_ids)} 超过单次上限 {SHARE_MAX_FILE_IDS}，" f"只取前 {SHARE_MAX_FILE_IDS} 个") body["fileIDList"] = ",".join(str(x) for x in file_ids[:SHARE_MAX_FILE_IDS]) resp = client._post("/api/v1/share/create", body=body) data = resp.get("data") or {} share_key = data.get("shareKey", "") if share_key: resp["_shareUrl"] = _share_url(client, share_key, getattr(args, "url_format", "official")) print_json(resp) if not JSON_OUTPUT and share_key: print_success(f"分享链接: {resp['_shareUrl']}") def _share_url(client, share_key, url_format): """按指定格式拼接分享链接。""" if url_format in ("uid", "official"): user_id = client.get_user_id_from_token() if user_id: return f"https://{user_id}.share.123pan.cn/123pan/{share_key}" # 拿不到 uid 时退回到官方默认格式 return f"https://www.123pan.com/s/{share_key}" def _list_direct_files(client, root_id): """只取文件夹直接子项中的文件（不递归），带全量分页。""" all_files = [] last_id = 0 while True: params = {"parentFileId": root_id, "limit": MAX_LIMIT} if last_id: params["lastFileId"] = last_id resp = client._get("/api/v2/file/list", params=params) data = client._extract(resp) if not isinstance(data, dict): break for item in data.get("fileList", []): if item.get("trashed") or item.get("type") == 1: continue if _is_rejected(item): continue all_files.append({ "fileId": item["fileId"], "filename": item.get("filename", ""), "path": "", "size": item.get("size", 0), "status": item.get("status", 0), }) nxt = data.get("lastFileId", -1) if not nxt or nxt == -1 or nxt == last_id: break last_id = nxt return all_files def cmd_share_create_batch(client, args): """批量创建分享：递归遍历文件夹，为所有文件创建分享链接""" root_id = args.parent_id expire = args.expire output_file = args.output if args.no_recursive: all_files = _list_direct_files(client, root_id) print(_colorize(f"📁 直接子项: {len(all_files)} 个文件", "36")) else: all_files, _ = bfs_list_files(client, root_id, use_cache=not args.no_cache) if not all_files: print_warn("未找到任何文件（可能全部是文件夹，或已被审核驳回）") return # 去掉重复 fileId seen, unique_files = set(), [] for f in all_files: if f["fileId"] in seen: continue seen.add(f["fileId"]) unique_files.append(f) all_files = unique_files total = len(all_files) if not JSON_OUTPUT: expire_desc = "永久" if expire == 0 else f"{expire}天" print(f"\n{_colorize(f'🔗 开始创建分享链接 (有效期: {expire_desc}, 共 {total} 个文件)', '36')}\n") # ── 断点续跑 ──────────────────────────────────────────────────────────── ckpt_path = None results = [] todo = all_files if args.resume: ckpt_path = os.path.join(CACHE_DIR, f"share_batch_{root_id}.json") os.makedirs(CACHE_DIR, exist_ok=True) if os.path.exists(ckpt_path): try: with open(ckpt_path, "r", encoding="utf-8") as f: ckpt = json.load(f) done_ids = {r["fileId"] for r in ckpt.get("results", [])} results = ckpt["results"] todo = [f for f in all_files if f["fileId"] not in done_ids] print(_colorize(f"♻️ 续跑模式：已完成 {len(results)} 个，剩余 {len(todo)} 个", "36")) except (json.JSONDecodeError, KeyError, IOError): results = [] def _save_checkpoint(): if ckpt_path: tmp = ckpt_path + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump({"root_id": root_id, "expire": expire, "results": results}, f, ensure_ascii=False) os.replace(tmp, ckpt_path) def _truncate_name(name): # 分享名称过长会被 API 拒绝，主动截断到安全长度 if len(name) > SHARE_NAME_MAX_LEN: return name[:SHARE_NAME_MAX_LEN - 3] + "..." return name url_format = getattr(args, "url_format", "official") start_time = time.time() fail_count = 0 fails = [] for i, f in enumerate(todo): display_name = f"{f['path']}/{f['filename']}" if f.get("path") else f["filename"] body = { "shareName": _truncate_name(display_name), "shareExpire": expire, "fileIDList": str(f["fileId"]), } resp = client._post("/api/v1/share/create", body=body) data = client._extract(resp) if isinstance(data, dict) and data.get("shareKey"): share_key = data["shareKey"] result = { "name": display_name, "filename": f.get("filename", ""), "path": f.get("path", ""), "url": _share_url(client, share_key, url_format), "fileId": f["fileId"], "shareKey": share_key, } results.append(result) else: code = resp.get("code", 0) if isinstance(resp, dict) else 0 msg = error_hint(code, resp.get("message", "未知错误")) if isinstance(resp, dict) else "未知错误" fail_count += 1 fails.append({"name": display_name, "fileId": f["fileId"], "error": msg}) result = {"name": display_name, "filename": f.get("filename", ""), "path": f.get("path", ""), "url": "", "fileId": f["fileId"], "shareKey": "", "error": msg} if not JSON_OUTPUT: pct = (i + 1) / total * 100 bar_len = 20 filled = int(bar_len * (i + 1) / total) bar = "█" * filled + "░" * (bar_len - filled) eta_str = "" elapsed = time.time() - start_time if i > 0 and elapsed > 0: remaining = elapsed / (i + 1) * (total - i - 1) if remaining >= 3600: eta_str = f" ETA:{int(remaining // 3600)}h{int((remaining % 3600) // 60)}m" elif remaining >= 60: eta_str = f" ETA:{int(remaining // 60)}m{int(remaining % 60)}s" else: eta_str = f" ETA:{int(remaining)}s" mark = "✓" if result["url"] else "✗" mark = _colorize(mark, "32") if result["url"] else _colorize(mark, "31") short = display_name if len(display_name) <= 35 else "..." + display_name[-32:] print(f"\r {bar} {pct:5.1f}% [{i+1}/{total}] {short} {mark}{eta_str}", end="", flush=True) _save_checkpoint() if args.no_recursive: time.sleep(0.35) elif (i + 1) % 5 == 0: time.sleep(0.5) if not JSON_OUTPUT: print() success = len(results) if not JSON_OUTPUT: print(f"\n{_colorize('=' * 50, '36')}") print(f"{_colorize(f'✅ 完成！成功 {success} 个，失败 {fail_count} 个', '32')}") if JSON_OUTPUT: print_json({"total": total, "success": success, "fail": fail_count, "results": results, "fails": fails}) return if fails: print(_colorize("❌ 失败明细：", "31")) for f in fails[:20]: print(f" [{f['fileId']}] {f['name']} — {f['error']}") if len(fails) > 20: print(f" ... 另有 {len(fails) - 20} 条失败") # ── 输出：链接数超过阈值时自动导出 Excel ──────────────────────────────── threshold = getattr(args, "excel_threshold", SHARE_EXCEL_THRESHOLD) link_count = sum(1 for r in results if r.get("url")) if args.no_excel and not output_file: for r in results: if r["url"]: print(f" {r['name']}: {r['url']}") return if output_file: fmt, n = save_share_table(results, output_file) print(f"\n📊 {fmt} 已保存: {output_file} ({n} 条)") return if link_count > threshold: auto_name = os.path.join(os.getcwd(), f"123pan_shares_{root_id}.xlsx") fmt, n = save_share_table(results, auto_name) print(f"\n📊 链接数 {link_count} 超过阈值 {threshold}，已自动生成 {fmt}：{auto_name} ({n} 条)") print(f" 列包含：文件名称（不含分类前缀） / 链接 / 路径") else: print() for r in results: if r["url"]: print(f" {r['name']}: {r['url']}") def cmd_share_list(client, args): """列出分享""" if args.all_pages: all_shares = [] last_id = args.last_share_id or "" while True: params = {"limit": args.limit} if last_id: params["lastShareId"] = last_id resp = client._get("/api/v1/share/list", params=params) data = client._extract(resp) if data and "shareList" in data: shares = data["shareList"] all_shares.extend(shares) last_id = str(data.get("lastShareId", "")) if len(shares) < args.limit or not last_id or last_id == "-1": break else: print_json(resp) return print(f"\033[36m🔗 共 {len(all_shares)} 个分享\033[0m") for s in all_shares: name = s.get("shareName", "未命名") sid = s.get("shareId", "") print(f" 🔗 [{sid}] {name}") return params = {"limit": args.limit} if args.last_share_id: params["lastShareId"] = args.last_share_id resp = client._get("/api/v1/share/list", params=params) print_json(resp) def cmd_share_update(client, args): """更新分享设置""" body = {"shareIdList": args.share_ids} if args.traffic_switch is not None: body["trafficSwitch"] = args.traffic_switch if args.traffic_limit_switch is not None: body["trafficLimitSwitch"] = args.traffic_limit_switch if args.traffic_limit is not None: body["trafficLimit"] = args.traffic_limit resp = client._put("/api/v1/share/list/info", body=body) print_json(resp) def cmd_share_pay_create(client, args): """创建付费分享""" body = { "shareName": args.name, "fileIDList": ",".join(str(x) for x in args.file_ids), "payAmount": args.amount, } if args.is_reward: body["isReward"] = 1 if args.desc: body["resourceDesc"] = args.desc resp = client._post("/api/v1/share/content-payment/create", body=body) print_json(resp) def cmd_share_pay_list(client, args): """列出付费分享""" if args.all_pages: all_shares = [] last_id = args.last_share_id or "" while True: params = {"limit": args.limit} if last_id: params["lastShareId"] = last_id resp = client._get("/api/v1/share/payment/list", params=params) data = client._extract(resp) if data and "shareList" in data: shares = data["shareList"] all_shares.extend(shares) last_id = str(data.get("lastShareId", "")) if len(shares) < args.limit or not last_id or last_id == "-1": break else: print_json(resp) return print(f"\033[36m💰 共 {len(all_shares)} 个付费分享\033[0m") for s in all_shares: name = s.get("shareName", "未命名") sid = s.get("shareId", "") amount = s.get("payAmount", "?") print(f" 💰 [{sid}] {name} ({amount}元)") return params = {"limit": args.limit} if args.last_share_id: params["lastShareId"] = args.last_share_id resp = client._get("/api/v1/share/payment/list", params=params) print_json(resp) def cmd_share_pay_update(client, args): """更新付费分享设置""" body = {"shareIdList": args.share_ids} if args.traffic_switch is not None: body["trafficSwitch"] = args.traffic_switch if args.traffic_limit_switch is not None: body["trafficLimitSwitch"] = args.traffic_limit_switch if args.traffic_limit is not None: body["trafficLimit"] = args.traffic_limit resp = client._put("/api/v1/share/list/payment/info", body=body) print_json(resp) def cmd_share_stats(client, args): """分享链接统计汇总""" all_shares = [] last_id = "" while True: params = {"limit": 100} if last_id: params["lastShareId"] = last_id resp = client._get("/api/v1/share/list", params=params) data = client._extract(resp) if not data or "shareList" not in data: break shares = data["shareList"] all_shares.extend(shares) last_id = str(data.get("lastShareId", "")) if len(shares) < 100 or not last_id or last_id == "-1": break time.sleep(0.3) if not all_shares: print_warn("没有分享链接") return total_preview = sum(s.get("previewCount", 0) for s in all_shares) total_download = sum(s.get("downloadCount", 0) for s in all_shares) total_save = sum(s.get("saveCount", 0) for s in all_shares) total_traffic = sum(s.get("bytesCharge", 0) for s in all_shares) print(f"\n\033[36m{'='*50}\033[0m") print(f"\033[36m📊 分享链接统计\033[0m") print(f"\033[36m{'='*50}\033[0m") print(f" 🔗 分享总数: {len(all_shares)} 个") print(f" 👁️ 总预览次数: {total_preview:,}") print(f" ⬇️ 总下载次数: {total_download:,}") print(f" 💾 总转存次数: {total_save:,}") print(f" 📦 总流量: {human_size(total_traffic)}") print(f"\033[36m{'='*50}\033[0m") if args.top: # 按下载量排序，显示 Top N sorted_shares = sorted(all_shares, key=lambda x: x.get("downloadCount", 0), reverse=True) print(f"\n\033[36m🏆 下载量 Top {args.top}:\033[0m") for i, s in enumerate(sorted_shares[:args.top]): name = s.get("shareName", "未命名") dl = s.get("downloadCount", 0) pv = s.get("previewCount", 0) sv = s.get("saveCount", 0) print(f" {i+1}. {name}") print(f" 预览 {pv:,} | 下载 {dl:,} | 转存 {sv:,}") def cmd_share_delete_batch(client, args): """批量删除分享链接""" if args.all: # 删除所有分享 print("\033[33m⚠️ 正在查找所有分享链接...\033[0m") all_share_ids = [] all_share_names = [] last_id = "" while True: params = {"limit": 100} if last_id: params["lastShareId"] = last_id resp = client._get("/api/v1/share/list", params=params) data = client._extract(resp) if not data or "shareList" not in data: break shares = data["shareList"] for s in shares: all_share_ids.append(str(s.get("shareId", ""))) all_share_names.append(s.get("shareName", "未命名")) last_id = str(data.get("lastShareId", "")) if len(shares) < 100 or not last_id or last_id == "-1": break time.sleep(0.3) if not all_share_ids: print_warn("没有分享链接可删除") return if args.dry_run: print(f"\033[36m🔍 预览模式：以下 {len(all_share_ids)} 个分享将被删除\033[0m") for i, (sid, name) in enumerate(zip(all_share_ids, all_share_names)): print(f" 🔗 [{sid}] {name}") print(f"\n\033[33m⚠️ 这是预览模式，未执行任何操作。去掉 --dry-run 才会真正删除。\033[0m") return print(f" 找到 {len(all_share_ids)} 个分享链接") if not args.yes: confirm = input(" 确认删除全部？(y/N): ").strip().lower() if confirm != "y": print("已取消") return # 批量删除（每批最多 100 个） success = 0 for i in range(0, len(all_share_ids), 100): batch = all_share_ids[i:i+100] resp = client._post("/api/v1/share/delete", body={"shareIdList": batch}) data = client._extract(resp) if resp.get("code") == 0: success += len(batch) time.sleep(0.5) print(f"\033[32m✅ 删除完成！成功 {success}/{len(all_share_ids)}\033[0m") elif args.share_ids: share_ids = [str(x) for x in args.share_ids] if args.dry_run: print(f"\033[36m🔍 预览模式：以下 {len(share_ids)} 个分享将被删除\033[0m") for sid in share_ids: resp = client._get("/api/v1/share/list", params={"limit": 100, "lastShareId": ""}) print(f" 🔗 [{sid}]") print(f"\n\033[33m⚠️ 这是预览模式，未执行任何操作。去掉 --dry-run 才会真正删除。\033[0m") return print(f" 删除 {len(share_ids)} 个分享链接...") if not args.yes: confirm = input(" 确认删除？(y/N): ").strip().lower() if confirm != "y": print("已取消") return resp = client._post("/api/v1/share/delete", body={"shareIdList": share_ids}) data = client._extract(resp) if resp.get("code") == 0: print_success(f"成功删除 {len(share_ids)} 个分享链接") else: msg = resp.get("message", "未知错误") print_error(f"删除失败: {msg}") else: print_warn("请指定 --share-ids 或 --all") # ─── 离线下载 ──────────────────────────────────────────────────────────────── def cmd_offline_create(client, args): """创建离线下载任务""" body = {"url": args.url, "parentFileID": args.parent_id} if args.filename: body["fileName"] = args.filename resp = client._post("/api/v1/offline/download", body=body) print_json(resp) def cmd_offline_progress(client, args): """查询离线下载进度""" resp = client._get("/api/v1/offline/download/process", params={"taskID": args.task_id}) print_json(resp) # ─── 用户信息 / 空间 ───────────────────────────────────────────────────────── def _get_user_info(client): """调用 user/info 并在出错时返回带中文提示的信息""" resp = client._get("/api/v1/user/info") if not isinstance(resp, dict) or resp.get("code") not in (0, None): msg = error_hint(resp.get("code", 0), resp.get("message", "未知错误")) return None, msg return resp, None def _parse_size(value): """user/info 的空间字段既有字节数也有字符串，统一转成 int""" try: return int(value or 0) except (TypeError, ValueError): return 0 def format_space_report(data): """把 user/info 的空间字段整理成结构化报告（含专业/标准空间拆分）""" professional_used = _parse_size(data.get("ProfessionalSpaceUsed")) professional_total = _parse_size(data.get("ProfessionalSpacePermanent")) standard_used = _parse_size(data.get("StandardSpaceUsed")) standard_total = _parse_size(data.get("StandardSpacePermanent")) total_used = _parse_size(data.get("SpaceUsedTotal")) or ( professional_used + standard_used) total_quota = _parse_size(data.get("SpacePermanentTotal")) or ( professional_total + standard_total) return { "uid": data.get("uid"), "nickname": data.get("nickname"), "vip": data.get("vip"), "vipLabel": ((data.get("vipInfo") or {}).get("vipLabel") if isinstance(data.get("vipInfo"), dict) else None), "total_quota": total_quota, "total_used": total_used, "space_temp": _parse_size(data.get("spaceTemp")), "space_temp_expr": data.get("spaceTempExpr", ""), "professional": {"used": professional_used, "quota": professional_total, "remaining_days": data.get("ProfessionalSpaceRemainingDays")}, "standard": {"used": standard_used, "quota": standard_total}, } def cmd_user_info(client, args): """获取用户信息""" resp, err = _get_user_info(client) if err: print_error(err) return if JSON_OUTPUT: print_json(resp) return space = format_space_report(client._extract(resp) or {}) vip = space.get("vipLabel") or ("是" if space.get("vip") else "否") print(f"\n{_colorize('👤 用户信息', '36')}") print(f" 用户ID: {space['uid']}") print(f" 昵称: {space['nickname']}") print(f" 会员: {vip}") print(f" 总空间: {human_size(space['total_quota'])}（已用 {human_size(space['total_used'])}）") print_space_details(space) def print_space_details(space, indent=" "): """打印专业空间 / 标准空间明细，含超额与清理倒计时预警""" pro, std = space["professional"], space["standard"] pct_pro = (pro["used"] / pro["quota"]) if pro["quota"] else 0 pct_std = (std["used"] / std["quota"]) if std["quota"] else 0 print(f"{indent}📦 专业空间: {human_size(pro['used'])} / {human_size(pro['quota'])} " f"({pct_pro * 100:.1f}%)") if pro["quota"] and pct_pro >= 0.9: print(f"{indent} 🔴 专业空间即将超额！超额后开放平台上传会被阻断") if pro["remaining_days"] not in (None, "", 0): days = pro["remaining_days"] print(f"{indent} ⏳ 距专业空间超额清理倒计时: {days} 天" f"{_colorize('（超额后超期未处理会被清理，且不可恢复）', '33')}") print(f"{indent}📦 标准空间: {human_size(std['used'])} / {human_size(std['quota'])} " f"({pct_std * 100:.1f}%)") if std["quota"] and pct_std >= 0.9: print(f"{indent} 🔴 标准空间即将超额！超额后新增上传/转存会被阻断") if space.get("space_temp"): print(f"{indent}🧊 临时空间: {human_size(space['space_temp'])}" f"{f"（{space['space_temp_expr']} 到期）" if space.get("space_temp_expr") else ""}") def cmd_space(client, args): """空间占用报告（专业空间 / 标准空间）""" resp, err = _get_user_info(client) if err: print_error(err) return data = client._extract(resp) or {} if JSON_OUTPUT: print_json(format_space_report(data)) return space = format_space_report(data) print(f"\n{_colorize('=' * 56, '36')}") print(f"{_colorize('💾 空间占用报告', '36')} {space['nickname']}") print(f"{_colorize('=' * 56, '36')}") print(f" 总配额: {human_size(space['total_quota'])}") print(f" 总已用: {human_size(space['total_used'])}" f"（剩余 {human_size(max(space['total_quota'] - space['total_used'], 0))}）") print_space_details(space) print(f"{_colorize('=' * 56, '36')}\n") # ─── 空间规则说明（用于上传前预检） ──────────────────────────────────────────── def warn_if_professional_space_low(space): """专业空间接近上限时给出预警：超过配额后开放平台上传会被直接阻断""" pro = space["professional"] if not pro["quota"]: return if pro["used"] >= pro["quota"]: print_warn("⚠️ 专业空间已满：开放平台上传 / 离线下载 / 转码 等入口会被阻断，" "请先清理专业空间文件") elif pro["used"] >= pro["quota"] * 0.9: print_warn(f"⚠️ 专业空间已用 {pro['used'] * 100 // pro['quota']}%，继续上传可能触发超额限制") # ─── 直链管理 ──────────────────────────────────────────────────────────────── def cmd_direct_enable(client, args): """启用文件直链""" resp = client._post("/api/v1/direct-link/enable", body={"fileID": args.file_id}) print_json(resp) def cmd_direct_disable(client, args): """禁用文件直链""" resp = client._post("/api/v1/direct-link/disable", body={"fileID": args.file_id}) print_json(resp) def cmd_direct_url(client, args): """获取文件直链URL""" resp = client._get("/api/v1/direct-link/url", params={"fileID": args.file_id}) data = client._extract(resp) if isinstance(data, dict) and "url" in data: if not JSON_OUTPUT: print_success(f"直链: {data['url']}") print_json(resp) return print_error(error_hint(resp.get("code", 0) if isinstance(resp, dict) else 0, "获取直链失败（文件可能不在已开启直链空间的文件夹中）")) print_json(resp) def cmd_direct_short_url(client, args): """生成短直链 URL（不暴露文件名，更隐私）""" user_id = client.get_user_id_from_token() if not user_id: # 回退：从 API 获取直链，再从中提取 user_id resp = client._get("/api/v1/direct-link/url", params={"fileID": args.file_id}) data = client._extract(resp) if data and "url" in data: # URL 格式: https://{uid}.v.123pan.cn/{uid}/{file_id}/{filename} parts = data["url"].split(".") if parts: user_id = parts[0].split("//")[-1] if not user_id: print_error("无法获取 user_id，无法生成短直链") return short_url = f"https://{user_id}.v.123pan.cn/{user_id}/{args.file_id}" print_json({"short_url": short_url, "user_id": user_id, "file_id": args.file_id}) def cmd_direct_cache_refresh(client, args): """刷新直链缓存""" resp = client._post("/api/v1/direct-link/cache/refresh", body={"fileID": args.file_id}) print_json(resp) def cmd_direct_ip_list(client, args): """获取IP白名单""" resp = client._get("/api/v1/developer/config/forbide-ip/list") print_json(resp) def cmd_direct_ip_switch(client, args): """切换IP白名单开关""" resp = client._post("/api/v1/developer/config/forbide-ip/switch", body={"switch": args.switch_state}) print_json(resp) def cmd_direct_ip_update(client, args): """更新IP白名单""" resp = client._post("/api/v1/developer/config/forbide-ip/update", body={"ipList": args.ips}) print_json(resp) def cmd_direct_offline_logs(client, args): """获取直链离线日志""" params = {} if args.file_id: params["fileID"] = args.file_id if args.limit: params["limit"] = args.limit if args.offset: params["offset"] = args.offset resp = client._get("/api/v1/direct-link/offline/logs", params=params) print_json(resp) def cmd_direct_traffic_logs(client, args): """获取直链流量日志""" params = {} if args.file_id: params["fileID"] = args.file_id if args.limit: params["limit"] = args.limit if args.offset: params["offset"] = args.offset resp = client._get("/api/v1/direct-link/log", params=params) print_json(resp) # ─── 图床管理 ──────────────────────────────────────────────────────────────── def cmd_oss_list(client, args): """列出图床文件""" body = {"parentFileID": args.parent_id, "limit": args.limit} if args.last_file_id: body["lastFileId"] = args.last_file_id if args.search: body["searchData"] = args.search resp = client._post("/api/v1/oss/file/list", body=body) print_json(resp) def cmd_oss_detail(client, args): """获取图床文件详情""" resp = client._get("/api/v1/oss/file/detail", params={"fileID": args.file_id}) print_json(resp) def cmd_oss_delete(client, args): """删除图床文件""" resp = client._post("/api/v1/oss/file/delete", body={"fileID": args.file_id}) print_json(resp) def cmd_oss_move(client, args): """移动图床文件""" resp = client._post("/api/v1/oss/file/move", body={"fileID": args.file_id, "targetParentFileID": args.target_parent_id}) print_json(resp) def cmd_oss_copy(client, args): """复制图床文件""" resp = client._post("/api/v1/oss/source/copy", body={"fileIDs": args.file_ids, "targetParentFileID": args.target_parent_id}) print_json(resp) def cmd_oss_copy_process(client, args): """查询图床复制进度""" resp = client._get("/api/v1/oss/source/copy/process", params={"taskID": args.task_id}) print_json(resp) def cmd_oss_copy_fail(client, args): """查询图床复制失败详情""" resp = client._get("/api/v1/oss/source/copy/fail", params={"taskID": args.task_id}) print_json(resp) # ─── 视频转码 ──────────────────────────────────────────────────────────────── def cmd_transcode_list(client, args): """列出转码视频""" params = {} if args.parent_file_id: params["parentFileID"] = args.parent_file_id params["limit"] = args.limit if args.last_file_id: params["lastFileId"] = args.last_file_id resp = client._get("/api/v1/video/transcode/list", params=params) print_json(resp) def cmd_transcode_folder(client, args): """获取转码文件夹信息""" resp = client._post("/api/v1/transcode/folder/info", body={"parentFileID": args.parent_file_id}) print_json(resp) def cmd_transcode_resolutions(client, args): """获取视频可用分辨率""" resp = client._post("/api/v1/transcode/video/resolutions", body={"fileID": args.file_id}) print_json(resp) def cmd_transcode_start(client, args): """开始视频转码""" body = {"fileID": args.file_id, "resolutions": args.resolutions} resp = client._post("/api/v1/transcode/video", body=body) print_json(resp) def cmd_transcode_record(client, args): """获取转码记录""" resp = client._post("/api/v1/transcode/video/record", body={"fileID": args.file_id}) print_json(resp) def cmd_transcode_result(client, args): """获取转码结果""" resp = client._post("/api/v1/transcode/video/result", body={"fileID": args.file_id}) print_json(resp) def cmd_transcode_delete(client, args): """删除转码结果""" resp = client._post("/api/v1/transcode/delete", body={"fileID": args.file_id}) print_json(resp) def cmd_transcode_download(client, args): """下载转码文件""" resp = client._post("/api/v1/transcode/file/download", body={"fileID": args.file_id}) data = client._extract(resp) if data and "downloadUrl" in data: url = data["downloadUrl"] filename = args.output or data.get("fileName", "transcoded_video") client._download_file(url, filename) else: print_json(resp) def cmd_transcode_download_m3u8(client, args): """下载转码M3U8文件""" resp = client._post("/api/v1/transcode/m3u8_ts/download", body={"fileID": args.file_id}) data = client._extract(resp) if data and "downloadUrl" in data: url = data["downloadUrl"] filename = args.output or "transcoded_video.m3u8" client._download_file(url, filename) else: print_json(resp) def cmd_transcode_download_all(client, args): """下载所有转码结果""" resp = client._post("/api/v1/transcode/file/download/all", body={"fileID": args.file_id}) data = client._extract(resp) if data and "downloadUrl" in data: url = data["downloadUrl"] filename = args.output or "transcoded_all.zip" client._download_file(url, filename) else: print_json(resp) def cmd_transcode_upload_cloud(client, args): """从云盘上传到转码""" resp = client._post("/api/v1/transcode/upload/from_cloud_disk", body={"fileID": args.file_id, "parentFileID": args.parent_file_id}) print_json(resp) # ─── 保险箱 ──────────────────────────────────────────────────────────────── def cmd_safebox_unlock(client, args): """解锁保险箱""" resp = client._get("/api/v1/safebox/id", params={"password": args.password}) print_json(resp) # ─── 缓存管理 ───────────────────────────────────────────────────────────── def cmd_cache_list(client, args): """列出所有缓存文件""" if not os.path.exists(CACHE_DIR): print_warn("缓存目录不存在") return cache_files = [] total_size = 0 for f in os.listdir(CACHE_DIR): if f.endswith(".json"): path = os.path.join(CACHE_DIR, f) size = os.path.getsize(path) mtime = os.path.getmtime(path) age = time.time() - mtime cache_files.append({ "filename": f, "size": human_size(size), "size_bytes": size, "age_seconds": int(age), "age_human": f"{int(age // 3600)}小时{int((age % 3600) // 60)}分钟" if age >= 3600 else f"{int(age // 60)}分钟{int(age % 60)}秒", "root_id": f.replace("files_", "").replace(".json", ""), }) total_size += size if not cache_files: print_warn("没有缓存文件") return print(f"\n\033[36m📂 缓存目录: {CACHE_DIR}\033[0m") print(f"\033[36m📊 共 {len(cache_files)} 个缓存文件，总计 {human_size(total_size)}\033[0m\n") for cf in sorted(cache_files, key=lambda x: x["age_seconds"]): print(f" 📄 {cf['filename']}") print(f" 文件夹ID: {cf['root_id']} | 大小: {cf['size']} | 创建于: {cf['age_human']}前") print() def cmd_cache_clear(client, args): """清除缓存""" if not os.path.exists(CACHE_DIR): print_warn("缓存目录不存在，无需清除") return if args.root_id: # 清除指定文件夹的缓存 cache_file = os.path.join(CACHE_DIR, f"files_{args.root_id}.json") if not os.path.exists(cache_file): print_warn(f"没有找到文件夹 {args.root_id} 的缓存") return if not args.yes: confirm = input(f" 确认清除文件夹 {args.root_id} 的缓存？(y/N): ").strip().lower() if confirm != "y": print("已取消") return os.remove(cache_file) print_success(f"已清除文件夹 {args.root_id} 的缓存") else: # 清除全部缓存 cache_files = [f for f in os.listdir(CACHE_DIR) if f.endswith(".json")] if not cache_files: print_warn("没有缓存文件") return if not args.yes: confirm = input(f" 确认清除全部 {len(cache_files)} 个缓存文件？(y/N): ").strip().lower() if confirm != "y": print("已取消") return for f in cache_files: os.remove(os.path.join(CACHE_DIR, f)) print_success(f"已清除全部 {len(cache_files)} 个缓存文件") def cmd_cache_info(client, args): """显示缓存目录信息""" if not os.path.exists(CACHE_DIR): print(f"缓存目录: {CACHE_DIR}") print("状态: 目录不存在") return cache_files = [f for f in os.listdir(CACHE_DIR) if f.endswith(".json")] total_size = sum(os.path.getsize(os.path.join(CACHE_DIR, f)) for f in cache_files) info = { "缓存目录": CACHE_DIR, "目录存在": "✅ 是", "缓存文件数": len(cache_files), "总大小": human_size(total_size), "缓存有效期": "5 分钟 (300秒)", } print_json(info) # ─── 高级命令 ──────────────────────────────────────────────────────────────── def cmd_file_tree(client, args): """以树状结构显示目录内容""" parent_id = args.parent_id max_depth = getattr(args, "max_depth", None) # BFS 获取所有文件夹和文件 all_files, all_folders = bfs_list_files(client, parent_id, use_cache=False, show_progress=True, max_depth=max_depth) # 构建树结构：将文件和文件夹挂到对应路径下 tree = {} # path -> [(name, size, is_dir)] tree[""] = [] for f in all_folders: p = f["path"] if p not in tree: tree[p] = [] tree[p].append((f["filename"], 0, True, f["fileId"])) for f in all_files: p = f["path"] if p not in tree: tree[p] = [] tree[p].append((f["filename"], f.get("size", 0), False, f["fileId"])) # 渲染树 def _render(prefix, path, items): items.sort(key=lambda x: (not x[2], x[0])) # 文件夹优先 for i, (name, size, is_dir, fid) in enumerate(items): is_last = (i == len(items) - 1) connector = "└── " if is_last else "├── " folder_icon = "📁" if is_dir else "📄" size_str = f" ({human_size(size)})" if not is_dir and size else "" print(f"{prefix}{connector}{folder_icon} {name}{size_str}") if is_dir: child_path = f"{path}/{name}" if path else name children = tree.get(child_path, []) if children: ext = " " if is_last else "│ " _render(prefix + ext, child_path, children) # 顶层输出 root_items = tree.get("", []) total_files = len(all_files) total_folders = len(all_folders) if JSON_OUTPUT: json_tree = {"folders": total_folders, "files": total_files, "children": []} for name, size, is_dir, fid in root_items: node = {"name": name, "size": size, "is_dir": is_dir, "fileId": fid} if is_dir: child_path = name node["children"] = tree.get(child_path, []) json_tree["children"].append(node) print_json(json_tree) return print(f"\n\033[36m🌳 目录树 (共 {total_folders} 个文件夹, {total_files} 个文件)\033[0m\n") _render("", "", root_items) def cmd_offline_list(client, args): """列出离线下载任务 注意：123云盘 API 没有提供批量列出离线任务的接口。 离线任务需要通过 offline create 获取 task_id 后， 使用 offline progress 逐个查询进度。 """ if JSON_OUTPUT: print_json({ "message": "123云盘 API 不支持列出离线下载任务", "hint": "请使用 offline create 创建任务，然后用 offline progress 查询进度" }) return print_warn("123云盘 API 没有提供列出所有离线下载任务的接口") print() print(" 离线下载流程：") print(" 1. 使用 offline create 创建离线下载任务，获取 task_id") print(" 2. 使用 offline progress 查询任务进度") print() print(" 示例：") print(" 123yunpan offline create --url <磁力链接或HTTP链接>") print(" 123yunpan offline progress --task-id ") def cmd_share_export(client, args): """导出所有分享链接到 Excel 或 JSON""" output_file = args.output if not output_file: print_error("请使用 --output 指定输出文件路径") return # 分页获取所有分享 all_shares = [] last_id = "" page = 1 print(f"\033[36m📦 正在导出分享链接...\033[0m") while True: params = {"limit": 100} if last_id: params["lastShareId"] = last_id resp = client._get("/api/v1/share/list", params=params) data = client._extract(resp) if data and "shareList" in data: shares = data["shareList"] all_shares.extend(shares) last_id = str(data.get("lastShareId", "")) print(f"\r 📥 第 {page} 页: 已获取 {len(all_shares)} 条分享", end="", flush=True) if len(shares) < 100 or not last_id or last_id == "-1": break else: break page += 1 time.sleep(0.3) print(f"\n\033[36m📊 共获取 {len(all_shares)} 条分享\033[0m") if not all_shares: print_warn("没有找到任何分享") return # 构建导出数据 # 官方 share/list 返回字段：shareId, shareKey, shareName, expiration, expired, # sharePwd, trafficSwitch, trafficLimitSwitch, trafficLimit, bytesCharge, # previewCount, downloadCount, saveCount url_format = getattr(args, "url_format", "official") export_data = [] for s in all_shares: share_key = s.get("shareKey", "") share_url = _share_url(client, share_key, url_format) if share_key else "" export_data.append({ "shareId": s.get("shareId", ""), "shareName": s.get("shareName", ""), "shareKey": share_key, "shareUrl": share_url, "expiration": s.get("expiration", ""), "expired": "已失效" if s.get("expired") == 1 else "有效", "previewCount": s.get("previewCount", 0), "downloadCount": s.get("downloadCount", 0), "saveCount": s.get("saveCount", 0), "bytesCharge": human_size(s.get("bytesCharge", 0)), }) if JSON_OUTPUT: print_json(export_data) return if output_file.endswith(".json"): with open(output_file, "w", encoding="utf-8") as f: json.dump(export_data, f, ensure_ascii=False, indent=2) print_success(f"JSON 已保存: {output_file} ({len(export_data)} 条)") return headers = ["分享ID", "名称", "分享链接", "过期时间", "状态", "预览", "下载", "转存", "流量"] widths = [14, 32, 52, 20, 10, 10, 10, 10, 14] rows = [[d["shareId"], d["shareName"], d["shareUrl"], d["expiration"], d["expired"], d["previewCount"], d["downloadCount"], d["saveCount"], d["bytesCharge"]] for d in export_data] write_xlsx(output_file, "分享链接", headers, rows, widths, hyperlink_cols=(2,)) print_success(f"Excel 已保存: {output_file} ({len(rows)} 条)") def cmd_config_export(client, args): """导出配置到 JSON 文件（不包含 access_token）""" output_file = args.output if not output_file: print_error("请使用 --output 指定输出文件路径") return config_data = { "client_id": client.config.client_id or "", "client_secret": client.config.client_secret or "", } if JSON_OUTPUT: print_json({"message": "配置已导出", "output": output_file}) return with open(output_file, "w", encoding="utf-8") as f: json.dump(config_data, f, ensure_ascii=False, indent=2) print_success(f"配置已导出到: {output_file}") print(f" client_id: {config_data['client_id'][:8]}..." if len(config_data["client_id"]) > 8 else f" client_id: {config_data['client_id']}") print(f" client_secret: {'*' * 8}...") def cmd_config_import(client, args): """从 JSON 文件导入配置""" input_file = args.file if not input_file: print_error("请使用 --file 指定配置文件路径") return if not os.path.exists(input_file): print_error(f"文件不存在: {input_file}") return try: with open(input_file, "r", encoding="utf-8") as f: config_data = json.load(f) except (json.JSONDecodeError, IOError) as e: print_error(f"读取配置文件失败: {e}") return if JSON_OUTPUT: print_json({"message": "配置已导入", "client_id": config_data.get("client_id", "")}) return if config_data.get("client_id"): client.config.client_id = config_data["client_id"] print_success(f"已设置 client_id: {config_data['client_id'][:8]}..." if len(config_data.get("client_id", "")) > 8 else f"已设置 client_id: {config_data['client_id']}") if config_data.get("client_secret"): client.config.client_secret = config_data["client_secret"] print_success("已设置 client_secret: " + "*" * 8) client.config.save() print_success(f"配置已保存到 {CONFIG_PATH}") print(" 运行 123yunpan setup 来获取新的 access_token") def cmd_file_download_batch(client, args): """批量下载多个文件""" file_ids = args.file_ids output_dir = args.output_dir or "." if not os.path.exists(output_dir): os.makedirs(output_dir, exist_ok=True) total = len(file_ids) success = 0 fail = 0 if JSON_OUTPUT: results = [] print(f"\033[36m📥 批量下载 {total} 个文件到 {output_dir}\033[0m\n") for i, file_id in enumerate(file_ids): print(f"\r [{i+1}/{total}] 获取文件 {file_id} 的下载链接...", end="", flush=True) detail_data = client._extract( client._get("/api/v1/file/detail", params={"fileID": file_id})) or {} filename = detail_data.get("filename") or f"{file_id}" file_size = detail_data.get("size", 0) data = get_download_info(client, file_id) if data is None: msg = "获取下载链接失败：文件不存在(5066) 或 自用下载流量不足(5113)" print(f"\n ❌ {filename}：{msg}") fail += 1 if JSON_OUTPUT: results.append({"fileId": file_id, "filename": filename, "status": "error", "message": msg}) continue download_url = data["downloadUrl"] save_path = os.path.join(output_dir, filename) # 文件名冲突时加 file_id 后缀，避免覆盖 if os.path.exists(save_path): base, ext = os.path.splitext(filename) save_path = os.path.join(output_dir, f"{base}_{file_id}{ext}") print(f"\n 📄 下载: {filename} ({human_size(file_size)})") try: client._download_file(download_url, save_path, file_size) except HTTPError as e: print(f"\n ❌ {filename} 下载失败：HTTP {e.code}") fail += 1 if JSON_OUTPUT: results.append({"fileId": file_id, "filename": filename, "status": "error", "message": f"HTTP {e.code}"}) continue except URLError as e: print(f"\n ❌ {filename} 下载失败：{e.reason}") fail += 1 if JSON_OUTPUT: results.append({"fileId": file_id, "filename": filename, "status": "error", "message": str(e.reason)}) continue success += 1 if JSON_OUTPUT: results.append({"fileId": file_id, "filename": filename, "path": save_path, "status": "success", "size": file_size}) print(f"\n\033[36m{'='*50}\033[0m") print(f"\033[32m✅ 完成！成功 {success} 个，失败 {fail} 个\033[0m") if JSON_OUTPUT: print_json({"total": total, "success": success, "fail": fail, "results": results}) def cmd_file_storage_report(client, args): """分析存储空间使用情况""" root_id = args.parent_id use_cache = not args.no_cache top_n = getattr(args, "top_n", 10) or 10 all_files, all_folders = bfs_list_files(client, root_id, use_cache=use_cache, show_progress=True) if not all_files: if JSON_OUTPUT: print_json({"total_files": 0, "total_size": 0, "folders": {}, "top_files": []}) else: print_warn("没有找到任何文件") return # 按顶级文件夹聚合 folder_stats = {} # folder_name -> {"size": bytes, "count": int} total_size = 0 for f in all_files: size = f.get("size", 0) total_size += size path_parts = f["path"].split("/") if f["path"] else [] top_folder = path_parts[0] if path_parts else "(根目录)" if top_folder not in folder_stats: folder_stats[top_folder] = {"size": 0, "count": 0} folder_stats[top_folder]["size"] += size folder_stats[top_folder]["count"] += 1 # 按大小排序的顶级文件夹 sorted_folders = sorted(folder_stats.items(), key=lambda x: x[1]["size"], reverse=True) # 按大小排序的文件 sorted_files = sorted(all_files, key=lambda x: x.get("size", 0), reverse=True)[:top_n] if JSON_OUTPUT: json_report = { "total_files": len(all_files), "total_folders": len(all_folders), "total_size": total_size, "total_size_human": human_size(total_size), "folders": [ {"name": name, "size": stat["size"], "size_human": human_size(stat["size"]), "count": stat["count"]} for name, stat in sorted_folders ], "top_files": [ {"filename": f["filename"], "path": f["path"], "fileId": f["fileId"], "size": f.get("size", 0), "size_human": human_size(f.get("size", 0))} for f in sorted_files ], } print_json(json_report) return # 漂亮打印 print(f"\n\033[36m{'='*60}\033[0m") print(f"\033[36m📊 存储空间分析报告\033[0m") print(f"\033[36m{'='*60}\033[0m") print(f"\n 📁 总文件数: {len(all_files)}") print(f" 📂 总文件夹数: {len(all_folders)}") print(f" 💾 总大小: {human_size(total_size)}") print(f"\n\033[36m📁 文件夹使用情况 (按大小排序)\033[0m\n") for name, stat in sorted_folders: bar_len = 30 pct = stat["size"] / total_size if total_size > 0 else 0 filled = int(bar_len * pct) bar = "█" * filled + "░" * (bar_len - filled) print(f" {name:.<30} {bar} {human_size(stat['size']):>10} ({stat['count']} 个文件)") print(f"\n\033[36m📄 最大的 {min(top_n, len(sorted_files))} 个文件\033[0m\n") for f in sorted_files: full_path = f"{f['path']}/{f['filename']}" if f["path"] else f["filename"] print(f" {human_size(f.get('size', 0)):>10} [{f['fileId']}] {full_path}") print() # ─── 帮助信息格式化 ────────────────────────────────────────────────────────── def format_file(f): """格式化文件信息（用于列表输出）""" ftype = "📁" if f.get("type") == 1 else "📄" return { "type": ftype, "fileId": f.get("fileId"), "filename": f.get("filename"), "size": human_size(f.get("size", 0)), "size_bytes": f.get("size"), "parentFileID": f.get("parentFileID"), "createAt": f.get("createAt"), "trashed": f.get("trashed"), } # ─── 主程序 ────────────────────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser( description="123云盘 (123 Cloud Drive) 开放平台 API 命令行工具", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 使用示例: %(prog)s setup --client-id YOUR_ID --client-secret YOUR_SECRET %(prog)s config check %(prog)s space %(prog)s file list --parent-id 0 %(prog)s file list --parent-id 0 --search "文档" --all-pages %(prog)s file detail --file-id 123456 %(prog)s file mkdir --name "新文件夹" --parent-id 0 %(prog)s file download --file-id 123456 %(prog)s upload --file /path/to/file --parent-id 0 %(prog)s share create --name "分享" --file-ids 123,456 --expire 7 %(prog)s share create-batch --parent-id 27172609 --expire 0 %(prog)s offline create --url "magnet:?xt=..." --parent-id 0 %(prog)s user-info """, ) parser.add_argument("--no-color", action="store_true", help="禁用彩色输出（也支持 NO_COLOR 环境变量）") parser.add_argument("--json", action="store_true", dest="json_output", help="JSON 机器可读输出（单行，无缩进）") subparsers = parser.add_subparsers(dest="command", help="子命令") # ── setup ────────────────────────────────────────────── p_setup = subparsers.add_parser("setup", help="配置客户端并获取 access_token") p_setup.add_argument("--client-id", help="Client ID") p_setup.add_argument("--client-secret", help="Client Secret") p_setup.set_defaults(func=cmd_setup) # ── user-info ────────────────────────────────────────── p_user = subparsers.add_parser("user-info", help="获取用户信息") p_user.set_defaults(func=cmd_user_info) # ── space ────────────────────────────────────────────── p_space = subparsers.add_parser("space", help="空间占用报告（专业空间 / 标准空间）") p_space.set_defaults(func=cmd_space) # ── file 子命令组 ────────────────────────────────────── p_file = subparsers.add_parser("file", help="文件管理") file_sub = p_file.add_subparsers(dest="file_cmd", help="文件操作") # file list p_fl = file_sub.add_parser("list", help="列出文件") p_fl.add_argument("--parent-id", type=int, default=0, help="父文件夹ID (默认0=根目录)") p_fl.add_argument("--limit", type=int, default=DEFAULT_LIMIT, help=f"每页数量 (默认100，上限 {MAX_LIMIT})") p_fl.add_argument("--last-file-id", help="上一页最后文件ID (分页)") p_fl.add_argument("--search", help="搜索关键词") p_fl.add_argument("--search-mode", type=int, default=0, choices=[0, 1], help="搜索模式: 0=模糊 1=精确") p_fl.add_argument("--all-pages", action="store_true", help="自动获取所有分页") p_fl.add_argument("--trashed", action="store_true", help="只显示回收站中的文件") p_fl.add_argument("--show-trashed", action="store_true", help="同时显示回收站中的文件") p_fl.set_defaults(func=cmd_file_list) # file search p_fs = file_sub.add_parser("search", help="跨文件夹递归搜索文件") p_fs.add_argument("keyword", help="搜索关键词（文件名包含即匹配）") p_fs.add_argument("--parent-id", type=int, default=0, help="搜索起点文件夹ID (默认0=根目录)") p_fs.add_argument("--max-depth", type=int, default=5, help="最大搜索深度 (默认5)") p_fs.add_argument("--no-cache", action="store_true", help="不使用缓存，强制重新遍历") p_fs.set_defaults(func=cmd_file_search) # file detail p_fd = file_sub.add_parser("detail", help="获取文件详情") p_fd.add_argument("--file-id", required=True, type=int, help="文件ID") p_fd.set_defaults(func=cmd_file_detail) # file details-multi p_fdm = file_sub.add_parser("details-multi", help="批量获取文件详情") p_fdm.add_argument("--file-ids", required=True, type=parse_int_list, help="文件ID (逗号分隔，如: 1,2,3)") p_fdm.set_defaults(func=cmd_file_details_multi) # file mkdir p_mkdir = file_sub.add_parser("mkdir", help="创建文件夹") p_mkdir.add_argument("--name", required=True, help="文件夹名称") p_mkdir.add_argument("--parent-id", type=int, default=0, help="父文件夹ID") p_mkdir.set_defaults(func=cmd_mkdir) # file rename p_rn = file_sub.add_parser("rename", help="重命名文件") p_rn.add_argument("--file-id", required=True, type=int, help="文件ID") p_rn.add_argument("--new-name", required=True, help="新名称") p_rn.set_defaults(func=cmd_rename) # file rename-batch p_rnb = file_sub.add_parser("rename-batch", help="批量重命名文件") p_rnb.add_argument("--file-ids", required=True, type=parse_int_list, help="文件ID (逗号分隔，如: 1,2,3)") p_rnb.add_argument("--names", required=True, nargs="+", help="新名称列表 (逗号或空格分隔，与文件ID一一对应)") p_rnb.set_defaults(func=cmd_rename_batch) # file move p_mv = file_sub.add_parser("move", help="移动文件") p_mv.add_argument("--file-ids", required=True, type=parse_int_list, help="文件ID (逗号分隔，如: 1,2,3)") p_mv.add_argument("--target-parent-id", required=True, type=int, help="目标文件夹ID") p_mv.set_defaults(func=cmd_move) # file copy p_cp = file_sub.add_parser("copy", help="复制文件") p_cp.add_argument("--file-id", required=True, type=int, help="文件ID") p_cp.add_argument("--target-parent-id", required=True, type=int, help="目标文件夹ID") p_cp.set_defaults(func=cmd_copy) # file copy-batch p_cpb = file_sub.add_parser("copy-batch", help="批量复制文件") p_cpb.add_argument("--file-ids", required=True, type=parse_int_list, help="文件ID (逗号分隔，如: 1,2,3)") p_cpb.add_argument("--target-parent-id", required=True, type=int, help="目标文件夹ID") p_cpb.set_defaults(func=cmd_copy_batch) # file copy-progress p_cpp = file_sub.add_parser("copy-progress", help="查询批量复制进度") p_cpp.add_argument("--task-id", required=True, help="任务ID") p_cpp.set_defaults(func=cmd_copy_progress) # file trash p_trash = file_sub.add_parser("trash", help="移入回收站") p_trash.add_argument("--file-ids", required=True, type=parse_int_list, help="文件ID (逗号分隔，如: 1,2,3)") p_trash.add_argument("--dry-run", action="store_true", help="预览模式：只显示将被删除的文件，不执行操作") p_trash.set_defaults(func=cmd_trash) # file recover p_rec = file_sub.add_parser("recover", help="从回收站恢复文件") p_rec.add_argument("--file-ids", required=True, type=parse_int_list, help="文件ID (逗号分隔，如: 1,2,3)") p_rec.set_defaults(func=cmd_recover) # file recover-by-path p_rbp = file_sub.add_parser("recover-by-path", help="按路径恢复文件") p_rbp.add_argument("--file-id", required=True, type=int, help="文件ID") p_rbp.add_argument("--target-parent-id", required=True, type=int, help="目标父文件夹ID") p_rbp.add_argument("--filename", required=True, help="文件名") p_rbp.set_defaults(func=cmd_recover_by_path) # file download p_dl = file_sub.add_parser("download", help="下载文件") p_dl.add_argument("--file-id", required=True, type=int, help="文件ID") p_dl.add_argument("--output", help="保存路径 (默认使用原始文件名)") p_dl.set_defaults(func=cmd_download) # file tree p_ft = file_sub.add_parser("tree", help="显示文件夹树形结构") p_ft.add_argument("--parent-id", type=int, default=0, help="父文件夹ID (默认0=根目录)") p_ft.add_argument("--max-depth", type=int, default=3, help="最大显示深度 (默认3)") p_ft.set_defaults(func=cmd_file_tree) # file download-batch p_fdb = file_sub.add_parser("download-batch", help="批量下载文件") p_fdb.add_argument("--file-ids", required=True, type=parse_int_list, help="文件ID (逗号分隔，如: 1,2,3)") p_fdb.add_argument("--output-dir", default=".", help="输出目录 (默认当前目录)") p_fdb.set_defaults(func=cmd_file_download_batch) # file storage-report p_fsr = file_sub.add_parser("storage-report", help="存储使用报告") p_fsr.add_argument("--parent-id", type=int, default=0, help="父文件夹ID (默认0=根目录)") p_fsr.add_argument("--top", type=int, default=20, help="显示最大的 Top N 文件 (默认20)") p_fsr.set_defaults(func=cmd_file_storage_report) # ── upload ───────────────────────────────────────────── p_up = subparsers.add_parser("upload", help="上传文件 (V2)") p_up.add_argument("--file", required=True, help="本地文件路径") p_up.add_argument("--parent-id", default="0", help="目标文件夹ID (默认0=根目录)") p_up.add_argument("--filename", help="指定上传后的文件名 (默认使用原始文件名)") p_up.add_argument("--duplicate", type=int, default=None, choices=[1, 2], help="重名处理: 1=重命名 2=覆盖 (默认不发送=覆盖)") p_up.set_defaults(func=cmd_upload) # ── share 子命令组 ───────────────────────────────────── p_share = subparsers.add_parser("share", help="分享管理") share_sub = p_share.add_subparsers(dest="share_cmd", help="分享操作") # share create p_sc = share_sub.add_parser("create", help="创建分享") p_sc.add_argument("--name", required=True, help="分享名称") p_sc.add_argument("--file-ids", required=True, type=parse_int_list, help="文件ID (逗号分隔，如: 1,2,3)") p_sc.add_argument("--expire", type=int, default=7, choices=[0, 1, 7, 30], help="有效期: 0=永久 1=1天 7=7天 30=30天 (默认7)") p_sc.add_argument("--pwd", help="分享密码 (可选)") p_sc.add_argument("--traffic-switch", type=int, help="流量限制开关: 0=关 1=开") p_sc.add_argument("--traffic-limit-switch", type=int, help="流量限制值开关: 0=关 1=开") p_sc.add_argument("--traffic-limit", type=int, help="流量限制值 (字节)") p_sc.add_argument("--url-format", choices=["official", "web", "uid", "short"], default="official", help="分享链接格式: official(默认,官方 https://{uid}.share.123pan.cn/123pan/{key}) " "或 web (https://www.123pan.com/s/{key})") p_sc.set_defaults(func=cmd_share_create) # share list p_sl = share_sub.add_parser("list", help="列出分享") p_sl.add_argument("--limit", type=int, default=DEFAULT_LIMIT, help="每页数量") p_sl.add_argument("--last-share-id", help="上一页最后分享ID") p_sl.add_argument("--all-pages", action="store_true", help="自动获取所有分页") p_sl.set_defaults(func=cmd_share_list) # share update p_su = share_sub.add_parser("update", help="更新分享设置") p_su.add_argument("--share-ids", required=True, type=parse_int_list, help="分享ID (逗号分隔，如: 1,2,3)") p_su.add_argument("--traffic-switch", type=int, help="流量限制开关") p_su.add_argument("--traffic-limit-switch", type=int, help="流量限制值开关") p_su.add_argument("--traffic-limit", type=int, help="流量限制值") p_su.set_defaults(func=cmd_share_update) # share pay-create p_spc = share_sub.add_parser("pay-create", help="创建付费分享") p_spc.add_argument("--name", required=True, help="分享名称") p_spc.add_argument("--file-ids", required=True, type=parse_int_list, help="文件ID (逗号分隔，如: 1,2,3)") p_spc.add_argument("--amount", required=True, type=int, help="付费金额 (1-1000)") p_spc.add_argument("--is-reward", action="store_true", help="是否为打赏模式") p_spc.add_argument("--desc", help="资源描述") p_spc.set_defaults(func=cmd_share_pay_create) # share pay-list p_spl = share_sub.add_parser("pay-list", help="列出付费分享") p_spl.add_argument("--limit", type=int, default=DEFAULT_LIMIT, help="每页数量") p_spl.add_argument("--last-share-id", help="上一页最后分享ID") p_spl.add_argument("--all-pages", action="store_true", help="自动获取所有分页") p_spl.set_defaults(func=cmd_share_pay_list) # share pay-update p_spu = share_sub.add_parser("pay-update", help="更新付费分享设置") p_spu.add_argument("--share-ids", required=True, type=parse_int_list, help="分享ID (逗号分隔，如: 1,2,3)") p_spu.add_argument("--traffic-switch", type=int, help="流量限制开关") p_spu.add_argument("--traffic-limit-switch", type=int, help="流量限制值开关") p_spu.add_argument("--traffic-limit", type=int, help="流量限制值") p_spu.set_defaults(func=cmd_share_pay_update) # share create-batch p_scb = share_sub.add_parser("create-batch", help="批量创建分享 (递归遍历文件夹)") p_scb.add_argument("--parent-id", required=True, type=int, help="根文件夹ID") p_scb.add_argument("--expire", type=int, default=0, choices=[0, 1, 7, 30], help="有效期: 0=永久 1=1天 7=7天 30=30天 (默认永久)") p_scb.add_argument("--output", "-o", help="输出文件路径 (.xlsx 默认, 或 .json)；" "不指定且链接数超过阈值时自动生成 xlsx") p_scb.add_argument("--no-recursive", action="store_true", help="不递归，只处理直接子项文件") p_scb.add_argument("--no-cache", action="store_true", help="不使用缓存，强制重新遍历") p_scb.add_argument("--url-format", choices=["official", "web", "uid", "short"], default="official", help="分享链接格式: official(默认) 或 web") p_scb.add_argument("--excel-threshold", type=int, default=SHARE_EXCEL_THRESHOLD, help=f"链接数超过该值时自动生成 Excel (默认 {SHARE_EXCEL_THRESHOLD})") p_scb.add_argument("--no-excel", action="store_true", help="关闭自动 Excel，超过阈值时也只打印链接") p_scb.add_argument("--resume", action="store_true", help="断点续跑：跳过上次已完成的文件，仅处理剩余部分") p_scb.set_defaults(func=cmd_share_create_batch) # share stats p_ss = share_sub.add_parser("stats", help="分享链接统计汇总") p_ss.add_argument("--top", type=int, default=10, help="显示下载量 Top N (默认10)") p_ss.set_defaults(func=cmd_share_stats) # share delete-batch p_sdb = share_sub.add_parser("delete-batch", help="批量删除分享链接") p_sdb.add_argument("--share-ids", type=parse_int_list, help="分享ID (逗号分隔，如: 1,2,3)") p_sdb.add_argument("--all", action="store_true", help="删除所有分享链接") p_sdb.add_argument("-y", "--yes", action="store_true", help="跳过确认提示") p_sdb.add_argument("--dry-run", action="store_true", help="预览模式：只显示将被删除的分享，不执行操作") p_sdb.set_defaults(func=cmd_share_delete_batch) # share export p_se = share_sub.add_parser("export", help="导出分享列表到文件") p_se.add_argument("--output", "-o", required=True, help="输出文件路径 (.xlsx 或 .json)") p_se.add_argument("--url-format", choices=["official", "web", "uid", "short"], default="official", help="分享链接格式 (默认 official)") p_se.set_defaults(func=cmd_share_export) # ── offline 子命令组 ─────────────────────────────────── p_off = subparsers.add_parser("offline", help="离线下载") off_sub = p_off.add_subparsers(dest="offline_cmd", help="离线下载操作") # offline create p_oc = off_sub.add_parser("create", help="创建离线下载任务") p_oc.add_argument("--url", required=True, help="下载链接 (HTTP/FTP/磁力链接)") p_oc.add_argument("--parent-id", type=int, default=0, help="保存到的文件夹ID") p_oc.add_argument("--filename", help="指定文件名 (可选)") p_oc.set_defaults(func=cmd_offline_create) # offline progress p_op = off_sub.add_parser("progress", help="查询离线下载进度") p_op.add_argument("--task-id", required=True, help="任务ID") p_op.set_defaults(func=cmd_offline_progress) # offline list p_ol = off_sub.add_parser("list", help="列出离线下载任务") p_ol.set_defaults(func=cmd_offline_list) # ── direct-link 子命令组 ─────────────────────────────── p_direct = subparsers.add_parser("direct-link", help="直链管理") direct_sub = p_direct.add_subparsers(dest="direct_cmd", help="直链操作") # direct enable p_de = direct_sub.add_parser("enable", help="启用文件直链") p_de.add_argument("--file-id", required=True, type=int, help="文件ID") p_de.set_defaults(func=cmd_direct_enable) # direct disable p_dd = direct_sub.add_parser("disable", help="禁用文件直链") p_dd.add_argument("--file-id", required=True, type=int, help="文件ID") p_dd.set_defaults(func=cmd_direct_disable) # direct url p_du = direct_sub.add_parser("url", help="获取文件直链URL") p_du.add_argument("--file-id", required=True, type=int, help="文件ID") p_du.set_defaults(func=cmd_direct_url) # direct short-url p_dsu = direct_sub.add_parser("short-url", help="生成短直链URL（更隐私，不暴露文件名）") p_dsu.add_argument("--file-id", required=True, type=int, help="文件ID") p_dsu.set_defaults(func=cmd_direct_short_url) # direct cache-refresh p_dcr = direct_sub.add_parser("cache-refresh", help="刷新直链缓存") p_dcr.add_argument("--file-id", required=True, type=int, help="文件ID") p_dcr.set_defaults(func=cmd_direct_cache_refresh) # direct ip-list p_dil = direct_sub.add_parser("ip-list", help="获取IP白名单") p_dil.set_defaults(func=cmd_direct_ip_list) # direct ip-switch p_dis = direct_sub.add_parser("ip-switch", help="切换IP白名单开关") p_dis.add_argument("--switch", required=True, type=int, choices=[0, 1], dest="switch_state", help="开关状态: 0=关闭 1=开启") p_dis.set_defaults(func=cmd_direct_ip_switch) # direct ip-update p_diu = direct_sub.add_parser("ip-update", help="更新IP白名单") p_diu.add_argument("--ips", required=True, nargs="+", help="IP地址列表") p_diu.set_defaults(func=cmd_direct_ip_update) # direct offline-logs p_dol = direct_sub.add_parser("offline-logs", help="获取直链离线日志") p_dol.add_argument("--file-id", type=int, help="文件ID") p_dol.add_argument("--limit", type=int, help="每页数量") p_dol.add_argument("--offset", type=int, help="偏移量") p_dol.set_defaults(func=cmd_direct_offline_logs) # direct traffic-logs p_dtl = direct_sub.add_parser("traffic-logs", help="获取直链流量日志") p_dtl.add_argument("--file-id", type=int, help="文件ID") p_dtl.add_argument("--limit", type=int, help="每页数量") p_dtl.add_argument("--offset", type=int, help="偏移量") p_dtl.set_defaults(func=cmd_direct_traffic_logs) # ── oss (图床) 子命令组 ──────────────────────────────── p_oss = subparsers.add_parser("oss", help="图床管理") oss_sub = p_oss.add_subparsers(dest="oss_cmd", help="图床操作") # oss list p_ol = oss_sub.add_parser("list", help="列出图床文件") p_ol.add_argument("--parent-id", type=int, default=0, help="父文件夹ID") p_ol.add_argument("--limit", type=int, default=DEFAULT_LIMIT, help="每页数量") p_ol.add_argument("--last-file-id", help="上一页最后文件ID") p_ol.add_argument("--search", help="搜索关键词") p_ol.set_defaults(func=cmd_oss_list) # oss detail p_od = oss_sub.add_parser("detail", help="获取图床文件详情") p_od.add_argument("--file-id", required=True, type=int, help="文件ID") p_od.set_defaults(func=cmd_oss_detail) # oss delete p_odel = oss_sub.add_parser("delete", help="删除图床文件") p_odel.add_argument("--file-id", required=True, type=int, help="文件ID") p_odel.set_defaults(func=cmd_oss_delete) # oss move p_omv = oss_sub.add_parser("move", help="移动图床文件") p_omv.add_argument("--file-id", required=True, type=int, help="文件ID") p_omv.add_argument("--target-parent-id", required=True, type=int, help="目标文件夹ID") p_omv.set_defaults(func=cmd_oss_move) # oss copy p_ocp = oss_sub.add_parser("copy", help="复制图床文件") p_ocp.add_argument("--file-ids", required=True, type=parse_int_list, help="文件ID (逗号分隔，如: 1,2,3)") p_ocp.add_argument("--target-parent-id", required=True, type=int, help="目标文件夹ID") p_ocp.set_defaults(func=cmd_oss_copy) # oss copy-process p_oco = oss_sub.add_parser("copy-process", help="查询图床复制进度") p_oco.add_argument("--task-id", required=True, help="任务ID") p_oco.set_defaults(func=cmd_oss_copy_process) # oss copy-fail p_ocf = oss_sub.add_parser("copy-fail", help="查询图床复制失败详情") p_ocf.add_argument("--task-id", required=True, help="任务ID") p_ocf.set_defaults(func=cmd_oss_copy_fail) # ── transcode (视频转码) 子命令组 ────────────────────── p_tc = subparsers.add_parser("transcode", help="视频转码") tc_sub = p_tc.add_subparsers(dest="transcode_cmd", help="视频转码操作") # transcode list p_tcl = tc_sub.add_parser("list", help="列出转码视频") p_tcl.add_argument("--parent-file-id", type=int, help="父文件夹ID") p_tcl.add_argument("--limit", type=int, default=DEFAULT_LIMIT, help="每页数量") p_tcl.add_argument("--last-file-id", help="上一页最后文件ID") p_tcl.set_defaults(func=cmd_transcode_list) # transcode folder p_tcf = tc_sub.add_parser("folder", help="获取转码文件夹信息") p_tcf.add_argument("--parent-file-id", required=True, type=int, help="父文件夹ID") p_tcf.set_defaults(func=cmd_transcode_folder) # transcode resolutions p_tcr = tc_sub.add_parser("resolutions", help="获取视频可用分辨率") p_tcr.add_argument("--file-id", required=True, type=int, help="文件ID") p_tcr.set_defaults(func=cmd_transcode_resolutions) # transcode start p_tcs = tc_sub.add_parser("transcode", help="开始视频转码") p_tcs.add_argument("--file-id", required=True, type=int, help="文件ID") p_tcs.add_argument("--resolutions", required=True, nargs="+", help="分辨率列表 (如: 360p 480p 720p 1080p)") p_tcs.set_defaults(func=cmd_transcode_start) # transcode record p_tcrd = tc_sub.add_parser("record", help="获取转码记录") p_tcrd.add_argument("--file-id", required=True, type=int, help="文件ID") p_tcrd.set_defaults(func=cmd_transcode_record) # transcode result p_tcres = tc_sub.add_parser("result", help="获取转码结果") p_tcres.add_argument("--file-id", required=True, type=int, help="文件ID") p_tcres.set_defaults(func=cmd_transcode_result) # transcode delete p_tcdel = tc_sub.add_parser("delete", help="删除转码结果") p_tcdel.add_argument("--file-id", required=True, type=int, help="文件ID") p_tcdel.set_defaults(func=cmd_transcode_delete) # transcode download p_tcdl = tc_sub.add_parser("download", help="下载转码文件") p_tcdl.add_argument("--file-id", required=True, type=int, help="文件ID") p_tcdl.add_argument("--output", help="保存路径") p_tcdl.set_defaults(func=cmd_transcode_download) # transcode download-m3u8 p_tcm3 = tc_sub.add_parser("download-m3u8", help="下载转码M3U8文件") p_tcm3.add_argument("--file-id", required=True, type=int, help="文件ID") p_tcm3.add_argument("--output", help="保存路径") p_tcm3.set_defaults(func=cmd_transcode_download_m3u8) # transcode download-all p_tca = tc_sub.add_parser("download-all", help="下载所有转码结果") p_tca.add_argument("--file-id", required=True, type=int, help="文件ID") p_tca.add_argument("--output", help="保存路径") p_tca.set_defaults(func=cmd_transcode_download_all) # transcode upload-cloud p_tcuc = tc_sub.add_parser("upload-cloud", help="从云盘上传到转码") p_tcuc.add_argument("--file-id", required=True, type=int, help="文件ID") p_tcuc.add_argument("--parent-file-id", required=True, type=int, help="父文件夹ID") p_tcuc.set_defaults(func=cmd_transcode_upload_cloud) # ── safebox (保险箱) 子命令组 ────────────────────────── p_sb = subparsers.add_parser("safebox", help="保险箱管理") sb_sub = p_sb.add_subparsers(dest="safebox_cmd", help="保险箱操作") # safebox unlock p_sbu = sb_sub.add_parser("unlock", help="解锁保险箱") p_sbu.add_argument("--password", required=True, help="保险箱密码") p_sbu.set_defaults(func=cmd_safebox_unlock) # ── cache 子命令组 ────────────────────────────────────── p_cache = subparsers.add_parser("cache", help="缓存管理") cache_sub = p_cache.add_subparsers(dest="cache_cmd", help="缓存操作") # cache list p_cl = cache_sub.add_parser("list", help="列出所有缓存文件") p_cl.set_defaults(func=cmd_cache_list) # cache clear p_cc = cache_sub.add_parser("clear", help="清除缓存") p_cc.add_argument("--root-id", type=int, help="清除指定文件夹的缓存 (不指定则清除全部)") p_cc.add_argument("-y", "--yes", action="store_true", help="跳过确认提示") p_cc.set_defaults(func=cmd_cache_clear) # cache info p_ci = cache_sub.add_parser("info", help="显示缓存目录信息") p_ci.set_defaults(func=cmd_cache_info) # ── config ───────────────────────────────────────────── p_config = subparsers.add_parser("config", help="查看/验证配置") config_sub = p_config.add_subparsers(dest="config_cmd", help="配置操作") # config show (默认) def cmd_config_show(client, args): """显示当前配置""" cfg = { "配置文件": CONFIG_PATH, "Client ID": client.config.client_id[:8] + "..." if len(client.config.client_id) > 8 else client.config.client_id or "未设置", "Client Secret": "***" if client.config.client_secret else "未设置", "Access Token": client.config.access_token[:16] + "..." if len(client.config.access_token) > 16 else client.config.access_token or "未设置", "Token 过期时间": client.config.expired_at or "未设置", "配置文件存在": os.path.exists(CONFIG_PATH), } missing = client.config.check() if missing: cfg["状态"] = "⚠️ 不完整" cfg["缺少项"] = missing else: cfg["状态"] = "✅ 已配置" print_json(cfg) p_cs = config_sub.add_parser("show", help="显示当前配置") p_cs.set_defaults(func=cmd_config_show) # config check def cmd_config_check(client, args): """验证配置有效性（调用 API 测试）""" print("🔍 正在验证配置...") missing = client.config.check() if missing: print_error(f"配置不完整: {', '.join(missing)}") return # 测试 token 有效性 print(f" ✅ Client ID: 已配置") print(f" ✅ Client Secret: 已配置") print(f" ✅ Access Token: 已配置") # 调用 user/info 验证 token resp, err = _get_user_info(client) if err: print_error(f"Token 无效或已过期: {err}") if "access_token" in err or "401" in err: print_warn("请运行: python3 123yunpan.py setup 重新获取 token") return data = client._extract(resp) or {} print(f"\n{_colorize('✅ Token 有效！', '32')}") space = format_space_report(data) print(f" 用户ID: {space['uid']} 昵称: {space['nickname']} " f"会员: {space.get('vipLabel') or ('是' if space['vip'] else '否')}") print(f" 总空间: {human_size(space['total_quota'])}（已用 {human_size(space['total_used'])}）") print_space_details(space) p_cc = config_sub.add_parser("check", help="验证配置有效性（测试 API 连接）") p_cc.set_defaults(func=cmd_config_check) # config export p_ce = config_sub.add_parser("export", help="导出配置到文件") p_ce.add_argument("--output", "-o", required=True, help="输出文件路径") p_ce.set_defaults(func=cmd_config_export) # config import p_ci = config_sub.add_parser("import", help="从文件导入配置") p_ci.add_argument("--file", required=True, help="配置文件路径") p_ci.set_defaults(func=cmd_config_import) # 兼容：直接 `config` 等同于 `config show` p_config.set_defaults(func=cmd_config_show) # 解析参数 args = parser.parse_args() if not args.command: parser.print_help() sys.exit(0) # 应用全局选项 global NO_COLOR, JSON_OUTPUT if args.no_color: NO_COLOR = True if args.json_output: JSON_OUTPUT = True # 加载配置 config = Config() # 检查配置（setup 和 config 命令不需要检查） if args.command not in ("setup", "config"): if not config.check_warning(): sys.exit(1) # 创建客户端 client = ApiClient(config) # 执行子命令 if hasattr(args, "func"): try: args.func(client, args) except KeyboardInterrupt: print("\n\033[33m操作已取消\033[0m") sys.exit(130) except Exception as e: print_error(f"执行出错: {e}") import traceback traceback.print_exc() sys.exit(1) else: parser.print_help() if __name__ == "__main__": main()