# -*- 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 ( BilibiliAPIError, 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 truncated: 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, "truncated": self.truncated, "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: all_refs = extract_video_refs(text, limit=None) result = ManualIngestResult(refs=len(all_refs)) if not all_refs: return result if len(all_refs) > self.max_refs: result.truncated = len(all_refs) - self.max_refs all_refs = all_refs[: self.max_refs] result.refs = len(all_refs) client = self.read_provider.get() ups: list[tuple[int | None, str]] = [] for ref in all_refs: try: detail = self._resolve_ref(client, ref) except BilibiliAPIError as exc: message = ( "无法解析该视频引用：BV 号/av 号无效或视频不存在（code=-400）" if exc.code == -400 else str(exc) ) result.results.append(IngestResult( ok=False, aid=None, state="error", message=message, )) self._log("warning", "ingest.manual_failed", error=str(exc)) continue 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")