170 lines
5.6 KiB
Python
170 lines
5.6 KiB
Python
# -*- 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,
|
|
)
|