diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index a2932d6..b4a0601 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -71,11 +71,12 @@ uv run python --version # 查看现有收藏夹工具帮助 uv run bili-fav help -# 扫码登录并原子更新 config.json -uv run bili-fav auth +# 扫码登录并原子更新指定账号 +uv run python app.py auth --account read +uv run python app.py auth --account write -# 检查现有 B 站登录状态 -uv run bili-fav auth --check +# 检查两套 B 站登录状态 +uv run python app.py auth-check --account all # 执行当前无网络依赖的测试 uv run python tests/test_auth.py @@ -200,7 +201,8 @@ PRAGMA synchronous = NORMAL; │ ├── logging_json.py # 脱敏 JSON 日志 │ ├── bilibili/ │ │ ├── client.py # HTTP、WBI、重试与限速 -│ │ ├── auth.py # 从 favsort 抽取的扫码授权能力 +│ │ ├── auth.py # Cookie、设备指纹与请求头工具 +│ │ ├── auth_flow.py # app.py auth 的扫码与原子写回流程 │ │ └── favorite.py # 加入收藏夹与排序定位 │ ├── services/ │ │ ├── discovery.py @@ -224,7 +226,7 @@ PRAGMA synchronous = NORMAL; └── data/ # 本地运行期数据，默认不提交 ``` -`favsort.py` 继续保持可独立运行。P0 实施时优先抽取其通用的 Cookie、请求头、扫码授权和收藏夹 API 逻辑；不要直接复制出第二套不一致的身份与请求实现。 +`favsort.py` 继续保持可独立运行，但只操作 `bilibili.write`；扫码授权统一由 `app.py auth --account read|write` 提供。Cookie、设备指纹与请求头逻辑必须共用 `reaction_board.bilibili.auth`。 --- @@ -275,7 +277,7 @@ PRAGMA synchronous = NORMAL; ``` - `folders` 中的年度夹由人工创建，系统只读； -- 兼容现有 `favsort.py` 的扁平 `mid` / `csrf` / `cookie` 写法；存在 `bilibili` 嵌套对象时以嵌套值为准； +- 配置必须为 `bilibili.read` / `bilibili.write` 双账号结构；不兼容顶层扁平 `mid` / `csrf` / `cookie`； - P0 已实现的命令行：`python app.py init`、`python app.py sync [--full]`、`python app.py status`。 ### 5.2 安全规则- Windows 本地开发时应限制 `config.json` 的文件访问权限；部署到 Linux 时权限必须为 `600`； diff --git a/README.md b/README.md index 779dfea..a9c1fac 100644 --- a/README.md +++ b/README.md @@ -97,8 +97,13 @@ uv run python app.py snapshot # 忽略当日完成标记，重新采集 uv run python app.py snapshot --force -# 检查 Cookie 登录态 -uv run python app.py auth-check +# 分别为只读和写入账号扫码授权（默认 read） +uv run python app.py auth +uv run python app.py auth --account write + +# 检查指定账号或两套账号的登录态 +uv run python app.py auth-check --account read +uv run python app.py auth-check --account all ``` - `snapshot` 对 `active` / `unfaved` 视频逐条请求 `view`，每条请求间隔 1.0～1.6 秒； @@ -108,7 +113,7 @@ uv run python app.py auth-check - 粉丝数写入 `up_stat`，失败同样写 `ok=0`； - 当日全部视频与粉丝采集成功后写入完成标记；服务启动时若当天未完成会自动补跑； - 每次采集后生成 `data/backup/app-YYYYMMDD.db`，按 `storage.backup_keep_days` 清理旧日备份； -- Cookie 失效只告警，不阻塞匿名 `view` 采集；`auth-check` 可单独检查。 +- Cookie 失效只告警，不阻塞匿名 `view` 采集；P0/P1 只使用 `bilibili.read`。 首次真实全量采集结果：795 条视频全部成功、0 失败、0 移除，耗时约 19 分钟；粉丝 `follower=2767270`；日备份为 `data/backup/app-20261031.db`。 @@ -120,9 +125,20 @@ uv run python app.py auth-check Copy-Item config.example.json config.json ``` -`config.json` 已被 `.gitignore` 忽略，**不得提交**。其中： +`config.json` 已被 `.gitignore` 忽略，**不得提交**。配置必须使用双账号格式： -- `mid` / `cookie` / `csrf`：B 站账号 UID、完整 Cookie 与 `bili_jct`（兼容 `favsort.py` 的扁平写法；也支持 `bilibili` 嵌套对象）； +```json +{ + "bilibili": { + "read": { "mid": 0, "cookie": "...", "csrf": "..." }, + "write": { "mid": 0, "cookie": "...", "csrf": "..." } + } +} +``` + +- `bilibili.read`：年度收藏夹同步、视频数据快照、粉丝数与其他只读查询使用； +- `bilibili.write`：后续加入目标收藏夹、回读验证与现有 `favsort.py` 排序使用； +- 两个账号的 Cookie、CSRF 与设备指纹严格隔离；顶层 `mid` / `cookie` / `csrf` **不再支持**； - `overrides`：可选，为代发或联合投稿视频指定归属 UP； - `folders`：年度收藏夹配置（`year` + `media_id` + `title`），系统不自动建夹； - `up_blacklist_seed`：首次启动写入数据库的 UP 黑名单种子； @@ -140,9 +156,9 @@ Copy-Item config.example.json config.json # 查看全部命令；无需 config.json uv run bili-fav help -# 扫码登录，更新 config.json 中失效的身份 Cookie -uv run bili-fav auth -uv run bili-fav auth --check +# 为写账号扫码登录与检查状态（认证入口统一在 app.py） +uv run python app.py auth --account write +uv run python app.py auth-check --account write # 列出本人创建的收藏夹 uv run bili-fav folders @@ -172,38 +188,39 @@ uv run bili-fav verify ## 扫码授权 -当接口返回 `-101` 或 Cookie 即将过期时，运行： +当接口返回 `-101` 或 Cookie 即将过期时，使用统一认证入口： ```powershell -uv run bili-fav auth -uv run bili-fav auth --check +uv run python app.py auth --account read +uv run python app.py auth --account write +uv run python app.py auth-check --account all ``` 授权流程在本地生成二维码图片并等待手机客户端扫码确认。成功后： - 更新 `SESSDATA`、`bili_jct`、`DedeUserID`、`sid` 等身份字段； - 保留 `buvid3`、`buvid4`、`_uuid`、`buvid_fp` 等设备指纹； -- 将 `csrf` 同步为新的 `bili_jct`； +- 将选中账号的 `csrf` 同步为新的 `bili_jct`，并从 `nav` 回写该账号实际 mid； - 原子写入 `config.json`，并保留 `config.json.bak`； -- 通过 `x/web-interface/nav` 验证新登录态，失败时自动回滚。 +- 通过 `x/web-interface/nav` 验证新登录态，失败时自动回滚；不会修改另一个账号。 常用选项： ```powershell -uv run bili-fav auth --no-browser # 只生成二维码图片并输出路径 -uv run bili-fav auth --timeout 180 # 修改扫码等待时间（秒） +uv run python app.py auth --account write --no-browser +uv run python app.py auth --account read --timeout 180 ``` ## B 站访问与安全约束 | 接口 | 当前用途 | 身份 Cookie | |---|---|---| -| `fav/resource/ids` | 收藏夹资源 ID | 不需要账号身份；可匿名读取。 | -| `fav/resource/infos` | 收藏夹元数据 | 不需要账号身份，但需要设备指纹，完全无 Cookie 时可能返回 HTTP 412。 | -| `fav/folder/created/list-all` | 列出自建收藏夹 | 需要完整登录 Cookie。 | -| `fav/resource/sort` | 收藏夹排序 | 写操作；需要完整 Cookie 和 `csrf`。 | +| `fav/resource/ids` / `infos` | 年度收藏夹同步与排序计划 | `read` 用于 P0；`favsort.py` 计划使用 `write`。 | +| `view` / `relation/stat` / `nav` | P1 快照与 read 登录态 | `read`。 | +| `fav/folder/created/list-all` | 列出目标自建收藏夹 | `write`。 | +| `fav/resource/deal` / `sort` | 加入收藏夹、收藏夹排序 | `write`，需要完整 Cookie 和 CSRF。 | -现有工具对 `ids` / `infos` 默认剥离账号身份 Cookie，仅保留设备指纹；如果只读请求失败，才使用完整 Cookie 重试。所有写操作使用完整 Cookie 与 CSRF。 +只读客户端对 `ids` / `infos` 默认剥离 `read` 身份 Cookie，仅保留该账号设备指纹；如果失败才使用完整 `read` Cookie 重试。写操作一律使用完整 `write` Cookie 与 CSRF。 `x/v3/fav/resource/sort` 的实测语义为： diff --git a/app.py b/app.py index f1ba40f..a5f56d6 100644 --- a/app.py +++ b/app.py @@ -1,7 +1,7 @@ # -*- coding: utf-8 -*- """看板服务入口。 -P1 提供 ``init`` / ``sync`` / ``snapshot`` / ``auth-check`` / ``run`` / ``status`` +P1 提供 ``init`` / ``sync`` / ``snapshot`` / ``auth`` / ``auth-check`` / ``run`` / ``status`` 子命令；后续阶段会在同一入口启动本地 Web 服务。 """ diff --git a/config.example.json b/config.example.json index 545b949..d40a124 100644 --- a/config.example.json +++ b/config.example.json @@ -1,7 +1,16 @@ { - "mid": 0, - "csrf": "从 cookie 中的 bili_jct 取值", - "cookie": "粘贴浏览器请求头里的完整 Cookie（需含 SESSDATA、bili_jct、DedeUserID 等）", + "bilibili": { + "read": { + "mid": 0, + "csrf": "从 read 账号 cookie 中的 bili_jct 取值", + "cookie": "只读账号的完整 Cookie（需含 SESSDATA、bili_jct、DedeUserID 等）" + }, + "write": { + "mid": 0, + "csrf": "从 write 账号 cookie 中的 bili_jct 取值", + "cookie": "写入账号的完整 Cookie（用于加入收藏夹和排序）" + } + }, "overrides": {}, "sleep": { "min": 2.5, diff --git a/favsort.py b/favsort.py index d351af9..d219971 100644 --- a/favsort.py +++ b/favsort.py @@ -10,7 +10,6 @@ bili-fav / favsort —— B 站收藏夹批量重排序工具 指令 python favsort.py help [子命令] - python favsort.py auth [--check] [--no-browser] [--timeout SEC] python favsort.py folders python favsort.py plan [--folder-mid N] [-v] python favsort.py show @@ -18,55 +17,28 @@ bili-fav / favsort —— B 站收藏夹批量重排序工具 python favsort.py run [--verify-every N] python favsort.py verify -鉴权 - auth 会申请扫码登录，在本地生成二维码图片（vendor/qrcodegen.py，纯 Python、MIT， - 无需 pip 安装），用默认看图工具打开给你扫；拿到新的身份 cookie 后只替换 - SESSDATA / bili_jct / DedeUserID 等身份字段，buvid3 / buvid4 / _uuid / buvid_fp - 等设备指纹一律保留。只有设备 cookie 缺失时才用公开的 finger/spi 接口补， - 拿不到的（如 buvid_fp）保持原样。 - ids / infos / folders 为只读请求，默认剥离账号身份 cookie；sort 为写操作， - 始终使用完整 cookie + csrf。 +本工具只处理写账号所属收藏夹的排序。扫码授权统一使用 +``python app.py auth --account write``；所有 API 请求均使用 ``bilibili.write``。 """ import argparse import json import os import random -import shutil -import struct import sys -import tempfile import time import urllib.parse import urllib.request -import webbrowser -import zlib sys.stdout.reconfigure(encoding="utf-8") BASE = "https://api.bilibili.com" -PASSPORT = "https://passport.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") -CONFIG_BACKUP = CONFIG_PATH + ".bak" - -sys.path.insert(0, os.path.join(ROOT, "vendor")) -try: - import qrcodegen # noqa: E402 纯 Python，MIT，见 vendor/qrcodegen.py -except ImportError: # pragma: no cover - qrcodegen = None from reaction_board.bilibili.auth import ( # noqa: E402 - IDENTITY_KEYS, - cookie_value, - cookies_from_set_cookie, - gen_uuid, make_headers, - merge_cookie, - parse_cookie, - sessdata_expiry, - strip_identity, ) CHUNK = 20 # infos 接口单次 resources 数量（实测 20 稳定） @@ -75,18 +47,11 @@ API = { "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", - "nav": "{base}/x/web-interface/nav", - "spi": "{base}/x/frontend/finger/spi", - "qr_generate": "{passport}/x/passport-login/web/qrcode/generate?source=main-fe-header", - "qr_poll": "{passport}/x/passport-login/web/qrcode/poll?qrcode_key={key}", } # ---------------------------------------------------------------- config DEFAULT_CONFIG = { - "mid": 0, - "csrf": "", - "cookie": "", "overrides": {}, "sleep": { "min": 2.5, @@ -107,13 +72,24 @@ def load_config(require_auth=True): f"请复制 config.example.json 为 config.json 并填入 cookie / csrf / mid。" ) with open(CONFIG_PATH, encoding="utf-8") as f: - cfg = json.load(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 中的 cookie / csrf 不能为空（可先运行 python favsort.py auth）") + raise SystemExit("config.json 中的 bilibili.write.cookie / csrf 不能为空（可先运行 python app.py auth --account write）") return cfg @@ -152,95 +128,6 @@ def post_form(cfg, url, data): return json.loads(r.read().decode("utf-8", "replace")) -# ---------------------------------------------------------------- cookies - -def ensure_device_cookies(cfg): - """补齐缺失的设备 cookie；已有的一律不动（设备指纹换新反而更容易触发风控）。""" - pairs = parse_cookie(cfg["cookie"]) - lower = {k.lower() for k, _ in pairs} - added = [] - - if "buvid3" not in lower or "buvid4" not in lower: - try: - r = get_json(cfg, API["spi"].format(base=BASE), auth=True) - data = r.get("data") or {} - except Exception: # noqa: BLE001 补不到就算了，保留原有数据 - data = {} - if data.get("b_3") and "buvid3" not in lower: - pairs.append(("buvid3", data["b_3"])) - lower.add("buvid3") - added.append("buvid3") - if data.get("b_4") and "buvid4" not in lower: - pairs.append(("buvid4", data["b_4"])) - lower.add("buvid4") - added.append("buvid4") - - if "_uuid" not in lower: - pairs.append(("_uuid", gen_uuid())) - lower.add("_uuid") - added.append("_uuid") - - if "b_nut" not in lower and "buvid3" in added: - pairs.append(("b_nut", str(int(time.time())))) - added.append("b_nut") - - cfg["cookie"] = "; ".join(f"{k}={v}" for k, v in pairs) - return added - - -def save_config(cfg): - """原子写入 config.json，并留一份 .bak。""" - if os.path.exists(CONFIG_PATH): - shutil.copy2(CONFIG_PATH, CONFIG_BACKUP) - tmp = CONFIG_PATH + ".tmp" - with open(tmp, "w", encoding="utf-8") as f: - json.dump(cfg, f, ensure_ascii=False, indent=2) - f.write("\n") - os.replace(tmp, CONFIG_PATH) - - -# ---------------------------------------------------------------- qr - -def write_qr_png(text, path, scale=8, border=4): - """把文本编码成 QR 并写成 8bit 灰度 PNG（不依赖任何图像库）。""" - if qrcodegen is None: - raise RuntimeError("缺少 vendor/qrcodegen.py") - qr = qrcodegen.QrCode.encode_text(text, qrcodegen.QrCode.Ecc.MEDIUM) - size = qr.get_size() - dim = (size + border * 2) * scale - - rows = [] - for y in range(dim): - my = y // scale - border - row = bytearray([0]) # PNG 行过滤类型 0 - for x in range(dim): - mx = x // scale - border - dark = 0 <= mx < size and 0 <= my < size and qr.get_module(mx, my) - row.append(0 if dark else 255) - rows.append(bytes(row)) - raw = b"".join(rows) - - def chunk(tag, data): - return (struct.pack(">I", len(data)) + tag + data - + struct.pack(">I", zlib.crc32(tag + data) & 0xFFFFFFFF)) - - png = b"\x89PNG\r\n\x1a\n" - png += chunk(b"IHDR", struct.pack(">IIBBBBB", dim, dim, 8, 0, 0, 0, 0)) - png += chunk(b"IDAT", zlib.compress(raw, 9)) - png += chunk(b"IEND", b"") - with open(path, "wb") as f: - f.write(png) - return path - - -def open_path(path): - """用系统默认程序打开文件（Windows 用 startfile，其它平台用 webbrowser）。""" - if os.name == "nt": - os.startfile(path) # type: ignore[attr-defined] - return True - return webbrowser.open("file://" + path) - - def fetch_ids(cfg, media_id): r = get_json(cfg, API["ids"].format(base=BASE, mid=media_id)) if r.get("code") != 0: @@ -419,185 +306,6 @@ def load_plan(media_id): # ---------------------------------------------------------------- commands -def poll_qr(cfg, qrcode_key): - """轮询扫码状态，同时拿到本轮的 Set-Cookie。""" - url = API["qr_poll"].format(passport=PASSPORT, key=urllib.parse.quote(qrcode_key)) - req = urllib.request.Request(url, headers=make_headers(cfg, auth=False)) - with urllib.request.urlopen(req, timeout=30) as r: - payload = json.loads(r.read().decode("utf-8", "replace")) - return payload, (r.headers.get_all("Set-Cookie") or []) - - -def fetch_nav(cfg): - return get_json(cfg, API["nav"].format(base=BASE), auth=True) - - -def auth_check(cfg): - """只读：检查当前 cookie 是否有效、SESSDATA 何时到期、设备 cookie 是否齐全。""" - device = [k for k in ("buvid3", "buvid4", "_uuid", "buvid_fp") - if cookie_value(cfg["cookie"], k)] - print(f"[auth] 设备 cookie: {', '.join(device) if device else '（缺失）'}") - if not device: - print("[auth] 提示：缺少 buvid3/buvid4 等设备指纹，infos 接口可能返回 412") - - exp = sessdata_expiry(cfg["cookie"]) - if exp: - days = (exp - time.time()) / 86400 - print(f"[auth] SESSDATA 到期: {time.strftime('%Y-%m-%d', time.localtime(exp))}" - f"（剩余 {days:.0f} 天）") - - if not cfg["cookie"]: - print("[auth] 未登录：config.json 里没有 cookie") - return 1 - - try: - r = fetch_nav(cfg) - except Exception as e: # noqa: BLE001 - print(f"[auth] 检查失败: {e}") - return 1 - data = r.get("data") or {} - if r.get("code") != 0 or not data.get("isLogin"): - print(f"[auth] 登录已失效 ❌ code={r.get('code')} {r.get('message')}") - print("[auth] 运行 `python favsort.py auth` 扫码重新登录") - return 1 - - print(f"[auth] 已登录 ✅ mid={data.get('mid')} uname={data.get('uname')}") - if exp and exp < time.time(): - print("[auth] 警告：SESSDATA 时间戳已过期但接口仍返回登录态，建议尽快更新") - return 0 - - -def finish_auth(cfg, pairs): - """合并新身份 cookie -> 补设备 cookie -> 写配置 -> 校验，失败回滚。""" - old_cookie = cfg["cookie"] - merged = merge_cookie(old_cookie, pairs) - if not cookie_value(merged, "SESSDATA") or not cookie_value(merged, "bili_jct"): - print("[auth] 合并后缺少 SESSDATA / bili_jct，中止") - return 2 - - cfg["cookie"] = merged - cfg["csrf"] = cookie_value(merged, "bili_jct") - updated = sorted(k for k in pairs if k.lower() in IDENTITY_KEYS) - print(f"[auth] 登录成功，更新身份字段: {', '.join(updated)}") - - added = ensure_device_cookies(cfg) - if added: - print(f"[auth] 补齐缺失的设备字段: {', '.join(added)}") - kept = [k for k in ("buvid3", "buvid4", "_uuid", "buvid_fp") - if cookie_value(cfg["cookie"], k)] - print(f"[auth] 保留设备指纹: {', '.join(kept) if kept else '（无）'}") - if not cookie_value(cfg["cookie"], "buvid3") or not cookie_value(cfg["cookie"], "buvid4"): - print("[auth] 警告：buvid3/buvid4 仍缺失，infos 接口可能返回 412") - - new_mid = cookie_value(cfg["cookie"], "DedeUserID") - if new_mid and new_mid.isdigit() and int(cfg.get("mid") or 0) != int(new_mid): - print(f"[auth] 注意：登录账号 mid={new_mid}，与 config 里的 mid={cfg.get('mid')} 不同，" - f"已同步为 {new_mid}") - cfg["mid"] = int(new_mid) - - exp = sessdata_expiry(cfg["cookie"]) - if exp: - print(f"[auth] 新 SESSDATA 到期: " - f"{time.strftime('%Y-%m-%d', time.localtime(exp))}") - - save_config(cfg) - print(f"[auth] 已写入 {CONFIG_PATH}（备份 {os.path.basename(CONFIG_BACKUP)}）") - - try: - r = fetch_nav(cfg) - ok = r.get("code") == 0 and (r.get("data") or {}).get("isLogin") - uname = (r.get("data") or {}).get("uname") - except Exception as e: # noqa: BLE001 - ok, uname = False, str(e) - - if not ok: - if os.path.exists(CONFIG_BACKUP): - shutil.copy2(CONFIG_BACKUP, CONFIG_PATH) - print("[auth] 登录校验失败，已回滚 config.json") - else: - print("[auth] 登录校验失败（没有备份可回滚）") - return 4 - - print(f"[auth] 校验通过 ✅ uname={uname}") - return 0 - - -def cmd_auth(cfg, args): - if args.check: - return auth_check(cfg) - - print("[auth] 申请扫码登录 ...") - gen = get_json(cfg, API["qr_generate"].format(passport=PASSPORT), auth=False) - data = gen.get("data") or {} - qrcode_key, url = data.get("qrcode_key"), data.get("url") - if gen.get("code") != 0 or not qrcode_key or not url: - print(f"[auth] 申请二维码失败: code={gen.get('code')} {gen.get('message')}") - return 2 - - qr_path = os.path.join(tempfile.gettempdir(), f"bili-fav-login-{qrcode_key[:8]}.png") - try: - write_qr_png(url, qr_path) - print(f"[auth] 二维码已生成: {qr_path}") - except Exception as e: # noqa: BLE001 - print(f"[auth] 本地生成二维码失败: {e}") - print(f"[auth] 请手动处理这个链接: {url}") - else: - if not args.no_browser: - try: - if open_path(qr_path): - print("[auth] 已用默认看图工具打开二维码") - except Exception as e: # noqa: BLE001 - print(f"[auth] 打开图片失败: {e}，请手动打开上面的路径") - - print(f"[auth] 请在 {args.timeout} 秒内用 B 站客户端扫码并确认（二维码约 180 秒有效）") - sys.stdout.flush() - - deadline = time.time() + args.timeout - last = None - while time.time() < deadline: - try: - payload, set_cookies = poll_qr(cfg, qrcode_key) - except Exception as e: # noqa: BLE001 - print(f"[auth] 轮询异常: {e}") - time.sleep(3) - continue - - data = payload.get("data") or {} - state = data.get("code") - if state is None: - print(f"[auth] 轮询失败: code={payload.get('code')} {payload.get('message')}") - return 2 - - if state == 0: - pairs = cookies_from_set_cookie(set_cookies) - # 老版本会把 cookie 放在跨域 URL 的 query 里，作为兜底保留 - if data.get("url"): - for part in urllib.parse.urlparse(data["url"]).query.split("&"): - k, _, v = part.partition("=") - if k and v: - pairs.setdefault(k, v) - got = "; ".join(f"{k}={v}" for k, v in pairs.items()) - if not cookie_value(got, "SESSDATA") or not cookie_value(got, "bili_jct"): - print("[auth] 扫码成功，但响应里没有 SESSDATA / bili_jct，无法继续") - return 2 - return finish_auth(cfg, pairs) - - if state == 86038: - print("[auth] 二维码已过期，请重新运行 auth") - return 3 - - if state != last: - msg = data.get("message") or { - 86101: "未扫码", - 86090: "已扫码，等待确认", - }.get(state, f"状态 {state}") - print(f"[auth] {msg} ...") - last = state - time.sleep(2) - - print("[auth] 等待超时，二维码已失效，请重新运行 auth") - return 3 - def cmd_folders(cfg, args): # 注意：list-all 是"列出我自己的收藏夹"，匿名调用只会返回公开收藏夹（实测 11 -> 3）， @@ -823,8 +531,7 @@ def cmd_verify(cfg, args): # ---------------------------------------------------------------- main COMMAND_HELP = { - "auth": "扫码登录并更新 config.json 里失效的身份 cookie（保留设备指纹）", - "folders": "列出自己（config.mid）的自建收藏夹及其 media_id", + "folders": "列出写账号（bilibili.write.mid）的自建收藏夹及其 media_id", "plan": "拉取收藏夹数据并生成排序计划（只读，不改动收藏夹）", "show": "查看已生成的计划与完整移动脚本（离线）", "replan": "离线重算移动脚本（不联网，不改目标顺序）", @@ -844,18 +551,10 @@ def build_parser(): sub.add_parser("folders", help=COMMAND_HELP["folders"]) - pa = sub.add_parser("auth", help=COMMAND_HELP["auth"]) - pa.add_argument("--check", action="store_true", - help="只检查当前登录状态与到期时间，不修改配置") - pa.add_argument("--no-browser", action="store_true", - help="不自动打开二维码图片，只打印图片路径") - pa.add_argument("--timeout", type=int, default=180, metavar="SEC", - help="等待扫码的超时时间（默认 180 秒，与二维码有效期一致）") - 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（默认取 config.json 的 mid）") + help="收藏夹归属用户 mid（默认取 bilibili.write.mid）") pp.add_argument("--verbose", "-v", action="store_true", help="打印完整移动脚本") ps = sub.add_parser("show", help=COMMAND_HELP["show"]) @@ -907,14 +606,13 @@ def print_help(parser, sub, topic=None): else: print(f" {a.dest.upper():<26} {a.help or ''}") print("\n示例：") - print(" python favsort.py auth") - print(" python favsort.py auth --check") + 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鉴权：auth 扫码后只更新身份 cookie，保留设备指纹；" - "只读请求自动去掉账号身份；run 的 sort 仍使用完整 cookie + csrf。") + print("\n鉴权：使用 python app.py auth --account write；" + "本工具的所有请求使用 bilibili.write，run 的 sort 使用完整 Cookie + csrf。") return 0 @@ -925,11 +623,10 @@ def main(): if args.cmd in (None, "help"): return print_help(p, sub, getattr(args, "topic", None)) - cfg = load_config(require_auth=args.cmd != "auth") + cfg = load_config(require_auth=True) ensure_data_dir() handler = { - "auth": cmd_auth, "folders": cmd_folders, "plan": cmd_plan, "show": cmd_show, diff --git a/project.md b/project.md index 7d88d4e..593cf26 100644 --- a/project.md +++ b/project.md @@ -90,9 +90,16 @@ ```jsonc { "bilibili": { - "cookie": "SESSDATA=...; bili_jct=...; buvid3=...; ...", - "csrf": "", - "mid": 28802351 + "read": { + "cookie": "SESSDATA=...; bili_jct=...; buvid3=...; ...", + "csrf": "", + "mid": 28802351 + }, + "write": { + "cookie": "SESSDATA=...; bili_jct=...; buvid3=...; ...", + "csrf": "", + "mid": 0 + } }, "telegram": { "bot_token": "123456:ABC...", @@ -126,7 +133,8 @@ - **黑名单本身放数据库**（`up_blacklist` 表），因为要从通知里动态拉黑；配置文件只做初始种子； - 日志里**永不打印 token / cookie**，启动时只打印"已配置/未配置"； - 容器里把配置目录挂成**可写卷**（`-v ./config:/config`，权限 600），因为 `auth` 扫码续期后要回写 cookie； -- `python app.py auth` 复用 `bili-fav` 的扫码流程，只替换身份字段、保留设备指纹； +- `python app.py auth --account read|write` 对指定账号扫码续期，只替换该账号身份字段、保留该账号设备指纹； +- `read` 只用于年度收藏夹同步、快照和查询；`write` 只用于目标收藏夹、`deal` 和 `sort` 写操作；不支持顶层扁平凭据； - 可选的 `ENV` 覆盖（`TG_BOT_TOKEN` 等）留给 CI/临时调试，不作为主路径。 ### 1.6 初始黑名单（已核对昵称，实测 2026-10-31） @@ -725,7 +733,7 @@ storage：SQLite WAL，单写线程 | 风控 `-352/-412` | 采集中断 | 低（实测未见） | 限速 1.2s + 抖动 + 指数退避 + 告警 | | 跨年切换遗漏 | 新视频进错夹 | 低 | 按 pubdate 选夹 + 1 月 1 日提醒 | | SQLite 并发 | 写冲突 | 低 | WAL + 单写线程 | -| cookie 过期 | 收藏夹相关接口失败 | 中 | 复用 `bili-fav auth` 扫码续期；`view`/搜索匿名可用作降级 | +| cookie 过期 | 收藏夹相关接口失败 | 中 | 使用 `python app.py auth --account read|write` 扫码续期；`view`/搜索匿名可用作降级 | --- diff --git a/reaction_board/bilibili/auth_flow.py b/reaction_board/bilibili/auth_flow.py new file mode 100644 index 0000000..abee8c2 --- /dev/null +++ b/reaction_board/bilibili/auth_flow.py @@ -0,0 +1,249 @@ +# -*- coding: utf-8 -*- +"""扫码授权流程；仅由 ``python app.py auth`` 调用。 + +认证结果只写入指定的 ``bilibili.read`` 或 ``bilibili.write`` 段，两个账号的 +身份及设备指纹 Cookie 绝不互相覆盖。 +""" + +from __future__ import annotations + +import json +import os +import shutil +import struct +import sys +import tempfile +import time +import urllib.parse +import urllib.request +import webbrowser +import zlib +from pathlib import Path +from typing import Any, Callable, Mapping + +from .auth import ( + IDENTITY_KEYS, + cookie_value, + cookies_from_set_cookie, + gen_uuid, + make_headers, + merge_cookie, + parse_cookie, + sessdata_expiry, +) + +ROOT = Path(__file__).resolve().parents[2] +VENDOR = ROOT / "vendor" +if str(VENDOR) not in sys.path: + sys.path.insert(0, str(VENDOR)) +import qrcodegen # noqa: E402 + +BASE = "https://api.bilibili.com" +PASSPORT = "https://passport.bilibili.com" + + +def write_qr_png(text: str, path: str | Path, scale: int = 8, border: int = 4) -> Path: + """将登录链接编码为不依赖图像库的 8-bit 灰度 QR PNG。""" + qr = qrcodegen.QrCode.encode_text(text, qrcodegen.QrCode.Ecc.MEDIUM) + size = qr.get_size() + dim = (size + border * 2) * scale + rows: list[bytes] = [] + for y in range(dim): + my = y // scale - border + row = bytearray([0]) + for x in range(dim): + mx = x // scale - border + dark = 0 <= mx < size and 0 <= my < size and qr.get_module(mx, my) + row.append(0 if dark else 255) + rows.append(bytes(row)) + + def chunk(tag: bytes, data: bytes) -> bytes: + return struct.pack(">I", len(data)) + tag + data + struct.pack( + ">I", zlib.crc32(tag + data) & 0xFFFFFFFF + ) + + output = Path(path) + output.write_bytes( + b"\x89PNG\r\n\x1a\n" + + chunk(b"IHDR", struct.pack(">IIBBBBB", dim, dim, 8, 0, 0, 0, 0)) + + chunk(b"IDAT", zlib.compress(b"".join(rows), 9)) + + chunk(b"IEND", b"") + ) + return output + + +def open_path(path: Path) -> bool: + if os.name == "nt": + os.startfile(path) # type: ignore[attr-defined] + return True + return webbrowser.open(path.as_uri()) + + +def ensure_device_cookies( + credentials: dict[str, Any], get_json: Callable[[str, bool], Mapping[str, Any]] +) -> list[str]: + """仅为当前账号补齐缺失设备字段，已有指纹不改动。""" + pairs = parse_cookie(str(credentials.get("cookie") or "")) + lower = {key.lower() for key, _ in pairs} + added: list[str] = [] + if "buvid3" not in lower or "buvid4" not in lower: + try: + data = get_json(f"{BASE}/x/frontend/finger/spi", True).get("data") or {} + except Exception: # noqa: BLE001 + data = {} + if data.get("b_3") and "buvid3" not in lower: + pairs.append(("buvid3", str(data["b_3"]))) + lower.add("buvid3") + added.append("buvid3") + if data.get("b_4") and "buvid4" not in lower: + pairs.append(("buvid4", str(data["b_4"]))) + lower.add("buvid4") + added.append("buvid4") + if "_uuid" not in lower: + pairs.append(("_uuid", gen_uuid())) + lower.add("_uuid") + added.append("_uuid") + if "b_nut" not in lower and "buvid3" in added: + pairs.append(("b_nut", str(int(time.time())))) + credentials["cookie"] = "; ".join(f"{key}={value}" for key, value in pairs) + return added + + +class AuthService: + def __init__(self, config_path: str | Path, *, account: str, logger=None) -> None: + if account not in {"read", "write"}: + raise ValueError(f"未知账号角色: {account}") + self.path = Path(config_path) + self.account = account + self.logger = logger + + @property + def backup_path(self) -> Path: + return self.path.with_name(self.path.name + ".bak") + + def _load_raw(self) -> dict[str, Any]: + with self.path.open(encoding="utf-8") as fh: + raw = json.load(fh) + if not isinstance(raw, dict): + raise ValueError("配置文件顶层必须是对象") + section = raw.get("bilibili") + if not isinstance(section, dict) or not isinstance(section.get(self.account), dict): + raise ValueError(f"缺少 bilibili.{self.account} 账号配置") + return raw + + @staticmethod + def _request(credentials: Mapping[str, Any], url: str, *, auth: bool) -> tuple[dict[str, Any], list[str]]: + req = urllib.request.Request(url, headers=make_headers(dict(credentials), auth=auth)) + with urllib.request.urlopen(req, timeout=30) as response: + return ( + json.loads(response.read().decode("utf-8", "replace")), + response.headers.get_all("Set-Cookie") or [], + ) + + def _get_json(self, credentials: Mapping[str, Any], url: str, auth: bool) -> Mapping[str, Any]: + return self._request(credentials, url, auth=auth)[0] + + def _write_raw(self, raw: Mapping[str, Any]) -> None: + tmp = self.path.with_name(self.path.name + ".tmp") + with tmp.open("w", encoding="utf-8") as fh: + json.dump(raw, fh, ensure_ascii=False, indent=2) + fh.write("\n") + os.replace(tmp, self.path) + + def finish(self, raw: dict[str, Any], pairs: Mapping[str, str]) -> int: + """写入单个账号、验证 nav；验证失败还原整个原配置文件。""" + credentials = raw["bilibili"][self.account] + old_cookie = str(credentials.get("cookie") or "") + merged = merge_cookie(old_cookie, dict(pairs)) + if not cookie_value(merged, "SESSDATA") or not cookie_value(merged, "bili_jct"): + print("[auth] 合并后缺少 SESSDATA / bili_jct，中止") + return 2 + credentials["cookie"] = merged + credentials["csrf"] = cookie_value(merged, "bili_jct") + added = ensure_device_cookies( + credentials, + lambda url, auth: self._get_json(credentials, url, auth), + ) + new_mid = cookie_value(str(credentials["cookie"]), "DedeUserID") + if new_mid.isdigit(): + credentials["mid"] = int(new_mid) + + shutil.copy2(self.path, self.backup_path) + self._write_raw(raw) + try: + nav = self._get_json(credentials, f"{BASE}/x/web-interface/nav", True) + data = nav.get("data") or {} + ok = nav.get("code") == 0 and bool(data.get("isLogin")) + except Exception as exc: # noqa: BLE001 + data, ok = {"uname": str(exc)}, False + if not ok: + shutil.copy2(self.backup_path, self.path) + print(f"[auth] {self.account} 登录校验失败，已回滚 config.json") + return 4 + + if isinstance(data.get("mid"), int): + credentials["mid"] = data["mid"] + self._write_raw(raw) + updated = sorted(key for key in pairs if key.lower() in IDENTITY_KEYS) + kept = [key for key in ("buvid3", "buvid4", "_uuid", "buvid_fp") + if cookie_value(str(credentials["cookie"]), key)] + print(f"[auth] {self.account} 已更新身份字段: {', '.join(updated)}") + if added: + print(f"[auth] {self.account} 补齐设备字段: {', '.join(added)}") + print(f"[auth] {self.account} 保留设备指纹: {', '.join(kept) if kept else '（无）'}") + expiry = sessdata_expiry(str(credentials["cookie"])) + if expiry: + print(f"[auth] 新 SESSDATA 到期: {time.strftime('%Y-%m-%d', time.localtime(expiry))}") + print(f"[auth] {self.account} 校验通过 ✅ mid={credentials['mid']} uname={data.get('uname')}") + return 0 + + def run(self, *, timeout: int, no_browser: bool) -> int: + raw = self._load_raw() + credentials = raw["bilibili"][self.account] + print(f"[auth] 为 {self.account} 账号申请扫码登录 ...") + generated, _ = self._request( + credentials, f"{PASSPORT}/x/passport-login/web/qrcode/generate?source=main-fe-header", auth=False + ) + data = generated.get("data") or {} + key, login_url = data.get("qrcode_key"), data.get("url") + if generated.get("code") != 0 or not key or not login_url: + print(f"[auth] 申请二维码失败: code={generated.get('code')} {generated.get('message')}") + return 2 + qr_path = Path(tempfile.gettempdir()) / f"bili-fav-{self.account}-{key[:8]}.png" + write_qr_png(str(login_url), qr_path) + print(f"[auth] 二维码已生成: {qr_path}") + if not no_browser: + try: + open_path(qr_path) + except Exception as exc: # noqa: BLE001 + print(f"[auth] 打开图片失败: {exc}，请手动打开二维码") + deadline, last = time.time() + timeout, None + while time.time() < deadline: + try: + payload, set_cookies = self._request( + credentials, + f"{PASSPORT}/x/passport-login/web/qrcode/poll?qrcode_key={urllib.parse.quote(str(key))}", + auth=False, + ) + except Exception as exc: # noqa: BLE001 + print(f"[auth] 轮询异常: {exc}") + time.sleep(3) + continue + poll = payload.get("data") or {} + state = poll.get("code") + if state == 0: + pairs = cookies_from_set_cookie(set_cookies) + for part in urllib.parse.urlparse(str(poll.get("url") or "")).query.split("&"): + key_name, _, value = part.partition("=") + if key_name and value: + pairs.setdefault(key_name, value) + return self.finish(raw, pairs) + if state == 86038: + print("[auth] 二维码已过期，请重新运行 auth") + return 3 + if state != last: + print(f"[auth] {(poll.get('message') or {86101: '未扫码', 86090: '已扫码，等待确认'}.get(state, f'状态 {state}'))} ...") + last = state + time.sleep(2) + print("[auth] 等待超时，二维码已失效，请重新运行 auth") + return 3 diff --git a/reaction_board/cli.py b/reaction_board/cli.py index 30fce28..c775697 100644 --- a/reaction_board/cli.py +++ b/reaction_board/cli.py @@ -19,6 +19,7 @@ import sys from pathlib import Path from typing import Sequence +from .bilibili.auth_flow import AuthService from .bilibili.client import BilibiliClient, Pacer from .config import ConfigError, ConfigStore from .database import Database @@ -58,7 +59,15 @@ def _build_parser() -> argparse.ArgumentParser: snapshot.add_argument( "--force", action="store_true", help="忽略当日完成标记，重新采集" ) - sub.add_parser("auth-check", parents=[common], help="检查 Cookie 登录态") + auth = sub.add_parser("auth", parents=[common], help="扫码授权并更新指定账号 Cookie") + auth.add_argument("--account", choices=("read", "write"), default="read", + help="要授权的账号角色（默认 read）") + auth.add_argument("--no-browser", action="store_true", help="不自动打开二维码图片") + auth.add_argument("--timeout", type=int, default=180, metavar="SEC", + help="等待扫码的超时时间（默认 180 秒）") + auth_check = sub.add_parser("auth-check", parents=[common], help="检查账号登录态") + auth_check.add_argument("--account", choices=("read", "write", "all"), default="read", + help="要检查的账号角色（默认 read）") run = sub.add_parser("run", parents=[common], help="按配置周期常驻运行同步与快照") run.add_argument( "--once", action="store_true", help="只执行一次到期任务后退出" @@ -82,11 +91,11 @@ def _print(payload) -> None: print(json.dumps(payload, ensure_ascii=False, indent=2), flush=True) -def _client(cfg, logger: JsonLogger, *, pacer=None, retries: int = 3) -> BilibiliClient: +def _client(credentials, logger: JsonLogger, *, pacer=None, retries: int = 3) -> BilibiliClient: return BilibiliClient( - cookie=cfg.bilibili.cookie, - csrf=cfg.bilibili.csrf, - mid=cfg.bilibili.mid, + cookie=credentials.cookie, + csrf=credentials.csrf, + mid=credentials.mid, pacer=pacer, retries=retries, logger=logger, @@ -120,7 +129,7 @@ def cmd_sync(args, logger: JsonLogger) -> int: db = _open_db(cfg, logger) try: repo = Repository(db) - client = _client(cfg, logger) + client = _client(cfg.bilibili.read, logger) service = FolderSyncService(store, repo, client, logger=logger) summary = service.run(full_refresh=args.full) payload = summary.as_dict() @@ -139,7 +148,7 @@ def cmd_snapshot(args, logger: JsonLogger) -> int: try: repo = Repository(db) client = _client( - cfg, logger, pacer=Pacer(1.0, 1.6), retries=1 + cfg.bilibili.read, logger, pacer=Pacer(1.0, 1.6), retries=1 ) service = SnapshotService(store, repo, client, logger=logger) summary = service.run_if_needed(force=args.force) @@ -153,23 +162,40 @@ def cmd_snapshot(args, logger: JsonLogger) -> int: db.close() -def cmd_auth_check(args, logger: JsonLogger) -> int: - store = ConfigStore(args.config, require_auth=False) - cfg = store.get() - if not cfg.bilibili.cookie: - _print({"ok": False, "is_login": False, "message": "未配置 cookie"}) - return 1 - client = _client(cfg, logger, retries=1) - state = client.check_login() - _print({ +def cmd_auth(args, logger: JsonLogger) -> int: + if args.timeout < 1: + raise ConfigError("auth --timeout 必须至少为 1 秒") + return AuthService(Path(args.config), account=args.account, logger=logger).run( + timeout=args.timeout, no_browser=args.no_browser + ) + + +def _auth_check(cfg, account: str, logger: JsonLogger) -> dict[str, object]: + credentials = getattr(cfg.bilibili, account) + if not credentials.cookie: + return {"account": account, "ok": False, "is_login": False, "message": "未配置 cookie"} + state = _client(credentials, logger, retries=1).check_login() + return { + "account": account, "ok": state.is_login, "is_login": state.is_login, "mid": state.mid, "uname": state.uname, "code": state.code, "message": state.message, - }) - return 0 if state.is_login else 1 + } + + +def cmd_auth_check(args, logger: JsonLogger) -> int: + store = ConfigStore(args.config, require_auth=False) + cfg = store.get() + if args.account != "all": + result = _auth_check(cfg, args.account, logger) + _print(result) + return 0 if result["ok"] else 1 + results = [_auth_check(cfg, account, logger) for account in ("read", "write")] + _print({"ok": all(result["ok"] for result in results), "accounts": results}) + return 0 if all(result["ok"] for result in results) else 1 def cmd_run(args, logger: JsonLogger) -> int: @@ -179,9 +205,9 @@ def cmd_run(args, logger: JsonLogger) -> int: last: dict[str, object] = {} try: repo = Repository(db) - folder_client = _client(cfg, logger) + folder_client = _client(cfg.bilibili.read, logger) folder_service = FolderSyncService(store, repo, folder_client, logger=logger) - snapshot_client = _client(cfg, logger, pacer=Pacer(1.0, 1.6), retries=1) + snapshot_client = _client(cfg.bilibili.read, logger, pacer=Pacer(1.0, 1.6), retries=1) snapshot_service = SnapshotService(store, repo, snapshot_client, logger=logger) def run_sync(): @@ -284,6 +310,8 @@ def main(argv: Sequence[str] | None = None) -> int: return cmd_sync(args, logger) if args.command == "snapshot": return cmd_snapshot(args, logger) + if args.command == "auth": + return cmd_auth(args, logger) if args.command == "auth-check": return cmd_auth_check(args, logger) if args.command == "run": diff --git a/reaction_board/config.py b/reaction_board/config.py index 3414a4c..aaffbc0 100644 --- a/reaction_board/config.py +++ b/reaction_board/config.py @@ -1,13 +1,8 @@ # -*- coding: utf-8 -*- """运行期配置加载、校验与热加载。 -支持两种写法： - -* 新格式：``{"bilibili": {"mid": ..., "cookie": ...}, ...}``； -* 旧 ``favsort.py`` 格式：顶层 ``mid`` / ``cookie`` / ``csrf``。 - -旧的扁平字段仍然被读取，因此 ``bili-fav auth`` 写回的 config.json 无需迁移即可 -被看板使用。敏感值只保存在配置对象中，日志输出必须先经过 ``logging_json`` 脱敏。 +B 站凭据必须采用 ``bilibili.read`` / ``bilibili.write`` 双账号格式。敏感值只 +保存在配置对象中，日志输出必须先经过 ``logging_json`` 脱敏。 """ from __future__ import annotations @@ -43,7 +38,10 @@ DEFAULT_HEAT_WEIGHTS: dict[str, float] = { } DEFAULT_CONFIG: dict[str, Any] = { - "bilibili": {"mid": 0, "csrf": "", "cookie": ""}, + "bilibili": { + "read": {"mid": 0, "csrf": "", "cookie": ""}, + "write": {"mid": 0, "csrf": "", "cookie": ""}, + }, "telegram": {"enabled": False, "bot_token": "", "chat_id": 0}, "web": { "host": "127.0.0.1", @@ -73,9 +71,12 @@ DEFAULT_CONFIG: dict[str, Any] = { # 环境变量只作为 CI / 临时调试的覆盖入口，不作为主路径。 ENV_OVERRIDES: dict[str, tuple[tuple[str, ...], type]] = { - "BILI_COOKIE": (("bilibili", "cookie"), str), - "BILI_CSRF": (("bilibili", "csrf"), str), - "BILI_MID": (("bilibili", "mid"), int), + "BILI_READ_COOKIE": (("bilibili", "read", "cookie"), str), + "BILI_READ_CSRF": (("bilibili", "read", "csrf"), str), + "BILI_READ_MID": (("bilibili", "read", "mid"), int), + "BILI_WRITE_COOKIE": (("bilibili", "write", "cookie"), str), + "BILI_WRITE_CSRF": (("bilibili", "write", "csrf"), str), + "BILI_WRITE_MID": (("bilibili", "write", "mid"), int), "TG_BOT_TOKEN": (("telegram", "bot_token"), str), "TG_CHAT_ID": (("telegram", "chat_id"), int), "DASHBOARD_AUTH_TOKEN": (("web", "auth_token"), str), @@ -91,12 +92,18 @@ class ConfigError(ValueError): @dataclass(frozen=True) -class BilibiliConfig: +class BilibiliCredentials: mid: int csrf: str cookie: str +@dataclass(frozen=True) +class BilibiliConfig: + read: BilibiliCredentials + write: BilibiliCredentials + + @dataclass(frozen=True) class TelegramConfig: enabled: bool @@ -234,18 +241,6 @@ def _as_str_list(value: Any, name: str) -> list[str]: return out -def _legacy_to_current(raw: Mapping[str, Any]) -> dict[str, Any]: - """把 favsort 的扁平字段映射到新格式，不修改调用方对象。""" - data = copy.deepcopy(dict(raw)) - if "bilibili" not in data: - data["bilibili"] = { - "mid": raw.get("mid", 0), - "csrf": raw.get("csrf", ""), - "cookie": raw.get("cookie", ""), - } - return data - - def _validate_folders(value: Any) -> tuple[FolderConfig, ...]: if not isinstance(value, list): raise ConfigError("folders 必须是数组") @@ -273,11 +268,17 @@ def _build_config(data: Mapping[str, Any], path: Path) -> AppConfig: b = data["bilibili"] if not isinstance(b, Mapping): raise ConfigError("bilibili 必须是对象") - bilibili = BilibiliConfig( - mid=_as_int(b.get("mid", 0), "bilibili.mid", minimum=0), - csrf=_as_str(b.get("csrf", "") or "", "bilibili.csrf"), - cookie=_as_str(b.get("cookie", "") or "", "bilibili.cookie"), - ) + def credentials(account: str) -> BilibiliCredentials: + value = b.get(account) + if not isinstance(value, Mapping): + raise ConfigError(f"bilibili.{account} 必须是对象") + return BilibiliCredentials( + mid=_as_int(value.get("mid", 0), f"bilibili.{account}.mid", minimum=0), + csrf=_as_str(value.get("csrf", "") or "", f"bilibili.{account}.csrf"), + cookie=_as_str(value.get("cookie", "") or "", f"bilibili.{account}.cookie"), + ) + + bilibili = BilibiliConfig(read=credentials("read"), write=credentials("write")) tg = data["telegram"] if not isinstance(tg, Mapping): @@ -431,8 +432,14 @@ def load_config( raise ConfigError(f"无法读取配置文件: {config_path}: {exc}") from exc if not isinstance(raw, Mapping): raise ConfigError("配置文件顶层必须是 JSON 对象") + raw_bilibili = raw.get("bilibili") + if not isinstance(raw_bilibili, Mapping): + raise ConfigError("缺少 bilibili.read / bilibili.write 双账号配置") + for account in ("read", "write"): + if account not in raw_bilibili: + raise ConfigError(f"缺少 bilibili.{account} 账号配置") - data = _deep_merge(DEFAULT_CONFIG, _legacy_to_current(raw)) + data = _deep_merge(DEFAULT_CONFIG, raw) env_map = os.environ if env is None else env for env_name, (target, caster) in ENV_OVERRIDES.items(): @@ -447,9 +454,9 @@ def load_config( _set_path(data, target, value) cfg = _build_config(data, config_path) - if require_auth and (not cfg.bilibili.cookie or not cfg.bilibili.csrf): + if require_auth and (not cfg.bilibili.read.cookie or not cfg.bilibili.read.csrf): raise ConfigError( - "config.json 中的 cookie / csrf 不能为空（可先运行 bili-fav auth）" + "config.json 中的 bilibili.read.cookie / csrf 不能为空（可先运行 python app.py auth）" ) return cfg diff --git a/reaction_board/services/folder_sync.py b/reaction_board/services/folder_sync.py index 4a0442b..d70f219 100644 --- a/reaction_board/services/folder_sync.py +++ b/reaction_board/services/folder_sync.py @@ -231,7 +231,7 @@ class FolderSyncService: infos = self.client.fav_infos( folder.media_id, sorted(targets), - folder_mid=cfg.bilibili.mid, + folder_mid=cfg.bilibili.read.mid, ) for aid, resource in infos.items.items(): if aid not in remote: diff --git a/reaction_board/services/snapshot.py b/reaction_board/services/snapshot.py index 2289fa5..8ff6c69 100644 --- a/reaction_board/services/snapshot.py +++ b/reaction_board/services/snapshot.py @@ -255,7 +255,7 @@ class SnapshotService: # ------------------------------------------------------------ internals def _check_auth(self, cfg: AppConfig, now: int) -> tuple[bool, str]: - if not cfg.bilibili.cookie: + if not cfg.bilibili.read.cookie: self.repo.kv_set("auth:last_check", str(now)) self.repo.kv_set("auth:ok", "0") self.repo.kv_set("auth:message", "未配置 cookie") diff --git a/tests/test_auth.py b/tests/test_auth.py index 35c44af..c0ece53 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -1,170 +1,88 @@ # -*- coding: utf-8 -*- -"""auth 相关纯函数自测（不联网）: python tests/test_auth.py""" +"""双账号 auth 纯函数与写回隔离测试（不联网）: uv run python tests/test_auth.py""" + +import json import os import sys +import tempfile +from pathlib import Path sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) sys.stdout.reconfigure(encoding="utf-8") -import favsort as f # noqa: E402 - -OLD = ( - "buvid3=OLD-B3; buvid4=OLD-B4; _uuid=OLD-UUID; buvid_fp=OLD-FP; " - "b_nut=1700000000; SESSDATA=old%2C1700000000%2Cabc; bili_jct=oldcsrf; " - "DedeUserID=28802351; DedeUserID__ckMd5=oldmd5; sid=oldsid; b_lsid=OLDL; " - "bili_ticket=ticket; home_feed_column=5" +from reaction_board.bilibili.auth import ( # noqa: E402 + cookie_value, + cookies_from_set_cookie, + merge_cookie, + parse_cookie, + sessdata_expiry, ) +from reaction_board.bilibili.auth_flow import AuthService, write_qr_png # noqa: E402 + +READ_OLD = "buvid3=READ-B3; {{SECRET_i5bghhpt}}readcsrf; DedeUserID=11; SESSDATA=old%2C1700000000%2Ca" +WRITE_OLD = "buvid3=WRITE-B3; bili_jct=writecsrf; DedeUserID=22; SESSDATA=old%2C1700000000%2Cb" +NEW = {"SESSDATA": "new%2C1800000000%2Cxyz", "bili_jct": "newcsrf", "DedeUserID": "99"} -def test_parse_and_merge(): - pairs = f.parse_cookie(OLD) - assert pairs[0] == ("buvid3", "OLD-B3") - assert f.cookie_value(OLD, "SESSDATA") == "old%2C1700000000%2Cabc" - - new = { - "SESSDATA": "new%2C1800000000%2Cxyz", - "bili_jct": "newcsrf", - "DedeUserID": "28802351", - "DedeUserID__ckMd5": "newmd5", - "sid": "newsid", +def raw_config(): + return { + "bilibili": { + "read": {"mid": 11, "csrf": "readcsrf", "cookie": READ_OLD}, + "write": {"mid": 22, "csrf": "writecsrf", "cookie": WRITE_OLD}, + }, + "folders": [], } - merged = f.merge_cookie(OLD, new) - # 身份字段被替换 - assert f.cookie_value(merged, "SESSDATA") == "new%2C1800000000%2Cxyz" - assert f.cookie_value(merged, "bili_jct") == "newcsrf" - assert f.cookie_value(merged, "sid") == "newsid" - # 设备指纹原样保留 - assert f.cookie_value(merged, "buvid3") == "OLD-B3" - assert f.cookie_value(merged, "buvid4") == "OLD-B4" - assert f.cookie_value(merged, "_uuid") == "OLD-UUID" - assert f.cookie_value(merged, "buvid_fp") == "OLD-FP" - assert f.cookie_value(merged, "b_nut") == "1700000000" - assert f.cookie_value(merged, "home_feed_column") == "5" - # 旧会话字段在没有新值时应被丢弃 - assert f.cookie_value(merged, "b_lsid") == "" -def test_set_cookie_and_url_fallback(): - headers = [ - "SESSDATA=abc%2C1800000000%2Cdef; Path=/; Domain=.bilibili.com; HttpOnly", - "bili_jct=csrf2; Domain=.bilibili.com", - "buvid3=SHOULD-NOT-OVERWRITE; Domain=.bilibili.com", - ] - got = f.cookies_from_set_cookie(headers) - assert got["SESSDATA"] == "abc%2C1800000000%2Cdef" - assert got["bili_jct"] == "csrf2" - - merged = f.merge_cookie(OLD, got) - assert f.cookie_value(merged, "buvid3") == "OLD-B3" # 设备字段不被 Set-Cookie 覆盖 +def test_cookie_helpers(): + assert parse_cookie(READ_OLD)[0] == ("buvid3", "READ-B3") + merged = merge_cookie(WRITE_OLD, NEW) + assert cookie_value(merged, "SESSDATA") == "new%2C1800000000%2Cxyz" + assert cookie_value(merged, "bili_jct") == "newcsrf" + assert cookie_value(merged, "buvid3") == "WRITE-B3" + assert sessdata_expiry(merged) == 1800000000 + headers = ["SESSDATA=x; Path=/", "bili_jct=y; Domain=.bilibili.com"] + assert cookies_from_set_cookie(headers) == {"SESSDATA": "x", "bili_jct": "y"} -def test_expiry(): - assert f.sessdata_expiry(OLD) == 1700000000 - assert f.sessdata_expiry("SESSDATA=bad") is None - assert f.sessdata_expiry("") is None +def test_finish_updates_only_selected_account_and_rolls_back(): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "config.json" + original = raw_config() + path.write_text(json.dumps(original, ensure_ascii=False), encoding="utf-8") + service = AuthService(path, account="write") + service._get_json = lambda credentials, url, auth: { # type: ignore[method-assign] + "code": 0, "data": {"isLogin": True, "mid": 99, "uname": "writer"} + } + assert service.finish(service._load_raw(), NEW) == 0 + saved = json.loads(path.read_text(encoding="utf-8")) + assert saved["bilibili"]["read"] == original["bilibili"]["read"] + assert saved["bilibili"]["write"]["mid"] == 99 + assert saved["bilibili"]["write"]["csrf"] == "newcsrf" + assert cookie_value(saved["bilibili"]["write"]["cookie"], "buvid3") == "WRITE-B3" + assert path.with_name("config.json.bak").exists() -def test_ensure_device_cookies(): - called = {"n": 0} - - def fake_get_json(cfg, url, auth=False): - called["n"] += 1 - return {"code": 0, "data": {"b_3": "NEW-B3", "b_4": "NEW-B4"}} - - orig = f.get_json - f.get_json = fake_get_json - try: - cfg = {"cookie": "SESSDATA=x; bili_jct=y"} - added = f.ensure_device_cookies(cfg) - assert "buvid3" in added and "buvid4" in added and "_uuid" in added - assert f.cookie_value(cfg["cookie"], "buvid3") == "NEW-B3" - assert f.cookie_value(cfg["cookie"], "buvid4") == "NEW-B4" - assert f.cookie_value(cfg["cookie"], "_uuid").endswith("infoc") - - # 已有设备字段时不应调用 spi，也不应改动 - called["n"] = 0 - cfg2 = {"cookie": OLD} - added2 = f.ensure_device_cookies(cfg2) - assert added2 == [] and called["n"] == 0 - assert cfg2["cookie"] == OLD - finally: - f.get_json = orig - - -def test_finish_auth_write_and_rollback(): - import json - import tempfile - - new = {"SESSDATA": "new%2C1800000000%2Cxyz", "bili_jct": "newcsrf", - "DedeUserID": "28802351", "DedeUserID__ckMd5": "newmd5"} - cfg = {"mid": 28802351, "csrf": "oldcsrf", "cookie": OLD, - "overrides": {}, "sleep": {}} - old_raw = json.dumps(cfg, ensure_ascii=False, indent=2) - - orig = (f.CONFIG_PATH, f.CONFIG_BACKUP, f.fetch_nav, f.ensure_device_cookies) - - def fake_nav_ok(c): - return {"code": 0, "data": {"isLogin": True, "uname": "tester", "mid": 28802351}} - - def fake_nav_bad(c): - return {"code": -101, "message": "账号未登录", "data": {"isLogin": False}} - - def fake_device(c): - return [] - - try: - with tempfile.TemporaryDirectory() as d: - path = os.path.join(d, "config.json") - f.CONFIG_PATH, f.CONFIG_BACKUP = path, path + ".bak" - f.ensure_device_cookies = fake_device - - with open(path, "w", encoding="utf-8") as fh: - fh.write(old_raw) - - # 成功：写入新身份字段，设备字段不变，并生成备份 - f.fetch_nav = fake_nav_ok - assert f.finish_auth(cfg, dict(new)) == 0 - with open(path, encoding="utf-8") as fh: - saved = json.load(fh) - assert "newcsrf" == saved["csrf"] - assert f.cookie_value(saved["cookie"], "buvid3") == "OLD-B3" - assert f.cookie_value(saved["cookie"], "SESSDATA") == "new%2C1800000000%2Cxyz" - assert os.path.exists(path + ".bak") - - # 失败：用上一份备份回滚 - before = open(path, encoding="utf-8").read() - cfg2 = dict(saved) - f.fetch_nav = fake_nav_bad - assert f.finish_auth(cfg2, {"SESSDATA": "x%2C1%2Cy", "bili_jct": "z"}) == 4 - after = open(path, encoding="utf-8").read() - assert after == before - finally: - f.CONFIG_PATH, f.CONFIG_BACKUP, f.fetch_nav, f.ensure_device_cookies = orig + before = path.read_text(encoding="utf-8") + service._get_json = lambda credentials, url, auth: { # type: ignore[method-assign] + "code": -101, "data": {"isLogin": False} + } + assert service.finish(service._load_raw(), NEW) == 4 + assert path.read_text(encoding="utf-8") == before def test_qr_png(): - import tempfile - - path = os.path.join(tempfile.gettempdir(), "favsort-test-qr.png") - try: - f.write_qr_png("https://example.com/login?key=abc&from=test", path) - with open(path, "rb") as fh: - data = fh.read() - assert data.startswith(b"\x89PNG\r\n\x1a\n") - assert len(data) > 500 - finally: - if os.path.exists(path): - os.remove(path) + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "qr.png" + write_qr_png("https://example.com/login?key=abc", path) + assert path.read_bytes().startswith(b"\x89PNG\r\n\x1a\n") def main(): - test_parse_and_merge() - test_set_cookie_and_url_fallback() - test_expiry() - test_ensure_device_cookies() - test_finish_auth_write_and_rollback() + test_cookie_helpers() + test_finish_updates_only_selected_account_and_rolls_back() test_qr_png() - print("OK: auth 纯函数测试全部通过") + print("OK: 双账号 auth Cookie 合并、写回隔离与回滚测试全部通过") if __name__ == "__main__": diff --git a/tests/test_cli.py b/tests/test_cli.py index 458ba42..626750f 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -22,9 +22,10 @@ import reaction_board.cli as cli # noqa: E402 def write_config(tmp, *, cookie="SESSDATA=x; buvid3=y", csrf="csrf"): path = Path(tmp) / "config.json" path.write_text(json.dumps({ - "mid": 1, - "csrf": csrf, - "cookie": cookie, + "bilibili": { + "read": {"mid": 1, "csrf": csrf, "cookie": cookie}, + "write": {"mid": 2, "csrf": "write-csrf", "cookie": "write-cookie"}, + }, "storage": { "database_path": str(Path(tmp) / "app.db"), "backup_dir": str(Path(tmp) / "backup"), @@ -112,6 +113,37 @@ def test_auth_check_without_cookie(): assert "未配置 cookie" in payload["message"] +def test_auth_check_all_and_auth_dispatch(): + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, cookie="", csrf="") + code, output = run_cli(["auth-check", "--account", "all", "--config", str(path)]) + assert code == 1 + payload = json.loads(output) + assert payload["ok"] is False + assert [item["account"] for item in payload["accounts"]] == ["read", "write"] + + original = cli.AuthService + calls = [] + + class FakeAuthService: + def __init__(self, path, *, account, logger): + calls.append((str(path), account)) + + def run(self, *, timeout, no_browser): + assert timeout == 9 and no_browser is True + return 7 + + cli.AuthService = FakeAuthService + try: + code, _ = run_cli([ + "auth", "--account", "write", "--timeout", "9", "--no-browser", "--config", str(path), + ]) + finally: + cli.AuthService = original + assert code == 7 + assert calls == [(str(path), "write")] + + def test_run_once_offline(): class FakeClient: def check_login(self): @@ -125,7 +157,13 @@ def test_run_once_offline(): path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") original = cli._client - cli._client = lambda *args, **kwargs: FakeClient() + selected = [] + + def fake_client(credentials, *args, **kwargs): + selected.append(credentials) + return FakeClient() + + cli._client = fake_client try: code, output = run_cli(["run", "--once", "--config", str(path)]) finally: @@ -133,6 +171,7 @@ def test_run_once_offline(): assert code == 0 assert '"name": "folder_sync"' in output assert '"name": "snapshot"' in output + assert selected and all(credentials.mid == 1 for credentials in selected) def main_test(): @@ -141,6 +180,7 @@ def main_test(): test_sync_requires_auth_without_network() test_snapshot_offline_empty_db() test_auth_check_without_cookie() + test_auth_check_all_and_auth_dispatch() test_run_once_offline() print("OK: CLI init/sync/snapshot/auth-check/run/status/错误路径测试全部通过") diff --git a/tests/test_config.py b/tests/test_config.py index a8f1ca5..544d513 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1,5 +1,5 @@ # -*- coding: utf-8 -*- -"""配置加载 / 校验 / 热加载测试（不联网）: uv run python tests/test_config.py""" +"""双账号配置加载 / 校验 / 热加载测试（不联网）: uv run python tests/test_config.py""" import json import os @@ -13,24 +13,22 @@ sys.stdout.reconfigure(encoding="utf-8") from reaction_board.config import ConfigError, ConfigStore, load_config # noqa: E402 -COOKIE = "SESSDATA=secret%2C1800000000%2Cabc; bili_jct=csrf-secret; buvid3=dev" - def write(path: Path, data) -> None: path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") def bump_mtime(path: Path) -> None: - """保证 mtime 严格大于上一次，避免 Windows 时间精度导致热加载测试偶发失败。""" target = max(path.stat().st_mtime_ns, time.time_ns()) + 1_000_000 os.utime(path, ns=(target, target)) -def base_legacy() -> dict: +def base_config() -> dict: return { - "mid": 28802351, - "csrf": "csrf-secret", - "cookie": COOKIE, + "bilibili": { + "read": {"mid": 11, "csrf": "read-csrf", "cookie": "read-cookie"}, + "write": {"mid": 22, "csrf": "write-csrf", "cookie": "write-cookie"}, + }, "overrides": {"1": {"mid": 2, "name": "x"}}, "sleep": {"min": 1}, "folders": [ @@ -42,168 +40,95 @@ def base_legacy() -> dict: } -def test_legacy_flat_mapping(): +def test_dual_accounts_and_env_override(): with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "config.json" - write(path, base_legacy()) - cfg = load_config(path, env={}) - assert cfg.bilibili.mid == 28802351 - assert cfg.bilibili.csrf == "csrf-secret" - assert cfg.bilibili.cookie == COOKIE - assert cfg.storage.database_path == Path(tmp) / "data" / "app.db" - assert cfg.storage.backup_dir == Path(tmp) / "data" / "backup" - assert cfg.schedule.snapshot_at == "00:01" - assert [f.media_id for f in cfg.folders] == [4451053851, 3700023351] - assert cfg.track_fans == (37754047,) - assert cfg.up_blacklist_seed == (1, 2) # 去重 - assert cfg.raw["overrides"] # 旧字段仍保留给 favsort - assert set(cfg.heat_weights) == { - "view", "like", "coin", "favorite", "share", "reply", "danmaku" - } - - -def test_nested_and_env_override(): - with tempfile.TemporaryDirectory() as tmp: - path = Path(tmp) / "config.json" - write(path, { - "bilibili": {"mid": 1, "csrf": "old", "cookie": "old-cookie"}, - "storage": {"database_path": "var/app.db", "backup_dir": "var/bak"}, - "folders": [], - }) + write(path, base_config()) env = { - "BILI_MID": "999", - "BILI_CSRF": "env-csrf", - "BILI_COOKIE": "env-cookie", - "TG_BOT_TOKEN": "123456:abcdefghijklmnopqrstuvwxyzABCDE", - "TG_CHAT_ID": "42", - "DASHBOARD_AUTH_TOKEN": "web-token", + "BILI_READ_MID": "101", "BILI_READ_CSRF": "env-read", "BILI_READ_COOKIE": "env-read-cookie", + "BILI_WRITE_MID": "202", "BILI_WRITE_CSRF": "env-write", "BILI_WRITE_COOKIE": "env-write-cookie", "DATABASE_PATH": "custom.db", } cfg = load_config(path, env=env) - assert cfg.bilibili.mid == 999 - assert cfg.bilibili.csrf == "env-csrf" - assert cfg.bilibili.cookie == "env-cookie" - assert cfg.telegram.chat_id == 42 - assert cfg.telegram.bot_token.startswith("123456:") - assert cfg.web.auth_token == "web-token" + assert (cfg.bilibili.read.mid, cfg.bilibili.read.csrf, cfg.bilibili.read.cookie) == (101, "env-read", "env-read-cookie") + assert (cfg.bilibili.write.mid, cfg.bilibili.write.csrf, cfg.bilibili.write.cookie) == (202, "env-write", "env-write-cookie") assert cfg.storage.database_path == Path(tmp) / "custom.db" - # 备份目录仍按配置文件目录解析 - assert cfg.storage.backup_dir == Path(tmp) / "var" / "bak" + assert cfg.up_blacklist_seed == (1, 2) -def test_absolute_paths_kept(): - with tempfile.TemporaryDirectory() as tmp: - db = str(Path(tmp) / "abs" / "app.db") - path = Path(tmp) / "config.json" - write(path, { - "bilibili": {"mid": 1, "csrf": "c", "cookie": "k"}, - "storage": {"database_path": db, "backup_dir": str(Path(tmp) / "b")}, - "folders": [], - }) - cfg = load_config(path, env={}) - assert cfg.storage.database_path == Path(db) - - -def test_invalid_configs(): +def test_flat_and_missing_account_rejected(): with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "config.json" - cases = [ - ({"folders": [{"year": 2025, "media_id": 1}, - {"year": 2026, "media_id": 1}]}, "media_id 重复"), - ({"folders": [{"year": 2025, "media_id": 1}, - {"year": 2025, "media_id": 2}]}, "year 重复"), - ({"folders": [{"year": 2025, "media_id": 0}]}, "不能小于 1"), - ({"schedule": {"snapshot_at": "24:00"}}, "snapshot_at"), - ({"schedule": {"snapshot_at": "1:01"}}, "snapshot_at"), - ({"notify": {"max_items_per_message": 0}}, "max_items_per_message"), - ({"heat_weights": {"view": -1}}, "heat_weights.view"), - ({"classify": {"require_title": ""}}, "require_title"), - ] - base = { - "bilibili": {"mid": 1, "csrf": "c", "cookie": "k"}, - "folders": [], - } - for patch, needle in cases: - data = json.loads(json.dumps(base)) - data.update(patch) - write(path, data) - try: - load_config(path, env={}) - except ConfigError as exc: - assert needle in str(exc), (needle, str(exc)) - else: - raise AssertionError(f"应当失败: {patch}") + write(path, {"mid": 1, "csrf": "x", "cookie": "y", "folders": []}) + try: + load_config(path, env={}) + except ConfigError as exc: + assert "bilibili.read" in str(exc) + else: + raise AssertionError("flat 格式必须拒绝") + write(path, {"bilibili": {"read": {}}, "folders": []}) + try: + load_config(path, env={}) + except ConfigError as exc: + assert "bilibili.write" in str(exc) + else: + raise AssertionError("缺少 write 必须拒绝") -def test_require_auth(): +def test_validation_require_read_auth_and_hot_reload(): with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "config.json" - write(path, {"bilibili": {"mid": 1, "csrf": "", "cookie": ""}, "folders": []}) + data = base_config() + data["bilibili"]["read"].update(cookie="", csrf="") + write(path, data) load_config(path, env={}, require_auth=False) try: load_config(path, env={}, require_auth=True) except ConfigError as exc: - assert "cookie" in str(exc) + assert "bilibili.read.cookie" in str(exc) else: - raise AssertionError("require_auth 应拒绝空凭据") + raise AssertionError("read 空凭据必须拒绝") - -def test_env_int_invalid(): - with tempfile.TemporaryDirectory() as tmp: - path = Path(tmp) / "config.json" - write(path, base_legacy()) - try: - load_config(path, env={"BILI_MID": "abc"}) - except ConfigError as exc: - assert "BILI_MID" in str(exc) - else: - raise AssertionError("非法整数环境变量应失败") - - -def test_hot_reload_keeps_last_good_config(): - with tempfile.TemporaryDirectory() as tmp: - path = Path(tmp) / "config.json" - write(path, base_legacy()) + write(path, base_config()) store = ConfigStore(path, env={}) - assert store.get().folders[0].title == "2025" - - data = base_legacy() - data["folders"][0]["title"] = "2025-新" + data = base_config() + data["folders"][0]["title"] = "2025-new" write(path, data) bump_mtime(path) - assert store.get().folders[0].title == "2025-新" - assert store.last_error is None - - path.write_text("{not json", encoding="utf-8") + assert store.get().folders[0].title == "2025-new" + path.write_text("{bad", encoding="utf-8") bump_mtime(path) - assert store.get().folders[0].title == "2025-新" - assert store.last_error is not None + assert store.get().folders[0].title == "2025-new" + assert store.last_error + +def test_invalid_values(): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "config.json" + data = base_config() + data["bilibili"]["write"]["mid"] = "bad" write(path, data) - bump_mtime(path) - assert store.get().folders[0].title == "2025-新" - assert store.last_error is None - - -def test_missing_file(): - try: - load_config(Path(tempfile.gettempdir()) / "no-such-bili-fav-config.json", env={}) - except ConfigError as exc: - assert "缺少配置文件" in str(exc) - else: - raise AssertionError("缺失配置应失败") + try: + load_config(path, env={}) + except ConfigError as exc: + assert "bilibili.write.mid" in str(exc) + else: + raise AssertionError("非法 mid 必须拒绝") + write(path, base_config()) + try: + load_config(path, env={"BILI_READ_MID": "bad"}) + except ConfigError as exc: + assert "BILI_READ_MID" in str(exc) + else: + raise AssertionError("非法环境变量必须拒绝") def main(): - test_legacy_flat_mapping() - test_nested_and_env_override() - test_absolute_paths_kept() - test_invalid_configs() - test_require_auth() - test_env_int_invalid() - test_hot_reload_keeps_last_good_config() - test_missing_file() - print("OK: 配置加载、校验与热加载测试全部通过") + test_dual_accounts_and_env_override() + test_flat_and_missing_account_rejected() + test_validation_require_read_auth_and_hot_reload() + test_invalid_values() + print("OK: 双账号配置加载、拒绝 flat 格式、ENV 覆盖与热加载测试全部通过") if __name__ == "__main__": diff --git a/tests/test_favsort_config.py b/tests/test_favsort_config.py new file mode 100644 index 0000000..dab7f3c --- /dev/null +++ b/tests/test_favsort_config.py @@ -0,0 +1,62 @@ +# -*- coding: utf-8 -*- +"""favsort 写账号配置隔离测试: uv run python tests/test_favsort_config.py""" + +import json +import os +import sys +import tempfile +from pathlib import Path + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +sys.stdout.reconfigure(encoding="utf-8") + +import favsort # noqa: E402 + + +def test_loads_write_account_only(): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "config.json" + path.write_text(json.dumps({ + "bilibili": { + "read": {"mid": 11, "csrf": "read-csrf", "cookie": "read-cookie"}, + "write": {"mid": 22, "csrf": "write-csrf", "cookie": "write-cookie"}, + }, + "overrides": {"1": {"mid": 2, "name": "UP"}}, + }), encoding="utf-8") + old = favsort.CONFIG_PATH + favsort.CONFIG_PATH = str(path) + try: + cfg = favsort.load_config() + finally: + favsort.CONFIG_PATH = old + assert cfg["mid"] == 22 + assert cfg["csrf"] == "write-csrf" + assert cfg["cookie"] == "write-cookie" + assert cfg["overrides"] == {"1": {"mid": 2, "name": "UP"}} + + +def test_rejects_flat_config(): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "config.json" + path.write_text(json.dumps({"mid": 1, "csrf": "x", "cookie": "y"}), encoding="utf-8") + old = favsort.CONFIG_PATH + favsort.CONFIG_PATH = str(path) + try: + try: + favsort.load_config() + except SystemExit as exc: + assert "bilibili.write" in str(exc) + else: + raise AssertionError("flat config 必须拒绝") + finally: + favsort.CONFIG_PATH = old + + +def main(): + test_loads_write_account_only() + test_rejects_flat_config() + print("OK: favsort 仅使用 bilibili.write 配置测试全部通过") + + +if __name__ == "__main__": + main() diff --git a/tests/test_folder_sync.py b/tests/test_folder_sync.py index 21bd602..40af003 100644 --- a/tests/test_folder_sync.py +++ b/tests/test_folder_sync.py @@ -74,9 +74,10 @@ class FakeFavTransport: def write_config(tmp, folders): data = { - "mid": 28802351, - "csrf": "csrf-secret", - "cookie": "SESSDATA=secret; bili_jct=csrf-secret; buvid3=device", + "bilibili": { + "read": {"mid": 28802351, "csrf": "csrf-secret", "cookie": "SESSDATA=secret; bili_jct=csrf-secret; buvid3=device"}, + "write": {"mid": 2, "csrf": "write-csrf", "cookie": "write-cookie"}, + }, "storage": { "database_path": str(Path(tmp) / "app.db"), "backup_dir": str(Path(tmp) / "backup"), diff --git a/tests/test_snapshot.py b/tests/test_snapshot.py index def3630..e87a013 100644 --- a/tests/test_snapshot.py +++ b/tests/test_snapshot.py @@ -75,9 +75,10 @@ def make_service(tmp, transport, *, track_fans=(37754047,), logger=None, retries=3, cookie="SESSDATA=secret; buvid3=device"): path = Path(tmp) / "config.json" path.write_text(json.dumps({ - "mid": 28802351, - "csrf": "csrf", - "cookie": cookie, + "bilibili": { + "read": {"mid": 28802351, "csrf": "csrf", "cookie": cookie}, + "write": {"mid": 2, "csrf": "write-csrf", "cookie": "write-cookie"}, + }, "storage": { "database_path": str(Path(tmp) / "app.db"), "backup_dir": str(Path(tmp) / "backup"),