# -*- coding: utf-8 -*- """ bili-fav / favsort —— B 站收藏夹批量重排序工具 排序规则 1) 同一个 upper 的视频必须连续(聚成一组),组内按发布时间倒序(并列按 avid 升序) 2) 组间按"该 upper 在本收藏夹中最早视频的发布时间"倒序(并列按该组最小 avid 升序) 3) 失效/异常资源置于末尾,保持原相对顺序 4) config.json 里的 overrides 可把指定视频强制归入另一个 upper(代发等场景) 指令 python favsort.py help [子命令] python favsort.py folders python favsort.py plan [--folder-mid N] [-v] python favsort.py show python favsort.py replan python favsort.py run [--verify-every N] python favsort.py verify 本工具只处理写账号所属收藏夹的排序。扫码授权统一使用 ``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 urllib.request 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") CONFIG_PATH = os.path.join(ROOT, "config.json") from reaction_board.bilibili.auth import ( # noqa: E402 make_headers, ) 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.json 为 config.json 并填入 cookie / csrf / mid。" ) with open(CONFIG_PATH, encoding="utf-8") as f: raw = json.load(f) bilibili = raw.get("bilibili") if not isinstance(bilibili, dict) or not isinstance(bilibili.get("write"), dict): raise SystemExit("config.json 必须配置 bilibili.write(扫码请运行 python app.py auth --account write)") write = bilibili["write"] cfg = { "mid": write.get("mid", 0), "csrf": write.get("csrf", ""), "cookie": write.get("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("config.json 中的 bilibili.write.cookie / csrf 不能为空(可先运行 python app.py auth --account write)") return cfg # ---------------------------------------------------------------- http def _get_json(cfg, url, auth): req = urllib.request.Request(url, headers=make_headers(cfg, auth=auth)) with urllib.request.urlopen(req, timeout=30) as r: return json.loads(r.read().decode("utf-8", "replace")) def get_json(cfg, url, auth=False): """只读请求:默认用去掉账号身份的 cookie。 万一匿名请求被拒(HTTP 错误,或 code != 0),自动带完整 cookie 重试一次, 保证功能不会因为剥离身份而回退。 """ if auth: return _get_json(cfg, url, auth=True) try: r = _get_json(cfg, url, auth=False) if r.get("code") == 0: return r except Exception: # noqa: BLE001 网络错误 / 412 风控都走回退 pass return _get_json(cfg, url, auth=True) def post_form(cfg, url, data): headers = make_headers(cfg, auth=True) headers["content-type"] = "application/x-www-form-urlencoded" req = urllib.request.Request( url, data=urllib.parse.urlencode(data).encode(), headers=headers, method="POST" ) with urllib.request.urlopen(req, timeout=30) as r: return json.loads(r.read().decode("utf-8", "replace")) def fetch_ids(cfg, media_id): r = get_json(cfg, API["ids"].format(base=BASE, mid=media_id)) 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 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) 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) 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, "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"] sc = cfg["sleep"] verify_every = args.verify_every 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"]) for i, m in enumerate(moves, 1): body = {"media_id": media_id, "sort": f"{m['pred']}:2:{m['moved']}:2", "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) 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) 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 build_parser(): p = argparse.ArgumentParser( prog="favsort", description="B 站收藏夹批量重排序:同 UP 聚在一起,组间按该 UP 最早投稿时间倒序。", formatter_class=argparse.RawDescriptionHelpFormatter, ) sub = p.add_subparsers(dest="cmd", title="子命令", metavar="") 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=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="", 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())