feat(up): capture follower snapshots at ingest and backfill history

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":

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),
             )
+            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)

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
 

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,
+        )

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__":

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()

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()

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()