# -*- coding: utf-8 -*- """每小时搜索发现与候选池写入。""" from __future__ import annotations import random import time from dataclasses import dataclass, field from typing import Any from ..bilibili.client import BilibiliError from ..bilibili.search import SearchClient, SearchItem from ..classify import passes_required_title, score_item from ..config import ConfigStore from ..repository import Repository @dataclass class DiscoverySummary: started_at: int finished_at: int = 0 searches: int = 0 items: int = 0 known: int = 0 old: int = 0 blacklisted: int = 0 title_miss: int = 0 scored: int = 0 candidates: int = 0 notified: int = 0 notify_failed: int = 0 errors: list[str] = field(default_factory=list) def as_dict(self) -> dict[str, Any]: return { "started_at": self.started_at, "finished_at": self.finished_at, "elapsed": max(0, self.finished_at - self.started_at), "searches": self.searches, "items": self.items, "known": self.known, "old": self.old, "blacklisted": self.blacklisted, "title_miss": self.title_miss, "scored": self.scored, "candidates": self.candidates, "notified": self.notified, "notify_failed": self.notify_failed, "errors": self.errors, } class DiscoveryService: def __init__( self, config_store: ConfigStore, repository: Repository, search_client: SearchClient, *, notifier=None, clock=time.time, sleeper=time.sleep, rng: random.Random | None = None, logger=None, ) -> None: self.config_store = config_store self.repo = repository self.search_client = search_client self.notifier = notifier self._clock = clock self._sleep = sleeper self._rng = rng or random.Random() self.logger = logger 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 run(self) -> DiscoverySummary: cfg = self.config_store.get() now = self._now() summary = DiscoverySummary(started_at=now) pending: list[dict[str, Any]] = [] first = True for keyword in cfg.discover.keywords: for page in range(1, cfg.discover.pages + 1): if not first: self._sleep(self._rng.uniform(1.5, 2.5)) first = False channel = f"search:{keyword}#p{page}" try: items = self.search_client.search(keyword, page) except BilibiliError as exc: summary.errors.append(f"{channel}: {exc}") self._log( "error", "discovery.search_failed", channel=channel, error=str(exc), ) continue summary.searches += 1 for item in items: self._handle_item(item, channel, cfg, summary, pending, now) self._notify(pending, summary, now) if summary.candidates or summary.notified: self.repo.bump_cache_version("videos") summary.finished_at = self._now() self._log( "info" if not summary.errors else "warning", "discovery.done", **summary.as_dict(), ) return summary def _handle_item( self, item: SearchItem, channel: str, cfg, summary: DiscoverySummary, pending: list[dict[str, Any]], now: int, ) -> None: summary.items += 1 existing = self.repo.get_candidate(item.aid) if existing is not None: summary.known += 1 if ( existing.get("notified_at") is None and existing.get("decision") is None and float(existing.get("score") or 0) >= cfg.notify.min_score ): self._queue_notify(pending, existing) return if self.repo.known_aid(item.aid): summary.known += 1 return if item.pubdate and now - item.pubdate > cfg.discover.window_days * 86400: summary.old += 1 return if not passes_required_title(item.title, cfg.classify["require_title"]): summary.title_miss += 1 return if item.up_mid is not None and self.repo.is_blacklisted(item.up_mid): summary.blacklisted += 1 return self.repo.record_seen(item.aid, source=channel, now=now) score = score_item( title=item.title, up_mid=item.up_mid, duration_seconds=item.duration_seconds, classify=cfg.classify, ) summary.scored += 1 if score < 0: return candidate = self.repo.upsert_candidate( aid=item.aid, bvid=item.bvid, title=item.title, up_mid=item.up_mid, up_name=item.up_name, pubdate=item.pubdate, cover=item.cover, score=score, channel=channel, duration=item.duration_seconds, play=item.play, like=item.like, favorite=item.favorite, danmaku=item.danmaku, review=item.review, now=now, ) summary.candidates += 1 if ( score >= cfg.notify.min_score and candidate.get("notified_at") is None and candidate.get("decision") is None ): self._queue_notify(pending, candidate) @staticmethod def _queue_notify( pending: list[dict[str, Any]], candidate: dict[str, Any] ) -> None: """同一次扫描中同一 aid 可能出现在多个关键词/页，按 aid 去重。""" if all(item["aid"] != candidate["aid"] for item in pending): pending.append(candidate) def _notify( self, pending: list[dict[str, Any]], summary: DiscoverySummary, now: int ) -> None: if not pending or self.notifier is None: return batch_id = f"{now}-{self._rng.randrange(1_000_000):06d}" error: str | None = None ok = False try: ok = bool(self.notifier.send_candidates(pending)) except Exception as exc: # noqa: BLE001 通知失败不回滚候选 error = str(exc) for candidate in pending: self.repo.mark_notified( candidate["aid"], channel="telegram", batch_id=batch_id, ok=ok, err=error, now=now, ) if ok: summary.notified = len(pending) else: summary.notify_failed = len(pending) summary.errors.append(f"telegram: {error or '发送失败'}") self._log( "error", "discovery.notify_failed", count=len(pending), error=error or "发送失败", )