bot-ci
8e7ead1669
Load business config as JSONC (comments + trailing commas) via an in-repo stripper then stdlib json — no new deps. Default path prefers config.jsonc, falls back to legacy config.json. Auth write-back targets only sibling secrets.json so JSONC comments are never wiped; secrets overlay merges on load. Examples, compose, Docker, docs, favsort and tools all share the same loader. Phase A only — no DB settings migration.
716 lines
25 KiB
Python
716 lines
25 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
bili-fav / favsort —— B 站收藏夹批量重排序工具
|
||
|
||
排序规则
|
||
1) 同一个 upper 的视频必须连续（聚成一组），组内按发布时间倒序（并列按 avid 升序）
|
||
2) 组间按"该 upper 在本收藏夹中最早视频的发布时间"倒序（并列按该组最小 avid 升序）
|
||
3) 失效/异常资源置于末尾，保持原相对顺序
|
||
4) 配置里的 overrides 可把指定视频强制归入另一个 upper（代发等场景）
|
||
|
||
指令
|
||
python favsort.py help [子命令]
|
||
python favsort.py folders
|
||
python favsort.py plan <media_id> [--folder-mid N] [-v]
|
||
python favsort.py show <media_id>
|
||
python favsort.py replan <media_id>
|
||
python favsort.py run <media_id> [--verify-every N]
|
||
python favsort.py verify <media_id>
|
||
|
||
本工具只处理写账号所属收藏夹的排序。扫码授权统一使用
|
||
``python app.py auth --account write``；所有 API 请求均使用 ``bilibili.write``。
|
||
"""
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import random
|
||
import sys
|
||
import time
|
||
import urllib.parse
|
||
|
||
import httpx
|
||
|
||
sys.stdout.reconfigure(encoding="utf-8")
|
||
|
||
BASE = "https://api.bilibili.com"
|
||
ROOT = os.path.dirname(os.path.abspath(__file__))
|
||
DATA_DIR = os.path.join(ROOT, "data")
|
||
from reaction_board.bilibili.auth import ( # noqa: E402
|
||
make_headers,
|
||
)
|
||
from reaction_board.config import ( # noqa: E402
|
||
ConfigError,
|
||
default_config_path,
|
||
load_config as load_app_config,
|
||
)
|
||
|
||
CONFIG_PATH = str(default_config_path(ROOT))
|
||
|
||
CHUNK = 20 # infos 接口单次 resources 数量（实测 20 稳定）
|
||
API = {
|
||
"ids": "{base}/x/v3/fav/resource/ids?media_id={mid}&platform=web",
|
||
"infos": "{base}/x/v3/fav/resource/infos?resources={res}&platform=web&folder_mid={fmid}&folder_id={mid}",
|
||
"sort": "{base}/x/v3/fav/resource/sort",
|
||
"folders": "{base}/x/v3/fav/folder/created/list-all?up_mid={mid}&jsonp=jsonp",
|
||
}
|
||
|
||
# ---------------------------------------------------------------- config
|
||
|
||
DEFAULT_CONFIG = {
|
||
"overrides": {},
|
||
"sleep": {
|
||
"min": 2.5,
|
||
"max": 7.5,
|
||
"long_every_min": 20,
|
||
"long_every_max": 40,
|
||
"long_min": 30,
|
||
"long_max": 90,
|
||
"retry_wait": 45,
|
||
},
|
||
}
|
||
|
||
|
||
def load_config(require_auth=True):
|
||
if not os.path.exists(CONFIG_PATH):
|
||
raise SystemExit(
|
||
f"缺少 {CONFIG_PATH}\n"
|
||
f"请复制 config.example.jsonc 为 config.jsonc，并复制 secrets.example.json 为 secrets.json 后填入凭据\n"
|
||
f"（或运行 python app.py auth --account write）。"
|
||
)
|
||
try:
|
||
app_cfg = load_app_config(CONFIG_PATH, require_auth=False)
|
||
except ConfigError as exc:
|
||
raise SystemExit(str(exc)) from exc
|
||
write = app_cfg.bilibili.write
|
||
raw = app_cfg.raw
|
||
cfg = {
|
||
"mid": write.mid,
|
||
"csrf": write.csrf,
|
||
"cookie": write.cookie,
|
||
"overrides": raw.get("overrides", {}),
|
||
"sleep": raw.get("sleep", {}),
|
||
}
|
||
for k, v in DEFAULT_CONFIG.items():
|
||
cfg.setdefault(k, v)
|
||
for k, v in DEFAULT_CONFIG["sleep"].items():
|
||
cfg["sleep"].setdefault(k, v)
|
||
if require_auth and (not cfg["cookie"] or not cfg["csrf"]):
|
||
raise SystemExit(
|
||
"bilibili.write.cookie / csrf 不能为空"
|
||
f"（写入 {app_cfg.secrets_path.name}，或运行 python app.py auth --account write）"
|
||
)
|
||
return cfg
|
||
|
||
|
||
# ---------------------------------------------------------------- http
|
||
|
||
_HTTP_CLIENT: httpx.Client | None = None
|
||
|
||
|
||
def _http_client() -> httpx.Client:
|
||
global _HTTP_CLIENT
|
||
if _HTTP_CLIENT is None:
|
||
_HTTP_CLIENT = httpx.Client(
|
||
timeout=httpx.Timeout(30, connect=10),
|
||
limits=httpx.Limits(
|
||
max_keepalive_connections=5,
|
||
keepalive_expiry=30,
|
||
),
|
||
follow_redirects=False,
|
||
trust_env=True,
|
||
)
|
||
return _HTTP_CLIENT
|
||
|
||
|
||
def _get_json(cfg, url, auth, referer=None):
|
||
response = _http_client().get(
|
||
url,
|
||
headers=make_headers(cfg, auth=auth, referer=referer),
|
||
timeout=30,
|
||
)
|
||
response.raise_for_status()
|
||
return response.json()
|
||
|
||
|
||
def get_json(cfg, url, auth=False, referer=None):
|
||
"""只读请求：默认用去掉账号身份的 cookie。
|
||
|
||
万一匿名请求被拒（HTTP 错误，或 code != 0），自动带完整 cookie 重试一次，
|
||
保证功能不会因为剥离身份而回退。
|
||
"""
|
||
if auth:
|
||
return _get_json(cfg, url, auth=True, referer=referer)
|
||
try:
|
||
r = _get_json(cfg, url, auth=False, referer=referer)
|
||
if r.get("code") == 0:
|
||
return r
|
||
except Exception: # noqa: BLE001 网络错误 / 412 风控都走回退
|
||
pass
|
||
return _get_json(cfg, url, auth=True, referer=referer)
|
||
|
||
|
||
def post_form(cfg, url, data, referer=None):
|
||
headers = make_headers(cfg, auth=True, referer=referer)
|
||
headers["content-type"] = "application/x-www-form-urlencoded"
|
||
response = _http_client().post(
|
||
url, headers=headers, data=data, timeout=30
|
||
)
|
||
response.raise_for_status()
|
||
return response.json()
|
||
|
||
|
||
def fetch_ids(cfg, media_id):
|
||
referer = (
|
||
f"https://space.bilibili.com/{cfg['mid']}/favlist"
|
||
f"?fid={media_id}&ftype=create"
|
||
)
|
||
r = get_json(cfg, API["ids"].format(base=BASE, mid=media_id), referer=referer)
|
||
if r.get("code") != 0:
|
||
raise SystemExit(f"ids 失败: code={r.get('code')} msg={r.get('message')}")
|
||
return r.get("data") or []
|
||
|
||
|
||
def fetch_infos(cfg, media_id, folder_mid, id_list):
|
||
items, calls = [], 0
|
||
referer = (
|
||
f"https://space.bilibili.com/{folder_mid}/favlist"
|
||
f"?fid={media_id}&ftype=create"
|
||
)
|
||
for i in range(0, len(id_list), CHUNK):
|
||
chunk = id_list[i:i + CHUNK]
|
||
res = ",".join(f"{x['id']}:{x['type']}" for x in chunk)
|
||
url = API["infos"].format(
|
||
base=BASE, res=urllib.parse.quote(res), fmid=folder_mid, mid=media_id
|
||
)
|
||
r = get_json(cfg, url, referer=referer)
|
||
calls += 1
|
||
if r.get("code") != 0:
|
||
raise SystemExit(f"infos 失败 @{i}: code={r.get('code')} msg={r.get('message')}")
|
||
m = {g["id"]: g for g in (r.get("data") or [])}
|
||
for c in chunk:
|
||
g = m.get(c["id"]) or {}
|
||
up = g.get("upper") or {}
|
||
items.append({
|
||
"avid": c["id"],
|
||
"type": c["type"],
|
||
"title": g.get("title") or "(缺失)",
|
||
"upper_mid": up.get("mid"),
|
||
"upper_name": up.get("name") or "(未知UP)",
|
||
"pubtime": g.get("pubtime") or g.get("ctime") or 0,
|
||
"attr": g.get("attr", 0),
|
||
"found": bool(g),
|
||
})
|
||
return items, calls
|
||
|
||
|
||
# ---------------------------------------------------------------- core
|
||
|
||
def build_target(items, overrides):
|
||
applied = []
|
||
for v in items:
|
||
w = dict(v)
|
||
o = overrides.get(str(v["avid"])) or overrides.get(v["avid"])
|
||
if o:
|
||
w["upper_mid"] = o["mid"]
|
||
w["upper_name"] = o["name"]
|
||
w["overridden"] = True
|
||
applied.append(w)
|
||
|
||
invalid = [x for x in applied if (not x["found"]) or x["attr"] != 0 or not x["upper_mid"]]
|
||
valid = [x for x in applied if x not in invalid]
|
||
|
||
groups = {}
|
||
for v in valid:
|
||
g = groups.setdefault(v["upper_mid"], {"mid": v["upper_mid"], "name": v["upper_name"], "vids": []})
|
||
g["vids"].append(v)
|
||
garr = list(groups.values())
|
||
for g in garr:
|
||
g["vids"].sort(key=lambda v: (-v["pubtime"], v["avid"]))
|
||
g["key"] = min(v["pubtime"] for v in g["vids"])
|
||
g["min_avid"] = min(v["avid"] for v in g["vids"])
|
||
garr.sort(key=lambda g: (-g["key"], g["min_avid"]))
|
||
|
||
target = [v for g in garr for v in g["vids"]] + invalid
|
||
return applied, garr, target
|
||
|
||
|
||
def lcs_keep(cur, tgt):
|
||
n = len(cur)
|
||
dp = [[0] * (n + 1) for _ in range(n + 1)]
|
||
for i in range(n - 1, -1, -1):
|
||
for j in range(n - 1, -1, -1):
|
||
dp[i][j] = dp[i + 1][j + 1] + 1 if cur[i] == tgt[j] else max(dp[i + 1][j], dp[i][j + 1])
|
||
i = j = 0
|
||
kept = set()
|
||
while i < n and j < n:
|
||
if cur[i] == tgt[j]:
|
||
kept.add(cur[i]); i += 1; j += 1
|
||
elif dp[i + 1][j] >= dp[i][j + 1]:
|
||
i += 1
|
||
else:
|
||
j += 1
|
||
return kept
|
||
|
||
|
||
def _one_pass(work, tgt):
|
||
"""单轮：保留 LCS 里的元素不动，其余元素逐个插到"目标前驱"之后。
|
||
|
||
原地修改 work，返回本轮动作。返回后 work 会非常接近 tgt（通常只差 0~5 个位置），
|
||
由 build_moves 多跑几轮收敛。
|
||
"""
|
||
kept = lcs_keep(work, tgt)
|
||
pos = {a: i for i, a in enumerate(tgt)}
|
||
moves = []
|
||
|
||
# 首位：API 只能"插到某人之后"，要占住第 1 位得走两步
|
||
if work[0] != tgt[0]:
|
||
a = tgt[0]
|
||
first = work[0]
|
||
f = work.index(a)
|
||
if f != 1:
|
||
work.pop(f)
|
||
work.insert(1, a)
|
||
moves.append({"moved": a, "pred": first, "from": f + 1, "to": 2})
|
||
k = pos[first]
|
||
work.remove(first)
|
||
pred = work[k - 1]
|
||
work.insert(k, first)
|
||
moves.append({"moved": first, "pred": pred, "from": 1, "to": k + 1})
|
||
kept.discard(a)
|
||
kept.discard(first)
|
||
|
||
for t in range(1, len(tgt)):
|
||
a = tgt[t]
|
||
if work[t] == a or a in kept:
|
||
continue
|
||
f = work.index(a)
|
||
work.pop(f)
|
||
i = work.index(tgt[t - 1])
|
||
work.insert(i + 1, a)
|
||
moves.append({"moved": a, "pred": tgt[t - 1], "from": f + 1, "to": i + 2})
|
||
kept.discard(a)
|
||
return moves
|
||
|
||
|
||
def build_moves(cur, tgt):
|
||
"""生成单步移动脚本。
|
||
|
||
服务端语义（实测）：sort=A:2:B:2 => 把 B 插到 A 之后（B 可以往前也可以往后）。
|
||
最少步数 ≈ n - LCS(cur, tgt)：保留最长公共子序列不动，只搬其余元素。
|
||
由于"保留元素"需要靠其它元素被搬走才会滑到正确位置，单轮往往差几个位置，
|
||
多跑一两轮即可收敛（实测 658 条约 100 步，正好等于理论下界）。
|
||
"""
|
||
work = list(cur)
|
||
moves = []
|
||
for _ in range(20):
|
||
if work == tgt:
|
||
break
|
||
part = _one_pass(work, tgt)
|
||
if not part:
|
||
break
|
||
moves.extend(part)
|
||
assert work == tgt, "内部错误：生成的动作未能到达目标顺序"
|
||
return moves
|
||
|
||
|
||
def apply_move(work, m):
|
||
work = list(work)
|
||
work.remove(m["moved"])
|
||
i = work.index(m["pred"])
|
||
work.insert(i + 1, m["moved"])
|
||
return work
|
||
|
||
|
||
# ---------------------------------------------------------------- io
|
||
|
||
def ensure_data_dir():
|
||
os.makedirs(DATA_DIR, exist_ok=True)
|
||
|
||
|
||
def plan_path(media_id):
|
||
return os.path.join(DATA_DIR, f"plan-{media_id}.json")
|
||
|
||
|
||
def journal_path(media_id):
|
||
return os.path.join(DATA_DIR, f"journal-{media_id}.jsonl")
|
||
|
||
|
||
def load_plan(media_id):
|
||
p = plan_path(media_id)
|
||
if not os.path.exists(p):
|
||
raise SystemExit(f"找不到计划文件 {p}，请先运行: python favsort.py plan {media_id}")
|
||
with open(p, encoding="utf-8") as f:
|
||
return json.load(f)
|
||
|
||
|
||
# ---------------------------------------------------------------- commands
|
||
|
||
|
||
def cmd_folders(cfg, args):
|
||
# 注意：list-all 是"列出我自己的收藏夹"，匿名调用只会返回公开收藏夹（实测 11 -> 3），
|
||
# 因此这里必须带完整 cookie（仍然只是 GET，不会改动任何数据）。
|
||
r = get_json(
|
||
cfg,
|
||
API["folders"].format(base=BASE, mid=cfg["mid"]),
|
||
auth=True,
|
||
referer=f"https://space.bilibili.com/{cfg['mid']}/favlist",
|
||
)
|
||
if r.get("code") != 0:
|
||
raise SystemExit(f"获取收藏夹列表失败: {r.get('code')} {r.get('message')}")
|
||
lst = (r.get("data") or {}).get("list") or []
|
||
print(f"共 {len(lst)} 个自建收藏夹（mid={cfg['mid']}）")
|
||
for f in sorted(lst, key=lambda x: -x.get("media_count", 0)):
|
||
print(f" {f['id']:<12} {f.get('media_count', 0):>5} 条 {f['title']}")
|
||
|
||
|
||
def cmd_plan(cfg, args):
|
||
media_id = args.media_id
|
||
folder_mid = args.folder_mid or cfg["mid"]
|
||
ensure_data_dir()
|
||
|
||
print(f"[plan] 拉取资源 id (media_id={media_id}) ...")
|
||
id_list = fetch_ids(cfg, media_id)
|
||
print(f"[plan] {len(id_list)} 条资源，分批拉取详情 ...")
|
||
items, calls = fetch_infos(cfg, media_id, folder_mid, id_list)
|
||
|
||
applied, garr, target = build_target(items, cfg["overrides"])
|
||
cur = [v["avid"] for v in applied]
|
||
tgt = [v["avid"] for v in target]
|
||
kept = lcs_keep(cur, tgt)
|
||
moves = build_moves(cur, tgt)
|
||
|
||
plan = {
|
||
"media_id": media_id,
|
||
"folder_mid": folder_mid,
|
||
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
|
||
"requests": {"ids": 1, "infos": calls},
|
||
"n": len(cur),
|
||
"groups": len(garr),
|
||
"lcs": len(kept),
|
||
"current_order": cur,
|
||
"target_order": tgt,
|
||
"moves": moves,
|
||
"types": {v["avid"]: v["type"] for v in applied},
|
||
"titles": {
|
||
v["avid"]: {
|
||
"title": v["title"],
|
||
"upper": v["upper_name"],
|
||
"pubtime": v["pubtime"],
|
||
"overridden": v.get("overridden", False),
|
||
}
|
||
for v in applied
|
||
},
|
||
"group_summary": [
|
||
{
|
||
"name": g["name"],
|
||
"mid": g["mid"],
|
||
"key": g["key"],
|
||
"count": len(g["vids"]),
|
||
"newest": max(v["pubtime"] for v in g["vids"]),
|
||
}
|
||
for g in garr
|
||
],
|
||
"unchanged": cur == tgt,
|
||
}
|
||
with open(plan_path(media_id), "w", encoding="utf-8") as f:
|
||
json.dump(plan, f, ensure_ascii=False, indent=1)
|
||
|
||
print(f"[plan] n={plan['n']} 分组={plan['groups']} 不动={plan['lcs']} "
|
||
f"需移动={len(moves)} 已是最优={plan['unchanged']}")
|
||
print(f"[plan] 已保存 -> {plan_path(media_id)}")
|
||
if moves and args.verbose:
|
||
local_titles = {v["avid"]: v["title"] for v in applied}
|
||
for i, m in enumerate(moves, 1):
|
||
print(f" {i:>4}. {local_titles[m['moved']][:44]:<45} {m['from']:>4} -> {m['to']:<4}")
|
||
if moves and not args.verbose:
|
||
print("[plan] 加 --verbose 可查看完整移动脚本")
|
||
|
||
|
||
def cmd_show(cfg, args):
|
||
plan = load_plan(args.media_id)
|
||
print(f"media={plan['media_id']} n={plan['n']} 分组={plan['groups']} 移动={len(plan['moves'])}")
|
||
print(f"plan 生成于 {plan.get('created_at')}，执行于 {plan.get('applied_at', '未执行')}")
|
||
for i, m in enumerate(plan["moves"], 1):
|
||
t = plan["titles"][str(m["moved"])]
|
||
print(f" {i:>4}. {t['title'][:44]:<45} {m['from']:>4} -> {m['to']:<4}")
|
||
|
||
|
||
def cmd_run(cfg, args):
|
||
media_id = args.media_id
|
||
plan = load_plan(media_id)
|
||
target = plan["target_order"]
|
||
titles = plan["titles"]
|
||
types = plan.get("types") or {}
|
||
sc = cfg["sleep"]
|
||
verify_every = args.verify_every
|
||
if verify_every < 1:
|
||
print("[run] --verify-every 必须 >= 1；未执行任何写操作")
|
||
return 2
|
||
|
||
print("[run] 读取当前顺序 ...")
|
||
actual = [x["id"] for x in fetch_ids(cfg, media_id)]
|
||
if actual == target:
|
||
print("[run] 顺序已与目标一致，无需操作")
|
||
return 0
|
||
|
||
if set(actual) != set(target):
|
||
added = set(actual) - set(target)
|
||
removed = set(target) - set(actual)
|
||
print(f"[run] 收藏夹内容已变化：计划 {len(target)} 条 / 现在 {len(actual)} 条，"
|
||
f"新增 {len(added)} 条、移除 {len(removed)} 条")
|
||
print("[run] 目标序已过期，请先重新生成计划: python favsort.py plan %d" % media_id)
|
||
return 6
|
||
|
||
moves = build_moves(actual, target)
|
||
if any(m["pred"] is None for m in moves):
|
||
print("[run] 存在需移动到第 1 位（无前驱锚点）的步骤，当前版本不支持")
|
||
return 5
|
||
|
||
print(f"[run] 从当前状态重算: {len(moves)} 步 / 目标 {len(target)} 条")
|
||
for i, m in enumerate(moves, 1):
|
||
t = titles[str(m["moved"])]
|
||
print(f" {i:>4}. {t['title'][:40]:<41} {m['from']:>4} -> {m['to']:<4}")
|
||
|
||
open(journal_path(media_id), "w", encoding="utf-8").close()
|
||
work = list(actual)
|
||
total = len(moves)
|
||
long_after = random.randint(sc["long_every_min"], sc["long_every_max"])
|
||
sort_referer = (
|
||
f"https://space.bilibili.com/{cfg['mid']}/favlist"
|
||
f"?fid={media_id}&ftype=create"
|
||
)
|
||
|
||
for i, m in enumerate(moves, 1):
|
||
pred_type = types.get(m["pred"], 2)
|
||
moved_type = types.get(m["moved"], 2)
|
||
body = {
|
||
"media_id": media_id,
|
||
"sort": f"{m['pred']}:{pred_type}:{m['moved']}:{moved_type}",
|
||
"csrf": cfg["csrf"],
|
||
}
|
||
print(f"[run] {i}/{total} {titles[str(m['moved'])]['title'][:30]} "
|
||
f"-> after {m['pred']}", end=" ", flush=True)
|
||
resp = post_form(
|
||
cfg, API["sort"].format(base=BASE), body, referer=sort_referer
|
||
)
|
||
code = resp.get("code")
|
||
print(f"code={code}")
|
||
if code != 0:
|
||
wait = sc["retry_wait"]
|
||
print(f"[run] 返回非 0，冷却 {wait}s 后重试一次")
|
||
time.sleep(wait)
|
||
resp = post_form(
|
||
cfg, API["sort"].format(base=BASE), body, referer=sort_referer
|
||
)
|
||
code = resp.get("code")
|
||
print(f"[run] retry code={code}")
|
||
if code != 0:
|
||
print("[run] 重试仍失败，中止（可用 run 续跑）")
|
||
return 2
|
||
|
||
work = apply_move(work, m)
|
||
with open(journal_path(media_id), "a", encoding="utf-8") as f:
|
||
f.write(json.dumps({
|
||
"i": i, "moved": m["moved"], "pred": m["pred"], "to": m["to"],
|
||
"code": code, "ts": time.strftime("%Y-%m-%d %H:%M:%S"),
|
||
}, ensure_ascii=False) + "\n")
|
||
|
||
time.sleep(random.uniform(sc["min"], sc["max"]))
|
||
|
||
if i <= 3 or i % verify_every == 0 or i == total:
|
||
now = [x["id"] for x in fetch_ids(cfg, media_id)]
|
||
if now != work:
|
||
print(f"[verify] 第 {i} 步后与模拟不符，停止")
|
||
for k in range(min(len(now), len(work))):
|
||
if now[k] != work[k]:
|
||
print(f" 首个差异 idx={k} 实际={now[k]} 预期={work[k]}")
|
||
break
|
||
return 3
|
||
print(f"[verify] 第 {i} 步 OK")
|
||
time.sleep(random.uniform(sc["min"], sc["max"]))
|
||
|
||
if i % long_after == 0 and i != total:
|
||
pause = random.uniform(sc["long_min"], sc["long_max"])
|
||
print(f"[run] 长冷却 {pause:.0f}s")
|
||
time.sleep(pause)
|
||
long_after = random.randint(sc["long_every_min"], sc["long_every_max"])
|
||
|
||
final = [x["id"] for x in fetch_ids(cfg, media_id)]
|
||
ok = final == target
|
||
print(f"[done] 最终顺序 {'与目标完全一致 ✅' if ok else '与目标不一致 ❌'}")
|
||
plan["moves"] = moves
|
||
plan["applied_at"] = time.strftime("%Y-%m-%d %H:%M:%S")
|
||
with open(plan_path(media_id), "w", encoding="utf-8") as f:
|
||
json.dump(plan, f, ensure_ascii=False, indent=1)
|
||
if not ok:
|
||
for k in range(min(len(final), len(target))):
|
||
if final[k] != target[k]:
|
||
print(f" 首个差异 idx={k} 实际={final[k]} 目标={target[k]}")
|
||
break
|
||
return 4
|
||
return 0
|
||
|
||
|
||
def cmd_replan(cfg, args):
|
||
"""不联网：用 plan 里保存的 current_order / target_order 重新生成移动脚本。"""
|
||
media_id = args.media_id
|
||
plan = load_plan(media_id)
|
||
cur, tgt = plan["current_order"], plan["target_order"]
|
||
if cur == tgt:
|
||
print("[replan] 原序即目标序，无需移动")
|
||
return 0
|
||
kept = lcs_keep(cur, tgt)
|
||
moves = build_moves(cur, tgt)
|
||
plan["moves"] = moves
|
||
plan["lcs"] = len(kept)
|
||
plan["replanned_at"] = time.strftime("%Y-%m-%d %H:%M:%S")
|
||
with open(plan_path(media_id), "w", encoding="utf-8") as f:
|
||
json.dump(plan, f, ensure_ascii=False, indent=1)
|
||
print(f"[replan] n={plan['n']} 不动={len(kept)} 理论下界={plan['n'] - len(kept)} "
|
||
f"实际步数={len(moves)}")
|
||
titles = plan["titles"]
|
||
for i, m in enumerate(moves, 1):
|
||
t = titles[str(m["moved"])]
|
||
print(f" {i:>4}. {t['title'][:40]:<41} {m['from']:>4} -> {m['to']:<4}")
|
||
return 0
|
||
|
||
|
||
def cmd_verify(cfg, args):
|
||
plan = load_plan(args.media_id)
|
||
target = plan["target_order"]
|
||
cur = [x["id"] for x in fetch_ids(cfg, args.media_id)]
|
||
if cur == target:
|
||
print(f"[verify] 与目标顺序完全一致 ✅（n={len(cur)}）")
|
||
return 0
|
||
diffs = [i for i in range(min(len(cur), len(target))) if cur[i] != target[i]]
|
||
print(f"[verify] 不一致 ❌ 差异位置 {len(diffs)} 处，前 5 处：")
|
||
for i in diffs[:5]:
|
||
t = plan["titles"]
|
||
print(f" idx={i} 当前={t.get(str(cur[i]), {}).get('title', '?')[:36]}")
|
||
print(f" 目标={t.get(str(target[i]), {}).get('title', '?')[:36]}")
|
||
return 1
|
||
|
||
|
||
# ---------------------------------------------------------------- main
|
||
|
||
COMMAND_HELP = {
|
||
"folders": "列出写账号（bilibili.write.mid）的自建收藏夹及其 media_id",
|
||
"plan": "拉取收藏夹数据并生成排序计划（只读，不改动收藏夹）",
|
||
"show": "查看已生成的计划与完整移动脚本（离线）",
|
||
"replan": "离线重算移动脚本（不联网，不改目标顺序）",
|
||
"run": "按计划逐步执行移动（写操作，每步一个 sort 请求）",
|
||
"verify": "对比当前实际顺序与计划目标",
|
||
"help": "列出所有子命令与可用选项",
|
||
}
|
||
|
||
|
||
def _positive_int(value):
|
||
try:
|
||
number = int(value)
|
||
except (TypeError, ValueError) as exc:
|
||
raise argparse.ArgumentTypeError("必须是整数") from exc
|
||
if number < 1:
|
||
raise argparse.ArgumentTypeError("必须 >= 1")
|
||
return number
|
||
|
||
|
||
def build_parser():
|
||
p = argparse.ArgumentParser(
|
||
prog="favsort",
|
||
description="B 站收藏夹批量重排序：同 UP 聚在一起，组间按该 UP 最早投稿时间倒序。",
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
)
|
||
sub = p.add_subparsers(dest="cmd", title="子命令", metavar="<command>")
|
||
|
||
sub.add_parser("folders", help=COMMAND_HELP["folders"])
|
||
|
||
pp = sub.add_parser("plan", help=COMMAND_HELP["plan"])
|
||
pp.add_argument("media_id", type=int, help="收藏夹 media_id（数字，见 folders 输出）")
|
||
pp.add_argument("--folder-mid", type=int, default=None, metavar="N",
|
||
help="收藏夹归属用户 mid（默认取 bilibili.write.mid）")
|
||
pp.add_argument("--verbose", "-v", action="store_true", help="打印完整移动脚本")
|
||
|
||
ps = sub.add_parser("show", help=COMMAND_HELP["show"])
|
||
ps.add_argument("media_id", type=int, help="收藏夹 media_id")
|
||
|
||
prp = sub.add_parser("replan", help=COMMAND_HELP["replan"])
|
||
prp.add_argument("media_id", type=int, help="收藏夹 media_id")
|
||
|
||
pr = sub.add_parser("run", help=COMMAND_HELP["run"])
|
||
pr.add_argument("media_id", type=int, help="收藏夹 media_id")
|
||
pr.add_argument("--verify-every", type=_positive_int, default=1, metavar="N",
|
||
help="每 N 步回读校验一次（默认 1；前 3 步始终逐步校验）")
|
||
|
||
pv = sub.add_parser("verify", help=COMMAND_HELP["verify"])
|
||
pv.add_argument("media_id", type=int, help="收藏夹 media_id")
|
||
|
||
ph = sub.add_parser("help", help=COMMAND_HELP["help"])
|
||
ph.add_argument("topic", nargs="?", default=None, metavar="<command>",
|
||
help="查看某个子命令的详细帮助，例如: help run")
|
||
return p, sub
|
||
|
||
|
||
def print_help(parser, sub, topic=None):
|
||
"""打印全部子命令与选项；指定 topic 时只打印该子命令的详细帮助。"""
|
||
if topic:
|
||
sp = sub.choices.get(topic)
|
||
if sp is None:
|
||
print(f"未知子命令: {topic}\n")
|
||
print(parser.format_help())
|
||
return 1
|
||
print(sp.format_help())
|
||
return 0
|
||
|
||
print(parser.format_help())
|
||
print("子命令与可用选项：")
|
||
for name, sp in sub.choices.items():
|
||
head = sp.format_usage().strip().removeprefix("usage: ")
|
||
print(f"\n {head}")
|
||
desc = COMMAND_HELP.get(name, "")
|
||
if desc:
|
||
print(f" {desc}")
|
||
for a in sp._actions: # noqa: SLF001 argparse 没有公开接口
|
||
if a.dest == "help":
|
||
continue
|
||
if a.option_strings:
|
||
flags = ", ".join(a.option_strings)
|
||
extra = "" if a.nargs == 0 else f" {a.metavar or a.dest.upper()}"
|
||
print(f" {flags + extra:<26} {a.help or ''}")
|
||
else:
|
||
print(f" {a.dest.upper():<26} {a.help or ''}")
|
||
print("\n示例：")
|
||
print(" python app.py auth --account write")
|
||
print(" python favsort.py folders")
|
||
print(" python favsort.py plan 3700023351 -v")
|
||
print(" python favsort.py run 3700023351 --verify-every 25")
|
||
print(" python favsort.py verify 3700023351")
|
||
print("\n鉴权：使用 python app.py auth --account write；"
|
||
"本工具的所有请求使用 bilibili.write，run 的 sort 使用完整 Cookie + csrf。")
|
||
return 0
|
||
|
||
|
||
def main():
|
||
p, sub = build_parser()
|
||
args = p.parse_args()
|
||
|
||
if args.cmd in (None, "help"):
|
||
return print_help(p, sub, getattr(args, "topic", None))
|
||
|
||
cfg = load_config(require_auth=True)
|
||
ensure_data_dir()
|
||
|
||
handler = {
|
||
"folders": cmd_folders,
|
||
"plan": cmd_plan,
|
||
"show": cmd_show,
|
||
"run": cmd_run,
|
||
"verify": cmd_verify,
|
||
"replan": cmd_replan,
|
||
}[args.cmd]
|
||
return handler(cfg, args) or 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|