227 lines
7.2 KiB
Python
227 lines
7.2 KiB
Python
# -*- 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 "发送失败",
|
|
)
|