# -*- coding: utf-8 -*- """UP 粉丝快照采集。 视频的播放、互动记在 bvid 上,而 UP 的粉丝数是视频之外的受众基数。为了让以后 能计算“播放 / 入库时粉丝数”,本服务在视频入库时记录一次粉丝快照,并支持对历史 数据做一次性回填。 ``up_profile.follower_first`` 永远保留首次采集值;``video.up_follower`` 保留每个 视频入库时的值,不会被后续刷新覆盖。 """ from __future__ import annotations import threading 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, 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()) def _log(self, level: str, event: str, **fields) -> None: 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], *, 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 {} total = len(ordered) for index, mid in enumerate(ordered, 1): try: fan = self._fetch_relation(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, ) 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( 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, )