From 519fc39fca16a89562208cc93293d5670da9e745 Mon Sep 17 00:00:00 2001 From: kaireewu Date: Sun, 1 Nov 2026 11:18:06 +0800 Subject: [PATCH] feat(up): capture follower snapshots at ingest and backfill history --- reaction_board/cli.py | 36 +++++++ reaction_board/repository.py | 37 +++++-- reaction_board/services/folder_sync.py | 33 ++++++- reaction_board/services/up_profile.py | 132 +++++++++++++++++++++++++ tests/test_cli.py | 18 +++- tests/test_folder_sync.py | 17 ++++ tests/test_repository.py | 10 +- tests/test_up_profile.py | 132 +++++++++++++++++++++++++ 8 files changed, 401 insertions(+), 14 deletions(-) create mode 100644 reaction_board/services/up_profile.py create mode 100644 tests/test_up_profile.py diff --git a/reaction_board/cli.py b/reaction_board/cli.py index 2357bf0..e47a43e 100644 --- a/reaction_board/cli.py +++ b/reaction_board/cli.py @@ -39,6 +39,7 @@ from .services.favorites import FavoriteActionService from .services.folder_sync import FolderSyncService from .services.snapshot import SnapshotService from .services.telegram_poller import TelegramPoller +from .services.up_profile import UpProfileService from .timeutil import next_daily_run, parse_hhmm, shanghai_date ROOT = Path(__file__).resolve().parent.parent @@ -120,6 +121,12 @@ def _build_parser() -> argparse.ArgumentParser: "--preview", action="store_true", help="以预览模式启动（要求数据库含 preview:seeded=1 标记）", ) + backfill = sub.add_parser( + "backfill-followers", parents=[common], + help="为历史视频回填 UP 粉丝数快照（只补空值）", + ) + backfill.add_argument("--limit", type=int, default=None, help="最多处理多少个 UP") + backfill.add_argument("--dry-run", action="store_true", help="只统计待回填数量，不发请求") sub.add_parser("status", parents=[common], help="查看数据库状态（离线）") return parser @@ -226,6 +233,33 @@ def cmd_sync(args, logger: JsonLogger) -> int: db.close() +def cmd_backfill_followers(args, logger: JsonLogger) -> int: + store = ConfigStore(args.config, require_auth=False) + cfg = store.get() + db = _open_db(cfg, logger) + try: + repo = Repository(db) + client = _client( + cfg.bilibili.read, logger, pacer=Pacer(1.0, 1.6), retries=2 + ) + service = UpProfileService(repo, lambda: client, logger=logger) + started = int(time.time()) + result = service.backfill(limit=args.limit, dry_run=args.dry_run) + payload = result.as_dict() + payload["dry_run"] = bool(args.dry_run) + payload["coverage"] = repo.up_follower_coverage() + payload["database"] = str(cfg.storage.database_path) + _record_task( + repo, "backfill_followers", ok=result.failed == 0, + started_at=started, finished_at=int(time.time()), payload=payload, + error="；".join(result.errors) if result.errors else None, + ) + _print(payload) + return 0 if result.failed == 0 else 1 + finally: + db.close() + + def cmd_snapshot(args, logger: JsonLogger) -> int: store = ConfigStore(args.config, require_auth=False) cfg = store.get() @@ -673,6 +707,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 == "backfill-followers": + return cmd_backfill_followers(args, logger) if args.command == "discover": return cmd_discover(args, logger) if args.command == "candidates": diff --git a/reaction_board/repository.py b/reaction_board/repository.py index ffb451a..ff4fdd5 100644 --- a/reaction_board/repository.py +++ b/reaction_board/repository.py @@ -410,12 +410,14 @@ class Repository: following: int | None = None, now: int | None = None, backfilled: bool = False, + live_since: int | None = None, ) -> int: """写入 UP 粉丝快照，并补齐该 UP 尚未记录粉丝数的视频。 ``follower_first`` 只在首次插入时写入；``follower_latest`` 每次刷新都会 - 更新。视频上的 ``up_follower`` 是入库时快照，已有值不会被覆盖。返回本次 - 补齐的视频数。 + 更新。视频上的 ``up_follower`` 是入库时快照，已有值不会被覆盖。传入 + ``live_since`` 时，只有 ``first_seen >= live_since`` 的视频视为实时采集， + 更早的视频标记为回填。返回本次补齐的视频数。 """ now = self._now() if now is None else int(now) mid = int(mid) @@ -436,14 +438,29 @@ class Repository: """, (mid, name, follower, following_value, now, follower, now), ) - cursor = conn.execute( - """ - UPDATE video - SET up_follower = ?, up_follower_at = ?, up_follower_backfilled = ? - WHERE up_mid = ? AND up_follower IS NULL - """, - (follower, now, backfill_flag, mid), - ) + if live_since is None: + cursor = conn.execute( + """ + UPDATE video + SET up_follower = ?, up_follower_at = ?, up_follower_backfilled = ? + WHERE up_mid = ? AND up_follower IS NULL + AND status IN ('active', 'unfaved') + """, + (follower, now, backfill_flag, mid), + ) + else: + cursor = conn.execute( + """ + UPDATE video + SET up_follower = ?, up_follower_at = ?, + up_follower_backfilled = CASE + WHEN first_seen >= ? THEN 0 ELSE 1 + END + WHERE up_mid = ? AND up_follower IS NULL + AND status IN ('active', 'unfaved') + """, + (follower, now, int(live_since), mid), + ) return max(cursor.rowcount, 0) return self.db.write(run) diff --git a/reaction_board/services/folder_sync.py b/reaction_board/services/folder_sync.py index b14450a..f41c7db 100644 --- a/reaction_board/services/folder_sync.py +++ b/reaction_board/services/folder_sync.py @@ -21,6 +21,7 @@ from typing import Any from ..config import AppConfig, ConfigStore, FolderConfig from ..repository import Repository +from .up_profile import UpProfileService CAPACITY_WARN = 950 @@ -61,6 +62,8 @@ class SyncSummary: finished_at: int = 0 folders: list[FolderSyncResult] = field(default_factory=list) blacklist_seeded: int = 0 + followers_updated: int = 0 + followers_failed: int = 0 @property def added(self) -> int: @@ -100,6 +103,8 @@ class SyncSummary: "refreshed": self.refreshed, "missing": self.missing, "blacklist_seeded": self.blacklist_seeded, + "followers_updated": self.followers_updated, + "followers_failed": self.followers_failed, "folders": [f.as_dict() for f in self.folders], "errors": self.errors, "warnings": self.warnings, @@ -119,6 +124,8 @@ class FolderSyncService: logger=None, capacity_warn: int = CAPACITY_WARN, allow_empty_folders: bool = False, + up_profile_service: UpProfileService | None = None, + capture_followers: bool = True, ) -> None: if client is None and client_provider is None: raise ValueError("必须提供 client 或 client_provider") @@ -130,6 +137,10 @@ class FolderSyncService: self.logger = logger self.capacity_warn = capacity_warn self.allow_empty_folders = allow_empty_folders + self.capture_followers = capture_followers + self.up_profile = up_profile_service or UpProfileService( + repository, self._client, clock=clock, logger=logger + ) def _client(self): if self.client_provider is not None: @@ -189,6 +200,26 @@ class FolderSyncService: ) summary.folders.append(result) + if self.capture_followers: + try: + missing = self.repo.up_mids_missing_follower() + if missing: + capture = self.up_profile.capture( + [int(row["up_mid"]) for row in missing], + names={ + int(row["up_mid"]): str(row.get("up_name") or "") + for row in missing + }, + live_since=summary.started_at - 3600, + ) + summary.followers_updated = capture.updated_videos + summary.followers_failed = capture.failed + except Exception as exc: # noqa: BLE001 粉丝采集失败不影响收藏夹同步 + summary.followers_failed += 1 + self._log( + "warning", "folder_sync.followers_failed", error=str(exc) + ) + summary.finished_at = self._now() self._log( "info" if summary.ok else "error", @@ -200,7 +231,7 @@ class FolderSyncService: errors=len(summary.errors), elapsed=max(0, summary.finished_at - summary.started_at), ) - if summary.added or summary.removed or summary.refreshed: + if summary.added or summary.removed or summary.refreshed or summary.followers_updated: self.repo.bump_cache_version("videos") return summary diff --git a/reaction_board/services/up_profile.py b/reaction_board/services/up_profile.py new file mode 100644 index 0000000..1c665b1 --- /dev/null +++ b/reaction_board/services/up_profile.py @@ -0,0 +1,132 @@ +# -*- coding: utf-8 -*- +"""UP 粉丝快照采集。 + +视频的播放、互动记在 bvid 上，而 UP 的粉丝数是视频之外的受众基数。为了让以后 +能计算“播放 / 入库时粉丝数”，本服务在视频入库时记录一次粉丝快照，并支持对历史 +数据做一次性回填。 + +``up_profile.follower_first`` 永远保留首次采集值；``video.up_follower`` 保留每个 +视频入库时的值，不会被后续刷新覆盖。 +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass, field +from typing import Any, Callable, Iterable, Mapping + +from ..repository import Repository + + +@dataclass +class CaptureResult: + requested: int = 0 + updated_videos: int = 0 + failed: int = 0 + errors: list[str] = field(default_factory=list) + + def as_dict(self) -> dict[str, Any]: + return { + "requested": self.requested, + "updated_videos": self.updated_videos, + "failed": self.failed, + "errors": self.errors, + } + + +class UpProfileService: + def __init__( + self, + repository: Repository, + client_provider: Callable[[], Any], + *, + clock=time.time, + logger=None, + ) -> None: + self.repo = repository + self.client_provider = client_provider + self._clock = clock + self.logger = logger + + def _now(self) -> int: + return int(self._clock()) + + def _log(self, level: str, event: str, **fields) -> None: + if self.logger is not None: + getattr(self.logger, level)(event, **fields) + + def capture( + self, + mids: Iterable[int], + *, + names: Mapping[int, str] | None = None, + live_since: int | None = None, + backfilled: bool = False, + ) -> CaptureResult: + """采集一组 UP 的当前粉丝数，并补齐其视频的空快照。 + + 同一个 mid 在一次调用里只请求一次；单个失败不会中断其余 UP。 + """ + ordered: list[int] = [] + seen: set[int] = set() + for mid in mids: + if mid is None or isinstance(mid, bool): + continue + try: + value = int(mid) + except (TypeError, ValueError): + continue + if value <= 0 or value in seen: + continue + seen.add(value) + ordered.append(value) + + result = CaptureResult(requested=len(ordered)) + if not ordered: + return result + now = self._now() + names = names or {} + for mid in ordered: + try: + fan = self.client_provider().relation_stat(mid) + updated = self.repo.apply_up_follower( + mid=mid, + name=names.get(mid) or getattr(fan, "uname", None), + follower=int(fan.follower), + following=int(getattr(fan, "following", 0) or 0), + now=now, + backfilled=backfilled, + live_since=live_since, + ) + except Exception as exc: # noqa: BLE001 单个 UP 失败不阻塞同步 + result.failed += 1 + result.errors.append(f"{mid}: {exc}") + self._log( + "warning", "up_profile.capture_failed", mid=mid, error=str(exc) + ) + else: + result.updated_videos += updated + self._log( + "info", "up_profile.captured", + mid=mid, follower=int(fan.follower), updated=updated, + ) + return result + + def backfill( + self, *, limit: int | None = None, dry_run: bool = False + ) -> CaptureResult: + """为历史视频回填粉丝快照；已有值的视频不会被覆盖。""" + rows = self.repo.up_mids_missing_follower() + if limit is not None: + rows = rows[: max(0, int(limit))] + if dry_run: + return CaptureResult(requested=len(rows)) + names = { + int(row["up_mid"]): str(row.get("up_name") or "") + for row in rows + } + return self.capture( + [int(row["up_mid"]) for row in rows], + names=names, + backfilled=True, + ) diff --git a/tests/test_cli.py b/tests/test_cli.py index 2a3b78e..a7e3c4b 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": 3' in output + assert '"schema_version": 4' in output db = Database(Path(tmp) / "app.db") db.initialize() @@ -319,6 +319,19 @@ def test_candidates_and_discover_commands(): assert '"notified": 1' in output +def test_backfill_followers_dry_run(): + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, cookie="", csrf="") + code, output = run_cli([ + "backfill-followers", "--dry-run", "--config", str(path), + ]) + assert code == 0 + assert '"requested": 0' in output + assert '"failed": 0' in output + assert '"dry_run": true' in output + assert '"total": 0' in output + + def main_test(): test_init_and_status() test_missing_config() @@ -331,7 +344,8 @@ def main_test(): test_default_command_starts_web_and_scheduler() test_snapshot_backup_failure_returns_error_and_retries() test_candidates_and_discover_commands() - print("OK: CLI init/sync/snapshot/auth-check/run/status/错误路径测试全部通过") + test_backfill_followers_dry_run() + print("OK: CLI init/sync/snapshot/auth-check/run/status/backfill/错误路径测试全部通过") if __name__ == "__main__": diff --git a/tests/test_folder_sync.py b/tests/test_folder_sync.py index aa51687..17f3cc0 100644 --- a/tests/test_folder_sync.py +++ b/tests/test_folder_sync.py @@ -45,9 +45,17 @@ class FakeFavTransport: self.omit_ids_data = omit_ids_data self.ids_calls = [] self.info_calls = [] + self.relation_calls = [] def request(self, method, url, headers, data=None, timeout=30): query = urllib.parse.parse_qs(urllib.parse.urlparse(url).query) + if "/x/relation/stat" in url: + mid = int(query["vmid"][0]) + self.relation_calls.append(mid) + return http_json({ + "code": 0, "message": "0", "ttl": 1, + "data": {"mid": mid, "following": 5, "follower": 1000 + mid}, + }) if "/x/v3/fav/resource/ids" in url: media_id = int(query["media_id"][0]) self.ids_calls.append(media_id) @@ -165,6 +173,13 @@ def test_initial_import_and_incremental_skip(): assert repo.get_video(1104)["bvid"] == "BV1104" assert repo.folder_active_aids(F1) == {1101, 1102, 1103, 1104} assert repo.is_blacklisted(999) + # 新入库视频记录入库时粉丝数；失效视频不再采集 + assert summary.followers_updated == 2 + assert summary.followers_failed == 0 + assert repo.get_video(1101)["up_follower"] == 1100 + assert repo.get_video(1101)["up_follower_backfilled"] == 0 + assert repo.get_video(1102)["up_follower"] == 1999 + assert sorted(transport.relation_calls) == [100, 999] folder = repo.get_folder(F1) assert folder.media_count == 4 @@ -173,10 +188,12 @@ def test_initial_import_and_incremental_skip(): # 6 小时内再次运行：没有新增，不请求 infos clock[0] += 60 info_before = len(transport.info_calls) + relation_before = len(transport.relation_calls) summary2 = service.run() assert summary2.added == 0 assert summary2.refreshed == 0 assert len(transport.info_calls) == info_before + assert len(transport.relation_calls) == relation_before assert summary2.folders[0].full_refresh is False finally: db.close() diff --git a/tests/test_repository.py b/tests/test_repository.py index fb4a5b0..6e39dd1 100644 --- a/tests/test_repository.py +++ b/tests/test_repository.py @@ -339,9 +339,17 @@ def test_up_follower_snapshot_and_backfill(): assert repo.get_video(3)["up_follower"] == 88 assert repo.get_video(3)["up_follower_backfilled"] == 1 + # live_since 之前的视频标为回填，之后的视为实时采集 + repo.upsert_from_fav(resource(5, up_mid=300), F1.media_id, was_added=True) + repo.apply_up_follower( + mid=300, name="UP3", follower=50, now=now + 180, + live_since=now + 1, + ) + assert repo.get_video(5)["up_follower_backfilled"] == 1 + coverage = repo.up_follower_coverage() assert coverage == { - "total": 4, "covered": 4, "missing": 0, "ups_covered": 2, + "total": 5, "covered": 5, "missing": 0, "ups_covered": 3, } finally: db.close() diff --git a/tests/test_up_profile.py b/tests/test_up_profile.py new file mode 100644 index 0000000..c7b73e0 --- /dev/null +++ b/tests/test_up_profile.py @@ -0,0 +1,132 @@ +# -*- coding: utf-8 -*- +"""UP 粉丝快照采集测试（不联网）: uv run python tests/test_up_profile.py""" + +import os +import sys +import tempfile +from pathlib import Path +from types import SimpleNamespace + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +sys.stdout.reconfigure(encoding="utf-8") + +from reaction_board.bilibili.models import FavResource # noqa: E402 +from reaction_board.config import FolderConfig # noqa: E402 +from reaction_board.database import Database # noqa: E402 +from reaction_board.repository import Repository # noqa: E402 +from reaction_board.services.up_profile import UpProfileService # noqa: E402 + +FOLDER = FolderConfig(year=2026, media_id=3700023351, title="2026") +NOW = 1_800_000_000 + + +def resource(aid, up_mid): + return FavResource( + aid=aid, + bvid=f"BV{aid}", + title=f"视频{aid}", + up_mid=up_mid, + up_name=f"UP{up_mid}", + pubdate=NOW - 86400, + duration=7200, + cover=None, + attr=0, + fav_time=None, + pages=2, + ) + + +class FakeClient: + def __init__(self, *, failing=()): + self.calls = [] + self.failing = set(failing) + + def relation_stat(self, mid): + self.calls.append(int(mid)) + if int(mid) in self.failing: + raise RuntimeError("simulated network error") + return SimpleNamespace(mid=int(mid), follower=1000 + int(mid), following=5) + + +def setup(tmp): + db = Database(Path(tmp) / "app.db") + db.initialize() + repo = Repository(db, clock=lambda: NOW) + repo.ensure_folder(FOLDER) + for aid, mid in ((1, 100), (2, 100), (3, 200)): + repo.upsert_from_fav(resource(aid, mid), FOLDER.media_id, was_added=True) + return db, repo + + +def test_capture_dedupes_and_continues_after_failure(): + with tempfile.TemporaryDirectory() as tmp: + db, repo = setup(tmp) + try: + client = FakeClient(failing={200}) + service = UpProfileService(repo, lambda: client, clock=lambda: NOW) + result = service.capture( + [100, 100, 200, None, 0], live_since=NOW - 3600 + ) + assert result.requested == 2 + assert client.calls == [100, 200] + assert result.updated_videos == 2 + assert result.failed == 1 + assert "200" in result.errors[0] + + assert repo.get_video(1)["up_follower"] == 1100 + assert repo.get_video(2)["up_follower"] == 1100 + assert repo.get_video(3)["up_follower"] is None + assert repo.get_up_profile(100)["follower_first"] == 1100 + assert repo.get_up_profile(200) is None + finally: + db.close() + + +def test_capture_marks_old_videos_as_backfilled(): + with tempfile.TemporaryDirectory() as tmp: + db, repo = setup(tmp) + try: + client = FakeClient() + service = UpProfileService(repo, lambda: client, clock=lambda: NOW) + service.capture([100], live_since=NOW + 1) + assert repo.get_video(1)["up_follower_backfilled"] == 1 + assert repo.get_video(2)["up_follower_backfilled"] == 1 + finally: + db.close() + + +def test_backfill_dry_run_and_limit(): + with tempfile.TemporaryDirectory() as tmp: + db, repo = setup(tmp) + try: + client = FakeClient() + service = UpProfileService(repo, lambda: client, clock=lambda: NOW) + + dry = service.backfill(dry_run=True) + assert dry.requested == 2 + assert client.calls == [] + assert repo.up_follower_coverage()["missing"] == 3 + + limited = service.backfill(limit=1) + assert limited.requested == 1 + assert client.calls == [100] + assert repo.get_video(1)["up_follower_backfilled"] == 1 + assert repo.get_video(3)["up_follower"] is None + + rest = service.backfill() + assert rest.requested == 1 + assert client.calls == [100, 200] + assert repo.up_follower_coverage()["missing"] == 0 + finally: + db.close() + + +def main(): + test_capture_dedupes_and_continues_after_failure() + test_capture_marks_old_videos_as_backfilled() + test_backfill_dry_run_and_limit() + print("OK: UP 粉丝快照采集与回填测试全部通过") + + +if __name__ == "__main__": + main()