nxxy335
f0ecfa6de0
- 完整 skill 本体：SKILL.md / README.md / manifest.json / config.json - scripts/123yunpan.py：零依赖 CLI（含空间双配额、QPS 限流、零依赖 XLSX） - references/ 参考资料 - 排除 .workbuddy/ 开发目录与 __pycache__
3407 lines
142 KiB
Python
3407 lines
142 KiB
Python
#!/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('<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
|
||
'<Types xmlns="http://schemas.openxmlformats.org/package/2006/content-types">'
|
||
'<Default Extension="rels" ContentType="application/vnd.openxmlformats-package.relationships+xml"/>'
|
||
'<Default Extension="xml" ContentType="application/xml"/>'
|
||
'<Override PartName="/xl/workbook.xml" ContentType="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet.main+xml"/>'
|
||
'<Override PartName="/xl/worksheets/sheet1.xml" ContentType="application/vnd.openxmlformats-officedocument.spreadsheetml.worksheet+xml"/>'
|
||
'<Override PartName="/xl/styles.xml" ContentType="application/vnd.openxmlformats-officedocument.spreadsheetml.styles+xml"/>'
|
||
'</Types>')
|
||
|
||
# _rels/.rels
|
||
parts.append('<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
|
||
'<Relationships xmlns="http://schemas.openxmlformats.org/package/2006/relationships">'
|
||
'<Relationship Id="rId1" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/officeDocument" Target="xl/workbook.xml"/>'
|
||
'</Relationships>')
|
||
|
||
# xl/workbook.xml
|
||
parts.append('<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
|
||
'<workbook xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main" '
|
||
'xmlns:r="http://schemas.openxmlformats.org/officeDocument/2006/relationships">'
|
||
f'<sheets><sheet name="{sheet_name}" sheetId="1" r:id="rId1"/></sheets>'
|
||
'</workbook>')
|
||
|
||
# xl/_rels/workbook.xml.rels
|
||
parts.append('<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
|
||
'<Relationships xmlns="http://schemas.openxmlformats.org/package/2006/relationships">'
|
||
'<Relationship Id="rId1" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/worksheet" Target="worksheets/sheet1.xml"/>'
|
||
'<Relationship Id="rId2" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/styles" Target="styles.xml"/>'
|
||
'</Relationships>')
|
||
|
||
# ── 样式表 ──────────────────────────────────────────────────────────────
|
||
style = (
|
||
'<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
|
||
'<styleSheet xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main">'
|
||
'<fonts count="3">'
|
||
'<font><sz val="11"/><color theme="1"/><name val="Microsoft YaHei"/><charset val="134"/></font>'
|
||
'<font><b/><sz val="11"/><color rgb="FFFFFFFF"/><name val="Microsoft YaHei"/><charset val="134"/></font>'
|
||
'<font><u/><sz val="11"/><color rgb="FF0563C1"/><name val="Microsoft YaHei"/><charset val="134"/></font>'
|
||
'</fonts>'
|
||
'<fills count="4">'
|
||
'<fill><patternFill patternType="none"/></fill>'
|
||
'<fill><patternFill patternType="gray125"/></fill>'
|
||
'<fill><patternFill patternType="solid"><fgColor rgb="FF4472C4"/><bgColor indexed="64"/></patternFill></fill>'
|
||
'<fill><patternFill patternType="solid"><fgColor rgb="FFD9E2F3"/><bgColor indexed="64"/></patternFill></fill>'
|
||
'</fills>'
|
||
'<borders count="2">'
|
||
'<border><left/><right/><top/><bottom/><diagonal/></border>'
|
||
'<border>'
|
||
'<left style="thin"><color rgb="FFBFBFBF"/></left>'
|
||
'<right style="thin"><color rgb="FFBFBFBF"/></right>'
|
||
'<top style="thin"><color rgb="FFBFBFBF"/></top>'
|
||
'<bottom style="thin"><color rgb="FFBFBFBF"/></bottom>'
|
||
'<diagonal/>'
|
||
'</border>'
|
||
'</borders>'
|
||
'<cellStyleXfs count="1"><xf numFmtId="0" fontId="0" fillId="0" borderId="0"/></cellStyleXfs>'
|
||
'<cellXfs count="5">'
|
||
f'<xf numFmtId="0" fontId="0" fillId="0" borderId="0" xfId="0" applyAlignment="1">'
|
||
'<alignment vertical="center"/></xf>'
|
||
f'<xf numFmtId="0" fontId="1" fillId="2" borderId="1" xfId="0" applyFont="1" applyFill="1" applyBorder="1" applyAlignment="1">'
|
||
'<alignment horizontal="center" vertical="center"/></xf>'
|
||
f'<xf numFmtId="0" fontId="0" fillId="3" borderId="1" xfId="0" applyFill="1" applyBorder="1" applyAlignment="1">'
|
||
'<alignment vertical="center"/></xf>'
|
||
f'<xf numFmtId="0" fontId="2" fillId="0" borderId="1" xfId="0" applyFont="1" applyBorder="1" applyAlignment="1">'
|
||
'<alignment vertical="center"/></xf>'
|
||
f'<xf numFmtId="0" fontId="0" fillId="0" borderId="1" xfId="0" applyFont="1" applyBorder="1" applyAlignment="1">'
|
||
'<alignment vertical="center"/></xf>'
|
||
'</cellXfs>'
|
||
'<cellStyles count="1"><cellStyle name="Normal" xfId="0" builtinId="0"/>'
|
||
'</cellStyles>'
|
||
'</styleSheet>'
|
||
)
|
||
parts.append(style)
|
||
|
||
# ── 工作表 ──────────────────────────────────────────────────────────────
|
||
last_col = _col_letter(max(n_cols, 1))
|
||
sheet = ['<?xml version="1.0" encoding="UTF-8" standalone="yes"?>',
|
||
'<worksheet xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main">',
|
||
f'<dimension ref="A1:{last_col}{n_rows + 1}"/>',
|
||
'<sheetViews><sheetView tabSelected="1" workbookViewId="0"/></sheetViews>',
|
||
'<sheetFormatPr defaultRowHeight="18"/>']
|
||
|
||
if widths:
|
||
sheet.append("<cols>")
|
||
for i, w in enumerate(widths[:n_cols], 1):
|
||
sheet.append(f'<col min="{i}" max="{i}" width="{w}" customWidth="1"/>')
|
||
sheet.append("</cols>")
|
||
|
||
sheet.append("<sheetData>")
|
||
|
||
# 表头行
|
||
sheet.append(f'<row r="1" ht="24" customHeight="1">')
|
||
for c, h in enumerate(headers, 1):
|
||
ref = f"{_col_letter(c)}1"
|
||
sheet.append(f'<c r="{ref}" s="{S_HEADER}" t="inlineStr"><is><t xml:space="preserve">'
|
||
f'{_xml_escape(h)}</t></is></c>')
|
||
sheet.append("</row>")
|
||
|
||
# 数据行
|
||
for r_i, row in enumerate(rows, start=2):
|
||
zebra = "1" if r_i % 2 == 0 else "0"
|
||
sheet.append(f'<row r="{r_i}" ht="18" customHeight="1">')
|
||
for c_i, value in enumerate(row[:n_cols]):
|
||
if value is None or value == "":
|
||
sheet.append(f'<c r="{_col_letter(c_i + 1)}{r_i}" s="{S_CELL}"/>')
|
||
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:
|
||
# 单元格里放可显示的文本，点击跳转由 <hyperlinks> 关系决定
|
||
sheet.append(f'<c r="{ref}" s="{style_idx}" t="inlineStr"><is>'
|
||
f'<t xml:space="preserve">{_xml_escape(value)}</t></is></c>')
|
||
elif isinstance(value, (int, float)) and not isinstance(value, bool):
|
||
sheet.append(f'<c r="{ref}" s="{style_idx}"><v>{value}</v></c>')
|
||
else:
|
||
sheet.append(f'<c r="{ref}" s="{style_idx}" t="inlineStr"><is>'
|
||
f'<t xml:space="preserve">{_xml_escape(value)}</t></is></c>')
|
||
sheet.append("</row>")
|
||
sheet.append("</sheetData>")
|
||
|
||
# 超链接关系
|
||
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("<hyperlinks>")
|
||
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'<hyperlink ref="{col}{r_i}" r:id="{rid}" '
|
||
f'xmlns:r="http://schemas.openxmlformats.org/officeDocument/2006/relationships"/>')
|
||
sheet.append("</hyperlinks>")
|
||
sheet.append("</worksheet>")
|
||
parts.append("".join(sheet))
|
||
|
||
# 超链接的关系文件（targetMode=External 不可省略）
|
||
rel_items = "".join(
|
||
f'<Relationship Id="rId{i}" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/hyperlink" '
|
||
f'Target="{_xml_attr(url)}" TargetMode="External"/>'
|
||
for i, (_, _, url) in enumerate(links, start=1))
|
||
parts.append('<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
|
||
'<Relationships xmlns="http://schemas.openxmlformats.org/package/2006/relationships">'
|
||
f'{rel_items}</Relationships>')
|
||
|
||
# 打包
|
||
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 <task_id> 查询进度"
|
||
})
|
||
return
|
||
|
||
print_warn("123云盘 API 没有提供列出所有离线下载任务的接口")
|
||
print()
|
||
print(" 离线下载流程：")
|
||
print(" 1. 使用 offline create 创建离线下载任务，获取 task_id")
|
||
print(" 2. 使用 offline progress <task_id> 查询任务进度")
|
||
print()
|
||
print(" 示例：")
|
||
print(" 123yunpan offline create --url <磁力链接或HTTP链接>")
|
||
print(" 123yunpan offline progress --task-id <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()
|