kairee
6a21b5d905
Backfill same-day snapshots for newly added or restored videos, preserve successful fan snapshots on resume, persist incomplete state before forced re-runs, and split collection completion from backup completion so failed backups can retry without API calls. Reject missing data payloads instead of treating them as empty folders, guard against mass-unfav on suspicious empty remote folders, rebuild Bilibili clients when credentials change, recompute primary folder ownership from active relations with per-folder fav_time, preserve original titles for dropped invalid videos, poll config every 60 seconds, and fix favsort verify interval and mixed resource types.
379 lines
14 KiB
Python
379 lines
14 KiB
Python
# -*- 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)
|
||
|
||
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 "未知错误"
|