# -*- coding: utf-8 -*- """每日视频七项快照与粉丝数采集。 规则（DEVELOPMENT.md §7.4）： * 对 ``active`` / ``unfaved`` 视频逐条请求 ``view``； * 成功写入七项指标；``ok=1`` 的同日快照直接跳过（断点续跑）； * 视频不可访问时标记 ``removed``； * 其他错误最多重试 3 次，仍失败写 ``ok=0`` 占位，不写假零值； * 对 ``track_fans`` 逐个请求 ``relation/stat``； * 完成后写入当日完成标记并生成日备份。 """ from __future__ import annotations import time from dataclasses import dataclass, field from typing import Any from ..bilibili.client import ( VIDEO_UNAVAILABLE_CODES, BilibiliAPIError, BilibiliError, BilibiliHTTPError, ) from ..config import AppConfig, ConfigStore from ..repository import Repository from ..timeutil import is_date, shanghai_date from .maintenance import backup_daily, prune_daily_backups @dataclass class SnapshotSummary: stat_date: str started_at: int total: int = 0 success: int = 0 failed: int = 0 skipped: int = 0 removed: int = 0 fans_ok: int = 0 fans_failed: int = 0 fans_skipped: int = 0 auth_ok: bool | None = None auth_message: str = "" backup_path: str | None = None backup_error: str | None = None completed: bool = False already_completed: bool = False finished_at: int = 0 errors: list[str] = field(default_factory=list) def as_dict(self) -> dict[str, Any]: return { "stat_date": self.stat_date, "started_at": self.started_at, "finished_at": self.finished_at, "elapsed": max(0, self.finished_at - self.started_at), "total": self.total, "success": self.success, "failed": self.failed, "skipped": self.skipped, "removed": self.removed, "fans_ok": self.fans_ok, "fans_failed": self.fans_failed, "fans_skipped": self.fans_skipped, "auth_ok": self.auth_ok, "auth_message": self.auth_message, "backup_path": self.backup_path, "backup_error": self.backup_error, "completed": self.completed, "already_completed": self.already_completed, "errors": self.errors, } class SnapshotService: def __init__( self, config_store: ConfigStore, repository: Repository, client=None, *, client_provider=None, clock=time.time, sleeper=time.sleep, logger=None, retries: int = 3, backoff: float = 5.0, progress_every: int = 50, ) -> None: if retries < 1: raise ValueError("retries 必须 >= 1") if client is None and client_provider is None: raise ValueError("必须提供 client 或 client_provider") self.config_store = config_store self.repo = repository self.client = client self.client_provider = client_provider self._clock = clock self._sleep = sleeper self.logger = logger self.retries = retries self.backoff = backoff self.progress_every = progress_every def _now(self) -> int: return int(self._clock()) def _client(self): if self.client_provider is not None: return self.client_provider.get() return self.client def _log(self, level: str, event: str, **fields) -> None: if self.logger is not None: getattr(self.logger, level)(event, **fields) @staticmethod def completed_key(stat_date: str) -> str: return f"snapshot:completed:{stat_date}" @staticmethod def backup_key(stat_date: str) -> str: return f"snapshot:backup:{stat_date}" @staticmethod def backup_error_key(stat_date: str) -> str: return f"snapshot:backup_error:{stat_date}" def run_if_needed( self, *, date: str | None = None, force: bool = False ) -> SnapshotSummary: now = self._now() stat_date = date or shanghai_date(now) if not is_date(stat_date): raise ValueError(f"日期格式必须是 YYYY-MM-DD: {stat_date!r}") if force: return self.run(date=stat_date, force=True) completed = self.repo.kv_get(self.completed_key(stat_date)) == "1" pending = self.repo.pending_snapshot_aids(stat_date) if completed and not pending: if self.repo.kv_get(self.backup_key(stat_date)) != "1": return self.backup_only(stat_date) self._log("info", "snapshot.skip", stat_date=stat_date, reason="already_completed") return SnapshotSummary( stat_date=stat_date, started_at=now, finished_at=now, completed=True, already_completed=True, ) return self.run(date=stat_date, force=False) def backup_only(self, stat_date: str) -> SnapshotSummary: """采集已完成但备份缺失时，只重试备份，不请求任何外部接口。""" now = self._now() cfg = self.config_store.get() summary = SnapshotSummary( stat_date=stat_date, started_at=now, completed=True, already_completed=True, ) self._log("info", "snapshot.backup_retry", stat_date=stat_date) self._run_backup(cfg, stat_date, summary) summary.finished_at = self._now() self._log("info", "snapshot.done", **summary.as_dict()) return summary def run(self, *, date: str | None = None, force: bool = False) -> SnapshotSummary: cfg = self.config_store.get() client = self._client() now = self._now() stat_date = date or shanghai_date(now) if not is_date(stat_date): raise ValueError(f"日期格式必须是 YYYY-MM-DD: {stat_date!r}") summary = SnapshotSummary(stat_date=stat_date, started_at=now) self._log("info", "snapshot.start", stat_date=stat_date, force=force) # 先落“未完成”，保证强制重采或中途中断后仍可续跑。 self.repo.kv_set(self.completed_key(stat_date), "0") if self.repo.kv_get("snapshot:last_completed") == stat_date: self.repo.kv_set("snapshot:last_completed", "") summary.auth_ok, summary.auth_message = self._check_auth(client, cfg, now) videos = self.repo.video_aids_for_snapshot() summary.total = len(videos) for index, aid in enumerate(videos, 1): existing = self.repo.get_snapshot(stat_date, aid) if existing is not None and int(existing.get("ok") or 0) == 1 and not force: summary.skipped += 1 continue stat, error, unavailable = self._fetch_view(client, aid) if unavailable: summary.removed += 1 self.repo.set_video_status(aid, "removed", now=now) self._log( "warning", "snapshot.removed", stat_date=stat_date, aid=aid, error=error, ) elif stat is not None: self.repo.upsert_snapshot( stat_date, aid, metrics=stat.metrics(), ok=True, err=None, captured_at=self._now(), ) summary.success += 1 else: summary.failed += 1 self.repo.upsert_snapshot( stat_date, aid, metrics=None, ok=False, err=error, captured_at=self._now(), ) summary.errors.append(f"aid {aid}: {error}") self._log( "error", "snapshot.failed", stat_date=stat_date, aid=aid, error=error, ) if self.progress_every > 0 and ( index % self.progress_every == 0 or index == summary.total ): self._log( "info", "snapshot.progress", stat_date=stat_date, done=index, total=summary.total, success=summary.success, failed=summary.failed, skipped=summary.skipped, removed=summary.removed, ) for mid in cfg.track_fans: existing_fan = self.repo.get_up_stat(stat_date, mid) if ( not force and existing_fan is not None and int(existing_fan.get("ok") or 0) == 1 ): summary.fans_skipped += 1 continue stat, error = self._fetch_fan(client, mid) if stat is not None: self.repo.upsert_up_stat( stat_date, mid, follower=stat.follower, following=stat.following, ok=True, captured_at=self._now(), ) summary.fans_ok += 1 else: summary.fans_failed += 1 self.repo.upsert_up_stat( stat_date, mid, follower=None, following=None, ok=False, captured_at=self._now(), ) summary.errors.append(f"fans mid {mid}: {error}") self._log( "error", "snapshot.fan_failed", stat_date=stat_date, mid=mid, error=error, ) summary.completed = summary.failed == 0 and summary.fans_failed == 0 self.repo.kv_set(self.completed_key(stat_date), "1" if summary.completed else "0") self.repo.kv_set("snapshot:last_attempt", stat_date) if summary.completed: self.repo.kv_set("snapshot:last_completed", stat_date) elif self.repo.kv_get("snapshot:last_completed") == stat_date: self.repo.kv_set("snapshot:last_completed", "") self._run_backup(cfg, stat_date, summary) if summary.success or summary.removed or summary.failed: self.repo.bump_cache_version("videos") if summary.fans_ok or summary.fans_failed: self.repo.bump_cache_version("fans") summary.finished_at = self._now() self._log( "info" if summary.completed else "warning", "snapshot.done", **summary.as_dict(), ) return summary # ------------------------------------------------------------ internals def _run_backup( self, cfg: AppConfig, stat_date: str, summary: SnapshotSummary ) -> None: try: path = backup_daily( self.repo.db, cfg.storage.backup_dir, stat_date, logger=self.logger, ) prune_daily_backups( cfg.storage.backup_dir, keep_days=cfg.storage.backup_keep_days, ) summary.backup_path = str(path) self.repo.kv_set(self.backup_key(stat_date), "1") self.repo.kv_set(self.backup_error_key(stat_date), "") except Exception as exc: # noqa: BLE001 备份失败不推翻已采集数据 summary.backup_error = str(exc) self.repo.kv_set(self.backup_key(stat_date), "0") self.repo.kv_set(self.backup_error_key(stat_date), str(exc)) self._log("error", "backup.failed", error=str(exc), stat_date=stat_date) def _check_auth( self, client, cfg: AppConfig, now: int ) -> tuple[bool, str]: if not cfg.bilibili.read.cookie: self.repo.kv_set("auth:last_check", str(now)) self.repo.kv_set("auth:ok", "0") self.repo.kv_set("auth:message", "未配置 cookie") return False, "未配置 cookie" try: state = client.check_login() except BilibiliError as exc: message = str(exc) self.repo.kv_set("auth:last_check", str(now)) self.repo.kv_set("auth:ok", "0") self.repo.kv_set("auth:message", message) self._log("warning", "auth.invalid", message=message) return False, message self.repo.kv_set("auth:last_check", str(now)) self.repo.kv_set("auth:ok", "1" if state.is_login else "0") self.repo.kv_set("auth:message", state.message) if state.mid is not None: self.repo.kv_set("auth:mid", str(state.mid)) if not state.is_login: self._log( "warning", "auth.invalid", code=state.code, message=state.message, ) return state.is_login, state.message def _fetch_view(self, client, aid: int): last_error = "" for attempt in range(1, self.retries + 1): try: return client.view(aid), None, False except BilibiliAPIError as exc: last_error = str(exc) if exc.code in VIDEO_UNAVAILABLE_CODES: return None, last_error, True except BilibiliHTTPError as exc: last_error = str(exc) if exc.status == 404: return None, last_error, True except BilibiliError as exc: last_error = str(exc) if attempt < self.retries: self._sleep(self.backoff * (2 ** (attempt - 1))) return None, last_error or "未知错误", False def _fetch_fan(self, client, mid: int): last_error = "" for attempt in range(1, self.retries + 1): try: return client.relation_stat(mid), None except BilibiliError as exc: last_error = str(exc) if attempt < self.retries: self._sleep(self.backoff * (2 ** (attempt - 1))) return None, last_error or "未知错误"