# -*- coding: utf-8 -*- """库内视频管理：入库（候选确认 / Telegram 手动）、忽略、拉黑。 入库只写本地 ``video`` 表并置为 ``active``，**不再调用 B 站收藏夹写接口** （``deal`` / ``sort``）。收藏夹由人工维护，系统仅通过 ``favsort.py`` 做重排序。 """ from __future__ import annotations import time from dataclasses import dataclass, field from typing import Any from ..bilibili.client import BilibiliError, BilibiliProtocolError from ..config import ConfigStore from ..repository import Repository from ..timeutil import shanghai_date from ..video_ref import DEFAULT_MAX_REFS, VideoRef, extract_video_refs @dataclass class IngestResult: ok: bool aid: int | None state: str title: str = "" up_name: str = "" message: str = "" def as_dict(self) -> dict[str, Any]: return { "ok": self.ok, "aid": self.aid, "state": self.state, "title": self.title, "up_name": self.up_name, "message": self.message, } @dataclass class ManualIngestResult: refs: int = 0 added: int = 0 results: list[IngestResult] = field(default_factory=list) @property def ok(self) -> bool: return any(result.ok for result in self.results) def as_dict(self) -> dict[str, Any]: return { "refs": self.refs, "added": self.added, "ok": self.ok, "results": [result.as_dict() for result in self.results], } @dataclass class ActionResult: ok: bool aid: int action: str state: str message: str = "" def as_dict(self) -> dict[str, Any]: return { "ok": self.ok, "aid": self.aid, "action": self.action, "state": self.state, "message": self.message, } class LibraryService: def __init__( self, config_store: ConfigStore, repository: Repository, read_provider, *, up_profile_service=None, clock=time.time, logger=None, max_refs: int = DEFAULT_MAX_REFS, ) -> None: self.config_store = config_store self.repo = repository self.read_provider = read_provider self.up_profile = up_profile_service self._clock = clock self.logger = logger self.max_refs = max(1, int(max_refs)) 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) # ------------------------------------------------------------ candidate def ingest_candidate(self, aid: int) -> IngestResult: aid = int(aid) candidate = self.repo.get_candidate(aid) if candidate is None: video = self.repo.get_video(aid) if video is not None: return IngestResult( ok=True, aid=aid, state="exists", title=str(video.get("title") or ""), up_name=str(video.get("up_name") or ""), message="已在库，无需重复入库", ) return IngestResult( ok=False, aid=aid, state="missing", message="候选不存在或已处理" ) detail = None try: detail = self.read_provider.get().view_detail(aid=aid) except BilibiliError as exc: # 网络失败时用候选元数据兜底 self._log("warning", "ingest.detail_failed", aid=aid, error=str(exc)) if detail is not None: outcome = self.repo.ingest_video( aid=detail.aid, bvid=detail.bvid, title=detail.title, up_mid=detail.up_mid, up_name=detail.up_name, pubdate=detail.pubdate, duration=detail.duration, pages=detail.pages, cover=detail.cover, up_face=detail.up_face, source="candidate:telegram", now=self._now(), ) self._write_snapshot(detail) result = self._result_from_outcome( outcome, title=detail.title, up_name=detail.up_name or "" ) up_mid = detail.up_mid up_name = detail.up_name or "" else: pubdate = int(candidate.get("pubdate") or 0) duration = int(candidate.get("duration") or 0) or None outcome = self.repo.ingest_video( aid=aid, bvid=candidate.get("bvid"), title=str(candidate.get("title") or ""), up_mid=candidate.get("up_mid"), up_name=candidate.get("up_name"), pubdate=pubdate, duration=duration, cover=candidate.get("cover"), source="candidate:telegram", now=self._now(), ) result = self._result_from_outcome( outcome, title=str(candidate.get("title") or ""), up_name=str(candidate.get("up_name") or ""), ) up_mid = candidate.get("up_mid") up_name = str(candidate.get("up_name") or "") self._post_ingest([(up_mid, up_name)]) self._log( "info", "ingest.candidate", aid=aid, state=result.state, folder=None, ) return result # ------------------------------------------------------------ manual def ingest_text(self, text: object) -> ManualIngestResult: refs = extract_video_refs(text, limit=self.max_refs) result = ManualIngestResult(refs=len(refs)) if not refs: return result client = self.read_provider.get() ups: list[tuple[int | None, str]] = [] for ref in refs: try: detail = self._resolve_ref(client, ref) except BilibiliError as exc: result.results.append(IngestResult( ok=False, aid=None, state="error", message=str(exc), )) self._log("warning", "ingest.manual_failed", error=str(exc)) continue outcome = self.repo.ingest_video( aid=detail.aid, bvid=detail.bvid, title=detail.title, up_mid=detail.up_mid, up_name=detail.up_name, pubdate=detail.pubdate, duration=detail.duration, pages=detail.pages, cover=detail.cover, up_face=detail.up_face, source="manual:telegram", now=self._now(), ) self._write_snapshot(detail) item = self._result_from_outcome( outcome, title=detail.title, up_name=detail.up_name or "" ) if item.state in ("added", "recovered"): result.added += 1 result.results.append(item) ups.append((detail.up_mid, detail.up_name or "")) if ups: self._post_ingest(ups) self._log("info", "ingest.manual", **result.as_dict()) return result def _resolve_ref(self, client, ref: VideoRef): if ref.kind == "bvid": return client.view_detail(bvid=str(ref.value)) if ref.kind == "aid": return client.view_detail(aid=int(ref.value)) if ref.kind == "short": final = client.resolve_short_link(f"https://b23.tv/{ref.value}") for sub in extract_video_refs(final): if sub.kind != "short": return self._resolve_ref(client, sub) raise BilibiliProtocolError(f"短链未解析出视频引用: {final}") raise ValueError(f"未知引用类型: {ref.kind}") # ------------------------------------------------------------ actions def ignore(self, aid: int) -> ActionResult: aid = int(aid) candidate = self.repo.get_candidate(aid) if candidate is None: return ActionResult( ok=False, aid=aid, action="ignore", state="missing", message="候选不存在或已处理", ) self.repo.mark_candidate_decision(aid, "rejected", now=self._now()) self.repo.bump_cache_version("videos") return ActionResult( ok=True, aid=aid, action="ignore", state="rejected", message="已忽略", ) def blacklist(self, aid: int) -> ActionResult: aid = int(aid) candidate = self.repo.get_candidate(aid) if candidate is None: return ActionResult( ok=False, aid=aid, action="blacklist", state="missing", message="候选不存在或已处理", ) up_mid = candidate.get("up_mid") if not isinstance(up_mid, int) or isinstance(up_mid, bool): return ActionResult( ok=False, aid=aid, action="blacklist", state="failed", message="候选缺少 UP mid，无法拉黑", ) self.repo.add_blacklist( up_mid, name=candidate.get("up_name"), reason="telegram", now=self._now(), ) rejected = self.repo.reject_pending_for_up(up_mid, now=self._now()) self.repo.mark_candidate_decision(aid, "rejected", now=self._now()) self.repo.bump_cache_version("videos") return ActionResult( ok=True, aid=aid, action="blacklist", state="rejected", message=f"已拉黑 UP，并忽略该 UP 的 {rejected} 条待决策候选", ) # ------------------------------------------------------------ internals def _result_from_outcome(self, outcome, *, title: str, up_name: str) -> IngestResult: if outcome.created: state, message = "added", "已入库" elif outcome.recovered: state, message = "recovered", "已重新入库" else: state, message = "exists", "已在库" return IngestResult( ok=True, aid=outcome.aid, state=state, title=title, up_name=up_name, message=message, ) def _write_snapshot(self, detail) -> None: """用已取到的 ``view`` 数据立即写一条当天快照（同日已有成功记录则跳过）。""" now = self._now() stat_date = shanghai_date(now) existing = self.repo.get_snapshot(stat_date, detail.aid) if existing is not None and int(existing.get("ok") or 0) == 1: return self.repo.upsert_snapshot( stat_date, detail.aid, metrics=detail.metrics(), ok=True, err=None, captured_at=now, ) def _post_ingest(self, ups: list[tuple[int | None, str]]) -> None: if ups and self.up_profile is not None: mids = [mid for mid, _ in ups if isinstance(mid, int) and not isinstance(mid, bool)] names = {mid: name for mid, name in ups if isinstance(mid, int) and not isinstance(mid, bool)} if mids: try: self.up_profile.capture(mids, names=names, live_since=self._now() - 3600) except Exception as exc: # noqa: BLE001 粉丝采集失败不影响入库 self._log("warning", "ingest.follower_failed", error=str(exc)) self.repo.bump_cache_version("videos")