feat(p3): add discovery, telegram notifications, and one-click favorites

Implement WBI video search with legacy fallback, classification scoring, candidate pool, permanent seen-aid dedupe, blacklist filtering, and batched Telegram notifications. Add write client for fav deal/sort, target-folder selection, insertion position calculation, read-back verification, fav_action locking, retry on failure, ignore and blacklist actions. Wire discovery, candidate cleanup, Telegram long polling, and new discover/candidates CLI commands into the scheduler. Add migration v3 for candidate duration and index stats.

reaction_board/bilibili/search.py

@@ -0,0 +1,206 @@
+# -*- coding: utf-8 -*-
+"""B 站视频搜索（WBI 签名 + 旧版降级）。"""
+
+from __future__ import annotations
+
+import hashlib
+import re
+import time
+import urllib.parse
+from dataclasses import dataclass
+from typing import Any, Callable, Mapping
+
+from ..classify import parse_duration
+from .client import BASE, BilibiliClient, BilibiliError, BilibiliProtocolError
+
+# B 站前端固定的 mixinKey 重排表。
+MIXIN_KEY_ENC_TAB = (
+    46, 47, 18, 2, 53, 8, 23, 32, 15, 50, 10, 31, 58, 3, 45, 35,
+    27, 43, 5, 49, 33, 9, 42, 19, 29, 28, 14, 39, 12, 38, 41, 13,
+    37, 48, 7, 16, 24, 55, 40, 61, 26, 17, 0, 1, 60, 51, 30, 4,
+    22, 25, 54, 21, 56, 59, 6, 63, 57, 62, 11, 36, 20, 34, 44, 52,
+)
+WBI_KEY_TTL = 6 * 3600
+_TAG_RE = re.compile(r"<[^>]+>")
+
+
+@dataclass(frozen=True)
+class SearchItem:
+    aid: int
+    bvid: str | None
+    title: str
+    up_mid: int | None
+    up_name: str | None
+    pubdate: int
+    duration_seconds: int
+    cover: str | None
+    play: int = 0
+    like: int = 0
+    favorite: int = 0
+    danmaku: int = 0
+    review: int = 0
+
+
+def key_from_url(url: str | None) -> str:
+    """从 ``.../bfs/wbi/xxx.png`` 提取 ``xxx``。"""
+    if not isinstance(url, str) or not url:
+        return ""
+    path = urllib.parse.urlparse(url).path
+    name = path.rsplit("/", 1)[-1]
+    return name.split(".", 1)[0]
+
+
+def mixin_key(img_key: str, sub_key: str) -> str:
+    raw = f"{img_key}{sub_key}"
+    if len(raw) < 64:
+        raise ValueError("WBI key 长度不足")
+    return "".join(raw[index] for index in MIXIN_KEY_ENC_TAB)[:32]
+
+
+def sign_params(
+    params: Mapping[str, Any], key: str, *, now: float | None = None
+) -> dict[str, Any]:
+    signed = {str(k): v for k, v in params.items()}
+    signed["wts"] = int(time.time() if now is None else now)
+    query = urllib.parse.urlencode(sorted(signed.items()))
+    signed["w_rid"] = hashlib.md5(f"{query}{key}".encode("utf-8")).hexdigest()
+    return signed
+
+
+def _as_int(value: Any) -> int:
+    if isinstance(value, bool) or not isinstance(value, int):
+        return 0
+    return value
+
+
+def _clean_title(value: Any) -> str:
+    text = value if isinstance(value, str) else ""
+    return _TAG_RE.sub("", text).strip()
+
+
+def _cover(value: Any) -> str | None:
+    if not isinstance(value, str) or not value:
+        return None
+    if value.startswith("//"):
+        return f"https:{value}"
+    return value
+
+
+def _parse_duration(value: Any) -> int:
+    return parse_duration(value)
+
+
+def parse_search_payload(payload: Mapping[str, Any]) -> list[SearchItem]:
+    data = payload.get("data")
+    if not isinstance(data, Mapping):
+        raise BilibiliProtocolError("search data 不是对象")
+    if "result" not in data:
+        raise BilibiliProtocolError("search data 缺少 result 字段")
+    rows = data["result"]
+    if rows is None:
+        return []
+    if not isinstance(rows, list):
+        raise BilibiliProtocolError("search data.result 不是数组")
+
+    items: list[SearchItem] = []
+    for row in rows:
+        if not isinstance(row, Mapping):
+            raise BilibiliProtocolError("search result item 不是对象")
+        raw_type = row.get("type")
+        if isinstance(raw_type, str) and raw_type and raw_type != "video":
+            continue
+        aid = row.get("id")
+        if isinstance(aid, bool) or not isinstance(aid, int):
+            continue
+        up_mid = row.get("mid")
+        if isinstance(up_mid, bool) or not isinstance(up_mid, int):
+            up_mid = None
+        bvid = row.get("bvid")
+        if not isinstance(bvid, str) or not bvid:
+            bvid = None
+        up_name = row.get("author")
+        if not isinstance(up_name, str) or not up_name:
+            up_name = None
+        pubdate = row.get("pubdate")
+        if isinstance(pubdate, bool) or not isinstance(pubdate, int):
+            pubdate = 0
+        items.append(SearchItem(
+            aid=aid,
+            bvid=bvid,
+            title=_clean_title(row.get("title")),
+            up_mid=up_mid,
+            up_name=up_name,
+            pubdate=pubdate,
+            duration_seconds=_parse_duration(row.get("duration")),
+            cover=_cover(row.get("pic")),
+            play=_as_int(row.get("play")),
+            like=_as_int(row.get("like")),
+            favorite=_as_int(row.get("favorites")),
+            danmaku=_as_int(row.get("danmaku")),
+            review=_as_int(row.get("review")),
+        ))
+    return items
+
+
+class SearchClient:
+    """带 WBI 密钥缓存与旧接口降级的搜索客户端。"""
+
+    def __init__(
+        self,
+        client_getter: Callable[[], BilibiliClient],
+        *,
+        clock=time.time,
+    ) -> None:
+        self._client_getter = client_getter
+        self._clock = clock
+        self._keys: tuple[str, str] | None = None
+        self._keys_at = 0.0
+
+    def _wbi_keys(self) -> tuple[str, str]:
+        now = self._clock()
+        if self._keys is not None and now - self._keys_at < WBI_KEY_TTL:
+            return self._keys
+        payload = self._client_getter().get_json(
+            f"{BASE}/x/web-interface/nav", auth=True
+        )
+        data = payload.get("data")
+        if not isinstance(data, Mapping):
+            raise BilibiliProtocolError("nav data 不是对象")
+        wbi = data.get("wbi_img")
+        if not isinstance(wbi, Mapping):
+            raise BilibiliProtocolError("nav data.wbi_img 缺失")
+        img_key = key_from_url(wbi.get("img_url"))
+        sub_key = key_from_url(wbi.get("sub_url"))
+        if not img_key or not sub_key:
+            raise BilibiliProtocolError("WBI key 提取失败")
+        self._keys = (img_key, sub_key)
+        self._keys_at = now
+        return self._keys
+
+    def search(
+        self, keyword: str, page: int, *, page_size: int = 42
+    ) -> list[SearchItem]:
+        params = {
+            "search_type": "video",
+            "keyword": keyword,
+            "order": "pubdate",
+            "page": int(page),
+            "page_size": int(page_size),
+        }
+        try:
+            img_key, sub_key = self._wbi_keys()
+            signed = sign_params(
+                params, mixin_key(img_key, sub_key), now=self._clock()
+            )
+            url = (
+                f"{BASE}/x/web-interface/wbi/search/type?"
+                f"{urllib.parse.urlencode(signed)}"
+            )
+            return parse_search_payload(self._client_getter().get_json(url))
+        except BilibiliError:
+            # WBI 失败时降级到旧接口；旧接口也失败则抛出。
+            url = (
+                f"{BASE}/x/web-interface/search/type?"
+                f"{urllib.parse.urlencode(params)}"
+            )
+            return parse_search_payload(self._client_getter().get_json(url))

reaction_board/bilibili/write_client.py

@@ -0,0 +1,60 @@
+# -*- coding: utf-8 -*-
+"""B 站收藏夹写客户端（``deal`` / ``sort``）。
+
+只应使用 ``bilibili.write`` 凭据构造。
+"""
+
+from __future__ import annotations
+
+import urllib.parse
+from typing import Any, Mapping
+
+from .client import (
+    BASE,
+    BilibiliClient,
+    BilibiliHTTPError,
+    BilibiliProtocolError,
+)
+
+
+class BilibiliWriteClient(BilibiliClient):
+    def post_form(self, url: str, data: Mapping[str, Any]) -> dict[str, Any]:
+        self.pacer.wait()
+        headers = self._headers(auth=True)
+        headers["content-type"] = "application/x-www-form-urlencoded"
+        body = urllib.parse.urlencode(data).encode("utf-8")
+        response = self.transport.request(
+            "POST", url, headers, data=body, timeout=self.timeout
+        )
+        if response.status >= 400:
+            raise BilibiliHTTPError(
+                response.status, response.body.decode("utf-8", "replace")[:200]
+            )
+        payload = response.json()
+        if not isinstance(payload, dict):
+            raise BilibiliProtocolError("写接口响应顶层不是 JSON 对象")
+        if payload.get("code") != 0:
+            raise self._api_error(payload)
+        return payload
+
+    def deal(self, aid: int, folder_id: int) -> dict[str, Any]:
+        return self.post_form(
+            f"{BASE}/x/v3/fav/resource/deal",
+            {
+                "rid": int(aid),
+                "type": 2,
+                "add_media_ids": int(folder_id),
+                "del_media_ids": "",
+                "csrf": self.csrf,
+            },
+        )
+
+    def sort(self, media_id: int, sort_spec: str) -> dict[str, Any]:
+        return self.post_form(
+            f"{BASE}/x/v3/fav/resource/sort",
+            {
+                "media_id": int(media_id),
+                "sort": sort_spec,
+                "csrf": self.csrf,
+            },
+        )

reaction_board/classify.py

@@ -0,0 +1,89 @@
+# -*- coding: utf-8 -*-
+"""候选分类与时长解析。
+
+规则见 project.md §3.3：分类只决定“是否通知”，不决定是否入库。
+"""
+
+from __future__ import annotations
+
+import re
+import unicodedata
+from typing import Any, Mapping
+
+_TAG_RE = re.compile(r"<[^>]+>")
+
+
+def normalize_title(text: str) -> str:
+    """全半角归一化并转小写，便于标题关键词匹配。"""
+    return unicodedata.normalize("NFKC", text or "").lower()
+
+
+def clean_title(text: str) -> str:
+    """去掉搜索结果里的 ``<em class="keyword">`` 高亮标签。"""
+    return _TAG_RE.sub("", text or "").strip()
+
+
+def parse_duration(value: Any) -> int:
+    """解析 B 站搜索结果时长。
+
+    支持 ``105:5``（分钟:秒）、``1:02:03``（时:分:秒）和整数秒。
+    """
+    if isinstance(value, bool):
+        return 0
+    if isinstance(value, int):
+        return max(0, value)
+    if not isinstance(value, str):
+        return 0
+    parts = value.strip().split(":")
+    if not parts or any(not part.isdigit() for part in parts):
+        return 0
+    numbers = [int(part) for part in parts]
+    if len(numbers) == 1:
+        return numbers[0]
+    if len(numbers) == 2:
+        return numbers[0] * 60 + numbers[1]
+    if len(numbers) == 3:
+        return numbers[0] * 3600 + numbers[1] * 60 + numbers[2]
+    return 0
+
+
+def contains_any(title: str, words: Any) -> bool:
+    normalized = normalize_title(title)
+    if not isinstance(words, (list, tuple)):
+        return False
+    return any(normalize_title(str(word)) in normalized for word in words if word)
+
+
+def passes_required_title(title: str, require_title: str) -> bool:
+    if not require_title:
+        return True
+    return normalize_title(require_title) in normalize_title(title)
+
+
+def score_item(
+    *,
+    title: str,
+    up_mid: int | None,
+    duration_seconds: int,
+    classify: Mapping[str, Any],
+) -> int:
+    """按 project.md §3.3 计算候选分数。"""
+    score = 0
+    if contains_any(title, classify.get("positive")):
+        score += 2
+    if contains_any(title, classify.get("charge")):
+        score += 2
+
+    boost = classify.get("duration_boost") or {}
+    gt_60 = int(boost.get("gt_60min", 2))
+    gt_30 = int(boost.get("gt_30min", 1))
+    if duration_seconds > 60 * 60:
+        score += gt_60
+    elif duration_seconds > 30 * 60:
+        score += gt_30
+
+    if contains_any(title, classify.get("negative")):
+        score -= 5
+    if up_mid is not None and int(up_mid) == int(classify.get("self_mid", 37754047)):
+        score -= 3
+    return score

reaction_board/cli.py

@@ -4,6 +4,8 @@
     python app.py init        # 建库 / 迁移 / 写入黑名单种子（离线）
     python app.py sync        # 收藏夹只读同步（首次导入或增量）
     python app.py snapshot    # 采集当日七项快照与粉丝数
+    python app.py discover    # 执行一次搜索发现并写入候选池
+    python app.py candidates  # 查看候选池（离线）
     python app.py auth-check  # 检查 Cookie 登录态
     python app.py run         # 按配置周期常驻运行（--once 只跑一次）
     python app.py status      # 查看库内状态（离线）
@@ -16,19 +18,26 @@ from __future__ import annotations
 import argparse
 import json
 import sys
+import threading
 from pathlib import Path
 from typing import Sequence
 
 from .bilibili.auth_flow import AuthService
 from .bilibili.client import BilibiliClient, Pacer
 from .bilibili.provider import ClientProvider
+from .bilibili.search import SearchClient
+from .bilibili.write_client import BilibiliWriteClient
 from .config import ConfigError, ConfigStore
 from .database import Database
 from .logging_json import JsonLogger
+from .notify.telegram import TelegramClient
 from .repository import Repository
 from .scheduler import ScheduledTask, Scheduler
+from .services.discovery import DiscoveryService
+from .services.favorites import FavoriteActionService
 from .services.folder_sync import FolderSyncService
 from .services.snapshot import SnapshotService
+from .services.telegram_poller import TelegramPoller
 from .timeutil import next_daily_run, parse_hhmm, shanghai_date
 
 ROOT = Path(__file__).resolve().parent.parent
@@ -65,6 +74,23 @@ def _build_parser() -> argparse.ArgumentParser:
     snapshot.add_argument(
         "--force", action="store_true", help="忽略当日完成标记，重新采集"
     )
+    discover = sub.add_parser(
+        "discover", parents=[common], help="执行一次搜索发现并写入候选池"
+    )
+    discover.add_argument(
+        "--no-notify",
+        action="store_true",
+        help="只写候选池，不发送 Telegram 通知",
+    )
+    candidates = sub.add_parser(
+        "candidates", parents=[common], help="查看候选池（离线）"
+    )
+    candidates.add_argument(
+        "--decision",
+        choices=("accepted", "rejected"),
+        default=None,
+        help="只查看指定决策状态的候选",
+    )
     auth = sub.add_parser("auth", parents=[common], help="扫码授权并更新指定账号 Cookie")
     auth.add_argument("--account", choices=("read", "write"), default="read",
                       help="要授权的账号角色（默认 read）")
@@ -179,6 +205,64 @@ def cmd_snapshot(args, logger: JsonLogger) -> int:
         db.close()
 
 
+def _telegram(cfg, logger: JsonLogger):
+    if not cfg.telegram.enabled:
+        return None
+    if not cfg.telegram.bot_token or not cfg.telegram.chat_id:
+        raise ConfigError(
+            "telegram.enabled=true 时必须配置 bot_token 和 chat_id"
+        )
+    return TelegramClient(
+        cfg.telegram.bot_token,
+        cfg.telegram.chat_id,
+        logger=logger,
+    )
+
+
+def cmd_discover(args, logger: JsonLogger) -> int:
+    store = ConfigStore(args.config, require_auth=True)
+    cfg = store.get()
+    db = _open_db(cfg, logger)
+    try:
+        repo = Repository(db)
+        provider = ClientProvider(
+            store, "read", lambda credentials: _client(credentials, logger)
+        )
+        service = DiscoveryService(
+            store,
+            repo,
+            SearchClient(provider.get),
+            notifier=None if args.no_notify else _telegram(cfg, logger),
+            logger=logger,
+        )
+        summary = service.run()
+        payload = summary.as_dict()
+        payload["candidate_total"] = len(repo.list_candidates())
+        payload["database"] = str(cfg.storage.database_path)
+        _print(payload)
+        return 0 if not summary.errors else 1
+    finally:
+        db.close()
+
+
+def cmd_candidates(args, logger: JsonLogger) -> int:
+    store = ConfigStore(args.config, require_auth=False)
+    cfg = store.get()
+    db = _open_db(cfg, logger)
+    try:
+        repo = Repository(db)
+        rows = repo.list_candidates(args.decision)
+        _print({
+            "ok": True,
+            "count": len(rows),
+            "candidates": rows,
+            "database": str(cfg.storage.database_path),
+        })
+        return 0
+    finally:
+        db.close()
+
+
 def cmd_auth(args, logger: JsonLogger) -> int:
     if args.timeout < 1:
         raise ConfigError("auth --timeout 必须至少为 1 秒")
@@ -242,6 +326,47 @@ def cmd_run(args, logger: JsonLogger) -> int:
         snapshot_service = SnapshotService(
             store, repo, client_provider=snapshot_provider, logger=logger
         )
+        write_provider = ClientProvider(
+            store,
+            "write",
+            lambda credentials: BilibiliWriteClient(
+                cookie=credentials.cookie,
+                csrf=credentials.csrf,
+                mid=credentials.mid,
+                logger=logger,
+            ),
+        )
+        favorite_service = FavoriteActionService(
+            store, repo, write_provider, logger=logger
+        )
+        notifier = _telegram(cfg, logger)
+        discovery_service = DiscoveryService(
+            store,
+            repo,
+            SearchClient(folder_provider.get),
+            notifier=notifier,
+            logger=logger,
+        )
+
+        def run_discover():
+            summary = discovery_service.run()
+            last["discover"] = summary
+            logger.info("discovery.summary", **summary.as_dict())
+            return summary
+
+        def run_cleanup():
+            removed = repo.delete_expired_candidates(
+                ttl_days=cfg.schedule.candidate_ttl_days
+            )
+            expired = repo.expire_fav_actions()
+            logger.info(
+                "cleanup.done",
+                candidates_removed=removed, fav_actions_expired=expired,
+            )
+            return {
+                "candidates_removed": removed,
+                "fav_actions_expired": expired,
+            }
 
         def run_snapshot():
             summary = snapshot_service.run_if_needed()
@@ -281,10 +406,24 @@ def cmd_run(args, logger: JsonLogger) -> int:
                     next_run=0.0,
                     next_run_fn=lambda started: next_daily_run(started, snapshot_at),
                 ),
+                ScheduledTask(
+                    "discovery",
+                    float(cfg.schedule.discover_minutes * 60),
+                    run_discover,
+                ),
+                ScheduledTask("candidate_cleanup", 86400.0, run_cleanup),
                 ScheduledTask("config_watch", 60.0, run_config_watch),
             ],
             logger=logger,
         )
+        if notifier is not None and not args.once:
+            poller = TelegramPoller(notifier, favorite_service, logger=logger)
+            threading.Thread(
+                target=poller.run_forever,
+                name="telegram-poller",
+                daemon=True,
+            ).start()
+            logger.info("telegram.polling_started")
         if args.once:
             results = scheduler.run_due()
             _print([result.as_dict() for result in results])
@@ -298,9 +437,11 @@ def cmd_run(args, logger: JsonLogger) -> int:
 
         folder_summary = last.get("folder")
         snapshot_summary = last.get("snapshot")
+        discover_summary = last.get("discover")
         folder_ok = folder_summary is None or getattr(folder_summary, "ok", True)
         snapshot_ok = snapshot_summary is None or getattr(snapshot_summary, "completed", True)
-        return 0 if folder_ok and snapshot_ok else 1
+        discover_ok = discover_summary is None or not getattr(discover_summary, "errors", [])
+        return 0 if folder_ok and snapshot_ok and discover_ok else 1
     finally:
         db.close()
 
@@ -356,6 +497,10 @@ def main(argv: Sequence[str] | None = None) -> int:
             return cmd_sync(args, logger)
         if args.command == "snapshot":
             return cmd_snapshot(args, logger)
+        if args.command == "discover":
+            return cmd_discover(args, logger)
+        if args.command == "candidates":
+            return cmd_candidates(args, logger)
         if args.command == "auth":
             return cmd_auth(args, logger)
         if args.command == "auth-check":

reaction_board/config.py

@@ -18,6 +18,7 @@ from typing import Any, Mapping
 
 DEFAULT_CLASSIFY: dict[str, Any] = {
     "require_title": "咻咻满",
+    "self_mid": 37754047,
     "positive": [
         "reaction", "re", "反应", "看", "听", "初听", "首次",
         "点评", "锐评", "路人", "震惊", "双人",
@@ -64,6 +65,11 @@ DEFAULT_CONFIG: dict[str, Any] = {
         "snapshot_at": "00:01",
         "candidate_ttl_days": 7,
     },
+    "discover": {
+        "keywords": ["咻咻满 reaction", "咻咻满 充电"],
+        "pages": 2,
+        "window_days": 14,
+    },
     "notify": {"min_score": 2, "max_items_per_message": 10},
     "classify": DEFAULT_CLASSIFY,
     "heat_weights": DEFAULT_HEAT_WEIGHTS,
@@ -148,6 +154,13 @@ class NotifyConfig:
     max_items_per_message: int
 
 
+@dataclass(frozen=True)
+class DiscoverConfig:
+    keywords: tuple[str, ...]
+    pages: int
+    window_days: int
+
+
 @dataclass(frozen=True)
 class AppConfig:
     path: Path
@@ -160,6 +173,7 @@ class AppConfig:
     up_blacklist_seed: tuple[int, ...]
     schedule: ScheduleConfig
     notify: NotifyConfig
+    discover: DiscoverConfig
     classify: Mapping[str, Any]
     heat_weights: Mapping[str, float]
     raw: Mapping[str, Any] = field(repr=False, compare=False)
@@ -367,6 +381,24 @@ def _build_config(data: Mapping[str, Any], path: Path) -> AppConfig:
         ),
     )
 
+    discover_raw = data["discover"]
+    if not isinstance(discover_raw, Mapping):
+        raise ConfigError("discover 必须是对象")
+    keywords = discover_raw.get("keywords", [])
+    if not isinstance(keywords, list) or not keywords:
+        raise ConfigError("discover.keywords 必须是非空数组")
+    discover = DiscoverConfig(
+        keywords=tuple(
+            _as_str(item, "discover.keywords[]") for item in keywords
+        ),
+        pages=_as_int(
+            discover_raw.get("pages", 2), "discover.pages", minimum=1, maximum=10
+        ),
+        window_days=_as_int(
+            discover_raw.get("window_days", 14), "discover.window_days", minimum=1
+        ),
+    )
+
     classify_raw = data["classify"]
     if not isinstance(classify_raw, Mapping):
         raise ConfigError("classify 必须是对象")
@@ -376,6 +408,9 @@ def _build_config(data: Mapping[str, Any], path: Path) -> AppConfig:
     )
     if not classify["require_title"]:
         raise ConfigError("classify.require_title 不能为空")
+    classify["self_mid"] = _as_int(
+        classify.get("self_mid", 37754047), "classify.self_mid", minimum=1
+    )
     for key in ("positive", "charge", "negative"):
         classify[key] = _as_str_list(classify.get(key, []), f"classify.{key}")
     boost = classify.get("duration_boost", {})
@@ -407,6 +442,7 @@ def _build_config(data: Mapping[str, Any], path: Path) -> AppConfig:
         up_blacklist_seed=_as_int_list(data["up_blacklist_seed"], "up_blacklist_seed"),
         schedule=schedule,
         notify=notify,
+        discover=discover,
         classify=classify,
         heat_weights=heat,
         raw=data,

reaction_board/fav_position.py

@@ -0,0 +1,82 @@
+# -*- coding: utf-8 -*-
+"""新视频在收藏夹中的插入位置计算（纯函数）。"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Mapping, Sequence
+
+
+@dataclass(frozen=True)
+class VideoMeta:
+    up_mid: int | None
+    pubdate: int
+
+
+@dataclass(frozen=True)
+class InsertionPlan:
+    target_position: int
+    moves: tuple[tuple[int, int], ...]  # (pred_aid, moved_aid)
+
+    @property
+    def needs_head_move(self) -> bool:
+        return len(self.moves) == 2
+
+
+def plan_insertion(
+    current_ids: Sequence[int],
+    new_aid: int,
+    metadata: Mapping[int, VideoMeta],
+    new_meta: VideoMeta,
+) -> InsertionPlan:
+    """按“同 UP 连续、组间按最早投稿倒序”计算插入点。
+
+    ``current_ids`` 可以包含 ``new_aid``（通常在末尾），函数会先排除它。
+    """
+    ordered = [int(aid) for aid in current_ids if int(aid) != int(new_aid)]
+    if not ordered:
+        return InsertionPlan(target_position=1, moves=())
+
+    groups: list[dict] = []
+    for index, aid in enumerate(ordered):
+        meta = metadata.get(aid)
+        up_mid = meta.up_mid if meta else None
+        if up_mid is not None and groups and groups[-1]["up_mid"] == up_mid:
+            groups[-1]["aids"].append(aid)
+        else:
+            groups.append({"up_mid": up_mid, "aids": [aid], "start": index})
+
+    for group in groups:
+        keys = [
+            metadata[aid].pubdate
+            for aid in group["aids"]
+            if aid in metadata and metadata[aid].pubdate > 0
+        ]
+        group["key"] = min(keys) if keys else 0
+
+    insert_index: int | None = None
+    if new_meta.up_mid is not None:
+        for group in groups:
+            if group["up_mid"] == new_meta.up_mid:
+                insert_index = group["start"]
+                break
+    if insert_index is None:
+        for group in groups:
+            if group["key"] < new_meta.pubdate:
+                insert_index = group["start"]
+                break
+    if insert_index is None:
+        insert_index = len(ordered)
+
+    target_position = insert_index + 1
+    if insert_index == 0:
+        first = ordered[0]
+        return InsertionPlan(
+            target_position=1,
+            moves=((first, int(new_aid)), (int(new_aid), first)),
+        )
+    predecessor = ordered[insert_index - 1]
+    return InsertionPlan(
+        target_position=target_position,
+        moves=((predecessor, int(new_aid)),),
+    )

reaction_board/migrations.py

@@ -174,6 +174,18 @@ MIGRATIONS: tuple[Migration, ...] = (
             "ALTER TABLE folder_resource ADD COLUMN fav_time INTEGER",
         ),
     ),
+    Migration(
+        version=3,
+        name="candidate stats",
+        statements=(
+            "ALTER TABLE candidate ADD COLUMN duration INTEGER",
+            "ALTER TABLE candidate ADD COLUMN play INTEGER",
+            "ALTER TABLE candidate ADD COLUMN like INTEGER",
+            "ALTER TABLE candidate ADD COLUMN favorite INTEGER",
+            "ALTER TABLE candidate ADD COLUMN danmaku INTEGER",
+            "ALTER TABLE candidate ADD COLUMN review INTEGER",
+        ),
+    ),
 )
 
 

reaction_board/notify/__init__.py

@@ -0,0 +1,2 @@
+# -*- coding: utf-8 -*-
+"""通知渠道。"""

reaction_board/notify/telegram.py

@@ -0,0 +1,136 @@
+# -*- coding: utf-8 -*-
+"""Telegram Bot 通知与回调。"""
+
+from __future__ import annotations
+
+import html
+import json
+import urllib.parse
+import urllib.request
+from typing import Any, Mapping, Protocol, Sequence
+
+MAX_CANDIDATES_PER_MESSAGE = 3
+
+
+class TelegramError(RuntimeError):
+    pass
+
+
+class TelegramTransport(Protocol):
+    def post(self, url: str, data: Mapping[str, Any], timeout: float = 30) -> Any: ...
+
+
+class UrllibTelegramTransport:
+    def post(self, url: str, data: Mapping[str, Any], timeout: float = 30) -> Any:
+        request = urllib.request.Request(
+            url,
+            data=urllib.parse.urlencode(data).encode("utf-8"),
+            method="POST",
+        )
+        with urllib.request.urlopen(request, timeout=timeout) as response:
+            return json.loads(response.read().decode("utf-8", "replace"))
+
+
+def format_duration(seconds: int) -> str:
+    seconds = max(0, int(seconds or 0))
+    hours, remainder = divmod(seconds, 3600)
+    minutes, secs = divmod(remainder, 60)
+    if hours:
+        return f"{hours}:{minutes:02d}:{secs:02d}"
+    return f"{minutes}:{secs:02d}"
+
+
+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"))
+    duration = format_duration(int(candidate.get("duration") or 0))
+    play = int(candidate.get("play") or 0)
+    like = int(candidate.get("like") or 0)
+    return (
+        f"🎬 <b>{title}</b>\n"
+        f"UP: {up_name}\n"
+        f"时长 {duration} · 播放 {play} · 点赞 {like}（索引值）"
+    )
+
+
+class TelegramClient:
+    def __init__(
+        self,
+        token: str,
+        chat_id: int,
+        *,
+        transport: TelegramTransport | None = None,
+        timeout: float = 30,
+        poll_timeout: int = 25,
+        logger=None,
+    ) -> None:
+        if not token:
+            raise ValueError("telegram.bot_token 不能为空")
+        if not chat_id:
+            raise ValueError("telegram.chat_id 不能为空")
+        self.token = token
+        self.chat_id = int(chat_id)
+        self.transport = transport or UrllibTelegramTransport()
+        self.timeout = timeout
+        self.poll_timeout = poll_timeout
+        self.logger = logger
+
+    def _call(self, method: str, params: Mapping[str, Any] | None = None) -> Any:
+        url = f"https://api.telegram.org/bot{self.token}/{method}"
+        payload = self.transport.post(url, params or {}, timeout=self.timeout)
+        if not isinstance(payload, Mapping):
+            raise TelegramError("Telegram 响应不是对象")
+        if not payload.get("ok"):
+            raise TelegramError(str(payload.get("description") or "Telegram API 错误"))
+        return payload.get("result")
+
+    def send_text(
+        self, text: str, *, reply_markup: Mapping[str, Any] | None = None
+    ) -> Any:
+        params: dict[str, Any] = {
+            "chat_id": self.chat_id,
+            "text": text,
+            "parse_mode": "HTML",
+            "disable_web_page_preview": True,
+        }
+        if reply_markup is not None:
+            params["reply_markup"] = json.dumps(reply_markup, ensure_ascii=False)
+        return self._call("sendMessage", params)
+
+    def send_candidates(self, candidates: Sequence[Mapping[str, Any]]) -> bool:
+        items = list(candidates)
+        if not items:
+            return True
+        for index in range(0, len(items), MAX_CANDIDATES_PER_MESSAGE):
+            chunk = items[index:index + MAX_CANDIDATES_PER_MESSAGE]
+            text = "\n\n".join(format_candidate(candidate) for candidate in chunk)
+            keyboard = {
+                "inline_keyboard": [
+                    [
+                        {"text": "⭐ 收藏", "callback_data": f"fav:{candidate['aid']}"},
+                        {"text": "🙈 忽略", "callback_data": f"ign:{candidate['aid']}"},
+                        {"text": "🚫 拉黑", "callback_data": f"blk:{candidate['aid']}"},
+                    ]
+                    for candidate in chunk
+                ]
+            }
+            self.send_text(text, reply_markup=keyboard)
+        return True
+
+    def get_updates(
+        self, offset: int | None = None, *, timeout: int | None = None
+    ) -> 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"]),
+        }
+        if offset is not None:
+            params["offset"] = int(offset)
+        result = self._call("getUpdates", params)
+        return result if isinstance(result, list) else []
+
+    def answer_callback(self, callback_id: str, text: str = "") -> Any:
+        return self._call(
+            "answerCallbackQuery",
+            {"callback_query_id": callback_id, "text": text[:200]},
+        )

reaction_board/repository.py

@@ -555,6 +555,296 @@ class Repository:
             (key, str(value)),
         ))
 
+    # ------------------------------------------------------------ candidate
+
+    def known_aid(self, aid: int) -> bool:
+        def run(conn: sqlite3.Connection) -> bool:
+            row = conn.execute(
+                "SELECT 1 FROM video WHERE aid = ? "
+                "UNION ALL SELECT 1 FROM seen_aid WHERE aid = ? LIMIT 1",
+                (int(aid), int(aid)),
+            ).fetchone()
+            return row is not None
+        return self.db.read(run)
+
+    def record_seen(
+        self, aid: int, *, source: str, now: int | None = None
+    ) -> None:
+        now = self._now() if now is None else int(now)
+        self.db.write(lambda conn: conn.execute(
+            "INSERT INTO seen_aid(aid, first_seen, source) VALUES (?, ?, ?) "
+            "ON CONFLICT(aid) DO NOTHING",
+            (int(aid), now, source),
+        ))
+
+    def upsert_candidate(
+        self,
+        *,
+        aid: int,
+        bvid: str | None,
+        title: str,
+        up_mid: int | None,
+        up_name: str | None,
+        pubdate: int,
+        cover: str | None,
+        score: int,
+        channel: str,
+        duration: int = 0,
+        play: int = 0,
+        like: int = 0,
+        favorite: int = 0,
+        danmaku: int = 0,
+        review: int = 0,
+        now: int | None = None,
+    ) -> dict[str, Any]:
+        now = self._now() if now is None else int(now)
+
+        def run(conn: sqlite3.Connection) -> dict[str, Any]:
+            conn.execute(
+                """
+                INSERT INTO candidate(aid, bvid, title, up_mid, up_name, pubdate,
+                                      cover, score, channel, first_seen,
+                                      notified_at, decision, decided_at,
+                                      duration, play, like, favorite, danmaku, review)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL, NULL,
+                        ?, ?, ?, ?, ?, ?)
+                ON CONFLICT(aid) DO UPDATE SET
+                  bvid = COALESCE(excluded.bvid, candidate.bvid),
+                  title = COALESCE(NULLIF(excluded.title, ''), candidate.title),
+                  up_mid = COALESCE(excluded.up_mid, candidate.up_mid),
+                  up_name = COALESCE(excluded.up_name, candidate.up_name),
+                  pubdate = CASE WHEN excluded.pubdate > 0
+                                 THEN excluded.pubdate ELSE candidate.pubdate END,
+                  cover = COALESCE(excluded.cover, candidate.cover),
+                  score = excluded.score,
+                  channel = excluded.channel,
+                  duration = excluded.duration,
+                  play = excluded.play,
+                  "like" = excluded."like",
+                  favorite = excluded.favorite,
+                  danmaku = excluded.danmaku,
+                  review = excluded.review
+                """,
+                (
+                    int(aid), bvid, title, up_mid, up_name, int(pubdate),
+                    cover, float(score), channel, now,
+                    int(duration or 0), int(play or 0), int(like or 0),
+                    int(favorite or 0), int(danmaku or 0), int(review or 0),
+                ),
+            )
+            row = conn.execute(
+                "SELECT * FROM candidate WHERE aid = ?", (int(aid),)
+            ).fetchone()
+            return dict(row) if row else {}
+        return self.db.write(run)
+
+    def get_candidate(self, aid: int) -> dict[str, Any] | None:
+        def run(conn: sqlite3.Connection) -> dict[str, Any] | None:
+            row = conn.execute(
+                "SELECT * FROM candidate WHERE aid = ?", (int(aid),)
+            ).fetchone()
+            return dict(row) if row else None
+        return self.db.read(run)
+
+    def list_candidates(self, decision: str | None = None) -> list[dict[str, Any]]:
+        def run(conn: sqlite3.Connection) -> list[dict[str, Any]]:
+            if decision is None:
+                rows = conn.execute(
+                    "SELECT * FROM candidate ORDER BY COALESCE(score, 0) DESC, "
+                    "first_seen DESC, aid"
+                ).fetchall()
+            else:
+                rows = conn.execute(
+                    "SELECT * FROM candidate WHERE decision = ? "
+                    "ORDER BY COALESCE(score, 0) DESC, first_seen DESC, aid",
+                    (decision,),
+                ).fetchall()
+            return [dict(row) for row in rows]
+        return self.db.read(run)
+
+    def mark_candidate_decision(
+        self, aid: int, decision: str, *, now: int | None = None
+    ) -> None:
+        now = self._now() if now is None else int(now)
+        self.db.write(lambda conn: conn.execute(
+            "UPDATE candidate SET decision = ?, decided_at = ? WHERE aid = ?",
+            (decision, now, int(aid)),
+        ))
+
+    def reject_pending_for_up(self, up_mid: int, *, now: int | None = None) -> int:
+        now = self._now() if now is None else int(now)
+        def run(conn: sqlite3.Connection) -> int:
+            cur = conn.execute(
+                "UPDATE candidate SET decision = 'rejected', decided_at = ? "
+                "WHERE up_mid = ? AND decision IS NULL",
+                (now, int(up_mid)),
+            )
+            return max(cur.rowcount, 0)
+        return self.db.write(run)
+
+    def delete_candidate(self, aid: int) -> None:
+        self.db.write(lambda conn: conn.execute(
+            "DELETE FROM candidate WHERE aid = ?", (int(aid),)
+        ))
+
+    def delete_expired_candidates(
+        self, *, ttl_days: int, now: int | None = None
+    ) -> int:
+        now = self._now() if now is None else int(now)
+        cutoff = now - int(ttl_days) * 86400
+        def run(conn: sqlite3.Connection) -> int:
+            cur = conn.execute(
+                "DELETE FROM candidate WHERE decision IS NULL AND first_seen < ?",
+                (cutoff,),
+            )
+            return max(cur.rowcount, 0)
+        return self.db.write(run)
+
+    def mark_notified(
+        self,
+        aid: int,
+        *,
+        channel: str,
+        batch_id: str | None,
+        ok: bool,
+        err: str | None = None,
+        now: int | None = None,
+    ) -> None:
+        now = self._now() if now is None else int(now)
+
+        def run(conn: sqlite3.Connection) -> None:
+            conn.execute(
+                """
+                INSERT INTO notify(aid, channel, batch_id, sent_at, ok, err)
+                VALUES (?, ?, ?, ?, ?, ?)
+                ON CONFLICT(aid, channel) DO UPDATE SET
+                  batch_id = excluded.batch_id,
+                  sent_at = excluded.sent_at,
+                  ok = excluded.ok,
+                  err = excluded.err
+                """,
+                (int(aid), channel, batch_id, now, 1 if ok else 0, err),
+            )
+            if ok:
+                conn.execute(
+                    "UPDATE candidate SET notified_at = ? WHERE aid = ?",
+                    (now, int(aid)),
+                )
+        self.db.write(run)
+
+    # ------------------------------------------------------------ fav action
+
+    def video_metadata(self, aids: Iterable[int]) -> dict[int, dict[str, Any]]:
+        ids = [int(aid) for aid in aids]
+        if not ids:
+            return {}
+
+        def run(conn: sqlite3.Connection) -> dict[int, dict[str, Any]]:
+            placeholders = ",".join("?" for _ in ids)
+            rows = conn.execute(
+                f"SELECT aid, up_mid, pubdate FROM video WHERE aid IN ({placeholders})",
+                ids,
+            ).fetchall()
+            return {
+                int(row["aid"]): {
+                    "up_mid": row["up_mid"],
+                    "pubdate": int(row["pubdate"] or 0),
+                }
+                for row in rows
+            }
+        return self.db.read(run)
+
+    def folder_for_up(self, up_mid: int) -> int | None:
+        def run(conn: sqlite3.Connection) -> int | None:
+            row = conn.execute(
+                "SELECT folder_id FROM video "
+                "WHERE up_mid = ? AND folder_id IS NOT NULL "
+                "AND status IN ('active', 'unfaved') "
+                "ORDER BY pubdate ASC, aid ASC LIMIT 1",
+                (int(up_mid),),
+            ).fetchone()
+            return int(row["folder_id"]) if row else None
+        return self.db.read(run)
+
+    def begin_fav_action(
+        self,
+        aid: int,
+        folder_id: int,
+        *,
+        now: int | None = None,
+        ttl: int = 86400,
+    ) -> tuple[str, str]:
+        now = self._now() if now is None else int(now)
+        token = f"fav:{int(aid)}"
+
+        def run(conn: sqlite3.Connection) -> tuple[str, str]:
+            row = conn.execute(
+                "SELECT state, expires_at FROM fav_action WHERE token = ?",
+                (token,),
+            ).fetchone()
+            if row is not None:
+                state = str(row["state"])
+                if state == "done":
+                    return token, "done"
+                if state == "processing" and int(row["expires_at"]) > now:
+                    return token, "busy"
+            conn.execute(
+                """
+                INSERT INTO fav_action(token, aid, folder_id, state, created_at,
+                                       expires_at, done_at, position, err)
+                VALUES (?, ?, ?, 'processing', ?, ?, NULL, NULL, NULL)
+                ON CONFLICT(token) DO UPDATE SET
+                  folder_id = excluded.folder_id,
+                  state = 'processing',
+                  expires_at = excluded.expires_at,
+                  done_at = NULL,
+                  position = NULL,
+                  err = NULL
+                """,
+                (token, int(aid), int(folder_id), now, now + int(ttl)),
+            )
+            return token, "processing"
+        return self.db.write(run)
+
+    def finish_fav_action(
+        self, token: str, *, position: int, now: int | None = None
+    ) -> None:
+        now = self._now() if now is None else int(now)
+        self.db.write(lambda conn: conn.execute(
+            "UPDATE fav_action SET state = 'done', done_at = ?, position = ?, err = NULL "
+            "WHERE token = ?",
+            (now, int(position), token),
+        ))
+
+    def fail_fav_action(
+        self, token: str, *, error: str, now: int | None = None
+    ) -> None:
+        now = self._now() if now is None else int(now)
+        self.db.write(lambda conn: conn.execute(
+            "UPDATE fav_action SET state = 'failed', done_at = ?, err = ? "
+            "WHERE token = ?",
+            (now, error, token),
+        ))
+
+    def get_fav_action(self, token: str) -> dict[str, Any] | None:
+        def run(conn: sqlite3.Connection) -> dict[str, Any] | None:
+            row = conn.execute(
+                "SELECT * FROM fav_action WHERE token = ?", (token,)
+            ).fetchone()
+            return dict(row) if row else None
+        return self.db.read(run)
+
+    def expire_fav_actions(self, *, now: int | None = None) -> int:
+        now = self._now() if now is None else int(now)
+        def run(conn: sqlite3.Connection) -> int:
+            cur = conn.execute(
+                "UPDATE fav_action SET state = 'failed', done_at = ?, "
+                "err = '处理超时' WHERE state = 'processing' AND expires_at < ?",
+                (now, now),
+            )
+            return max(cur.rowcount, 0)
+        return self.db.write(run)
+
     # ------------------------------------------------------------ blacklist
 
     def seed_blacklist(self, mids: Iterable[int], *, now: int | None = None) -> int:

reaction_board/services/discovery.py

@@ -0,0 +1,224 @@
+# -*- 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)
+        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 "发送失败",
+            )

reaction_board/services/favorites.py

@@ -0,0 +1,234 @@
+# -*- coding: utf-8 -*-
+"""候选收藏、忽略、拉黑动作。"""
+
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass
+from typing import Any
+
+from ..bilibili.models import FavResource
+from ..config import ConfigStore
+from ..fav_position import VideoMeta, plan_insertion
+from ..repository import Repository
+from ..timeutil import shanghai_datetime
+
+
+@dataclass
+class ActionResult:
+    ok: bool
+    aid: int
+    action: str
+    state: str
+    position: int | None = None
+    already_in_folder: bool = False
+    message: str = ""
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "ok": self.ok,
+            "aid": self.aid,
+            "action": self.action,
+            "state": self.state,
+            "position": self.position,
+            "already_in_folder": self.already_in_folder,
+            "message": self.message,
+        }
+
+
+class FavoriteActionService:
+    def __init__(
+        self,
+        config_store: ConfigStore,
+        repository: Repository,
+        write_provider,
+        *,
+        clock=time.time,
+        logger=None,
+    ) -> None:
+        self.config_store = config_store
+        self.repo = repository
+        self.write_provider = write_provider
+        self._clock = clock
+        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 _target_folder(self, candidate: dict[str, Any]):
+        cfg = self.config_store.get()
+        up_mid = candidate.get("up_mid")
+        if isinstance(up_mid, int) and not isinstance(up_mid, bool):
+            folder_id = self.repo.folder_for_up(up_mid)
+            if folder_id is not None:
+                row = self.repo.get_folder(folder_id)
+                if row is not None and row.enabled:
+                    return row
+        year = shanghai_datetime(self._clock()).year
+        folder = cfg.folder_for_year(year)
+        if folder is not None:
+            return folder
+        latest = cfg.latest_folder()
+        if latest is not None:
+            return latest
+        raise RuntimeError("没有可用的目标收藏夹，请检查 folders 配置")
+
+    def favorite(self, aid: int) -> ActionResult:
+        aid = int(aid)
+        candidate = self.repo.get_candidate(aid)
+        if candidate is None:
+            return ActionResult(
+                ok=False, aid=aid, action="favorite", state="missing",
+                message="候选不存在或已处理",
+            )
+        try:
+            folder = self._target_folder(candidate)
+        except Exception as exc:  # noqa: BLE001
+            return ActionResult(
+                ok=False, aid=aid, action="favorite", state="failed",
+                message=str(exc),
+            )
+
+        token, state = self.repo.begin_fav_action(
+            aid, folder.media_id, now=self._now()
+        )
+        if state == "done":
+            action = self.repo.get_fav_action(token) or {}
+            return ActionResult(
+                ok=True, aid=aid, action="favorite", state="done",
+                position=action.get("position"),
+                message="已在收藏夹",
+            )
+        if state == "busy":
+            return ActionResult(
+                ok=False, aid=aid, action="favorite", state="busy",
+                message="该候选正在处理中，请稍后重试",
+            )
+
+        already = False
+        try:
+            client = self.write_provider.get()
+            ids = client.fav_ids(folder.media_id)
+            if aid in {item.aid for item in ids}:
+                already = True
+                position = [item.aid for item in ids].index(aid) + 1
+            else:
+                client.deal(aid, folder.media_id)
+                ids = client.fav_ids(folder.media_id)
+                aid_order = [item.aid for item in ids]
+                if aid not in aid_order:
+                    raise RuntimeError("deal 成功但回读收藏夹未找到该视频")
+                metadata = {
+                    item_aid: VideoMeta(
+                        up_mid=(info.get("up_mid") if info else None),
+                        pubdate=int((info or {}).get("pubdate") or 0),
+                    )
+                    for item_aid, info in self.repo.video_metadata(aid_order).items()
+                }
+                plan = plan_insertion(
+                    aid_order,
+                    aid,
+                    metadata,
+                    VideoMeta(
+                        up_mid=candidate.get("up_mid"),
+                        pubdate=int(candidate.get("pubdate") or 0),
+                    ),
+                )
+                type_map = {item.aid: item.type for item in ids}
+                for predecessor, moved in plan.moves:
+                    client.sort(
+                        folder.media_id,
+                        f"{predecessor}:{type_map.get(predecessor, 2)}:"
+                        f"{moved}:{type_map.get(moved, 2)}",
+                    )
+                final_order = [item.aid for item in client.fav_ids(folder.media_id)]
+                if aid not in final_order:
+                    raise RuntimeError("排序后回读收藏夹未找到该视频")
+                position = final_order.index(aid) + 1
+                if position != plan.target_position:
+                    raise RuntimeError(
+                        f"回读位置不符：期望 #{plan.target_position}，实际 #{position}"
+                    )
+
+            resource = FavResource(
+                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=int(candidate.get("pubdate") or 0),
+                duration=None,
+                cover=candidate.get("cover"),
+                attr=0,
+                fav_time=None,
+            )
+            self.repo.upsert_from_fav(
+                resource, folder.media_id, was_added=True, now=self._now()
+            )
+            self.repo.delete_candidate(aid)
+            self.repo.finish_fav_action(token, position=position, now=self._now())
+            self._log(
+                "info", "favorite.done",
+                aid=aid, folder_id=folder.media_id, position=position,
+                already=already,
+            )
+            return ActionResult(
+                ok=True, aid=aid, action="favorite", state="done",
+                position=position, already_in_folder=already,
+                message="已在收藏夹" if already else "已加入收藏夹",
+            )
+        except Exception as exc:  # noqa: BLE001  失败保留候选，允许重试
+            self.repo.fail_fav_action(token, error=str(exc), now=self._now())
+            self._log(
+                "error", "favorite.failed",
+                aid=aid, folder_id=folder.media_id, error=str(exc),
+            )
+            return ActionResult(
+                ok=False, aid=aid, action="favorite", state="failed",
+                message=str(exc),
+            )
+
+    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())
+        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())
+        return ActionResult(
+            ok=True, aid=aid, action="blacklist", state="rejected",
+            message=f"已拉黑 UP，并忽略该 UP 的 {rejected} 条待决策候选",
+        )

reaction_board/services/folder_sync.py

@@ -260,6 +260,9 @@ class FolderSyncService:
                 outcome = self.repo.upsert_from_fav(
                     resource, folder.media_id, was_added=aid in added, now=now
                 )
+                if aid in added:
+                    # 人工已加入收藏夹：候选池不再保留同一条视频
+                    self.repo.delete_candidate(aid)
                 if outcome.moved_folder:
                     self._log(
                         "warning",

reaction_board/services/telegram_poller.py

@@ -0,0 +1,85 @@
+# -*- coding: utf-8 -*-
+"""Telegram getUpdates long polling 与按钮回调分发。"""
+
+from __future__ import annotations
+
+import time
+from typing import Any, Mapping
+
+
+class TelegramPoller:
+    def __init__(self, bot, actions, *, logger=None, sleeper=time.sleep) -> None:
+        self.bot = bot
+        self.actions = actions
+        self.logger = logger
+        self._sleep = sleeper
+
+    def _log(self, level: str, event: str, **fields) -> None:
+        if self.logger is not None:
+            getattr(self.logger, level)(event, **fields)
+
+    def run_once(self, offset: int | None = None) -> int | None:
+        updates = self.bot.get_updates(offset)
+        for update in updates:
+            update_id = update.get("update_id")
+            if isinstance(update_id, int) and not isinstance(update_id, bool):
+                offset = max(offset or 0, update_id + 1)
+            callback = update.get("callback_query")
+            if isinstance(callback, Mapping):
+                self.handle_callback(callback)
+        return offset
+
+    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:
+            self._log(
+                "warning", "telegram.unauthorized",
+                chat_id=chat_id, from_id=from_id,
+            )
+            self._answer(callback_id, "无权限")
+            return
+
+        action, _, raw_aid = str(callback.get("data") or "").partition(":")
+        try:
+            aid = int(raw_aid)
+        except (TypeError, ValueError):
+            self._answer(callback_id, "未知操作")
+            return
+
+        if action == "fav":
+            result = self.actions.favorite(aid)
+        elif action == "ign":
+            result = self.actions.ignore(aid)
+        elif action == "blk":
+            result = self.actions.blacklist(aid)
+        else:
+            self._answer(callback_id, "未知操作")
+            return
+
+        text = result.message
+        if result.ok and result.position is not None:
+            text = f"{text}（位置 #{result.position}）"
+        self._log(
+            "info", "telegram.action",
+            action=action, aid=aid, ok=result.ok, state=result.state,
+            position=result.position,
+        )
+        self._answer(callback_id, text)
+
+    def _answer(self, callback_id: str, text: str) -> None:
+        try:
+            self.bot.answer_callback(callback_id, text)
+        except Exception as exc:  # noqa: BLE001
+            self._log("error", "telegram.answer_failed", error=str(exc))
+
+    def run_forever(self, stop_event=None) -> None:
+        offset: int | None = None
+        while stop_event is None or not stop_event.is_set():
+            try:
+                offset = self.run_once(offset)
+            except Exception as exc:  # noqa: BLE001  长轮询失败后退避重试
+                self._log("error", "telegram.poll_failed", error=str(exc))
+                self._sleep(3)