diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index 1dc5785..4aa8a6a 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -199,28 +199,29 @@ PRAGMA synchronous = NORMAL; │ ├── migrations.py # schema_version 与迁移 │ ├── scheduler.py # 内置任务调度 │ ├── logging_json.py # 脱敏 JSON 日志 +│ ├── classify.py # 标题归一化、时长解析与打分 +│ ├── fav_position.py # 新视频插入位置计算 +│ ├── timeutil.py # 上海时区与每日时间 │ ├── bilibili/ -│ │ ├── client.py # HTTP、WBI、重试与限速 +│ │ ├── client.py # 只读 HTTP、重试与限速 +│ │ ├── write_client.py # deal / sort 写客户端 +│ │ ├── search.py # WBI 搜索与旧接口降级 +│ │ ├── provider.py # 凭据变化重建客户端 │ │ ├── auth.py # Cookie、设备指纹与请求头工具 │ │ ├── auth_flow.py # app.py auth 的扫码与原子写回流程 -│ │ └── favorite.py # 加入收藏夹与排序定位 +│ │ └── models.py # 接口响应模型 │ ├── services/ -│ │ ├── discovery.py +│ │ ├── discovery.py # 搜索发现与候选池 +│ │ ├── favorites.py # 收藏 / 忽略 / 拉黑 +│ │ ├── telegram_poller.py # getUpdates 与回调分发 │ │ ├── folder_sync.py │ │ ├── snapshot.py -│ │ ├── fans.py -│ │ ├── candidates.py -│ │ ├── rankings.py │ │ └── maintenance.py -│ ├── notify/telegram.py -│ └── web/ +│ ├── notify/telegram.py # Telegram 消息与按钮 +│ └── web/ # P2 预留 │ ├── server.py │ ├── api.py │ └── static/ -│ ├── index.html -│ ├── app.js -│ ├── styles.css -│ └── vendor/echarts.min.js │ ├── tests/ └── data/ # 本地运行期数据，默认不提交 @@ -379,6 +380,14 @@ score >= 2：发送通知 score < 0：丢弃 ``` +P3 实现说明： + +- WBI 密钥来自 `nav.data.wbi_img`，缓存 6 小时；WBI 失败自动降级旧版 `search/type`； +- 关键词、页数和时间窗由 `discover` 配置控制； +- 同一 aid 在多个关键词/页重复出现时，候选与通知都按 aid 去重； +- 已通知但未决策的候选不会重复通知；未通知的候选会在后续发现轮次补发； +- 人工把候选加入收藏夹后，收藏夹同步会入库并删除同 aid 候选。 + ### 7.3 收藏夹同步 每小时对所有已配置年度夹执行： @@ -451,6 +460,16 @@ Telegram `callback_data` 只包含短操作 ID，例如 `fav:`；真实 aid GET 页面或链接预览只能展示信息，绝不能触发任何写操作。 +P3 实现说明： + +- `callback_data` 使用 `fav:` / `ign:` / `blk:`，真实状态存 `fav_action`； +- `begin_fav_action` 把 `pending/failed` 原子切换为 `processing`；并发点击返回 `busy`； +- 已在目标夹时直接成功，不重复 `deal`； +- 目标夹优先同 UP 在库最早视频所在夹，其次当前自然年夹，最后最新年度夹； +- 插入位置由 `fav_position.plan_insertion` 计算；组头需要两次 `sort`； +- `deal` 或 `sort` 失败写 `fav_action.state='failed'`，保留候选，用户可再次点击重试； +- Telegram 只接受 `chat_id` 与 `from.id` 都等于配置私聊用户的回调。 + --- ## 9. Web API 与页面 @@ -608,18 +627,30 @@ uv run python tests/test_moves.py - `favsort.py`：`--verify-every` 必须为正数；`sort` 使用真实资源类型，不再硬编码 `2`； - 配置每 60 秒检查 mtime；调度参数变更需重启。 +### P3：Telegram 与一键加入 ✅ + +- long polling、候选通知、收藏/忽略/拉黑； +- `deal`、`sort`、回读验证和失败重试； +- 对真实写入操作使用独立测试视频完成授权验证后再启用。 + +完成情况（2026-10-31）： + +- 已实现 `bilibili/search.py`：WBI 签名、密钥缓存 6 小时、旧版搜索降级； +- 已实现 `classify.py`：标题归一化、时长解析与打分； +- 已实现 `services/discovery.py`：关键词/页数/时间窗过滤、黑名单、候选池、通知批次； +- 已实现 `notify/telegram.py` 与 `services/telegram_poller.py`：消息分片、Inline Keyboard、回调授权、offset 推进； +- 已实现 `bilibili/write_client.py`、`fav_position.py`、`services/favorites.py`：目标夹选择、`deal`、`sort`、回读校验、失败保留候选； +- `app.py` 新增 `discover` / `candidates`；`run` 注册发现与候选清理任务，并启动 Telegram poller； +- 真实只读联调：4 次 WBI 搜索、168 条结果、1 条候选（score=6），Telegram 通知发送成功； +- 真实 `deal` / `sort` 尚未在收藏夹执行，等待用户点击 Telegram 按钮或授权测试视频后验证； +- 自动化测试见 `tests/test_classify.py`、`test_search.py`、`test_candidates.py`、`test_discovery.py`、`test_fav_position.py`、`test_favorites.py`、`test_telegram.py`。 + ### P2：本地看板 - 总览、榜单、详情、UP、粉丝、候选和系统状态页面； - 原生前端和 ECharts； - 本机默认绑定 `127.0.0.1:8080`。 -### P3：Telegram 与一键加入 - -- long polling、候选通知、收藏/忽略/拉黑； -- `deal`、`sort`、回读验证和失败重试； -- 对真实写入操作使用独立测试视频完成授权验证后再启用。 - ### P4：可选部署 - 反向代理、TLS、远程访问鉴权； diff --git a/README.md b/README.md index 95b43f6..2cc923d 100644 --- a/README.md +++ b/README.md @@ -4,9 +4,9 @@ 项目以 B 站年度收藏夹为可信视频来源，补充关键词发现和 Telegram 人工确认，持续采集 reaction 视频的七项数据与咻咻满粉丝数，并提供本地数据看板。现有的收藏夹排序工具仍保留并作为一键加入后的排序能力基础。 -> 当前状态：**P0（数据底座与收藏夹同步）、P1（快照与粉丝追踪）已实施；P2 待开始。** +> 当前状态：**P0（数据底座与收藏夹同步）、P1（快照与粉丝追踪）、P3（Telegram 与一键加入）已实施；P2 待开始。** > -> 当前已可用：扫码授权、收藏夹读取、排序计划生成、执行与校验；SQLite 建库迁移、收藏夹首次导入与增量同步、失效识别、黑名单种子；每日七项视频快照、粉丝数快照、失败占位、断点续跑与日备份。 +> 当前已可用：扫码授权、收藏夹读取、排序计划生成、执行与校验；SQLite 建库迁移、收藏夹首次导入与增量同步、失效识别、黑名单种子；每日七项视频快照、粉丝数快照、失败占位、断点续跑与日备份；搜索发现、候选池、Telegram 通知与收藏/忽略/拉黑回调。 > > 详细业务口径、接口验证和风险记录见 [project.md](project.md)；开发架构与实施计划见 [DEVELOPMENT.md](DEVELOPMENT.md)。 @@ -19,8 +19,9 @@ | 收藏夹同步与首次导入 | 可用（P0） | 监控已配置年度夹，自动发现人工新收藏的视频与失效资源。 | | 视频日快照 | 可用（P1） | 逐条采集播放、点赞、投币、收藏、转发、评论、弹幕。 | | 粉丝追踪 | 可用（P1） | 每日记录咻咻满等配置 UP 的粉丝数。 | +| 搜索发现与候选池 | 可用（P3） | 每小时搜索前 2 页，分类打分后写入候选池。 | +| Telegram 通知与一键加入 | 可用（P3） | 收藏 / 忽略 / 拉黑按钮，`deal` + `sort` 回读校验。 | | 本地数据看板 | 规划（P2） | 总览、榜单、视频/UP 详情、粉丝曲线、候选池和系统状态。 | -| Telegram 发现与一键加入 | 规划（P3） | 每小时发现候选，支持收藏、忽略、拉黑和收藏夹内定位。 | ## 核心工作流 @@ -120,6 +121,35 @@ uv run python app.py auth-check --account all 首次真实全量采集结果：795 条视频全部成功、0 失败、0 移除，耗时约 19 分钟；粉丝 `follower=2767270`；日备份为 `data/backup/app-20261031.db`。 +## P3：搜索发现与 Telegram 一键加入 + +P3 已实现 WBI 搜索发现、候选池、Telegram 通知与按钮回调，以及使用写账号完成的 `deal` / `sort` 加入流程。 + +```powershell +# 执行一次发现；--no-notify 只写候选池，不发 Telegram +uv run python app.py discover +uv run python app.py discover --no-notify + +# 查看候选池（离线） +uv run python app.py candidates +uv run python app.py candidates --decision rejected + +# 常驻运行：同步 + 快照 + 发现 + Telegram long polling +uv run python app.py run +``` + +- 搜索使用 `x/web-interface/wbi/search/type`，WBI 密钥缓存 6 小时，失败时降级旧版 `search/type`； +- 每轮按 `discover.keywords` 搜索前 `discover.pages` 页，过滤已入库/已见/超时间窗/黑名单/标题不含要求词的结果； +- 分数规则与 `classify` 配置一致；`score >= notify.min_score` 才发送 Telegram； +- 通知按钮：**收藏**、**忽略**、**拉黑**；回调只接受 `telegram.chat_id` 对应的私聊用户； +- 收藏使用 `bilibili.write`：先查目标夹是否已存在，再 `deal` 加入、`sort` 插入同 UP 组头、回读校验位置； +- 失败保留候选并写 `fav_action.state='failed'`，可直接重试；不自动换夹； +- 忽略写 `candidate.decision='rejected'`；拉黑写 `up_blacklist` 并拒绝该 UP 的其他待决策候选； +- 候选超过 `schedule.candidate_ttl_days` 且未决策时自动删除，`seen_aid` 防止重复通知； +- 人工把候选加入收藏夹后，收藏夹同步会入库并删除同 aid 候选。 + +> 真实 `deal` / `sort` 需要用户点击 Telegram 按钮才会触发；离线测试使用契约一致的 Mock 覆盖。 + ## 配置 复制样例配置，并填写 B 站账号信息： @@ -146,6 +176,7 @@ Copy-Item config.example.json config.json - `folders`：年度收藏夹配置（`year` + `media_id` + `title`），系统不自动建夹； - `up_blacklist_seed`：首次启动写入数据库的 UP 黑名单种子； - `track_fans`：需要每日追踪粉丝数的 UP mid； +- `discover`：搜索关键词、页数与时间窗（`keywords` / `pages` / `window_days`）； - `storage`：数据库与备份目录； - `schedule` / `notify` / `classify` / `heat_weights`：任务节奏、通知阈值、分类规则和热度权重。 @@ -247,6 +278,13 @@ uv run python tests/test_migrations.py # 迁移、备份、回滚与保留 uv run python tests/test_database.py # 单写队列、事务与并发 uv run python tests/test_repository.py # 收藏夹/视频状态机与黑名单 uv run python tests/test_bilibili_client.py # ids/infos/view/fans/nav 契约与重试 +uv run python tests/test_classify.py # 标题归一化、时长解析与打分 +uv run python tests/test_search.py # WBI 签名、搜索解析与降级 +uv run python tests/test_candidates.py # 候选池、去重、通知与 fav_action +uv run python tests/test_discovery.py # 发现过滤、候选池与通知批次 +uv run python tests/test_fav_position.py # 新视频插入位置 +uv run python tests/test_favorites.py # 收藏/忽略/拉黑与失败重试 +uv run python tests/test_telegram.py # 消息、分片、长轮询与回调授权 uv run python tests/test_folder_sync.py # 首次导入、差集、失效识别、容量预警 uv run python tests/test_snapshot.py # 七项快照、补采、断点续跑、粉丝、备份重试 uv run python tests/test_maintenance.py # 日备份与保留期 diff --git a/app.py b/app.py index a5f56d6..4c072ed 100644 --- a/app.py +++ b/app.py @@ -1,8 +1,9 @@ # -*- coding: utf-8 -*- """看板服务入口。 -P1 提供 ``init`` / ``sync`` / ``snapshot`` / ``auth`` / ``auth-check`` / ``run`` / ``status`` -子命令；后续阶段会在同一入口启动本地 Web 服务。 +P1/P3 提供 ``init`` / ``sync`` / ``snapshot`` / ``discover`` / ``candidates`` / +``auth`` / ``auth-check`` / ``run`` / ``status`` 子命令；P2 起在同一入口启动本地 +Web 服务。 """ import sys diff --git a/config.example.json b/config.example.json index d40a124..96841c8 100644 --- a/config.example.json +++ b/config.example.json @@ -60,12 +60,18 @@ "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": { "require_title": "咻咻满", + "self_mid": 37754047, "positive": [ "reaction", "re", diff --git a/project.md b/project.md index 8747d17..3b81402 100644 --- a/project.md +++ b/project.md @@ -1,8 +1,8 @@ # 咻咻满 Reaction 视频数据看板 —— 可行性分析与技术路线 -> 状态：**设计定稿；P0（数据底座与收藏夹同步）、P1（每日快照与粉丝追踪）已实施** -> 最后更新：2026-10-31（第 6 轮：P1 落地） -> 全部探测均为只读（GET），未调用 `fav/resource/deal`、`fav/resource/sort` +> 状态：**设计定稿；P0（数据底座与收藏夹同步）、P1（每日快照与粉丝追踪）、P3（Telegram 与一键加入）已实施；P2 待开始** +> 最后更新：2026-10-31（第 7 轮：P3 落地） +> B 站探测均为只读（GET），尚未调用 `fav/resource/deal`、`fav/resource/sort` --- @@ -788,11 +788,15 @@ storage：SQLite WAL，单写线程 | 27 | `x/web-interface/nav` 登录态 | `code=0`、`isLogin=true`、mid=28802351 | | 28 | **P1 真实全量快照** | 795 条全部 `ok=1`、0 失败、0 移除；耗时 1124s；粉丝 `follower=2767270`；日备份 `app-20261031.db` 含 795 条快照 | | 29 | **空收藏夹 `ids` 契约** | `4413147139` 返回 `{"code":0,"data":[]}`，`data` 键存在且为数组；缺失/`null` 按协议异常处理 | -| 30 | **迁移 v2 与增量同步** | 真实库升级到 schema 2，自动生成 `app-pre-v1-*.db`；同步 167+664 条无新增/移除，主归属无矛盾 | +| 30 | **迁移 v2/v3 与增量同步** | 真实库升级到 schema 3，自动生成 `app-pre-v1-*` / `app-pre-v2-*`；同步 167+664 条无新增/移除，主归属无矛盾 | +| 31 | **P3 真实发现** | 4 次 WBI 搜索、168 条结果、161 已入库、6 条超时间窗、1 条候选（score=6） | +| 32 | **P3 Telegram 通知** | 真实 Bot 发送成功；发现重复结果后已修复按 aid 去重，离线测试覆盖 | +| 33 | **P3 同步联动** | 候选被人工加入收藏夹后，同步识别 1 条新增并当天补采快照；候选行自动删除 | **未验证（需授权或外部配合）**： -- `fav/resource/deal` 实际加入 + `sort` 定位（会写收藏夹）； +- `fav/resource/deal` 实际加入 + `sort` 定位（会写收藏夹；需用户点击 Telegram 按钮或授权测试视频）； +- Telegram 按钮真实回调（长轮询代码与授权逻辑已离线测试）； - 手机扫码 `auth` 完整链路； - 各通知渠道的按钮/链接实际表现； - 长期（>72h）每小时扫描的风控表现； @@ -802,19 +806,11 @@ storage：SQLite WAL，单写线程 ## 14. 下一步 -所有决策点已确认（§12 无遗留）。下一步是**方案定稿**，产出： +P0、P1、P3 已实施。下一步是 **P2 本地看板**： -1. 最终 DDL + 迁移脚本设计（含 `up_stat`、`up_blacklist`）； -2. 四个任务的调度/限速/重试/幂等伪代码（发现 / 收藏夹同步 / 快照 / 粉丝）； -3. 分类规则 v1（含时长权重、UP 黑名单）+ 用 831 条收藏回测方案； -4. 热度算法参数表与榜单口径； -5. 前端线框 + API 契约（含粉丝页、候选池页）； -6. 一键加入时序图（Telegram callback 与 签名链接 两条入口）； -7. 分阶段实施计划： - - P0 收藏夹同步 + 导入 + 黑名单种子（先把库建起来） - - P1 每日快照 + 粉丝追踪（开始积累数据） - - P2 看板（榜单/详情/粉丝页） - - P3 Telegram 通知 + 一键加入 - - P4 容器化 + 反代 + 备份 +1. 总览、榜单、视频/UP 详情、粉丝曲线、候选池和系统状态页面； +2. 原生前端 + ECharts，本机默认绑定 `127.0.0.1:8080`； +3. 复用现有 SQLite 查询与状态机，不在前端复制业务判断； +4. P3 的真实 `deal` / `sort` 待用户点击 Telegram 按钮或授权测试视频后补充验证记录。 -> P0 已实施；P1 起继续按上述阶段推进。 +> P4（容器化 + 反代）继续保持可选。 diff --git a/reaction_board/bilibili/search.py b/reaction_board/bilibili/search.py new file mode 100644 index 0000000..58e5f4a --- /dev/null +++ b/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)) diff --git a/reaction_board/bilibili/write_client.py b/reaction_board/bilibili/write_client.py new file mode 100644 index 0000000..4345a69 --- /dev/null +++ b/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, + }, + ) diff --git a/reaction_board/classify.py b/reaction_board/classify.py new file mode 100644 index 0000000..e0c851f --- /dev/null +++ b/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: + """去掉搜索结果里的 ```` 高亮标签。""" + 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 diff --git a/reaction_board/cli.py b/reaction_board/cli.py index fe883d8..4ec1240 100644 --- a/reaction_board/cli.py +++ b/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": diff --git a/reaction_board/config.py b/reaction_board/config.py index 500cd6b..89acd26 100644 --- a/reaction_board/config.py +++ b/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, diff --git a/reaction_board/fav_position.py b/reaction_board/fav_position.py new file mode 100644 index 0000000..19d58f5 --- /dev/null +++ b/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)),), + ) diff --git a/reaction_board/migrations.py b/reaction_board/migrations.py index 66ad659..8f304d4 100644 --- a/reaction_board/migrations.py +++ b/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", + ), + ), ) diff --git a/reaction_board/notify/__init__.py b/reaction_board/notify/__init__.py new file mode 100644 index 0000000..05cd737 --- /dev/null +++ b/reaction_board/notify/__init__.py @@ -0,0 +1,2 @@ +# -*- coding: utf-8 -*- +"""通知渠道。""" diff --git a/reaction_board/notify/telegram.py b/reaction_board/notify/telegram.py new file mode 100644 index 0000000..18f9bc1 --- /dev/null +++ b/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"🎬 {title}\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]}, + ) diff --git a/reaction_board/repository.py b/reaction_board/repository.py index 94e568b..4681b45 100644 --- a/reaction_board/repository.py +++ b/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: diff --git a/reaction_board/services/discovery.py b/reaction_board/services/discovery.py new file mode 100644 index 0000000..0773d82 --- /dev/null +++ b/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 "发送失败", + ) diff --git a/reaction_board/services/favorites.py b/reaction_board/services/favorites.py new file mode 100644 index 0000000..e5a3c7d --- /dev/null +++ b/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} 条待决策候选", + ) diff --git a/reaction_board/services/folder_sync.py b/reaction_board/services/folder_sync.py index 29dfab2..4aebafb 100644 --- a/reaction_board/services/folder_sync.py +++ b/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", diff --git a/reaction_board/services/telegram_poller.py b/reaction_board/services/telegram_poller.py new file mode 100644 index 0000000..566a093 --- /dev/null +++ b/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) diff --git a/tests/test_candidates.py b/tests/test_candidates.py new file mode 100644 index 0000000..f5db665 --- /dev/null +++ b/tests/test_candidates.py @@ -0,0 +1,194 @@ +# -*- coding: utf-8 -*- +"""候选池、去重、通知记录与 fav_action 状态机测试（不联网）: +uv run python tests/test_candidates.py +""" + +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 reaction_board.bilibili.models import FavResource # noqa: E402 +from reaction_board.config import FolderConfig # noqa: E402 +from reaction_board.database import Database # noqa: E402 +from reaction_board.repository import Repository # noqa: E402 + +FOLDER = FolderConfig(year=2026, media_id=3700023351, title="2026") + + +def setup(tmp): + db = Database(Path(tmp) / "app.db", backup_dir=Path(tmp) / "backup") + db.initialize() + repo = Repository(db, clock=lambda: 1_800_000_000) + repo.ensure_folder(FOLDER) + return db, repo + + +def candidate(repo, aid=1, *, score=3, now=1_800_000_000): + return repo.upsert_candidate( + aid=aid, + bvid=f"BV{aid}", + title=f"咻咻满 reaction {aid}", + up_mid=100, + up_name="UP", + pubdate=1_799_000_000, + cover="https://example/cover.jpg", + score=score, + channel="search:test#p1", + now=now, + ) + + +def test_seen_and_known_aid(): + with tempfile.TemporaryDirectory() as tmp: + db, repo = setup(tmp) + try: + assert repo.known_aid(1) is False + repo.record_seen(1, source="search:test#p1") + repo.record_seen(1, source="search:test#p2") # 幂等 + assert repo.known_aid(1) is True + repo.upsert_from_fav( + FavResource( + aid=2, bvid="BV2", title="视频", up_mid=100, up_name="UP", + pubdate=1, duration=60, cover=None, attr=0, fav_time=None, + ), + FOLDER.media_id, + was_added=True, + ) + assert repo.known_aid(2) is True + finally: + db.close() + + +def test_candidate_upsert_preserves_decision_and_notified(): + with tempfile.TemporaryDirectory() as tmp: + db, repo = setup(tmp) + try: + first = candidate(repo, 1, score=2) + assert first["decision"] is None and first["notified_at"] is None + + repo.mark_notified( + 1, channel="telegram", batch_id="b1", ok=True, now=1_800_000_100 + ) + repo.mark_candidate_decision(1, "rejected", now=1_800_000_200) + + second = candidate(repo, 1, score=4, now=1_800_000_300) + assert second["score"] == 4 + assert second["decision"] == "rejected" + assert second["notified_at"] == 1_800_000_100 + assert second["first_seen"] == 1_800_000_000 + assert len(repo.list_candidates()) == 1 + assert len(repo.list_candidates("rejected")) == 1 + assert repo.list_candidates("accepted") == [] + finally: + db.close() + + +def test_notify_and_reject_for_up(): + with tempfile.TemporaryDirectory() as tmp: + db, repo = setup(tmp) + try: + candidate(repo, 1) + candidate(repo, 2) + repo.mark_notified( + 1, channel="telegram", batch_id="b1", ok=False, err="boom", + now=1_800_000_100, + ) + assert repo.get_candidate(1)["notified_at"] is None + repo.mark_notified( + 1, channel="telegram", batch_id="b2", ok=True, now=1_800_000_200 + ) + assert repo.get_candidate(1)["notified_at"] == 1_800_000_200 + + assert repo.reject_pending_for_up(100, now=1_800_000_300) == 2 + assert repo.get_candidate(2)["decision"] == "rejected" + finally: + db.close() + + +def test_delete_expired_candidates_only_pending(): + with tempfile.TemporaryDirectory() as tmp: + db, repo = setup(tmp) + try: + candidate(repo, 1, now=1_800_000_000) + candidate(repo, 2, now=1_800_000_000) + candidate(repo, 3, now=1_800_000_000) + repo.mark_candidate_decision(3, "rejected") + deleted = repo.delete_expired_candidates( + ttl_days=7, now=1_800_000_000 + 8 * 86400 + ) + assert deleted == 2 + assert repo.get_candidate(1) is None + assert repo.get_candidate(3) is not None + finally: + db.close() + + +def test_folder_for_up_uses_earliest_video(): + with tempfile.TemporaryDirectory() as tmp: + db, repo = setup(tmp) + try: + for aid, pubdate in ((1, 1_700_000_500), (2, 1_700_000_100)): + repo.upsert_from_fav( + FavResource( + aid=aid, bvid=f"BV{aid}", title="视频", up_mid=100, + up_name="UP", pubdate=pubdate, duration=60, + cover=None, attr=0, fav_time=None, + ), + FOLDER.media_id, + was_added=True, + ) + assert repo.folder_for_up(100) == FOLDER.media_id + assert repo.folder_for_up(999) is None + finally: + db.close() + + +def test_fav_action_state_machine(): + with tempfile.TemporaryDirectory() as tmp: + db, repo = setup(tmp) + try: + token, state = repo.begin_fav_action(1, FOLDER.media_id) + assert token == "fav:1" + assert state == "processing" + + token2, state2 = repo.begin_fav_action(1, FOLDER.media_id) + assert token2 == token and state2 == "busy" + + repo.finish_fav_action(token, position=3) + assert repo.get_fav_action(token)["state"] == "done" + assert repo.get_fav_action(token)["position"] == 3 + assert repo.begin_fav_action(1, FOLDER.media_id)[1] == "done" + + token3, state3 = repo.begin_fav_action(2, FOLDER.media_id) + assert state3 == "processing" + repo.fail_fav_action(token3, error="夹满") + assert repo.get_fav_action(token3)["state"] == "failed" + assert repo.get_fav_action(token3)["err"] == "夹满" + + # failed 可重试 + _, retry = repo.begin_fav_action(2, FOLDER.media_id) + assert retry == "processing" + + # 过期的 processing 自动失败 + assert repo.expire_fav_actions(now=1_800_000_000 + 2 * 86400) >= 1 + assert repo.get_fav_action("fav:2")["state"] == "failed" + finally: + db.close() + + +def main(): + test_seen_and_known_aid() + test_candidate_upsert_preserves_decision_and_notified() + test_notify_and_reject_for_up() + test_delete_expired_candidates_only_pending() + test_folder_for_up_uses_earliest_video() + test_fav_action_state_machine() + print("OK: 候选池、去重、通知记录与 fav_action 状态机测试全部通过") + + +if __name__ == "__main__": + main() diff --git a/tests/test_classify.py b/tests/test_classify.py new file mode 100644 index 0000000..5803bc2 --- /dev/null +++ b/tests/test_classify.py @@ -0,0 +1,77 @@ +# -*- coding: utf-8 -*- +"""候选分类纯函数测试: uv run python tests/test_classify.py""" + +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +sys.stdout.reconfigure(encoding="utf-8") + +from reaction_board.classify import ( # noqa: E402 + clean_title, + contains_any, + normalize_title, + parse_duration, + passes_required_title, + score_item, +) + +CLASSIFY = { + "require_title": "咻咻满", + "self_mid": 37754047, + "positive": ["reaction", "re", "反应", "看", "听", "初听", "首次", "点评", "锐评", "路人", "震惊", "双人"], + "charge": ["充电", "加急", "专场"], + "negative": ["歌切", "切片", "纯享", "搬运", "完整版"], + "duration_boost": {"gt_30min": 1, "gt_60min": 2}, +} + + +def test_title_helpers(): + assert normalize_title("咻咻满 REACTION") == "咻咻满 reaction" + assert clean_title('咻咻满 reaction') == "咻咻满 reaction" + assert passes_required_title("【咻咻满】初听", "咻咻满") + assert passes_required_title("咻咻满", "咻咻满") + assert not passes_required_title("其他 UP 的 reaction", "咻咻满") + assert contains_any("咻咻满 REACTION", ["reaction"]) + assert not contains_any("咻咻满", []) + + +def test_parse_duration(): + assert parse_duration("105:5") == 105 * 60 + 5 + assert parse_duration("1:02:03") == 3723 + assert parse_duration("15:30") == 930 + assert parse_duration(600) == 600 + assert parse_duration("bad") == 0 + assert parse_duration(None) == 0 + assert parse_duration(-5) == 0 + + +def test_score_rules(): + def score(title, up_mid=100, duration=0): + return score_item( + title=title, up_mid=up_mid, + duration_seconds=duration, classify=CLASSIFY, + ) + + assert score("咻咻满 reaction", duration=60) == 2 + assert score("咻咻满 充电", duration=60) == 2 + assert score("咻咻满 reaction 充电", duration=60) == 4 + assert score("咻咻满 reaction", duration=61 * 60) == 4 + assert score("咻咻满 reaction", duration=31 * 60) == 3 + assert score("咻咻满 reaction", duration=30 * 60) == 2 + assert score("咻咻满 切片", duration=60) == -5 + assert score("咻咻满 reaction 切片", duration=60) == -3 + assert score("咻咻满 reaction", up_mid=37754047, duration=60) == -1 + assert score("咻咻满", duration=0) == 0 + assert score("咻咻满", duration=61 * 60) == 2 + + +def main(): + test_title_helpers() + test_parse_duration() + test_score_rules() + print("OK: 候选标题归一化、时长解析与分类打分测试全部通过") + + +if __name__ == "__main__": + main() diff --git a/tests/test_cli.py b/tests/test_cli.py index 3af56b2..407fef6 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -51,7 +51,7 @@ def test_init_and_status(): path = write_config(tmp) code, output = run_cli(["init", "--config", str(path)]) assert code == 0 - assert '"schema_version": 2' in output + assert '"schema_version": 3' in output db = Database(Path(tmp) / "app.db") db.initialize() @@ -149,6 +149,14 @@ def test_run_once_offline(): def check_login(self): return LoginState(True, 1, "tester", 0, "OK") + class FakeDiscovery: + def __init__(self, *args, **kwargs): + pass + + def run(self): + from reaction_board.services.discovery import DiscoverySummary + return DiscoverySummary(started_at=1, finished_at=1) + with tempfile.TemporaryDirectory() as tmp: path = write_config(tmp) data = json.loads(path.read_text(encoding="utf-8")) @@ -157,6 +165,7 @@ def test_run_once_offline(): path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") original = cli._client + original_discovery = cli.DiscoveryService selected = [] def fake_client(credentials, *args, **kwargs): @@ -164,10 +173,12 @@ def test_run_once_offline(): return FakeClient() cli._client = fake_client + cli.DiscoveryService = FakeDiscovery try: code, output = run_cli(["run", "--once", "--config", str(path)]) finally: cli._client = original + cli.DiscoveryService = original_discovery assert code == 0 assert '"name": "folder_sync"' in output assert '"name": "snapshot"' in output @@ -209,6 +220,49 @@ def test_snapshot_backup_failure_returns_error_and_retries(): assert Path(payload["backup_path"]).exists() +def test_candidates_and_discover_commands(): + from reaction_board.services.discovery import DiscoverySummary + + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp) + assert run_cli(["init", "--config", str(path)])[0] == 0 + db = Database(Path(tmp) / "app.db") + db.initialize() + try: + repo = Repository(db) + repo.upsert_candidate( + aid=123, bvid="BV123", title="咻咻满 reaction", + up_mid=100, up_name="UP", pubdate=1, + cover=None, score=3, channel="search:test#p1", + ) + finally: + db.close() + + code, output = run_cli(["candidates", "--config", str(path)]) + assert code == 0 + assert '"count": 1' in output + assert '"aid": 123' in output + + class FakeDiscovery: + def __init__(self, *args, **kwargs): + pass + + def run(self): + return DiscoverySummary( + started_at=1, finished_at=2, candidates=3, notified=1 + ) + + original = cli.DiscoveryService + cli.DiscoveryService = FakeDiscovery + try: + code, output = run_cli(["discover", "--config", str(path)]) + finally: + cli.DiscoveryService = original + assert code == 0 + assert '"candidates": 3' in output + assert '"notified": 1' in output + + def main_test(): test_init_and_status() test_missing_config() @@ -218,6 +272,7 @@ def main_test(): test_auth_check_all_and_auth_dispatch() test_run_once_offline() test_snapshot_backup_failure_returns_error_and_retries() + test_candidates_and_discover_commands() print("OK: CLI init/sync/snapshot/auth-check/run/status/错误路径测试全部通过") diff --git a/tests/test_config.py b/tests/test_config.py index 544d513..f883f4d 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -54,6 +54,9 @@ def test_dual_accounts_and_env_override(): assert (cfg.bilibili.write.mid, cfg.bilibili.write.csrf, cfg.bilibili.write.cookie) == (202, "env-write", "env-write-cookie") assert cfg.storage.database_path == Path(tmp) / "custom.db" assert cfg.up_blacklist_seed == (1, 2) + assert cfg.discover.keywords == ("咻咻满 reaction", "咻咻满 充电") + assert cfg.discover.pages == 2 + assert cfg.classify["self_mid"] == 37754047 def test_flat_and_missing_account_rejected(): diff --git a/tests/test_database.py b/tests/test_database.py index f0d1054..3c01ad6 100644 --- a/tests/test_database.py +++ b/tests/test_database.py @@ -24,7 +24,7 @@ def test_write_read_transaction(): with tempfile.TemporaryDirectory() as tmp: db = make_db(tmp) try: - assert db.schema_version == 2 + assert db.schema_version == 3 db.write(lambda c: c.execute("INSERT INTO kv(k, v) VALUES ('a', '1')")) assert db.read(lambda c: c.execute("SELECT v FROM kv WHERE k='a'").fetchone()[0]) == "1" @@ -94,8 +94,8 @@ def test_close_and_reinitialize(): db2 = make_db(tmp) try: - assert db2.schema_version == 2 - assert db2.read(lambda c: c.execute("SELECT MAX(version) FROM schema_version").fetchone()[0]) == 2 + assert db2.schema_version == 3 + assert db2.read(lambda c: c.execute("SELECT MAX(version) FROM schema_version").fetchone()[0]) == 3 finally: db2.close() @@ -123,7 +123,7 @@ def test_migration_backup_on_existing_db(): db = Database(path, backup_dir=Path(tmp) / "backup") try: applied = db.initialize() - assert applied == [1, 2] + assert applied == [1, 2, 3] assert len(list((Path(tmp) / "backup").glob("app-pre-v0-*.db"))) == 1 finally: db.close() diff --git a/tests/test_discovery.py b/tests/test_discovery.py new file mode 100644 index 0000000..dd1a53a --- /dev/null +++ b/tests/test_discovery.py @@ -0,0 +1,228 @@ +# -*- coding: utf-8 -*- +"""发现任务测试（假搜索客户端，不联网）: uv run python tests/test_discovery.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 reaction_board.bilibili.client import BilibiliError # noqa: E402 +from reaction_board.bilibili.search import SearchItem # 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.discovery import DiscoveryService # noqa: E402 + +NOW = 1_800_000_000 + + +def write_config(tmp, **discover): + 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"), + }, + "folders": [{"year": 2026, "media_id": 3700023351, "title": "2026"}], + "discover": { + "keywords": ["咻咻满 reaction"], + "pages": 1, + "window_days": 14, + **discover, + }, + "classify": { + "require_title": "咻咻满", + "self_mid": 37754047, + "positive": ["reaction"], + "charge": ["充电"], + "negative": ["切片"], + "duration_boost": {"gt_30min": 1, "gt_60min": 2}, + }, + "notify": {"min_score": 2, "max_items_per_message": 10}, + } + path = Path(tmp) / "config.json" + path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") + return path + + +class FakeSearch: + def __init__(self, items, error=None): + self.items = items + self.error = error + self.calls = [] + + def search(self, keyword, page): + self.calls.append((keyword, page)) + if self.error: + raise self.error + return list(self.items) + + +class FakeNotifier: + def __init__(self, ok=True): + self.ok = ok + self.batches = [] + + def send_candidates(self, candidates): + self.batches.append(list(candidates)) + return self.ok + + +def item(aid, *, title="咻咻满 reaction", up_mid=100, pubdate=NOW - 3600, + duration=600): + return SearchItem( + aid=aid, + bvid=f"BV{aid}", + title=title, + up_mid=up_mid, + up_name="UP", + pubdate=pubdate, + duration_seconds=duration, + cover="https://example/cover.jpg", + ) + + +def make_service(tmp, items, *, notifier=None): + path = write_config(tmp) + store = ConfigStore(path, env={}) + db = Database(Path(tmp) / "app.db", backup_dir=Path(tmp) / "backup") + db.initialize() + repo = Repository(db, clock=lambda: NOW) + service = DiscoveryService( + store, repo, FakeSearch(items), + notifier=notifier, + clock=lambda: NOW, + sleeper=lambda seconds: None, + logger=None, + ) + return db, repo, service + + +def test_filters_and_scoring(): + with tempfile.TemporaryDirectory() as tmp: + notifier = FakeNotifier() + db, repo, service = make_service(tmp, [ + item(1), # 通知 + item(2, title="咻咻满 reaction 切片"), # 负分丢弃 + item(3, title="其他 UP reaction"), # 标题不含 + item(4, pubdate=NOW - 20 * 86400), # 超时间窗 + item(5, up_mid=999), # 黑名单 + item(6, title="咻咻满 日常"), # 0 分入池不通知 + ], notifier=notifier) + try: + repo.seed_blacklist([999], now=NOW) + repo.record_seen(1, source="old", now=NOW) + # aid=1 已 seen，改为验证已知跳过；另用 7 验证通知 + summary = service.run() + assert summary.items == 6 + assert summary.known == 1 + assert summary.title_miss == 1 + assert summary.old == 1 + assert summary.blacklisted == 1 + assert summary.candidates == 1 # aid 6 + assert summary.notified == 0 + assert repo.get_candidate(6)["score"] == 0 + assert repo.get_candidate(2) is None + assert repo.get_candidate(4) is None + assert repo.get_candidate(5) is None + assert notifier.batches == [] + finally: + db.close() + + +def test_notification_batch_and_failure(): + with tempfile.TemporaryDirectory() as tmp: + notifier = FakeNotifier() + db, repo, service = make_service(tmp, [item(1), item(2)], notifier=notifier) + try: + summary = service.run() + assert summary.candidates == 2 + assert summary.notified == 2 + assert len(notifier.batches) == 1 + assert {c["aid"] for c in notifier.batches[0]} == {1, 2} + assert repo.get_candidate(1)["notified_at"] == NOW + finally: + db.close() + + with tempfile.TemporaryDirectory() as tmp: + notifier = FakeNotifier(ok=False) + db, repo, service = make_service(tmp, [item(3)], notifier=notifier) + try: + summary = service.run() + assert summary.notify_failed == 1 + assert repo.get_candidate(3)["notified_at"] is None + assert repo.get_candidate(3)["decision"] is None + finally: + db.close() + + +def test_search_error_does_not_abort_other_keywords(): + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, keywords=["a", "b"]) + store = ConfigStore(path, env={}) + db = Database(Path(tmp) / "app.db", backup_dir=Path(tmp) / "backup") + db.initialize() + repo = Repository(db, clock=lambda: NOW) + search = FakeSearch([item(1)], error=BilibiliError("network")) + service = DiscoveryService( + store, repo, search, clock=lambda: NOW, sleeper=lambda s: None + ) + try: + summary = service.run() + assert summary.searches == 0 + assert len(summary.errors) == 2 + finally: + db.close() + + +def test_pending_candidate_notified_on_later_run(): + with tempfile.TemporaryDirectory() as tmp: + db, repo, service = make_service(tmp, [item(1)], notifier=None) + try: + first = service.run() + assert first.candidates == 1 + assert repo.get_candidate(1)["notified_at"] is None + + notifier = FakeNotifier() + service.notifier = notifier + second = service.run() + assert second.notified == 1 + assert len(notifier.batches) == 1 + assert repo.get_candidate(1)["notified_at"] == NOW + finally: + db.close() + + +def test_duplicate_search_results_notified_once(): + with tempfile.TemporaryDirectory() as tmp: + notifier = FakeNotifier() + db, repo, service = make_service( + tmp, [item(1), item(1)], notifier=notifier + ) + try: + summary = service.run() + assert summary.notified == 1 + assert len(notifier.batches) == 1 + assert len(notifier.batches[0]) == 1 + finally: + db.close() + + +def main(): + test_filters_and_scoring() + test_notification_batch_and_failure() + test_search_error_does_not_abort_other_keywords() + test_pending_candidate_notified_on_later_run() + test_duplicate_search_results_notified_once() + print("OK: 发现过滤、打分、候选池与通知批次测试全部通过") + + +if __name__ == "__main__": + main() diff --git a/tests/test_fav_position.py b/tests/test_fav_position.py new file mode 100644 index 0000000..a9405f8 --- /dev/null +++ b/tests/test_fav_position.py @@ -0,0 +1,80 @@ +# -*- coding: utf-8 -*- +"""新视频插入位置纯函数测试: uv run python tests/test_fav_position.py""" + +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +sys.stdout.reconfigure(encoding="utf-8") + +from reaction_board.fav_position import VideoMeta, plan_insertion # noqa: E402 + + +def meta(up_mid, pubdate): + return VideoMeta(up_mid=up_mid, pubdate=pubdate) + + +def test_empty_and_single(): + assert plan_insertion([], 9, {}, meta(100, 200)).moves == () + assert plan_insertion([], 9, {}, meta(100, 200)).target_position == 1 + plan = plan_insertion([1], 9, {1: meta(200, 50)}, meta(100, 200)) + assert plan.target_position == 1 + assert plan.moves == ((1, 9), (9, 1)) + + +def test_same_up_group(): + current = [1, 2, 3, 4] + metadata = { + 1: meta(100, 100), + 2: meta(100, 90), + 3: meta(200, 80), + 4: meta(200, 70), + } + # 同 UP 组在中间：插到组头 + plan = plan_insertion(current, 9, metadata, meta(200, 95)) + assert plan.target_position == 3 + assert plan.moves == ((2, 9),) + + # 同 UP 组在首位：两次移动 + plan = plan_insertion(current, 9, metadata, meta(100, 200)) + assert plan.target_position == 1 + assert plan.moves == ((1, 9), (9, 1)) + + +def test_new_group_by_earliest_pubdate(): + current = [1, 2, 3, 4] + metadata = { + 1: meta(100, 100), + 2: meta(100, 90), + 3: meta(200, 80), + 4: meta(200, 70), + } + # 新 UP 的组键 85，应插到组键 70 的组之前 + plan = plan_insertion(current, 9, metadata, meta(300, 85)) + assert plan.target_position == 3 + assert plan.moves == ((2, 9),) + + # 比所有组都旧：追加到末尾 + plan = plan_insertion(current, 9, metadata, meta(300, 60)) + assert plan.target_position == 5 + assert plan.moves == ((4, 9),) + + +def test_excludes_new_aid_if_present(): + current = [1, 9, 2] + metadata = {1: meta(100, 100), 2: meta(100, 90)} + plan = plan_insertion(current, 9, metadata, meta(100, 95)) + assert plan.target_position == 1 + assert plan.moves == ((1, 9), (9, 1)) + + +def main(): + test_empty_and_single() + test_same_up_group() + test_new_group_by_earliest_pubdate() + test_excludes_new_aid_if_present() + print("OK: 新视频插入位置计算测试全部通过") + + +if __name__ == "__main__": + main() diff --git a/tests/test_favorites.py b/tests/test_favorites.py new file mode 100644 index 0000000..e08f5e5 --- /dev/null +++ b/tests/test_favorites.py @@ -0,0 +1,211 @@ +# -*- coding: utf-8 -*- +"""收藏/忽略/拉黑动作测试（假写客户端，不联网）: uv run python tests/test_favorites.py""" + +import json +import os +import sys +import tempfile +from dataclasses import dataclass +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 reaction_board.bilibili.client import BilibiliAPIError # noqa: E402 +from reaction_board.bilibili.models import FavResource # 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.favorites import FavoriteActionService # noqa: E402 + +NOW = 1_793_444_393 +FOLDER_ID = 3700023351 + + +@dataclass +class FakeFavId: + aid: int + type: int = 2 + + +class FakeWriteClient: + def __init__(self, order, *, fail_deal=False): + self.order = list(order) + self.fail_deal = fail_deal + self.dealt = [] + self.sorts = [] + + def fav_ids(self, media_id): + return [FakeFavId(aid) for aid in self.order] + + def deal(self, aid, folder_id): + if self.fail_deal: + raise BilibiliAPIError(-1, "收藏夹已满") + self.dealt.append((aid, folder_id)) + self.order.append(aid) + + def sort(self, media_id, spec): + predecessor, _, moved, _ = (int(part) for part in spec.split(":")) + self.sorts.append(spec) + self.order.remove(moved) + index = self.order.index(predecessor) + self.order.insert(index + 1, moved) + + +class FakeProvider: + def __init__(self, client): + self.client = client + + def get(self): + return self.client + + +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"), + }, + "folders": [{"year": 2026, "media_id": FOLDER_ID, "title": "2026"}], + "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 add_video(repo, aid, up_mid, pubdate): + repo.upsert_from_fav( + FavResource( + aid=aid, bvid=f"BV{aid}", title=f"视频{aid}", up_mid=up_mid, + up_name=f"UP{up_mid}", pubdate=pubdate, duration=600, + cover=None, attr=0, fav_time=None, + ), + FOLDER_ID, + was_added=True, + ) + + +def add_candidate(repo, aid, up_mid=100, pubdate=NOW - 3600): + return repo.upsert_candidate( + aid=aid, + bvid=f"BV{aid}", + title=f"咻咻满 reaction {aid}", + up_mid=up_mid, + up_name=f"UP{up_mid}", + pubdate=pubdate, + cover="https://example/cover.jpg", + score=3, + channel="search:test#p1", + now=NOW, + ) + + +def setup(tmp, order, *, fail_deal=False): + path = write_config(tmp) + store = ConfigStore(path, env={}) + db = Database(Path(tmp) / "app.db", backup_dir=Path(tmp) / "backup") + db.initialize() + repo = Repository(db, clock=lambda: NOW) + add_video(repo, 1, 100, 1_792_000_000) + add_video(repo, 2, 200, 1_791_000_000) + add_candidate(repo, 9, up_mid=100, pubdate=1_793_000_000) + client = FakeWriteClient(order, fail_deal=fail_deal) + service = FavoriteActionService( + store, repo, FakeProvider(client), clock=lambda: NOW + ) + return db, repo, client, service + + +def test_favorite_inserts_and_positions(): + with tempfile.TemporaryDirectory() as tmp: + db, repo, client, service = setup(tmp, [1, 2]) + try: + result = service.favorite(9) + assert result.ok is True + assert result.position == 1 + assert result.already_in_folder is False + assert client.dealt == [(9, FOLDER_ID)] + assert client.sorts == ["1:2:9:2", "9:2:1:2"] + assert client.order == [9, 1, 2] + assert repo.get_candidate(9) is None + video = repo.get_video(9) + assert video["status"] == "active" + assert video["folder_id"] == FOLDER_ID + action = repo.get_fav_action("fav:9") + assert action["state"] == "done" + assert action["position"] == 1 + finally: + db.close() + + +def test_favorite_already_in_folder_skips_writes(): + with tempfile.TemporaryDirectory() as tmp: + db, repo, client, service = setup(tmp, [1, 9, 2]) + try: + result = service.favorite(9) + assert result.ok is True + assert result.already_in_folder is True + assert result.position == 2 + assert client.dealt == [] + assert client.sorts == [] + assert repo.get_candidate(9) is None + finally: + db.close() + + +def test_favorite_failure_keeps_candidate_and_allows_retry(): + with tempfile.TemporaryDirectory() as tmp: + db, repo, client, service = setup(tmp, [1, 2], fail_deal=True) + try: + result = service.favorite(9) + assert result.ok is False + assert "收藏夹已满" in result.message + assert repo.get_candidate(9) is not None + action = repo.get_fav_action("fav:9") + assert action["state"] == "failed" + assert "收藏夹已满" in action["err"] + + client.fail_deal = False + retry = service.favorite(9) + assert retry.ok is True + assert retry.position == 1 + assert repo.get_candidate(9) is None + finally: + db.close() + + +def test_ignore_and_blacklist(): + with tempfile.TemporaryDirectory() as tmp: + db, repo, client, service = setup(tmp, [1, 2]) + try: + add_candidate(repo, 11, up_mid=300) + add_candidate(repo, 12, up_mid=300) + ignored = service.ignore(11) + assert ignored.ok is True + 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 repo.get_video(1)["status"] == "active" + finally: + db.close() + + +def main(): + test_favorite_inserts_and_positions() + test_favorite_already_in_folder_skips_writes() + test_favorite_failure_keeps_candidate_and_allows_retry() + test_ignore_and_blacklist() + print("OK: 收藏、忽略、拉黑与失败重试测试全部通过") + + +if __name__ == "__main__": + main() diff --git a/tests/test_folder_sync.py b/tests/test_folder_sync.py index 7574670..aa51687 100644 --- a/tests/test_folder_sync.py +++ b/tests/test_folder_sync.py @@ -143,11 +143,19 @@ def test_initial_import_and_incremental_skip(): clock = [1_800_000_000] db, repo, service = make_service(tmp, transport, clock) try: + repo.upsert_candidate( + aid=1101, bvid="BV1101", title="咻咻满 reaction", + up_mid=100, up_name="UP", pubdate=1_799_000_000, + cover=None, score=3, channel="search:test#p1", + now=1_799_999_000, + ) summary = service.run() assert summary.ok assert summary.added == 4 assert summary.missing == 1 assert summary.refreshed == 3 + # 人工加入收藏夹后，同 aid 候选自动移除 + assert repo.get_candidate(1101) is None assert repo.seed_blacklist([999], now=0) == 0 # 已由 service 写入 assert repo.get_video(1101)["status"] == "active" diff --git a/tests/test_migrations.py b/tests/test_migrations.py index 7be6959..44aaaed 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -37,18 +37,22 @@ def test_fresh_and_idempotent(): conn = sqlite3.connect(Path(tmp) / "app.db") try: applied = run_migrations(conn) - assert applied == [1, 2] - assert current_version(conn) == 2 + assert applied == [1, 2, 3] + assert current_version(conn) == 3 assert EXPECTED_TABLES <= table_names(conn) assert "schema_version" in table_names(conn) columns = { row[1] for row in conn.execute("PRAGMA table_info(folder_resource)") } assert "fav_time" in columns + candidate_columns = { + row[1] for row in conn.execute("PRAGMA table_info(candidate)") + } + assert {"duration", "play", "like", "favorite", "danmaku", "review"} <= candidate_columns # 再跑一次不应重复应用 assert run_migrations(conn) == [] row = conn.execute("SELECT COUNT(*) FROM schema_version").fetchone() - assert row[0] == 2 + assert row[0] == 3 finally: conn.close() @@ -63,15 +67,15 @@ def test_backup_before_upgrade(): conn.execute("INSERT INTO kv(k, v) VALUES ('hello', 'world')") conn.commit() - v3 = Migration(3, "add probe", ("CREATE TABLE probe (id INTEGER PRIMARY KEY)",)) + v4 = Migration(4, "add probe", ("CREATE TABLE probe (id INTEGER PRIMARY KEY)",)) applied = run_migrations( conn, backup_dir=backup_dir, now=1_800_000_000, - migrations=MIGRATIONS + (v3,), + migrations=MIGRATIONS + (v4,), ) - assert applied == [3] - backups = list(backup_dir.glob("app-pre-v2-*.db")) + assert applied == [4] + backups = list(backup_dir.glob("app-pre-v3-*.db")) assert len(backups) == 1 check = sqlite3.connect(backups[0]) try: diff --git a/tests/test_search.py b/tests/test_search.py new file mode 100644 index 0000000..c4c0be7 --- /dev/null +++ b/tests/test_search.py @@ -0,0 +1,139 @@ +# -*- coding: utf-8 -*- +"""WBI 签名与搜索解析测试（不联网）: uv run python tests/test_search.py""" + +import os +import sys +import urllib.parse + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +sys.stdout.reconfigure(encoding="utf-8") + +from reaction_board.bilibili.client import BilibiliError, BilibiliProtocolError # noqa: E402 +from reaction_board.bilibili.search import ( # noqa: E402 + SearchClient, + key_from_url, + mixin_key, + parse_search_payload, + sign_params, +) + +IMG_URL = "https://i0.hdslb.com/bfs/wbi/7cd084941338484aae1ad9425b84077c.png" +SUB_URL = "https://i0.hdslb.com/bfs/wbi/4932caff0ff746eab6f01bf08b70ac45.png" + +NAV_PAYLOAD = { + "code": 0, "message": "0", "ttl": 1, + "data": { + "isLogin": True, + "wbi_img": {"img_url": IMG_URL, "sub_url": SUB_URL}, + }, +} + +SEARCH_PAYLOAD = { + "code": 0, "message": "0", "ttl": 1, + "data": { + "result": [ + { + "type": "video", + "id": 117530286889638, + "bvid": "BV1xvxi6LEZD", + "title": '咻咻满 reaction 初听', + "author": "测试UP", + "mid": 434442072, + "pubdate": 1793370433, + "duration": "105:5", + "pic": "//i0.hdslb.com/bfs/archive/example.jpg", + "play": 12345, + "like": 678, + "favorites": 90, + "danmaku": 12, + "review": 3, + }, + {"type": "article", "id": 999, "title": "非视频"}, + ], + "numPages": 1, + "numResults": 1, + }, +} + + +class FakeClient: + def __init__(self, responses): + self.responses = list(responses) + self.urls = [] + + def get_json(self, url, *, auth=False): + self.urls.append((url, auth)) + item = self.responses.pop(0) + if isinstance(item, BaseException): + raise item + return item + + +def test_wbi_helpers(): + assert key_from_url(IMG_URL) == "7cd084941338484aae1ad9425b84077c" + assert key_from_url(None) == "" + key = mixin_key( + "7cd084941338484aae1ad9425b84077c", + "4932caff0ff746eab6f01bf08b70ac45", + ) + assert key == "ea1db124af3c7062474693fa704f4ff8" + + signed = sign_params({"keyword": "咻咻满", "page": 1}, key, now=1_700_000_000) + assert signed["wts"] == 1_700_000_000 + assert len(signed["w_rid"]) == 32 + changed = sign_params({"keyword": "咻咻满", "page": 2}, key, now=1_700_000_000) + assert changed["w_rid"] != signed["w_rid"] + + +def test_parse_search_payload(): + items = parse_search_payload(SEARCH_PAYLOAD) + assert len(items) == 1 + item = items[0] + assert item.aid == 117530286889638 + assert item.title == "咻咻满 reaction 初听" + assert item.duration_seconds == 105 * 60 + 5 + assert item.cover.startswith("https://") + assert item.play == 12345 and item.like == 678 + assert item.favorite == 90 and item.danmaku == 12 and item.review == 3 + + +def test_wbi_search_and_fallback(): + fake = FakeClient([NAV_PAYLOAD, SEARCH_PAYLOAD]) + client = SearchClient(lambda: fake, clock=lambda: 1_700_000_000) + items = client.search("咻咻满 reaction", 1) + assert len(items) == 1 + assert "/x/web-interface/nav" in fake.urls[0][0] + assert "/wbi/search/type" in fake.urls[1][0] + query = urllib.parse.parse_qs(urllib.parse.urlparse(fake.urls[1][0]).query) + assert query["w_rid"][0] and query["wts"][0] == "1700000000" + + # nav 失败时降级旧接口 + fake = FakeClient([BilibiliError("wbi unavailable"), SEARCH_PAYLOAD]) + client = SearchClient(lambda: fake, clock=lambda: 1_700_000_000) + items = client.search("咻咻满 reaction", 2) + assert len(items) == 1 + assert "/x/web-interface/search/type" in fake.urls[1][0] + + +def test_missing_result_is_protocol_error(): + bad = {"code": 0, "message": "0", "data": {}} + fake = FakeClient([NAV_PAYLOAD, bad, bad]) + client = SearchClient(lambda: fake, clock=lambda: 1_700_000_000) + try: + client.search("x", 1) + except BilibiliProtocolError as exc: + assert "result" in str(exc) + else: + raise AssertionError("缺少 result 应报协议错误") + + +def main(): + test_wbi_helpers() + test_parse_search_payload() + test_wbi_search_and_fallback() + test_missing_result_is_protocol_error() + print("OK: WBI 签名、搜索解析与降级测试全部通过") + + +if __name__ == "__main__": + main() diff --git a/tests/test_telegram.py b/tests/test_telegram.py new file mode 100644 index 0000000..b6f6952 --- /dev/null +++ b/tests/test_telegram.py @@ -0,0 +1,156 @@ +# -*- coding: utf-8 -*- +"""Telegram 消息构造与回调分发测试（不联网）: +uv run python tests/test_telegram.py +""" + +import json +import os +import sys + +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 + TelegramClient, + format_duration, +) +from reaction_board.services.favorites import ActionResult # noqa: E402 +from reaction_board.services.telegram_poller import TelegramPoller # noqa: E402 + + +class FakeTransport: + def __init__(self, result=None): + self.result = result + self.calls = [] + + def post(self, url, data, timeout=30): + self.calls.append({"url": url, "data": dict(data), "timeout": timeout}) + return {"ok": True, "result": self.result if self.result is not None else {"message_id": 1}} + + +def candidate(aid, title="咻咻满 reaction", up_name="UP", duration=3725): + return { + "aid": aid, + "title": title, + "up_name": up_name, + "duration": duration, + "play": 12345, + "like": 678, + } + + +def test_format_duration(): + assert format_duration(0) == "0:00" + assert format_duration(65) == "1:05" + assert format_duration(3725) == "1:02:05" + + +def test_send_candidates_and_chunking(): + transport = FakeTransport() + bot = TelegramClient("token", 42, transport=transport) + assert bot.send_candidates([candidate(1), candidate(2)]) is True + assert len(transport.calls) == 1 + call = transport.calls[0] + assert call["url"].endswith("/sendMessage") + assert call["data"]["chat_id"] == 42 + assert "咻咻满 reaction" in call["data"]["text"] + 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", + ] + + transport = FakeTransport() + bot = TelegramClient("token", 42, transport=transport) + bot.send_candidates([candidate(i) for i in range(4)]) + assert len(transport.calls) == 2 + assert len(json.loads(transport.calls[1]["data"]["reply_markup"])["inline_keyboard"]) == 1 + + +def test_get_updates_and_answer(): + transport = FakeTransport(result=[]) + bot = TelegramClient("token", 42, transport=transport, poll_timeout=17) + assert bot.get_updates(offset=99) == [] + data = transport.calls[0]["data"] + assert data["offset"] == 99 + assert data["timeout"] == 17 + assert json.loads(data["allowed_updates"]) == ["callback_query"] + + transport = FakeTransport(result=True) + bot = TelegramClient("token", 42, transport=transport) + bot.answer_callback("cb1", "已加入") + assert transport.calls[0]["url"].endswith("/answerCallbackQuery") + assert transport.calls[0]["data"]["callback_query_id"] == "cb1" + + +class FakeBot: + chat_id = 42 + + def __init__(self, updates): + self.updates = updates + self.answers = [] + + def get_updates(self, offset=None): + return list(self.updates) + + def answer_callback(self, callback_id, text=""): + self.answers.append((callback_id, text)) + + +class FakeActions: + def __init__(self): + self.calls = [] + + def favorite(self, aid): + self.calls.append(("fav", aid)) + return ActionResult(True, aid, "favorite", "done", position=3, message="已加入收藏夹") + + def ignore(self, aid): + self.calls.append(("ign", aid)) + return ActionResult(True, aid, "ignore", "rejected", message="已忽略") + + def blacklist(self, aid): + self.calls.append(("blk", aid)) + return ActionResult(True, aid, "blacklist", "rejected", message="已拉黑") + + +def update(update_id, data, *, chat_id=42, from_id=42): + return { + "update_id": update_id, + "callback_query": { + "id": f"cb{update_id}", + "data": data, + "from": {"id": from_id}, + "message": {"chat": {"id": chat_id}}, + }, + } + + +def test_poller_dispatch_and_authorization(): + bot = FakeBot([ + update(1, "fav:9"), + update(2, "ign:10"), + update(3, "blk:11"), + update(4, "fav:12", chat_id=999), + update(5, "bad"), + ]) + 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 bot.answers[3] == ("cb4", "无权限") + assert bot.answers[4] == ("cb5", "未知操作") + + +def main(): + test_format_duration() + test_send_candidates_and_chunking() + test_get_updates_and_answer() + test_poller_dispatch_and_authorization() + print("OK: Telegram 消息、分片、长轮询与回调授权测试全部通过") + + +if __name__ == "__main__": + main()