diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index b4a0601..1dc5785 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -281,8 +281,9 @@ PRAGMA synchronous = NORMAL; - P0 已实现的命令行：`python app.py init`、`python app.py sync [--full]`、`python app.py status`。 ### 5.2 安全规则- Windows 本地开发时应限制 `config.json` 的文件访问权限；部署到 Linux 时权限必须为 `600`； -- 配置文件每 60 秒检查一次 mtime，收藏夹、分类规则和权重变更后无需重启； -- Cookie 或 CSRF 改变时重建 B 站请求客户端； +- 配置文件每 60 秒检查一次 mtime，凭据、收藏夹、分类规则和权重变更后无需重启； +- Cookie / CSRF 变化时由 `ClientProvider` 重建 B 站请求客户端，并保留该角色的限速与重试策略； +- `schedule.folder_sync_minutes` 与 `schedule.snapshot_at` 在启动时固化，修改后需重启 `run`； - Telegram 操作必须同时验证 `from.id` 和 `chat.id` 是否为配置的私聊用户； - 仅允许环境变量临时覆盖敏感配置，不能将真实凭据硬编码进源代码； - `GET` 路由绝不进行收藏、忽略、拉黑等写操作。 @@ -310,7 +311,8 @@ PRAGMA synchronous = NORMAL; - `snapshot` 主键为 `(stat_date, aid)`，保证同日幂等； - `up_stat` 主键为 `(stat_date, mid)`； - `folder_resource(media_id, aid)` 记录每个收藏夹实际拥有的资源，不能仅依赖 `video.folder_id`； -- `video.folder_id` 表示视频的主归属夹，用于新视频同 UP 归属和排序； +- `folder_resource.fav_time` 记录该视频在该夹的加入时间，用于跨夹主归属选择； +- `video.folder_id` 表示视频的主归属夹，按有效收藏关系重算，不依赖收藏夹遍历顺序； - `up_blacklist` 只在发现任务中生效，收藏夹同步不得用它过滤既有视频； - `seen_aid` 永久保留搜索去重标记，候选过期删除后也不重复通知； - `fav_action` 保存 Telegram 或签名链接的动作状态，防止重复点击造成重复写入。 @@ -409,8 +411,11 @@ P1 实现说明： - 时区使用固定 UTC+8（中国无夏令时），避免 Windows 缺少 `tzdata`； - 当日全部视频与粉丝采集成功后在 `kv` 写入 `snapshot:completed:=1`；启动时未完成会补跑； -- `ok=0` 的当日记录会在下次补跑时重试；`ok=1` 直接跳过； +- 完成标记不是唯一门槛：每次检查 `pending_snapshot_aids(date)`，新入库或从 `dropped` 恢复的视频会立即补采当天快照； +- `ok=0` 的当日记录会在下次补跑时重试；`ok=1` 直接跳过；粉丝同样跳过已成功的 `(stat_date, mid)`； +- 开始采集前先写 `snapshot:completed:=0`，强制重采中断后不会残留旧的完成标记； - 每次采集后生成 `data/backup/app-YYYYMMDD.db` 并清理超过 `storage.backup_keep_days` 的日备份； +- 采集完成与备份完成分开记录；备份失败时 `snapshot` 返回 1，下次执行只重试备份、不重复请求 B 站； - Cookie 登录态通过 `x/web-interface/nav` 检查，失效只告警，不阻塞匿名 `view`； - Telegram 摘要推迟到 P3，P1 先写结构化 JSON 日志和 `kv` 状态。 @@ -589,6 +594,20 @@ uv run python tests/test_moves.py - 真实全量联调：795 条视频全部 `ok=1`，0 失败、0 移除，耗时 1124 秒；粉丝 `follower=2767270`；日备份 `data/backup/app-20261031.db` 含 795 条快照；二次执行正确命中完成标记并跳过； - 自动化测试见 `tests/test_snapshot.py`、`test_maintenance.py`、`test_timeutil.py`，并扩展了 `test_bilibili_client.py` 与 `test_scheduler.py`。 +### P1 补丁：数据完整性与热更新修复 ✅ + +- 新入库/恢复视频当天补采：`pending_snapshot_aids()` + 同步后立即触发快照； +- 凭据热更新：`ClientProvider` 检测 `mid/cookie/csrf` 变化后重建客户端； +- 协议防御：`ids` / `infos` 缺失 `data` 或 `data=null` 时中止，不再当作空结果； +- 空远端保护：远端 0 条而本地非空时跳过差集，需 `--allow-empty-folder` 才执行清空； +- 粉丝续跑：非强制模式跳过已成功的 `(stat_date, mid)`； +- 强制重采：开始前先写未完成标记，中断后可续跑； +- 备份恢复：采集完成与备份完成分离，备份失败返回 1 并支持无请求重试； +- 跨夹主归属：迁移 v2 增加 `folder_resource.fav_time`，按有效关系重算 `video.folder_id`； +- `dropped` 视频失效时保留原始标题； +- `favsort.py`：`--verify-every` 必须为正数；`sort` 使用真实资源类型，不再硬编码 `2`； +- 配置每 60 秒检查 mtime；调度参数变更需重启。 + ### P2：本地看板 - 总览、榜单、详情、UP、粉丝、候选和系统状态页面； diff --git a/README.md b/README.md index a9c1fac..95b43f6 100644 --- a/README.md +++ b/README.md @@ -84,7 +84,8 @@ uv run python app.py status - `infos` 静默缺失或 `attr=9` 的资源标记为 `removed`，保留历史； - 从收藏夹移除的资源标记为 `unfaved`，仍保留数据； - 黑名单只作为发现阶段的种子写入，**不会过滤收藏夹同步**； -- 收藏夹达到 950 条时预警，系统绝不自动换夹。 +- 收藏夹达到 950 条时预警，系统绝不自动换夹； +- 远端返回 0 条但本地仍有记录时默认按异常跳过差集，确认收藏夹确实被清空后才使用 `--allow-empty-folder`。 ## P1：每日快照与粉丝追踪 @@ -108,11 +109,13 @@ uv run python app.py auth-check --account all - `snapshot` 对 `active` / `unfaved` 视频逐条请求 `view`，每条请求间隔 1.0～1.6 秒； - 成功写入 `snapshot` 的七项指标；同日 `ok=1` 的记录直接跳过，可断点续跑； +- 当天已完成后新入库或从 `dropped` 恢复的视频会立即补采当天快照，不必等到次日； - 视频不可访问（`-404` / `62002` / `62012`）标记为 `removed`； - 其他错误重试 3 次，仍失败写入 `ok=0` 和错误信息，不写伪造的零值； -- 粉丝数写入 `up_stat`，失败同样写 `ok=0`； +- 粉丝数写入 `up_stat`，失败同样写 `ok=0`；续跑时已成功的 `(stat_date, mid)` 会跳过，不会被失败结果覆盖； - 当日全部视频与粉丝采集成功后写入完成标记；服务启动时若当天未完成会自动补跑； - 每次采集后生成 `data/backup/app-YYYYMMDD.db`，按 `storage.backup_keep_days` 清理旧日备份； +- 备份失败时 `snapshot` 返回 1 便于任务告警；再次执行只重试备份，不重复请求 B 站； - Cookie 失效只告警，不阻塞匿名 `view` 采集；P0/P1 只使用 `bilibili.read`。 首次真实全量采集结果：795 条视频全部成功、0 失败、0 移除，耗时约 19 分钟；粉丝 `follower=2767270`；日备份为 `data/backup/app-20261031.db`。 @@ -146,6 +149,8 @@ Copy-Item config.example.json config.json - `storage`：数据库与备份目录； - `schedule` / `notify` / `classify` / `heat_weights`：任务节奏、通知阈值、分类规则和热度权重。 +配置每 60 秒检查一次 mtime，凭据、收藏夹、分类规则等变更无需重启即可在下次任务读取；但 `schedule.folder_sync_minutes` 与 `schedule.snapshot_at` 在启动时固化，修改后需重启 `run` 进程。 + 完整配置口径以 [project.md](project.md) 和 [DEVELOPMENT.md](DEVELOPMENT.md) 为准。 ## 当前可用：收藏夹排序 CLI @@ -243,10 +248,12 @@ uv run python tests/test_database.py # 单写队列、事务与并发 uv run python tests/test_repository.py # 收藏夹/视频状态机与黑名单 uv run python tests/test_bilibili_client.py # ids/infos/view/fans/nav 契约与重试 uv run python tests/test_folder_sync.py # 首次导入、差集、失效识别、容量预警 -uv run python tests/test_snapshot.py # 七项快照、断点续跑、失败占位、粉丝 +uv run python tests/test_snapshot.py # 七项快照、补采、断点续跑、粉丝、备份重试 uv run python tests/test_maintenance.py # 日备份与保留期 uv run python tests/test_timeutil.py # 上海时区与每日任务时间 uv run python tests/test_scheduler.py # 内置调度器到期/失败隔离 +uv run python tests/test_provider.py # 凭据变化重建客户端与配置 revision +uv run python tests/test_favsort_run.py # 排序参数校验与混合资源类型 uv run python tests/test_logging.py # JSON 日志脱敏 uv run python tests/test_cli.py # CLI 离线路径 ``` diff --git a/favsort.py b/favsort.py index d219971..9c78cd2 100644 --- a/favsort.py +++ b/favsort.py @@ -346,6 +346,7 @@ def cmd_plan(cfg, args): "current_order": cur, "target_order": tgt, "moves": moves, + "types": {v["avid"]: v["type"] for v in applied}, "titles": { v["avid"]: { "title": v["title"], @@ -395,8 +396,12 @@ def cmd_run(cfg, args): 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)] @@ -428,7 +433,13 @@ def cmd_run(cfg, args): 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"]} + 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) @@ -541,6 +552,16 @@ COMMAND_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", @@ -565,7 +586,7 @@ def build_parser(): 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", + 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"]) diff --git a/project.md b/project.md index 593cf26..8747d17 100644 --- a/project.md +++ b/project.md @@ -129,7 +129,7 @@ } ``` -- 支持热加载（`SIGHUP` 或每 60 秒检查 mtime），改收藏夹/黑名单不用重启； +- 支持热加载（每 60 秒检查 mtime），改收藏夹/黑名单不用重启；`schedule` 中的周期与每日时间需重启生效； - **黑名单本身放数据库**（`up_blacklist` 表），因为要从通知里动态拉黑；配置文件只做初始种子； - 日志里**永不打印 token / cookie**，启动时只打印"已配置/未配置"； - 容器里把配置目录挂成**可写卷**（`-v ./config:/config`，权限 600），因为 `auth` 扫码续期后要回写 cookie； @@ -332,6 +332,7 @@ for folder in 配置的年度夹（2025、2026、…）: - 被用户手动"移出库"（`dropped`）的视频若又出现在收藏夹（`diff_add`）→ **自动恢复为 `active`**，并补当天快照；这是唯一会"复活"的状态。 - 监控的是**所有配置的年度夹**（2025、2026…），所以人工把新视频加进旧夹也能被发现； - `video.folder_id` 以"实际所在夹"为准，若同一 aid 同时出现在多个夹（极少见），取最新加入的那个并告警。 +- 上述 `enqueue_snapshot_today` 已实现为 `pending_snapshot_aids()` 检查：完成标记为 1 时仍会为新入库/恢复视频补采当天快照。 ### 4.3 首次导入 @@ -376,6 +377,8 @@ for folder in 配置的年度夹（2025、2026、…）: - 唯一键 `(stat_date, aid)`，`INSERT ... ON CONFLICT DO UPDATE`； - `stat_date` 用本地日期，不用 UTC（避免 00:01 落到前一天）； - 失败重试 3 次后写 `ok=0` 占位，**不写假数据**；看板显示"缺失"而不是 0； +- 完成标记之外还要检查当天待采集合：新入库/恢复视频立即补采，`ok=0` 项在续跑时重试； +- 强制重采开始前先写未完成标记，中断后仍可续跑； - 日增量查询时用 `LAG()` 跳过缺失日，不插值。 ### 5.4 粉丝数追踪（新增） @@ -388,6 +391,7 @@ for folder in 配置的年度夹（2025、2026、…）: - 接口实测：`{"mid":37754047,"following":95,"follower":2767264}`，**匿名`code=0`**，无需登录； - 唯一键 `(stat_date, mid)`，同样幂等； +- 续跑时已成功的 `(stat_date, mid)` 会跳过，不会因后续失败覆盖为 NULL； - 只加 1 次请求/天，成本可忽略； - 支持配置多个 mid（`track_fans: [37754047]`），但看板默认只突出咻咻满； - 粉丝日增 = 今日 - 昨日，缺失日不插值，曲线断开。 @@ -783,6 +787,8 @@ storage：SQLite WAL，单写线程 | 26 | `x/relation/stat` 粉丝数 | `code=0`，匿名可用，实测 `follower=2767268` | | 27 | `x/web-interface/nav` 登录态 | `code=0`、`isLogin=true`、mid=28802351 | | 28 | **P1 真实全量快照** | 795 条全部 `ok=1`、0 失败、0 移除；耗时 1124s；粉丝 `follower=2767270`；日备份 `app-20261031.db` 含 795 条快照 | +| 29 | **空收藏夹 `ids` 契约** | `4413147139` 返回 `{"code":0,"data":[]}`，`data` 键存在且为数组；缺失/`null` 按协议异常处理 | +| 30 | **迁移 v2 与增量同步** | 真实库升级到 schema 2，自动生成 `app-pre-v1-*.db`；同步 167+664 条无新增/移除，主归属无矛盾 | **未验证（需授权或外部配合）**： diff --git a/reaction_board/bilibili/client.py b/reaction_board/bilibili/client.py index 0c3747c..513fa13 100644 --- a/reaction_board/bilibili/client.py +++ b/reaction_board/bilibili/client.py @@ -262,6 +262,20 @@ class BilibiliClient: raise self._api_error(payload) return payload + @staticmethod + def _data_list(payload: Mapping[str, Any], name: str) -> list[Any]: + """区分“真实空数组”与“协议缺字段”。 + + 实测空收藏夹返回 ``{"code":0,"data":[]}``；缺少 ``data`` 或为 ``null`` + 都不是已知契约，必须中止，避免被误判成空收藏夹后批量改状态。 + """ + if "data" not in payload: + raise BilibiliProtocolError(f"{name} 响应缺少 data 字段") + data = payload["data"] + if not isinstance(data, list): + raise BilibiliProtocolError(f"{name} data 不是数组") + return data + # ------------------------------------------------------------ fav def fav_ids(self, media_id: int) -> list[FavId]: @@ -269,11 +283,7 @@ class BilibiliClient: f"{BASE}/x/v3/fav/resource/ids?media_id={int(media_id)}&platform=web" ) payload = self.get_json(url) - data = payload.get("data") - if data is None: - data = [] - if not isinstance(data, list): - raise BilibiliProtocolError("fav/resource/ids data 不是数组") + data = self._data_list(payload, "fav/resource/ids") out: list[FavId] = [] for index, item in enumerate(data): if not isinstance(item, Mapping): @@ -320,11 +330,7 @@ class BilibiliClient: f"&platform=web&folder_mid={fmid}&folder_id={int(media_id)}" ) payload = self.get_json(url) - data = payload.get("data") - if data is None: - data = [] - if not isinstance(data, list): - raise BilibiliProtocolError("fav/resource/infos data 不是数组") + data = self._data_list(payload, "fav/resource/infos") chunk_set = set(chunk) for item in data: if not isinstance(item, Mapping): diff --git a/reaction_board/bilibili/provider.py b/reaction_board/bilibili/provider.py new file mode 100644 index 0000000..b6d2583 --- /dev/null +++ b/reaction_board/bilibili/provider.py @@ -0,0 +1,38 @@ +# -*- coding: utf-8 -*- +"""按账号角色提供 B 站客户端，并在凭据变化时自动重建。""" + +from __future__ import annotations + +from typing import Any, Callable + +from ..config import ConfigStore +from .client import BilibiliClient + + +class ClientProvider: + """缓存客户端；仅当 ``mid`` / ``cookie`` / ``csrf`` 变化时重建。 + + ``factory`` 负责固化该角色的限速与重试策略，因此重建不会丢失这些设置。 + """ + + def __init__( + self, + store: ConfigStore, + role: str, + factory: Callable[[Any], BilibiliClient], + ) -> None: + if role not in {"read", "write"}: + raise ValueError(f"未知账号角色: {role}") + self.store = store + self.role = role + self.factory = factory + self._key: tuple[int, str, str] | None = None + self._client: BilibiliClient | None = None + + def get(self) -> BilibiliClient: + credentials = getattr(self.store.get().bilibili, self.role) + key = (credentials.mid, credentials.cookie, credentials.csrf) + if self._client is None or key != self._key: + self._client = self.factory(credentials) + self._key = key + return self._client diff --git a/reaction_board/cli.py b/reaction_board/cli.py index c775697..fe883d8 100644 --- a/reaction_board/cli.py +++ b/reaction_board/cli.py @@ -21,6 +21,7 @@ from typing import Sequence from .bilibili.auth_flow import AuthService from .bilibili.client import BilibiliClient, Pacer +from .bilibili.provider import ClientProvider from .config import ConfigError, ConfigStore from .database import Database from .logging_json import JsonLogger @@ -53,6 +54,11 @@ def _build_parser() -> argparse.ArgumentParser: sync.add_argument( "--full", action="store_true", help="强制全量刷新元数据（忽略 6 小时周期）" ) + sync.add_argument( + "--allow-empty-folder", + action="store_true", + help="允许把远端 0 条视为真实清空（默认按异常跳过差集）", + ) snapshot = sub.add_parser( "snapshot", parents=[common], help="采集当日七项快照与粉丝数" ) @@ -72,6 +78,11 @@ def _build_parser() -> argparse.ArgumentParser: run.add_argument( "--once", action="store_true", help="只执行一次到期任务后退出" ) + run.add_argument( + "--allow-empty-folder", + action="store_true", + help="允许把远端 0 条视为真实清空（默认按异常跳过差集）", + ) sub.add_parser("status", parents=[common], help="查看数据库状态（离线）") return parser @@ -130,7 +141,13 @@ def cmd_sync(args, logger: JsonLogger) -> int: try: repo = Repository(db) client = _client(cfg.bilibili.read, logger) - service = FolderSyncService(store, repo, client, logger=logger) + service = FolderSyncService( + store, + repo, + client, + logger=logger, + allow_empty_folders=args.allow_empty_folder, + ) summary = service.run(full_refresh=args.full) payload = summary.as_dict() payload["database"] = str(cfg.storage.database_path) @@ -157,7 +174,7 @@ def cmd_snapshot(args, logger: JsonLogger) -> int: payload["snapshot_counts"] = repo.snapshot_counts(summary.stat_date) payload["counts"] = repo.counts() _print(payload) - return 0 if summary.completed else 1 + return 0 if summary.completed and not summary.backup_error else 1 finally: db.close() @@ -205,16 +222,26 @@ def cmd_run(args, logger: JsonLogger) -> int: last: dict[str, object] = {} try: repo = Repository(db) - folder_client = _client(cfg.bilibili.read, logger) - folder_service = FolderSyncService(store, repo, folder_client, logger=logger) - 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(): - summary = folder_service.run() - last["folder"] = summary - logger.info("folder_sync.summary", **summary.as_dict()) - return summary + folder_provider = ClientProvider( + store, "read", lambda credentials: _client(credentials, logger) + ) + snapshot_provider = ClientProvider( + store, + "read", + lambda credentials: _client( + credentials, logger, pacer=Pacer(1.0, 1.6), retries=1 + ), + ) + folder_service = FolderSyncService( + store, + repo, + client_provider=folder_provider, + logger=logger, + allow_empty_folders=args.allow_empty_folder, + ) + snapshot_service = SnapshotService( + store, repo, client_provider=snapshot_provider, logger=logger + ) def run_snapshot(): summary = snapshot_service.run_if_needed() @@ -222,6 +249,22 @@ def cmd_run(args, logger: JsonLogger) -> int: logger.info("snapshot.summary", **summary.as_dict()) return summary + def run_sync(): + summary = folder_service.run() + last["folder"] = summary + logger.info("folder_sync.summary", **summary.as_dict()) + # 新入库/恢复的视频立即补当天快照，而不是等到次日。 + if summary.added > 0: + run_snapshot() + return summary + + def run_config_watch(): + before = store.revision + store.get() + if store.revision != before: + logger.info("config.reloaded", revision=store.revision) + return {"revision": store.revision} + hh, mm = parse_hhmm(cfg.schedule.snapshot_at) snapshot_at = f"{hh:02d}:{mm:02d}" scheduler = Scheduler( @@ -238,6 +281,7 @@ def cmd_run(args, logger: JsonLogger) -> int: next_run=0.0, next_run_fn=lambda started: next_daily_run(started, snapshot_at), ), + ScheduledTask("config_watch", 60.0, run_config_watch), ], logger=logger, ) @@ -275,6 +319,8 @@ def cmd_status(args, logger: JsonLogger) -> int: "last_completed": repo.kv_get("snapshot:last_completed"), "auth_ok": repo.kv_get("auth:ok"), "auth_message": repo.kv_get("auth:message"), + "backup_ok": repo.kv_get(f"snapshot:backup:{today}"), + "backup_error": repo.kv_get(f"snapshot:backup_error:{today}"), }) _print({ "ok": True, diff --git a/reaction_board/config.py b/reaction_board/config.py index aaffbc0..500cd6b 100644 --- a/reaction_board/config.py +++ b/reaction_board/config.py @@ -478,6 +478,7 @@ class ConfigStore: self._config = load_config(path, env=env, require_auth=require_auth) self._mtime_ns = self._read_mtime_ns() self.last_error: str | None = None + self.revision = 0 def _read_mtime_ns(self) -> int | None: try: @@ -499,4 +500,5 @@ class ConfigStore: self.last_error = str(exc) return self._config self.last_error = None + self.revision += 1 return self._config diff --git a/reaction_board/migrations.py b/reaction_board/migrations.py index cfddd9e..66ad659 100644 --- a/reaction_board/migrations.py +++ b/reaction_board/migrations.py @@ -167,6 +167,13 @@ MIGRATIONS: tuple[Migration, ...] = ( """, ), ), + Migration( + version=2, + name="folder_resource fav_time", + statements=( + "ALTER TABLE folder_resource ADD COLUMN fav_time INTEGER", + ), + ), ) diff --git a/reaction_board/repository.py b/reaction_board/repository.py index 3ec4f61..94e568b 100644 --- a/reaction_board/repository.py +++ b/reaction_board/repository.py @@ -166,6 +166,35 @@ class Repository: # ------------------------------------------------------------ video + @staticmethod + def _recompute_primary_folder(conn: sqlite3.Connection, aid: int) -> int | None: + """按有效收藏关系重算主归属，避免依赖收藏夹遍历顺序。 + + 优先使用逐夹 ``fav_time``（后加入者优先），其次 ``first_seen``， + 最后用年度与 media_id 做确定性决胜。 + """ + row = conn.execute( + """ + SELECT fr.media_id + FROM folder_resource fr + JOIN folder f ON f.media_id = fr.media_id + WHERE fr.aid = ? AND fr.active = 1 AND f.enabled = 1 + ORDER BY COALESCE(fr.fav_time, 0) DESC, + fr.first_seen DESC, + f.year DESC, + fr.media_id DESC + LIMIT 1 + """, + (int(aid),), + ).fetchone() + if row is None: + return None + folder_id = int(row["media_id"]) + conn.execute( + "UPDATE video SET folder_id = ? WHERE aid = ?", (folder_id, int(aid)) + ) + return folder_id + def upsert_from_fav( self, resource, @@ -206,7 +235,8 @@ class Repository: ON CONFLICT(aid) DO UPDATE SET bvid = COALESCE(excluded.bvid, video.bvid), title = CASE - WHEN excluded.status = 'removed' AND video.title != '' + WHEN (excluded.attr = 9 OR excluded.status = 'removed') + AND video.title != '' THEN video.title WHEN excluded.title != '' THEN excluded.title ELSE video.title @@ -219,7 +249,10 @@ class Repository: END, duration = COALESCE(excluded.duration, video.duration), cover = COALESCE(excluded.cover, video.cover), - folder_id = excluded.folder_id, + folder_id = CASE + WHEN video.folder_id IS NULL THEN excluded.folder_id + ELSE video.folder_id + END, fav_time = COALESCE(excluded.fav_time, video.fav_time), status = excluded.status, source = COALESCE(video.source, excluded.source), @@ -246,19 +279,22 @@ class Repository: ) conn.execute( """ - INSERT INTO folder_resource(media_id, aid, first_seen, active, last_seen) - VALUES (?, ?, ?, 1, ?) + INSERT INTO folder_resource(media_id, aid, first_seen, active, + last_seen, fav_time) + VALUES (?, ?, ?, 1, ?, ?) ON CONFLICT(media_id, aid) DO UPDATE SET active = 1, - last_seen = excluded.last_seen + last_seen = excluded.last_seen, + fav_time = COALESCE(excluded.fav_time, folder_resource.fav_time) """, - (folder_id, resource.aid, now, now), + (folder_id, resource.aid, now, now, resource.fav_time), ) + primary = self._recompute_primary_folder(conn, resource.aid) return UpsertOutcome( aid=resource.aid, status=status, previous_folder_id=previous_folder, - moved_folder=previous_folder is not None and previous_folder != folder_id, + moved_folder=previous_folder is not None and previous_folder != primary, ) return self.db.write(run) @@ -304,6 +340,7 @@ class Repository: """, (folder_id, aid, now, now), ) + self._recompute_primary_folder(conn, aid) return status return self.db.write(run) @@ -320,6 +357,7 @@ class Repository: "WHERE media_id = ? AND aid = ?", (now, media_id, aid), ) + self._recompute_primary_folder(conn, aid) other = conn.execute( """ SELECT 1 @@ -379,6 +417,25 @@ class Repository: return [int(row["aid"]) for row in rows] return self.db.read(run) + def pending_snapshot_aids(self, stat_date: str) -> list[int]: + """当天仍缺成功快照的追踪视频；用于新入库/恢复后的同日补采。""" + def run(conn: sqlite3.Connection) -> list[int]: + rows = conn.execute( + """ + SELECT v.aid + FROM video v + WHERE v.status IN ('active', 'unfaved') + AND NOT EXISTS ( + SELECT 1 FROM snapshot s + WHERE s.stat_date = ? AND s.aid = v.aid AND s.ok = 1 + ) + ORDER BY v.aid + """, + (stat_date,), + ).fetchall() + return [int(row["aid"]) for row in rows] + return self.db.read(run) + def get_snapshot(self, stat_date: str, aid: int) -> dict[str, Any] | None: def run(conn: sqlite3.Connection) -> dict[str, Any] | None: row = conn.execute( diff --git a/reaction_board/services/folder_sync.py b/reaction_board/services/folder_sync.py index d70f219..29dfab2 100644 --- a/reaction_board/services/folder_sync.py +++ b/reaction_board/services/folder_sync.py @@ -19,7 +19,6 @@ import time from dataclasses import dataclass, field from typing import Any -from ..bilibili.client import BilibiliClient from ..config import AppConfig, ConfigStore, FolderConfig from ..repository import Repository @@ -113,18 +112,29 @@ class FolderSyncService: self, config_store: ConfigStore, repository: Repository, - client: BilibiliClient, + client=None, *, + client_provider=None, clock=time.time, logger=None, capacity_warn: int = CAPACITY_WARN, + allow_empty_folders: bool = False, ) -> None: + if client is None and client_provider is None: + raise ValueError("必须提供 client 或 client_provider") self.config_store = config_store self.repo = repository self.client = client + self.client_provider = client_provider self._clock = clock self.logger = logger self.capacity_warn = capacity_warn + self.allow_empty_folders = allow_empty_folders + + def _client(self): + if self.client_provider is not None: + return self.client_provider.get() + return self.client def _now(self) -> int: return int(self._clock()) @@ -135,6 +145,7 @@ class FolderSyncService: def run(self, *, full_refresh: bool = False) -> SyncSummary: cfg = self.config_store.get() + client = self._client() now = self._now() summary = SyncSummary(started_at=now) self._log( @@ -151,7 +162,9 @@ class FolderSyncService: for folder in cfg.folders: try: - result = self._sync_folder(cfg, folder, now=now, force_full=full_refresh) + result = self._sync_folder( + client, cfg, folder, now=now, force_full=full_refresh + ) except Exception as exc: # noqa: BLE001 单个夹失败不阻塞其它夹 result = FolderSyncResult( media_id=folder.media_id, @@ -191,6 +204,7 @@ class FolderSyncService: def _sync_folder( self, + client, cfg: AppConfig, folder: FolderConfig, *, @@ -206,13 +220,20 @@ class FolderSyncService: >= cfg.schedule.full_folder_refresh_hours * 3600 ) - resources = self.client.fav_ids(folder.media_id) + resources = client.fav_ids(folder.media_id) remote_map = {item.aid: item for item in resources if item.type == 2} remote = set(remote_map) local = self.repo.folder_active_aids(folder.media_id) added = remote - local removed = local - remote + if not remote and local and not self.allow_empty_folders: + raise RuntimeError( + f"收藏夹 {folder.media_id} 远端返回 0 条，但本地仍有 {len(local)} 条；" + f"疑似接口异常，已跳过差集。确认收藏夹确实被清空后，" + f"可使用 --allow-empty-folder 重试" + ) + result = FolderSyncResult( media_id=folder.media_id, year=folder.year, @@ -228,7 +249,7 @@ class FolderSyncService: targets |= remote if targets: - infos = self.client.fav_infos( + infos = client.fav_infos( folder.media_id, sorted(targets), folder_mid=cfg.bilibili.read.mid, diff --git a/reaction_board/services/snapshot.py b/reaction_board/services/snapshot.py index 8ff6c69..f171e61 100644 --- a/reaction_board/services/snapshot.py +++ b/reaction_board/services/snapshot.py @@ -20,7 +20,6 @@ from typing import Any from ..bilibili.client import ( VIDEO_UNAVAILABLE_CODES, BilibiliAPIError, - BilibiliClient, BilibiliError, BilibiliHTTPError, ) @@ -41,6 +40,7 @@ class SnapshotSummary: removed: int = 0 fans_ok: int = 0 fans_failed: int = 0 + fans_skipped: int = 0 auth_ok: bool | None = None auth_message: str = "" backup_path: str | None = None @@ -63,6 +63,7 @@ class SnapshotSummary: "removed": self.removed, "fans_ok": self.fans_ok, "fans_failed": self.fans_failed, + "fans_skipped": self.fans_skipped, "auth_ok": self.auth_ok, "auth_message": self.auth_message, "backup_path": self.backup_path, @@ -78,8 +79,9 @@ class SnapshotService: self, config_store: ConfigStore, repository: Repository, - client: BilibiliClient, + client=None, *, + client_provider=None, clock=time.time, sleeper=time.sleep, logger=None, @@ -89,9 +91,12 @@ class SnapshotService: ) -> None: if retries < 1: raise ValueError("retries 必须 >= 1") + if client is None and client_provider is None: + raise ValueError("必须提供 client 或 client_provider") self.config_store = config_store self.repo = repository self.client = client + self.client_provider = client_provider self._clock = clock self._sleep = sleeper self.logger = logger @@ -102,6 +107,11 @@ class SnapshotService: def _now(self) -> int: return int(self._clock()) + def _client(self): + if self.client_provider is not None: + return self.client_provider.get() + return self.client + def _log(self, level: str, event: str, **fields) -> None: if self.logger is not None: getattr(self.logger, level)(event, **fields) @@ -110,6 +120,14 @@ class SnapshotService: def completed_key(stat_date: str) -> str: return f"snapshot:completed:{stat_date}" + @staticmethod + def backup_key(stat_date: str) -> str: + return f"snapshot:backup:{stat_date}" + + @staticmethod + def backup_error_key(stat_date: str) -> str: + return f"snapshot:backup_error:{stat_date}" + def run_if_needed( self, *, date: str | None = None, force: bool = False ) -> SnapshotSummary: @@ -117,7 +135,14 @@ class SnapshotService: stat_date = date or shanghai_date(now) if not is_date(stat_date): raise ValueError(f"日期格式必须是 YYYY-MM-DD: {stat_date!r}") - if not force and self.repo.kv_get(self.completed_key(stat_date)) == "1": + if force: + return self.run(date=stat_date, force=True) + + completed = self.repo.kv_get(self.completed_key(stat_date)) == "1" + pending = self.repo.pending_snapshot_aids(stat_date) + if completed and not pending: + if self.repo.kv_get(self.backup_key(stat_date)) != "1": + return self.backup_only(stat_date) self._log("info", "snapshot.skip", stat_date=stat_date, reason="already_completed") return SnapshotSummary( stat_date=stat_date, @@ -126,10 +151,27 @@ class SnapshotService: completed=True, already_completed=True, ) - return self.run(date=stat_date, force=force) + return self.run(date=stat_date, force=False) + + def backup_only(self, stat_date: str) -> SnapshotSummary: + """采集已完成但备份缺失时，只重试备份，不请求任何外部接口。""" + now = self._now() + cfg = self.config_store.get() + summary = SnapshotSummary( + stat_date=stat_date, + started_at=now, + completed=True, + already_completed=True, + ) + self._log("info", "snapshot.backup_retry", stat_date=stat_date) + self._run_backup(cfg, stat_date, summary) + summary.finished_at = self._now() + self._log("info", "snapshot.done", **summary.as_dict()) + return summary def run(self, *, date: str | None = None, force: bool = False) -> SnapshotSummary: cfg = self.config_store.get() + client = self._client() now = self._now() stat_date = date or shanghai_date(now) if not is_date(stat_date): @@ -137,8 +179,12 @@ class SnapshotService: summary = SnapshotSummary(stat_date=stat_date, started_at=now) self._log("info", "snapshot.start", stat_date=stat_date, force=force) + # 先落“未完成”，保证强制重采或中途中断后仍可续跑。 + self.repo.kv_set(self.completed_key(stat_date), "0") + if self.repo.kv_get("snapshot:last_completed") == stat_date: + self.repo.kv_set("snapshot:last_completed", "") - summary.auth_ok, summary.auth_message = self._check_auth(cfg, now) + summary.auth_ok, summary.auth_message = self._check_auth(client, cfg, now) videos = self.repo.video_aids_for_snapshot() summary.total = len(videos) @@ -148,7 +194,7 @@ class SnapshotService: summary.skipped += 1 continue - stat, error, unavailable = self._fetch_view(aid) + stat, error, unavailable = self._fetch_view(client, aid) if unavailable: summary.removed += 1 self.repo.set_video_status(aid, "removed", now=now) @@ -193,7 +239,15 @@ class SnapshotService: ) for mid in cfg.track_fans: - stat, error = self._fetch_fan(mid) + existing_fan = self.repo.get_up_stat(stat_date, mid) + if ( + not force + and existing_fan is not None + and int(existing_fan.get("ok") or 0) == 1 + ): + summary.fans_skipped += 1 + continue + stat, error = self._fetch_fan(client, mid) if stat is not None: self.repo.upsert_up_stat( stat_date, @@ -228,6 +282,21 @@ class SnapshotService: elif self.repo.kv_get("snapshot:last_completed") == stat_date: self.repo.kv_set("snapshot:last_completed", "") + self._run_backup(cfg, stat_date, summary) + + summary.finished_at = self._now() + self._log( + "info" if summary.completed else "warning", + "snapshot.done", + **summary.as_dict(), + ) + return summary + + # ------------------------------------------------------------ internals + + def _run_backup( + self, cfg: AppConfig, stat_date: str, summary: SnapshotSummary + ) -> None: try: path = backup_daily( self.repo.db, @@ -240,28 +309,24 @@ class SnapshotService: keep_days=cfg.storage.backup_keep_days, ) summary.backup_path = str(path) + self.repo.kv_set(self.backup_key(stat_date), "1") + self.repo.kv_set(self.backup_error_key(stat_date), "") except Exception as exc: # noqa: BLE001 备份失败不推翻已采集数据 summary.backup_error = str(exc) + self.repo.kv_set(self.backup_key(stat_date), "0") + self.repo.kv_set(self.backup_error_key(stat_date), str(exc)) self._log("error", "backup.failed", error=str(exc), stat_date=stat_date) - summary.finished_at = self._now() - self._log( - "info" if summary.completed else "warning", - "snapshot.done", - **summary.as_dict(), - ) - return summary - - # ------------------------------------------------------------ internals - - def _check_auth(self, cfg: AppConfig, now: int) -> tuple[bool, str]: + def _check_auth( + self, client, cfg: AppConfig, now: int + ) -> tuple[bool, str]: 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") return False, "未配置 cookie" try: - state = self.client.check_login() + state = client.check_login() except BilibiliError as exc: message = str(exc) self.repo.kv_set("auth:last_check", str(now)) @@ -282,11 +347,11 @@ class SnapshotService: ) return state.is_login, state.message - def _fetch_view(self, aid: int): + def _fetch_view(self, client, aid: int): last_error = "" for attempt in range(1, self.retries + 1): try: - return self.client.view(aid), None, False + return client.view(aid), None, False except BilibiliAPIError as exc: last_error = str(exc) if exc.code in VIDEO_UNAVAILABLE_CODES: @@ -301,11 +366,11 @@ class SnapshotService: self._sleep(self.backoff * (2 ** (attempt - 1))) return None, last_error or "未知错误", False - def _fetch_fan(self, mid: int): + def _fetch_fan(self, client, mid: int): last_error = "" for attempt in range(1, self.retries + 1): try: - return self.client.relation_stat(mid), None + return client.relation_stat(mid), None except BilibiliError as exc: last_error = str(exc) if attempt < self.retries: diff --git a/tests/test_bilibili_client.py b/tests/test_bilibili_client.py index 798a49d..0c7a2f2 100644 --- a/tests/test_bilibili_client.py +++ b/tests/test_bilibili_client.py @@ -361,6 +361,42 @@ def test_check_login_without_cookie_makes_no_request(): assert transport.calls == [] +def test_missing_data_is_protocol_error(): + for payload in ( + {"code": 0, "message": "OK"}, + {"code": 0, "message": "OK", "data": None}, + ): + transport = QueueTransport([http_json(payload)]) + client, _ = make_client(transport) + try: + client.fav_ids(1) + except BilibiliProtocolError as exc: + assert "data" in str(exc) + else: + raise AssertionError(f"ids 缺少 data 应报协议错误: {payload}") + + transport = QueueTransport([http_json(payload)]) + client, _ = make_client(transport) + try: + client.fav_infos(1, [123]) + except BilibiliProtocolError as exc: + assert "data" in str(exc) + else: + raise AssertionError(f"infos 缺少 data 应报协议错误: {payload}") + + +def test_empty_data_list_is_valid(): + transport = QueueTransport([http_json({"code": 0, "message": "OK", "data": []})]) + client, _ = make_client(transport) + assert client.fav_ids(4413147139) == [] + + transport = QueueTransport([http_json({"code": 0, "message": "OK", "data": []})]) + client, _ = make_client(transport) + result = client.fav_infos(1, [123, 456]) + assert result.items == {} + assert result.missing == (123, 456) + + def main(): test_fav_ids_parse_and_anonymous_header() test_anonymous_failure_falls_back_to_full_cookie() @@ -380,6 +416,8 @@ def main(): test_relation_stat() test_check_login() test_check_login_without_cookie_makes_no_request() + test_missing_data_is_protocol_error() + test_empty_data_list_is_valid() print("OK: B 站只读客户端契约与重试测试全部通过") diff --git a/tests/test_cli.py b/tests/test_cli.py index 626750f..3af56b2 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -51,7 +51,7 @@ def test_init_and_status(): path = write_config(tmp) code, output = run_cli(["init", "--config", str(path)]) assert code == 0 - assert '"schema_version": 1' in output + assert '"schema_version": 2' in output db = Database(Path(tmp) / "app.db") db.initialize() @@ -174,6 +174,41 @@ def test_run_once_offline(): assert selected and all(credentials.mid == 1 for credentials in selected) +def test_snapshot_backup_failure_returns_error_and_retries(): + import reaction_board.services.snapshot as snapshot_module + + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, cookie="", csrf="") + data = json.loads(path.read_text(encoding="utf-8")) + data["folders"] = [] + data["track_fans"] = [] + path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") + + original = snapshot_module.backup_daily + + def fail_backup(*args, **kwargs): + raise OSError("disk full") + + snapshot_module.backup_daily = fail_backup + try: + code, output = run_cli(["snapshot", "--config", str(path)]) + finally: + snapshot_module.backup_daily = original + assert code == 1 + marker = output.find("{\n") + assert marker >= 0, output + payload = json.loads(output[marker:]) + assert payload["completed"] is True + assert "disk full" in payload["backup_error"] + + code, output = run_cli(["snapshot", "--config", str(path)]) + assert code == 0 + marker = output.find("{\n") + payload = json.loads(output[marker:]) + assert payload["backup_error"] is None + assert Path(payload["backup_path"]).exists() + + def main_test(): test_init_and_status() test_missing_config() @@ -182,6 +217,7 @@ def main_test(): test_auth_check_without_cookie() test_auth_check_all_and_auth_dispatch() test_run_once_offline() + test_snapshot_backup_failure_returns_error_and_retries() print("OK: CLI init/sync/snapshot/auth-check/run/status/错误路径测试全部通过") diff --git a/tests/test_database.py b/tests/test_database.py index ee37229..f0d1054 100644 --- a/tests/test_database.py +++ b/tests/test_database.py @@ -24,7 +24,7 @@ def test_write_read_transaction(): with tempfile.TemporaryDirectory() as tmp: db = make_db(tmp) try: - assert db.schema_version == 1 + assert db.schema_version == 2 db.write(lambda c: c.execute("INSERT INTO kv(k, v) VALUES ('a', '1')")) assert db.read(lambda c: c.execute("SELECT v FROM kv WHERE k='a'").fetchone()[0]) == "1" @@ -94,8 +94,8 @@ def test_close_and_reinitialize(): db2 = make_db(tmp) try: - assert db2.schema_version == 1 - assert db2.read(lambda c: c.execute("SELECT MAX(version) FROM schema_version").fetchone()[0]) == 1 + assert db2.schema_version == 2 + assert db2.read(lambda c: c.execute("SELECT MAX(version) FROM schema_version").fetchone()[0]) == 2 finally: db2.close() @@ -123,7 +123,7 @@ def test_migration_backup_on_existing_db(): db = Database(path, backup_dir=Path(tmp) / "backup") try: applied = db.initialize() - assert applied == [1] + assert applied == [1, 2] assert len(list((Path(tmp) / "backup").glob("app-pre-v0-*.db"))) == 1 finally: db.close() diff --git a/tests/test_favsort_run.py b/tests/test_favsort_run.py new file mode 100644 index 0000000..00f62e2 --- /dev/null +++ b/tests/test_favsort_run.py @@ -0,0 +1,100 @@ +# -*- coding: utf-8 -*- +"""favsort run 参数校验与资源类型测试（不联网）: uv run python tests/test_favsort_run.py""" + +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 + + +class Args: + media_id = 1 + verify_every = 1 + + +def config(): + return { + "mid": 1, + "csrf": "csrf", + "cookie": "cookie", + "overrides": {}, + "sleep": { + "min": 0, "max": 0, + "long_every_min": 1000, "long_every_max": 1000, + "long_min": 0, "long_max": 0, + "retry_wait": 0, + }, + } + + +def test_zero_verify_every_rejected_before_writes(): + calls = [] + originals = (favsort.load_plan, favsort.fetch_ids, favsort.post_form) + favsort.load_plan = lambda media_id: {"target_order": [], "titles": {}} + favsort.fetch_ids = lambda cfg, media_id: calls.append("fetch") or [] + favsort.post_form = lambda *args, **kwargs: calls.append("post") or {} + args = Args() + args.verify_every = 0 + try: + code = favsort.cmd_run(config(), args) + finally: + favsort.load_plan, favsort.fetch_ids, favsort.post_form = originals + assert code == 2 + assert calls == [] + + +def test_sort_preserves_resource_types(): + with tempfile.TemporaryDirectory() as tmp: + plan = { + "media_id": 1, + "target_order": [1, 2], + "titles": {"1": {"title": "音频"}, "2": {"title": "视频"}}, + "types": {1: 2, 2: 12}, + } + bodies = [] + orders = [[2, 1]] + originals = ( + favsort.load_plan, + favsort.fetch_ids, + favsort.post_form, + favsort.plan_path, + favsort.journal_path, + ) + favsort.load_plan = lambda media_id: plan + favsort.plan_path = lambda media_id: str(Path(tmp) / "plan.json") + favsort.journal_path = lambda media_id: str(Path(tmp) / "journal.jsonl") + + def fetch_ids(cfg, media_id): + order = orders.pop(0) if orders else [1, 2] + return [{"id": aid} for aid in order] + + favsort.fetch_ids = fetch_ids + favsort.post_form = lambda cfg, url, body: bodies.append(body) or {"code": 0} + args = Args() + try: + code = favsort.cmd_run(config(), args) + finally: + ( + favsort.load_plan, + favsort.fetch_ids, + favsort.post_form, + favsort.plan_path, + favsort.journal_path, + ) = originals + assert code == 0 + assert [body["sort"] for body in bodies] == ["1:2:2:12"] + + +def main(): + test_zero_verify_every_rejected_before_writes() + test_sort_preserves_resource_types() + print("OK: favsort run 参数校验与资源类型测试全部通过") + + +if __name__ == "__main__": + main() diff --git a/tests/test_folder_sync.py b/tests/test_folder_sync.py index 40af003..7574670 100644 --- a/tests/test_folder_sync.py +++ b/tests/test_folder_sync.py @@ -38,10 +38,11 @@ def http_json(payload): class FakeFavTransport: - def __init__(self, folders, infos, *, fail_media=None): + def __init__(self, folders, infos, *, fail_media=None, omit_ids_data=None): self.folders = {k: list(v) for k, v in folders.items()} self.infos = dict(infos) self.fail_media = fail_media + self.omit_ids_data = omit_ids_data self.ids_calls = [] self.info_calls = [] @@ -52,6 +53,8 @@ class FakeFavTransport: self.ids_calls.append(media_id) if self.fail_media == media_id: raise TransportError("ids 网络错误") + if self.omit_ids_data == media_id: + return http_json({"code": 0, "message": "OK", "ttl": 1}) data_rows = [ {"id": aid, "type": 2, "bv_id": f"BV{aid}", "bvid": f"BV{aid}"} for aid in self.folders.get(media_id, []) @@ -284,6 +287,51 @@ def test_blacklist_seed_only_once(): db.close() +def test_empty_remote_requires_explicit_opt_in(): + with tempfile.TemporaryDirectory() as tmp: + transport = FakeFavTransport( + {F1: [1101, 1102], F2: []}, + {1101: valid(1101), 1102: valid(1102)}, + ) + clock = [1_800_000_000] + db, repo, service = make_service(tmp, transport, clock) + try: + service.run() + transport.folders[F1] = [] + clock[0] += 3600 + summary = service.run() + assert summary.ok is False + assert "疑似接口异常" in summary.folders[0].error + assert repo.get_video(1101)["status"] == "active" + assert repo.folder_active_aids(F1) == {1101, 1102} + + service.allow_empty_folders = True + summary = service.run() + assert summary.ok is True + assert repo.get_video(1101)["status"] == "unfaved" + assert repo.folder_active_aids(F1) == set() + finally: + db.close() + + +def test_missing_ids_data_aborts_folder(): + with tempfile.TemporaryDirectory() as tmp: + transport = FakeFavTransport( + {F1: [1101], F2: []}, + {1101: valid(1101)}, + omit_ids_data=F1, + ) + clock = [1_800_000_000] + db, repo, service = make_service(tmp, transport, clock) + try: + summary = service.run() + assert summary.ok is False + assert "data" in summary.folders[0].error + assert repo.get_video(1101) is None + finally: + db.close() + + def main(): test_initial_import_and_incremental_skip() test_diff_add_and_remove() @@ -291,6 +339,8 @@ def main(): test_capacity_warning_does_not_move_folder() test_folder_error_does_not_block_other_folders() test_blacklist_seed_only_once() + test_empty_remote_requires_explicit_opt_in() + test_missing_ids_data_aborts_folder() print("OK: 收藏夹初始导入、差集同步、失效识别与容量预警测试全部通过") diff --git a/tests/test_migrations.py b/tests/test_migrations.py index 50f36ff..7be6959 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -37,14 +37,18 @@ def test_fresh_and_idempotent(): conn = sqlite3.connect(Path(tmp) / "app.db") try: applied = run_migrations(conn) - assert applied == [1] - assert current_version(conn) == 1 + assert applied == [1, 2] + assert current_version(conn) == 2 assert EXPECTED_TABLES <= table_names(conn) assert "schema_version" in table_names(conn) + columns = { + row[1] for row in conn.execute("PRAGMA table_info(folder_resource)") + } + assert "fav_time" in columns # 再跑一次不应重复应用 assert run_migrations(conn) == [] row = conn.execute("SELECT COUNT(*) FROM schema_version").fetchone() - assert row[0] == 1 + assert row[0] == 2 finally: conn.close() @@ -59,15 +63,15 @@ def test_backup_before_upgrade(): conn.execute("INSERT INTO kv(k, v) VALUES ('hello', 'world')") conn.commit() - v2 = Migration(2, "add probe", ("CREATE TABLE probe (id INTEGER PRIMARY KEY)",)) + v3 = Migration(3, "add probe", ("CREATE TABLE probe (id INTEGER PRIMARY KEY)",)) applied = run_migrations( conn, backup_dir=backup_dir, now=1_800_000_000, - migrations=MIGRATIONS + (v2,), + migrations=MIGRATIONS + (v3,), ) - assert applied == [2] - backups = list(backup_dir.glob("app-pre-v1-*.db")) + assert applied == [3] + backups = list(backup_dir.glob("app-pre-v2-*.db")) assert len(backups) == 1 check = sqlite3.connect(backups[0]) try: diff --git a/tests/test_provider.py b/tests/test_provider.py new file mode 100644 index 0000000..cc64c06 --- /dev/null +++ b/tests/test_provider.py @@ -0,0 +1,106 @@ +# -*- coding: utf-8 -*- +"""凭据变化时客户端重建测试（不联网）: uv run python tests/test_provider.py""" + +import json +import os +import sys +import tempfile +import time +from pathlib import Path + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +sys.stdout.reconfigure(encoding="utf-8") + +from reaction_board.bilibili.client import BilibiliClient, Pacer # noqa: E402 +from reaction_board.bilibili.provider import ClientProvider # noqa: E402 +from reaction_board.config import ConfigStore # noqa: E402 + + +def write_config(path: Path, *, cookie: str, mid: int) -> None: + path.write_text(json.dumps({ + "bilibili": { + "read": {"mid": mid, "csrf": "read-csrf", "cookie": cookie}, + "write": {"mid": 2, "csrf": "write-csrf", "cookie": "write-cookie"}, + }, + "folders": [], + }, ensure_ascii=False), encoding="utf-8") + + +def bump(path: Path) -> None: + target = max(path.stat().st_mtime_ns, time.time_ns()) + 1_000_000 + os.utime(path, ns=(target, target)) + + +def test_rebuild_only_when_credentials_change(): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "config.json" + write_config(path, cookie="cookie-a", mid=1) + store = ConfigStore(path, env={}) + created = [] + + def factory(credentials): + created.append((credentials.mid, credentials.cookie)) + return BilibiliClient( + cookie=credentials.cookie, + csrf=credentials.csrf, + mid=credentials.mid, + pacer=Pacer(1.0, 1.6), + ) + + provider = ClientProvider(store, "read", factory) + first = provider.get() + assert provider.get() is first + assert created == [(1, "cookie-a")] + + write_config(path, cookie="cookie-b", mid=1) + bump(path) + second = provider.get() + assert second is not first + assert created == [(1, "cookie-a"), (1, "cookie-b")] + assert second.pacer.min_delay == 1.0 + assert second.pacer.max_delay == 1.6 + + # 仅 mtime 变化但凭据相同：不重建 + bump(path) + assert provider.get() is second + assert len(created) == 2 + + +def test_mid_change_rebuilds(): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "config.json" + write_config(path, cookie="cookie-a", mid=1) + store = ConfigStore(path, env={}) + provider = ClientProvider( + store, "read", + lambda credentials: BilibiliClient(mid=credentials.mid), + ) + assert provider.get().mid == 1 + write_config(path, cookie="cookie-a", mid=9) + bump(path) + assert provider.get().mid == 9 + + +def test_config_revision_increments(): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "config.json" + write_config(path, cookie="cookie-a", mid=1) + store = ConfigStore(path, env={}) + assert store.revision == 0 + write_config(path, cookie="cookie-b", mid=1) + bump(path) + store.get() + assert store.revision == 1 + store.get() + assert store.revision == 1 + + +def main(): + test_rebuild_only_when_credentials_change() + test_mid_change_rebuilds() + test_config_revision_increments() + print("OK: 凭据变化重建客户端与配置 revision 测试全部通过") + + +if __name__ == "__main__": + main() diff --git a/tests/test_repository.py b/tests/test_repository.py index d7a182d..98cade6 100644 --- a/tests/test_repository.py +++ b/tests/test_repository.py @@ -18,7 +18,7 @@ F1 = FolderConfig(year=2025, media_id=4451053851, title="2025") F2 = FolderConfig(year=2026, media_id=3700023351, title="2026") -def resource(aid, *, attr=0, up_mid=100, title="视频", bvid=None): +def resource(aid, *, attr=0, up_mid=100, title="视频", bvid=None, fav_time=None): return FavResource( aid=aid, bvid=bvid or f"BV{aid}", @@ -29,7 +29,7 @@ def resource(aid, *, attr=0, up_mid=100, title="视频", bvid=None): duration=600, cover="http://example/cover.jpg", attr=attr, - fav_time=None, + fav_time=fav_time, ) @@ -227,6 +227,72 @@ def test_folder_sync_metadata_and_counts(): db.close() +def test_primary_folder_is_order_independent(): + with tempfile.TemporaryDirectory() as tmp: + db, repo = setup(tmp) + try: + repo.upsert_from_fav(resource(1, fav_time=100), F1.media_id, was_added=True) + assert repo.get_video(1)["folder_id"] == F1.media_id + repo.upsert_from_fav(resource(1, fav_time=200), F2.media_id, was_added=True) + assert repo.get_video(1)["folder_id"] == F2.media_id + + # 全量刷新按相反顺序遍历，主归属仍应指向 fav_time 更新的夹 + repo.upsert_from_fav(resource(1, fav_time=100), F1.media_id, was_added=False) + assert repo.get_video(1)["folder_id"] == F2.media_id + repo.upsert_from_fav(resource(1, fav_time=200), F2.media_id, was_added=False) + assert repo.get_video(1)["folder_id"] == F2.media_id + + # 从主夹移除、另一个夹仍持有时，主归属切换 + status = repo.mark_resource_removed(F2.media_id, 1) + assert status == "active" + assert repo.get_video(1)["folder_id"] == F1.media_id + finally: + db.close() + + +def test_dropped_invalid_preserves_original_title(): + with tempfile.TemporaryDirectory() as tmp: + db, repo = setup(tmp) + try: + repo.upsert_from_fav( + resource(1, title="原始标题"), F1.media_id, was_added=True + ) + repo.set_video_status(1, "dropped") + repo.upsert_from_fav( + resource(1, title="已失效视频", attr=9), + F1.media_id, + was_added=False, + ) + video = repo.get_video(1) + assert video["status"] == "dropped" + assert video["title"] == "原始标题" + finally: + db.close() + + +def test_pending_snapshot_aids(): + with tempfile.TemporaryDirectory() as tmp: + db, repo = setup(tmp) + try: + repo.upsert_from_fav(resource(1), F1.media_id, was_added=True) + repo.upsert_from_fav(resource(2), F1.media_id, was_added=True) + repo.upsert_from_fav(resource(3), F1.media_id, was_added=True) + repo.set_video_status(3, "unfaved") + repo.upsert_snapshot( + "2026-10-31", 1, + metrics={"view": 1, "like": 1, "favorite": 1, "coin": 1, + "share": 1, "reply": 1, "danmaku": 1}, + ok=True, err=None, captured_at=1, + ) + repo.upsert_snapshot( + "2026-10-31", 2, + metrics=None, ok=False, err="x", captured_at=1, + ) + assert repo.pending_snapshot_aids("2026-10-31") == [2, 3] + finally: + db.close() + + def main(): test_upsert_and_folder_resource() test_invalid_attr_and_missing() @@ -236,6 +302,9 @@ def main(): test_multi_folder_moves_primary_and_warns() test_blacklist_seed_idempotent() test_folder_sync_metadata_and_counts() + test_primary_folder_is_order_independent() + test_dropped_invalid_preserves_original_title() + test_pending_snapshot_aids() print("OK: 收藏夹/视频状态机与黑名单测试全部通过") diff --git a/tests/test_snapshot.py b/tests/test_snapshot.py index e87a013..7b11b47 100644 --- a/tests/test_snapshot.py +++ b/tests/test_snapshot.py @@ -60,6 +60,8 @@ class SnapshotTransport: urllib.parse.urlparse(url).query )["aid"][0]) payload = self.views[aid] + if isinstance(payload, BaseException): + raise payload if isinstance(payload, HTTPResponse): return payload return http_json(payload) @@ -316,6 +318,115 @@ def test_force_failure_clears_last_completed(): db.close() +def test_new_video_after_completion_backfills_same_day(): + with tempfile.TemporaryDirectory() as tmp: + transport = SnapshotTransport({1: view_for(1, view=100)}) + db, repo, service = make_service(tmp, transport, track_fans=()) + try: + add_video(repo, 1) + first = service.run(date=DATE) + assert first.completed is True + assert transport.count("/web-interface/view") == 1 + + # 当天 11:00 新入库：完成标记仍是 1，但必须补采新视频 + add_video(repo, 2) + transport.views[2] = view_for(2, view=222) + second = service.run_if_needed(date=DATE) + assert second.already_completed is False + assert second.success == 1 + assert second.skipped == 1 + assert transport.count("/web-interface/view") == 2 + assert repo.get_snapshot(DATE, 2)["view"] == 222 + assert repo.get_snapshot(DATE, 1)["view"] == 100 + finally: + db.close() + + +def test_fan_resume_preserves_successful_value(): + with tempfile.TemporaryDirectory() as tmp: + transport = SnapshotTransport( + {1: {"code": -500, "message": "x", "data": None}}, + ) + db, repo, service = make_service(tmp, transport, retries=1) + try: + add_video(repo, 1) + first = service.run(date=DATE) + assert first.failed == 1 + assert first.fans_ok == 1 + assert repo.get_up_stat(DATE, 37754047)["follower"] == 2767268 + + transport.views[1] = view_for(1, view=111) + transport.fan = {"code": -500, "message": "x", "data": None} + second = service.run(date=DATE) + assert second.fans_skipped == 1 + assert second.fans_failed == 0 + fan = repo.get_up_stat(DATE, 37754047) + assert fan["follower"] == 2767268 + assert fan["ok"] == 1 + finally: + db.close() + + +def test_force_interrupt_leaves_incomplete_marker(): + with tempfile.TemporaryDirectory() as tmp: + transport = SnapshotTransport({ + 1: view_for(1), + 2: RuntimeError("simulated interrupt"), + }) + db, repo, service = make_service(tmp, transport, track_fans=()) + try: + add_video(repo, 1) + add_video(repo, 2) + repo.kv_set(service.completed_key(DATE), "1") + repo.kv_set("snapshot:last_completed", DATE) + try: + service.run(date=DATE, force=True) + except RuntimeError: + pass + else: + raise AssertionError("应模拟中断") + assert repo.kv_get(service.completed_key(DATE)) == "0" + assert repo.kv_get("snapshot:last_completed") == "" + + transport.views[2] = view_for(2) + resumed = service.run_if_needed(date=DATE) + assert resumed.completed is True + assert resumed.success == 1 + assert resumed.skipped == 1 + finally: + db.close() + + +def test_backup_failure_retries_without_api(): + import reaction_board.services.snapshot as snapshot_module + + with tempfile.TemporaryDirectory() as tmp: + transport = SnapshotTransport({1: view_for(1)}) + db, repo, service = make_service(tmp, transport, track_fans=()) + original = snapshot_module.backup_daily + + def fail_backup(*args, **kwargs): + raise OSError("disk full") + + snapshot_module.backup_daily = fail_backup + try: + first = service.run(date=DATE) + finally: + snapshot_module.backup_daily = original + assert first.completed is True + assert first.backup_error + assert repo.kv_get(service.backup_key(DATE)) == "0" + view_calls = transport.count("/web-interface/view") + + retried = service.run_if_needed(date=DATE) + assert retried.already_completed is True + assert retried.backup_error is None + assert Path(retried.backup_path or "").exists() + assert repo.kv_get(service.backup_key(DATE)) == "1" + assert transport.count("/web-interface/view") == view_calls + db.close() + + def main(): test_snapshot_success_resume_and_force() test_resume_skips_ok_and_retries_failed() @@ -325,6 +436,10 @@ def main(): test_auth_invalid_is_warned_but_snapshot_continues() test_no_cookie_skips_nav_request() test_force_failure_clears_last_completed() + test_new_video_after_completion_backfills_same_day() + test_fan_resume_preserves_successful_value() + test_force_interrupt_leaves_incomplete_marker() + test_backup_failure_retries_without_api() print("OK: 每日快照、断点续跑、失败占位与粉丝采集测试全部通过")