From 62abbd7ea171af403f932b49c8f73a8dd63c73a9 Mon Sep 17 00:00:00 2001 From: kaireewu Date: Sun, 1 Nov 2026 13:50:03 +0800 Subject: [PATCH] feat(ingest): db-only library ingest and telegram manual ingest --- reaction_board/notify/telegram.py | 41 ++- reaction_board/repository.py | 89 ++++++ reaction_board/services/library.py | 322 +++++++++++++++++++++ reaction_board/services/telegram_poller.py | 63 +++- tests/test_library.py | 296 +++++++++++++++++++ tests/test_telegram.py | 83 ++++-- 6 files changed, 866 insertions(+), 28 deletions(-) create mode 100644 reaction_board/services/library.py create mode 100644 tests/test_library.py diff --git a/reaction_board/notify/telegram.py b/reaction_board/notify/telegram.py index 41ce7a9..222054e 100644 --- a/reaction_board/notify/telegram.py +++ b/reaction_board/notify/telegram.py @@ -53,6 +53,43 @@ def format_duration(seconds: int) -> str: return f"{minutes}:{secs:02d}" +def _result_line(result: Any) -> str: + state = getattr(result, "state", "") + title = html.escape(str(getattr(result, "title", "") or "")) + message = str(getattr(result, "message", "") or "") + aid = getattr(result, "aid", None) + suffix = f"《{title}》" if title else "" + ref = f"（aid {aid}）" if aid else "" + icon = { + "added": "✅", "recovered": "♻️", "exists": "ℹ️", + "error": "⚠️", "missing": "❔", + }.get(state, "•") + body = message or "完成" + return f"{icon} {body}{suffix}{ref}".strip() + + +def format_ingest_result(result: Any) -> str: + """候选确认后的回执（候选确认 / 手动入库单条）。""" + return _result_line(result) + + +def format_manual_reply(manual_result: Any) -> str: + """私聊文本手动入库的汇总回执。""" + refs = int(getattr(manual_result, "refs", 0) or 0) + if refs <= 0: + return "未识别到 B 站视频引用，请发送视频链接或 BV 号，例如 BV1x8x166EzY。" + results = list(getattr(manual_result, "results", []) or []) + lines = [_result_line(item) for item in results] + header = f"共识别 {refs} 个引用，入库 {int(getattr(manual_result, 'added', 0) or 0)} 条：" + return "\n".join([header, *lines]) + + +HELP_TEXT = ( + "发送 B 站视频链接、BV 号或 av 号即可手动入库并开始追踪。\n" + "例如：https://www.bilibili.com/video/BV1x8x166EzY 或 BV1x8x166EzY" +) + + def format_candidate(candidate: Mapping[str, Any]) -> str: title = html.escape(str(candidate.get("title") or "(无标题)")) up_name = html.escape(str(candidate.get("up_name") or "未知UP")) @@ -184,7 +221,7 @@ class TelegramClient: keyboard = { "inline_keyboard": [ [ - {"text": "⭐ 收藏", "callback_data": f"fav:{candidate['aid']}"}, + {"text": "📥 入库", "callback_data": f"add:{candidate['aid']}"}, {"text": "🙈 忽略", "callback_data": f"ign:{candidate['aid']}"}, {"text": "🚫 拉黑", "callback_data": f"blk:{candidate['aid']}"}, ] @@ -199,7 +236,7 @@ class TelegramClient: ) -> list[dict[str, Any]]: params: dict[str, Any] = { "timeout": self.poll_timeout if timeout is None else int(timeout), - "allowed_updates": json.dumps(["callback_query"]), + "allowed_updates": json.dumps(["callback_query", "message"]), } if offset is not None: params["offset"] = int(offset) diff --git a/reaction_board/repository.py b/reaction_board/repository.py index ff4fdd5..b8a25ee 100644 --- a/reaction_board/repository.py +++ b/reaction_board/repository.py @@ -24,6 +24,17 @@ class UpsertOutcome: moved_folder: bool +@dataclass(frozen=True) +class IngestOutcome: + """入库（候选确认或手动）结果。""" + + aid: int + status: str + previous_status: str | None + created: bool + recovered: bool + + @dataclass(frozen=True) class FolderRow: media_id: int @@ -399,6 +410,84 @@ class Repository: return [dict(row) for row in rows] return self.db.read(run) + def ingest_video( + self, + *, + aid: int, + bvid: str | None, + title: str, + up_mid: int | None, + up_name: str | None, + pubdate: int, + duration: int | None = None, + pages: int | None = None, + cover: str | None = None, + source: str = "manual", + now: int | None = None, + ) -> IngestOutcome: + """把一条视频写入库并置为 ``active``（候选确认 / Telegram 手动入库共用）。 + + 不写 B 站收藏夹、不设置 ``folder_id``；同时删除同 aid 的候选行。 + 既有的 ``removed`` / ``dropped`` / ``unfaved`` 状态都会被重新激活。 + """ + now = self._now() if now is None else int(now) + aid = int(aid) + + def run(conn: sqlite3.Connection) -> IngestOutcome: + previous = conn.execute( + "SELECT status FROM video WHERE aid = ?", (aid,) + ).fetchone() + previous_status = str(previous["status"]) if previous else None + conn.execute( + """ + INSERT INTO video(aid, bvid, title, up_mid, up_name, pubdate, + duration, pages, cover, folder_id, fav_time, status, + source, first_seen, last_sync, attr) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL, 'active', ?, ?, ?, 0) + ON CONFLICT(aid) DO UPDATE SET + bvid = COALESCE(excluded.bvid, video.bvid), + title = CASE + WHEN excluded.title != '' THEN excluded.title + ELSE video.title + END, + up_mid = COALESCE(excluded.up_mid, video.up_mid), + up_name = COALESCE(excluded.up_name, video.up_name), + pubdate = CASE + WHEN excluded.pubdate > 0 THEN excluded.pubdate + ELSE video.pubdate + END, + duration = COALESCE(excluded.duration, video.duration), + pages = COALESCE(excluded.pages, video.pages), + cover = COALESCE(excluded.cover, video.cover), + status = 'active', + source = COALESCE(video.source, excluded.source), + last_sync = excluded.last_sync + """, + ( + aid, + bvid, + title or "(无标题)", + up_mid, + up_name, + int(pubdate or 0), + None if duration is None else int(duration), + None if pages is None else int(pages), + cover, + source, + now, + now, + ), + ) + conn.execute("DELETE FROM candidate WHERE aid = ?", (aid,)) + return IngestOutcome( + aid=aid, + status="active", + previous_status=previous_status, + created=previous is None, + recovered=previous_status in {"removed", "dropped", "unfaved"}, + ) + return self.db.write(run) + # ------------------------------------------------------- up profile def apply_up_follower( diff --git a/reaction_board/services/library.py b/reaction_board/services/library.py new file mode 100644 index 0000000..78ddc2b --- /dev/null +++ b/reaction_board/services/library.py @@ -0,0 +1,322 @@ +# -*- 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, + 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, + 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") \ No newline at end of file diff --git a/reaction_board/services/telegram_poller.py b/reaction_board/services/telegram_poller.py index 215830c..584f612 100644 --- a/reaction_board/services/telegram_poller.py +++ b/reaction_board/services/telegram_poller.py @@ -1,11 +1,17 @@ # -*- coding: utf-8 -*- -"""Telegram getUpdates long polling 与按钮回调分发。""" +"""Telegram getUpdates long polling 与回调/文本消息分发。 + +回调按钮处理候选「入库 / 忽略 / 拉黑」；私聊文本消息按视频链接、BV 号或 av 号 +手动入库。两类操作都只写本地库，不写 B 站收藏夹。 +""" from __future__ import annotations import time from typing import Any, Mapping +from ..notify.telegram import HELP_TEXT, format_ingest_result, format_manual_reply + class TelegramPoller: def __init__(self, bot, actions, *, logger=None, sleeper=time.sleep) -> None: @@ -27,14 +33,21 @@ class TelegramPoller: callback = update.get("callback_query") if isinstance(callback, Mapping): self.handle_callback(callback) + continue + message = update.get("message") + if isinstance(message, Mapping): + self.handle_message(message) return offset + def _authorized(self, chat_id, from_id) -> bool: + return chat_id == self.bot.chat_id and from_id == self.bot.chat_id + def handle_callback(self, callback: Mapping[str, Any]) -> None: callback_id = str(callback.get("id") or "") message = callback.get("message") or {} chat_id = (message.get("chat") or {}).get("id") from_id = (callback.get("from") or {}).get("id") - if chat_id != self.bot.chat_id or from_id != self.bot.chat_id: + if not self._authorized(chat_id, from_id): self._log( "warning", "telegram.unauthorized", chat_id=chat_id, from_id=from_id, @@ -49,8 +62,9 @@ class TelegramPoller: self._answer(callback_id, "未知操作") return - if action == "fav": - result = self.actions.favorite(aid) + if action in ("add", "fav"): + # fav 为旧按钮回调，兼容已发出的历史消息。 + result = self.actions.ingest_candidate(aid) elif action == "ign": result = self.actions.ignore(aid) elif action == "blk": @@ -59,13 +73,10 @@ class TelegramPoller: self._answer(callback_id, "未知操作") return - text = result.message - if result.ok and result.position is not None: - text = f"{text}（位置 #{result.position}）" + text = format_ingest_result(result) self._log( "info", "telegram.action", action=action, aid=aid, ok=result.ok, state=result.state, - position=result.position, ) if not self._answer(callback_id, text): # 回调 ID 过期时 answerCallbackQuery 会 400；改用普通消息告知结果。 @@ -80,6 +91,40 @@ class TelegramPoller: except Exception as exc: # noqa: BLE001 self._log("warning", "telegram.edit_failed", error=str(exc)) + def handle_message(self, message: Mapping[str, Any]) -> None: + chat_id = (message.get("chat") or {}).get("id") + from_id = (message.get("from") or {}).get("id") + if not self._authorized(chat_id, from_id): + self._log( + "warning", "telegram.unauthorized_message", + chat_id=chat_id, from_id=from_id, + ) + return + text = message.get("text") + if not isinstance(text, str) or not text.strip(): + return + text = text.strip() + if text.startswith("/start") or text.startswith("/help"): + self._reply(HELP_TEXT) + return + try: + result = self.actions.ingest_text(text) + except Exception as exc: # noqa: BLE001 单条消息失败不影响长轮询 + self._log("error", "telegram.ingest_error", error=str(exc)) + self._reply(f"入库失败：{exc}") + return + self._log( + "info", "telegram.manual_ingest", + refs=result.refs, added=result.added, + ) + self._reply(format_manual_reply(result)) + + def _reply(self, text: str) -> None: + try: + self.bot.send_text(text) + except Exception as exc: # noqa: BLE001 + self._log("warning", "telegram.reply_failed", error=str(exc)) + def _answer(self, callback_id: str, text: str) -> bool: try: self.bot.answer_callback(callback_id, text) @@ -125,4 +170,4 @@ class TelegramPoller: delay = initial_delay iterations += 1 if max_iterations is not None and iterations >= max_iterations: - break + break \ No newline at end of file diff --git a/tests/test_library.py b/tests/test_library.py new file mode 100644 index 0000000..13e4a33 --- /dev/null +++ b/tests/test_library.py @@ -0,0 +1,296 @@ +# -*- coding: utf-8 -*- +"""入库（候选确认 / 手动）、忽略、拉黑动作测试（假只读客户端，不联网）: +uv run python tests/test_library.py +""" + +import json +import os +import sys +import tempfile +from pathlib import Path + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +sys.stdout.reconfigure(encoding="utf-8") + +from fixtures import view_for # noqa: E402 +from reaction_board.bilibili.client import BilibiliError # noqa: E402 +from reaction_board.bilibili.models import VideoDetail # noqa: E402 +from reaction_board.config import ConfigStore # noqa: E402 +from reaction_board.database import Database # noqa: E402 +from reaction_board.repository import Repository # noqa: E402 +from reaction_board.services.library import LibraryService # noqa: E402 +from reaction_board.timeutil import shanghai_date # noqa: E402 + +NOW = 1_793_444_393 +DATE = shanghai_date(NOW) +BV = "BV1xvxi6LEZD" + + +def detail( + aid, + *, + bvid=BV, + title="咻咻满 reaction", + up_mid=100, + up_name="UP100", + pubdate=NOW - 3600, + view=819, + duration=9504, + pages=2, +): + payload = view_for(aid, view=view) + data = payload["data"] + data["bvid"] = bvid + data["title"] = title + data["pubdate"] = pubdate + data["ctime"] = pubdate + data["duration"] = duration + data["owner"] = {"mid": up_mid, "name": up_name, "face": ""} + data["pages"] = [{"cid": 1, "page": i + 1} for i in range(pages)] + return VideoDetail.from_api(data) + + +class FakeReadClient: + def __init__(self, by_bvid=None, by_aid=None, short_map=None, fail_bvids=()): + self.by_bvid = by_bvid or {} + self.by_aid = by_aid or {} + self.short_map = short_map or {} + self.fail_bvids = set(fail_bvids) + + def view_detail(self, *, aid=None, bvid=None): + if bvid is not None: + if bvid in self.fail_bvids: + raise BilibiliError(f"view 失败: {bvid}") + if bvid in self.by_bvid: + return self.by_bvid[bvid] + raise BilibiliError(f"未知 bvid: {bvid}") + if aid in self.by_aid: + return self.by_aid[aid] + raise BilibiliError(f"未知 aid: {aid}") + + def resolve_short_link(self, url): + if url not in self.short_map: + raise BilibiliError(f"未知短链: {url}") + return self.short_map[url] + + +class FakeProvider: + def __init__(self, client): + self.client = client + + def get(self): + return self.client + + +class FakeUpProfile: + def __init__(self): + self.calls = [] + + def capture(self, mids, *, names=None, live_since=None, backfilled=False): + self.calls.append((list(mids), dict(names or {}))) + + class Result: + requested = len(mids) + updated_videos = len(mids) + failed = 0 + + return Result() + + +def write_config(tmp): + data = { + "bilibili": { + "read": {"mid": 1, "csrf": "r", "cookie": "read-cookie"}, + "write": {"mid": 2, "csrf": "w", "cookie": "write-cookie"}, + }, + "storage": { + "database_path": str(Path(tmp) / "app.db"), + "backup_dir": str(Path(tmp) / "backup"), + }, + "discover": {"keywords": ["x"], "pages": 1, "window_days": 14}, + } + path = Path(tmp) / "config.json" + path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") + return path + + +def setup(tmp, client, up_profile=None): + store = ConfigStore(write_config(tmp), env={}) + db = Database(Path(tmp) / "app.db", backup_dir=Path(tmp) / "backup") + db.initialize() + repo = Repository(db, clock=lambda: NOW) + service = LibraryService( + store, repo, FakeProvider(client), + up_profile_service=up_profile, clock=lambda: NOW, + ) + return db, repo, service + + +def add_candidate(repo, aid, *, up_mid=100, pubdate=NOW - 3600): + return repo.upsert_candidate( + aid=aid, bvid=BV, title="咻咻满 reaction", up_mid=up_mid, + up_name=f"UP{up_mid}", pubdate=pubdate, cover=None, score=3, + channel="search:test#p1", now=NOW, + ) + + +def test_ingest_candidate_added_and_snapshot(): + with tempfile.TemporaryDirectory() as tmp: + ups = FakeUpProfile() + client = FakeReadClient(by_aid={9: detail(9)}) + db, repo, service = setup(tmp, client, ups) + try: + add_candidate(repo, 9) + result = service.ingest_candidate(9) + assert result.ok is True and result.state == "added" + assert result.aid == 9 + assert repo.get_candidate(9) is None + video = repo.get_video(9) + assert video["status"] == "active" + assert video["source"] == "candidate:telegram" + assert video["folder_id"] is None + assert video["up_mid"] == 100 + snap = repo.get_snapshot(DATE, 9) + assert snap is not None and snap["ok"] == 1 and snap["view"] == 819 + assert ups.calls == [([100], {100: "UP100"})] + finally: + db.close() + + +def test_ingest_candidate_exists_and_missing(): + with tempfile.TemporaryDirectory() as tmp: + client = FakeReadClient(by_aid={9: detail(9)}) + db, repo, service = setup(tmp, client) + try: + add_candidate(repo, 9) + assert service.ingest_candidate(9).state == "added" + # 候选已被删除，但视频已在库 + again = service.ingest_candidate(9) + assert again.ok is True and again.state == "exists" + missing = service.ingest_candidate(999) + assert missing.ok is False and missing.state == "missing" + finally: + db.close() + + +def test_ingest_candidate_recovers_dropped(): + with tempfile.TemporaryDirectory() as tmp: + client = FakeReadClient(by_aid={9: detail(9)}) + db, repo, service = setup(tmp, client) + try: + add_candidate(repo, 9) + assert service.ingest_candidate(9).state == "added" + repo.set_video_status(9, "dropped", now=NOW) + add_candidate(repo, 9) + result = service.ingest_candidate(9) + assert result.ok is True and result.state == "recovered" + assert repo.get_video(9)["status"] == "active" + finally: + db.close() + + +def test_ingest_candidate_falls_back_when_detail_fails(): + with tempfile.TemporaryDirectory() as tmp: + client = FakeReadClient(fail_bvids={BV}) + db, repo, service = setup(tmp, client) + try: + add_candidate(repo, 9) + result = service.ingest_candidate(9) + assert result.ok is True and result.state == "added" + assert repo.get_video(9)["title"] == "咻咻满 reaction" + assert repo.get_snapshot(DATE, 9) is None + finally: + db.close() + + +def test_manual_ingest_forms_and_multiple(): + with tempfile.TemporaryDirectory() as tmp: + ups = FakeUpProfile() + client = FakeReadClient( + by_bvid={BV: detail(9)}, + by_aid={12345: detail(77, bvid="BV77", up_mid=200, up_name="UP200")}, + short_map={"https://b23.tv/abc": f"https://www.bilibili.com/video/{BV}"}, + ) + db, repo, service = setup(tmp, client, ups) + try: + result = service.ingest_text( + f"看这个 https://www.bilibili.com/video/{BV} 还有 av12345" + ) + assert result.refs == 2 and result.added == 2 + assert [r.aid for r in result.results] == [9, 77] + assert repo.get_video(9)["source"] == "manual:telegram" + assert repo.get_video(77)["status"] == "active" + + short = service.ingest_text("https://b23.tv/abc") + assert short.added == 0 + assert short.results[0].state == "exists" + + none = service.ingest_text("没有视频") + assert none.refs == 0 and none.results == [] + finally: + db.close() + + +def test_manual_ingest_error_isolation(): + with tempfile.TemporaryDirectory() as tmp: + client = FakeReadClient( + by_bvid={BV: detail(9)}, + fail_bvids={"BVfail000000"}, + ) + db, repo, service = setup(tmp, client) + try: + result = service.ingest_text(f"{BV} BVfail000000") + assert result.added == 1 + states = [r.state for r in result.results] + assert states == ["added", "error"] + finally: + db.close() + + +def test_snapshot_not_overwritten(): + with tempfile.TemporaryDirectory() as tmp: + client = FakeReadClient(by_bvid={BV: detail(9, view=819)}) + db, repo, service = setup(tmp, client) + try: + repo.upsert_snapshot( + DATE, 9, metrics={"view": 5000}, ok=True, err=None, captured_at=NOW, + ) + service.ingest_text(BV) + snap = repo.get_snapshot(DATE, 9) + assert snap["view"] == 5000 # 已有成功快照不覆盖 + finally: + db.close() + + +def test_ignore_and_blacklist(): + with tempfile.TemporaryDirectory() as tmp: + client = FakeReadClient() + db, repo, service = setup(tmp, client) + try: + add_candidate(repo, 11, up_mid=300) + add_candidate(repo, 12, up_mid=300) + assert service.ignore(11).state == "rejected" + assert repo.get_candidate(11)["decision"] == "rejected" + result = service.blacklist(12) + assert result.ok is True + assert repo.is_blacklisted(300) + assert repo.get_candidate(12)["decision"] == "rejected" + assert service.ignore(999).state == "missing" + finally: + db.close() + + +def main(): + test_ingest_candidate_added_and_snapshot() + test_ingest_candidate_exists_and_missing() + test_ingest_candidate_recovers_dropped() + test_ingest_candidate_falls_back_when_detail_fails() + test_manual_ingest_forms_and_multiple() + test_manual_ingest_error_isolation() + test_snapshot_not_overwritten() + test_ignore_and_blacklist() + print("OK: 入库、手动入库、忽略、拉黑测试全部通过") + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/tests/test_telegram.py b/tests/test_telegram.py index bcd8eac..d88daef 100644 --- a/tests/test_telegram.py +++ b/tests/test_telegram.py @@ -11,11 +11,16 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) sys.stdout.reconfigure(encoding="utf-8") from reaction_board.notify.telegram import ( # noqa: E402 + HELP_TEXT, TelegramClient, TelegramError, format_duration, ) -from reaction_board.services.favorites import ActionResult # noqa: E402 +from reaction_board.services.library import ( # noqa: E402 + ActionResult, + IngestResult, + ManualIngestResult, +) from reaction_board.services.telegram_poller import TelegramPoller # noqa: E402 @@ -58,7 +63,7 @@ def test_send_candidates_and_chunking(): keyboard = json.loads(call["data"]["reply_markup"]) assert len(keyboard["inline_keyboard"]) == 2 assert [button["callback_data"] for button in keyboard["inline_keyboard"][0]] == [ - "fav:1", "ign:1", "blk:1", + "add:1", "ign:1", "blk:1", ] transport = FakeTransport() @@ -76,7 +81,7 @@ def test_send_candidates_dedupes_by_aid(): keyboard = json.loads(transport.calls[0]["data"]["reply_markup"]) assert len(keyboard["inline_keyboard"]) == 2 assert [row[0]["callback_data"] for row in keyboard["inline_keyboard"]] == [ - "fav:1", "fav:2", + "add:1", "add:2", ] @@ -88,7 +93,7 @@ def test_get_updates_and_answer(): assert data["offset"] == 99 assert data["timeout"] == 17 assert transport.calls[0]["timeout"] == 32 - assert json.loads(data["allowed_updates"]) == ["callback_query"] + assert json.loads(data["allowed_updates"]) == ["callback_query", "message"] transport = FakeTransport(result=True) bot = TelegramClient("token", 42, transport=transport) @@ -148,10 +153,13 @@ class FakeBot: class FakeActions: def __init__(self): self.calls = [] + self.manual = [] - def favorite(self, aid): - self.calls.append(("fav", aid)) - return ActionResult(True, aid, "favorite", "done", position=3, message="已加入收藏夹") + def ingest_candidate(self, aid): + self.calls.append(("add", aid)) + return IngestResult( + True, aid, "added", title="咻咻满 reaction", message="已入库" + ) def ignore(self, aid): self.calls.append(("ign", aid)) @@ -161,6 +169,16 @@ class FakeActions: self.calls.append(("blk", aid)) return ActionResult(True, aid, "blacklist", "rejected", message="已拉黑") + def ingest_text(self, text): + self.manual.append(text) + return ManualIngestResult( + refs=1, added=1, + results=[IngestResult(True, 9, "added", title="咻咻满 reaction", message="已入库")], + ) + + +ADD_REPLY = "✅ 已入库《咻咻满 reaction》（aid 9）" + def update(update_id, data, *, chat_id=42, from_id=42): return { @@ -174,33 +192,63 @@ def update(update_id, data, *, chat_id=42, from_id=42): } +def message_update(update_id, text, *, chat_id=42, from_id=42): + return { + "update_id": update_id, + "message": { + "message_id": update_id * 10, + "chat": {"id": chat_id}, + "from": {"id": from_id}, + "text": text, + }, + } + + def test_poller_dispatch_and_authorization(): bot = FakeBot([ - update(1, "fav:9"), + update(1, "add:9"), update(2, "ign:10"), update(3, "blk:11"), - update(4, "fav:12", chat_id=999), + update(4, "add:12", chat_id=999), update(5, "bad"), + update(6, "fav:13"), ]) actions = FakeActions() poller = TelegramPoller(bot, actions) offset = poller.run_once(None) - assert offset == 6 - assert actions.calls == [("fav", 9), ("ign", 10), ("blk", 11)] - assert bot.answers[0] == ("cb1", "已加入收藏夹（位置 #3）") + assert offset == 7 + assert actions.calls == [("add", 9), ("ign", 10), ("blk", 11), ("add", 13)] + assert bot.answers[0] == ("cb1", ADD_REPLY) + assert bot.answers[1] == ("cb2", "• 已忽略（aid 10）") + assert bot.answers[2] == ("cb3", "• 已拉黑（aid 11）") assert bot.answers[3] == ("cb4", "无权限") assert bot.answers[4] == ("cb5", "未知操作") - assert bot.edits == [10, 20, 30] + assert bot.edits == [10, 20, 30, 60] -def test_answer_failure_falls_back_to_text(): - bot = FakeBot([update(1, "fav:9")], fail_answer=True) +def test_poller_manual_message(): + bot = FakeBot([ + message_update(1, "https://www.bilibili.com/video/BV1x8x166EzY"), + message_update(2, "hello", chat_id=999), + message_update(3, "/help"), + ]) actions = FakeActions() poller = TelegramPoller(bot, actions) poller.run_once(None) - assert actions.calls == [("fav", 9)] + assert actions.manual == ["https://www.bilibili.com/video/BV1x8x166EzY"] + assert "入库" in bot.sent[0] + assert HELP_TEXT in bot.sent + assert len(bot.sent) == 2 + + +def test_answer_failure_falls_back_to_text(): + bot = FakeBot([update(1, "add:9")], fail_answer=True) + actions = FakeActions() + poller = TelegramPoller(bot, actions) + poller.run_once(None) + assert actions.calls == [("add", 9)] assert bot.answers == [] - assert bot.sent == ["已加入收藏夹（位置 #3）"] + assert bot.sent == [ADD_REPLY] assert bot.edits == [10] @@ -257,6 +305,7 @@ def main(): test_get_updates_and_answer() test_telegram_error_description_preserved() test_poller_dispatch_and_authorization() + test_poller_manual_message() test_answer_failure_falls_back_to_text() test_poller_backoff_and_recovery() print("OK: Telegram 消息、分片、长轮询与回调授权测试全部通过")