diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index 5ab3ada..1407e18 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -104,6 +104,9 @@ uv run python app.py snapshot --force uv run python app.py backfill-followers uv run python app.py backfill-followers --dry-run +# 全量同步默认会在结尾采集缺失粉丝数；可用 --no-capture-followers 跳过 +uv run python app.py sync --full --no-capture-followers + # 检查 Cookie 登录态（P1） uv run python app.py auth-check @@ -734,7 +737,8 @@ uv run python tests/test_moves.py ### 第 9 轮：时长效率榜、粉丝快照与导航拆分 ✅（2026-11-01） - 迁移 v4：`video.pages`、`video.up_follower` / `up_follower_at` / `up_follower_backfilled`、`up_profile`； -- 收藏夹同步结束时采集缺失 UP 的粉丝数（`relation/stat`，匿名可用）；`first_seen >= 运行开始 - 1h` 记为实时，更早的记为回填； +- 收藏夹同步结束时采集缺失 UP 的粉丝数（`relation/stat`，匿名可用）；`first_seen >= 运行开始 - 1h` 记为实时，更早的记为回填；`sync --no-capture-followers` 可跳过； +- 粉丝采集对每条 `relation/stat` 加 45s 硬超时看门狗（防 DNS/传输层挂起），每 25 条输出 `up_profile.progress` 进度；单条失败不阻塞整批； - `app.py backfill-followers`：一次性回填历史视频粉丝数，支持 `--limit` / `--dry-run`； - 新增内容效率/互动效率榜与 UP 内容效率（Gamma–Poisson EB 收缩），过滤 < 600 秒与未满 2 天的视频； - `analytics` 配置段：`min_duration_seconds`、`maturity_days`、`min_view_for_ranking`、`video_shrink_minutes`、`shrink_minutes`； diff --git a/README.md b/README.md index 07b7bce..e2ced48 100644 --- a/README.md +++ b/README.md @@ -103,6 +103,9 @@ uv run python app.py snapshot --force uv run python app.py backfill-followers uv run python app.py backfill-followers --dry-run +# 全量同步默认也会在结尾采集缺失的 UP 粉丝数；如只想刷新元数据可跳过 +uv run python app.py sync --full --no-capture-followers + # 分别为只读和写入账号扫码授权（默认 read） uv run python app.py auth uv run python app.py auth --account write diff --git a/reaction_board/cli.py b/reaction_board/cli.py index e47a43e..e87b882 100644 --- a/reaction_board/cli.py +++ b/reaction_board/cli.py @@ -75,6 +75,11 @@ def _build_parser() -> argparse.ArgumentParser: action="store_true", help="允许把远端 0 条视为真实清空（默认按异常跳过差集）", ) + sync.add_argument( + "--no-capture-followers", + action="store_true", + help="跳过同步末尾的 UP 粉丝采集（改用 backfill-followers 单独回填）", + ) snapshot = sub.add_parser( "snapshot", parents=[common], help="采集当日七项快照与粉丝数" ) @@ -217,6 +222,7 @@ def cmd_sync(args, logger: JsonLogger) -> int: client, logger=logger, allow_empty_folders=args.allow_empty_folder, + capture_followers=not getattr(args, "no_capture_followers", False), ) started = int(time.time()) summary = service.run(full_refresh=args.full) diff --git a/reaction_board/services/up_profile.py b/reaction_board/services/up_profile.py index 1c665b1..c63cb9f 100644 --- a/reaction_board/services/up_profile.py +++ b/reaction_board/services/up_profile.py @@ -11,6 +11,7 @@ from __future__ import annotations +import threading import time from dataclasses import dataclass, field from typing import Any, Callable, Iterable, Mapping @@ -42,11 +43,15 @@ class UpProfileService: *, clock=time.time, logger=None, + request_timeout: float = 45.0, + heartbeat: int = 25, ) -> None: self.repo = repository self.client_provider = client_provider self._clock = clock self.logger = logger + self.request_timeout = max(1.0, float(request_timeout)) + self.heartbeat = max(1, int(heartbeat)) def _now(self) -> int: return int(self._clock()) @@ -55,6 +60,32 @@ class UpProfileService: if self.logger is not None: getattr(self.logger, level)(event, **fields) + def _fetch_relation(self, mid: int): + """带硬超时的单条请求。 + + 正常网络错误已由 ``BilibiliClient`` 的 timeout/retry 覆盖；但 DNS 解析或 + 传输层偶发挂起可能不触发 timeout，这里用工作线程兜底，避免整批任务静默 + 卡死。 + """ + box: dict[str, Any] = {} + + def work() -> None: + try: + box["value"] = self.client_provider().relation_stat(mid) + except BaseException as exc: # noqa: BLE001 结果由主线程统一处理 + box["error"] = exc + + worker = threading.Thread(target=work, name=f"relation-{mid}", daemon=True) + worker.start() + worker.join(self.request_timeout) + if worker.is_alive(): + raise TimeoutError( + f"relation/stat timeout after {self.request_timeout:.0f}s" + ) + if "error" in box: + raise box["error"] + return box["value"] + def capture( self, mids: Iterable[int], @@ -86,9 +117,10 @@ class UpProfileService: return result now = self._now() names = names or {} - for mid in ordered: + total = len(ordered) + for index, mid in enumerate(ordered, 1): try: - fan = self.client_provider().relation_stat(mid) + fan = self._fetch_relation(mid) updated = self.repo.apply_up_follower( mid=mid, name=names.get(mid) or getattr(fan, "uname", None), @@ -110,6 +142,11 @@ class UpProfileService: "info", "up_profile.captured", mid=mid, follower=int(fan.follower), updated=updated, ) + if index % self.heartbeat == 0 or index == total: + self._log( + "info", "up_profile.progress", + processed=index, total=total, failed=result.failed, + ) return result def backfill( diff --git a/tests/test_up_profile.py b/tests/test_up_profile.py index c7b73e0..1b85ff2 100644 --- a/tests/test_up_profile.py +++ b/tests/test_up_profile.py @@ -82,6 +82,41 @@ def test_capture_dedupes_and_continues_after_failure(): db.close() +class BlockingClient: + """模拟 DNS/传输层挂起：请求永不返回。""" + + def __init__(self, *, block=()): + self.calls = [] + self.block = set(block) + + def relation_stat(self, mid): + self.calls.append(int(mid)) + if int(mid) in self.block: + time.sleep(5) + raise AssertionError("不应返回") + return SimpleNamespace(mid=int(mid), follower=1000 + int(mid), following=5) + + +def test_capture_times_out_and_continues(): + with tempfile.TemporaryDirectory() as tmp: + db, repo = setup(tmp) + try: + client = BlockingClient(block={100}) + service = UpProfileService( + repo, lambda: client, clock=lambda: NOW, + request_timeout=0.05, heartbeat=1, + ) + result = service.capture([100, 200]) + assert result.requested == 2 + assert result.failed == 1 + assert "timeout" in result.errors[0] + # 卡住的 UP 不影响后续 + assert repo.get_video(3)["up_follower"] == 1200 + assert repo.get_video(1)["up_follower"] is None + finally: + db.close() + + def test_capture_marks_old_videos_as_backfilled(): with tempfile.TemporaryDirectory() as tmp: db, repo = setup(tmp)