From 5c2ab7b1873f5da5aa79322190544b9a817d4eac Mon Sep 17 00:00:00 2001 From: kaireewu Date: Sat, 31 Oct 2026 19:19:24 +0800 Subject: [PATCH] feat(p1): add daily snapshots and fan tracking Implement P1: seven-metric view snapshots, relation/stat fan snapshots, failure placeholders, resume by daily completion markers, Cookie login warnings, daily SQLite backups, Shanghai daily scheduling, and CLI snapshot/auth-check commands. Real read-only run captured 795/795 videos successfully over 1124s with a verified daily backup. --- DEVELOPMENT.md | 32 ++- README.md | 46 +++- app.py | 4 +- project.md | 9 +- reaction_board/bilibili/client.py | 73 +++++- reaction_board/bilibili/models.py | 80 ++++++ reaction_board/cli.py | 152 +++++++++--- reaction_board/database.py | 20 ++ reaction_board/repository.py | 132 +++++++++- reaction_board/scheduler.py | 6 +- reaction_board/services/maintenance.py | 50 ++++ reaction_board/services/snapshot.py | 313 +++++++++++++++++++++++ reaction_board/timeutil.py | 59 +++++ tests/fixtures.py | 105 ++++++++ tests/test_bilibili_client.py | 92 +++++++ tests/test_cli.py | 49 +++- tests/test_maintenance.py | 78 ++++++ tests/test_scheduler.py | 19 ++ tests/test_snapshot.py | 331 +++++++++++++++++++++++++ tests/test_timeutil.py | 84 +++++++ 20 files changed, 1680 insertions(+), 54 deletions(-) create mode 100644 reaction_board/services/maintenance.py create mode 100644 reaction_board/services/snapshot.py create mode 100644 reaction_board/timeutil.py create mode 100644 tests/test_maintenance.py create mode 100644 tests/test_snapshot.py create mode 100644 tests/test_timeutil.py diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index 23d6427..a2932d6 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -1,6 +1,6 @@ # 咻咻满 Reaction 视频数据看板开发文档 -> **状态：** P0（数据底座与收藏夹同步）已实施并通过只读联调；P1 待开始 +> **状态：** P0（数据底座与收藏夹同步）、P1（每日快照与粉丝追踪）已实施并通过只读联调；P2 待开始 > > **技术基线：** Python 3.14、Python 标准库、SQLite、原生 HTML/CSS/JavaScript、`uv` > @@ -95,6 +95,13 @@ uv run python app.py sync uv run python app.py run uv run python app.py run --once +# 每日七项快照与粉丝数（P1） +uv run python app.py snapshot +uv run python app.py snapshot --force + +# 检查 Cookie 登录态（P1） +uv run python app.py auth-check + # 查看数据库状态（离线，P0） uv run python app.py status @@ -388,7 +395,7 @@ score < 0：丢弃 - 对 `status IN ('active', 'unfaved')` 的视频逐条请求 `view?aid=`； - 成功写入播放、点赞、收藏、投币、转发、评论、弹幕七项数据； -- `-404` 时标记 `removed`； +- `-404` / `62002` / `62012` 时标记 `removed`（62002/62012 为当前实测返回）； - 其他错误最多重试 3 次，仍失败则写入 `ok=0` 与错误信息，不写伪造的零值； - 每条视频请求之间随机等待 1.0～1.6 秒； - 对 `track_fans` 中的每个 mid 请求一次 `relation/stat` 并写入 `up_stat`； @@ -396,6 +403,15 @@ score < 0：丢弃 服务启动后若当天快照尚未成功完成，应补跑当日任务；已成功的 `(stat_date, aid)` 不重复请求。 +P1 实现说明： + +- 时区使用固定 UTC+8（中国无夏令时），避免 Windows 缺少 `tzdata`； +- 当日全部视频与粉丝采集成功后在 `kv` 写入 `snapshot:completed:=1`；启动时未完成会补跑； +- `ok=0` 的当日记录会在下次补跑时重试；`ok=1` 直接跳过； +- 每次采集后生成 `data/backup/app-YYYYMMDD.db` 并清理超过 `storage.backup_keep_days` 的日备份； +- Cookie 登录态通过 `x/web-interface/nav` 检查，失效只告警，不阻塞匿名 `view`； +- Telegram 摘要推迟到 P3，P1 先写结构化 JSON 日志和 `kv` 状态。 + --- ## 8. Telegram 与收藏操作 @@ -555,12 +571,22 @@ uv run python tests/test_moves.py - 黑名单 9 个 mid 已写入 `up_blacklist`，且不参与收藏夹同步过滤； - 自动化测试见 `tests/test_config.py`、`test_migrations.py`、`test_database.py`、`test_repository.py`、`test_bilibili_client.py`、`test_folder_sync.py`、`test_scheduler.py`、`test_logging.py`、`test_cli.py`。 -### P1：快照与粉丝追踪 +### P1：快照与粉丝追踪 ✅ - 每日七项视频快照； - 粉丝快照； - 失败占位、断点续跑、Cookie 失效告警、SQLite 备份。 +完成情况（2026-10-31）： + +- 已实现 `services/snapshot.py`：`view` 七项指标、`relation/stat` 粉丝数、`ok=0` 失败占位、`ok=1` 断点跳过； +- 实测不可访问视频返回 `62012`，已和旧口径 `-404`、`62002` 一起按 `removed` 处理； +- `run` 调度器同时注册收藏夹同步与每日 `snapshot_at` 任务，启动时立即检查当日是否完成以支持补跑； +- 每次快照后生成日备份并清理超过 14 天的日备份； +- `auth-check` 与快照启动时检查 Cookie 登录态，失效写 `kv` 并告警； +- 真实全量联调：795 条视频全部 `ok=1`，0 失败、0 移除，耗时 1124 秒；粉丝 `follower=2767270`；日备份 `data/backup/app-20261031.db` 含 795 条快照；二次执行正确命中完成标记并跳过； +- 自动化测试见 `tests/test_snapshot.py`、`test_maintenance.py`、`test_timeutil.py`，并扩展了 `test_bilibili_client.py` 与 `test_scheduler.py`。 + ### P2：本地看板 - 总览、榜单、详情、UP、粉丝、候选和系统状态页面； diff --git a/README.md b/README.md index c938953..779dfea 100644 --- a/README.md +++ b/README.md @@ -4,9 +4,9 @@ 项目以 B 站年度收藏夹为可信视频来源，补充关键词发现和 Telegram 人工确认，持续采集 reaction 视频的七项数据与咻咻满粉丝数，并提供本地数据看板。现有的收藏夹排序工具仍保留并作为一键加入后的排序能力基础。 -> 当前状态：**P0（数据底座与收藏夹同步）已实施，P1 待开始。** +> 当前状态：**P0（数据底座与收藏夹同步）、P1（快照与粉丝追踪）已实施；P2 待开始。** > -> 当前已可用：扫码授权、收藏夹读取、排序计划生成、执行与校验；SQLite 建库迁移、收藏夹首次导入与增量同步、失效识别、黑名单种子。 +> 当前已可用：扫码授权、收藏夹读取、排序计划生成、执行与校验；SQLite 建库迁移、收藏夹首次导入与增量同步、失效识别、黑名单种子；每日七项视频快照、粉丝数快照、失败占位、断点续跑与日备份。 > > 详细业务口径、接口验证和风险记录见 [project.md](project.md)；开发架构与实施计划见 [DEVELOPMENT.md](DEVELOPMENT.md)。 @@ -17,8 +17,8 @@ | 收藏夹排序 | 可用 | 同一 UP 连续，组内按发布时间倒序，组间按该 UP 最早视频发布时间倒序。 | | 扫码授权与 Cookie 续期 | 可用 | 本地生成二维码；更新身份 Cookie 时保留设备指纹。 | | 收藏夹同步与首次导入 | 可用（P0） | 监控已配置年度夹，自动发现人工新收藏的视频与失效资源。 | -| 视频日快照 | 规划（P1） | 逐条采集播放、点赞、投币、收藏、转发、评论、弹幕。 | -| 粉丝追踪 | 规划（P1） | 每日记录咻咻满等配置 UP 的粉丝数。 | +| 视频日快照 | 可用（P1） | 逐条采集播放、点赞、投币、收藏、转发、评论、弹幕。 | +| 粉丝追踪 | 可用（P1） | 每日记录咻咻满等配置 UP 的粉丝数。 | | 本地数据看板 | 规划（P2） | 总览、榜单、视频/UP 详情、粉丝曲线、候选池和系统状态。 | | Telegram 发现与一键加入 | 规划（P3） | 每小时发现候选，支持收藏、忽略、拉黑和收藏夹内定位。 | @@ -68,7 +68,7 @@ uv run python app.py init # 首次导入或增量同步配置中的年度收藏夹（只读联网） uv run python app.py sync -# 按 folder_sync_minutes 常驻周期同步；--once 只跑一次 +# 按配置周期常驻运行收藏夹同步 + 每日快照；--once 只跑一次到期任务 uv run python app.py run uv run python app.py run --once @@ -86,6 +86,32 @@ uv run python app.py status - 黑名单只作为发现阶段的种子写入，**不会过滤收藏夹同步**； - 收藏夹达到 950 条时预警，系统绝不自动换夹。 +## P1：每日快照与粉丝追踪 + +P1 已实现每日七项视频指标快照、咻咻满粉丝数快照、失败占位、断点续跑、Cookie 失效告警和 SQLite 日备份。 + +```powershell +# 采集当日快照与粉丝数（只读联网，约 18～20 分钟 / 795 条） +uv run python app.py snapshot + +# 忽略当日完成标记，重新采集 +uv run python app.py snapshot --force + +# 检查 Cookie 登录态 +uv run python app.py auth-check +``` + +- `snapshot` 对 `active` / `unfaved` 视频逐条请求 `view`，每条请求间隔 1.0～1.6 秒； +- 成功写入 `snapshot` 的七项指标；同日 `ok=1` 的记录直接跳过，可断点续跑； +- 视频不可访问（`-404` / `62002` / `62012`）标记为 `removed`； +- 其他错误重试 3 次，仍失败写入 `ok=0` 和错误信息，不写伪造的零值； +- 粉丝数写入 `up_stat`，失败同样写 `ok=0`； +- 当日全部视频与粉丝采集成功后写入完成标记；服务启动时若当天未完成会自动补跑； +- 每次采集后生成 `data/backup/app-YYYYMMDD.db`，按 `storage.backup_keep_days` 清理旧日备份； +- Cookie 失效只告警，不阻塞匿名 `view` 采集；`auth-check` 可单独检查。 + +首次真实全量采集结果：795 条视频全部成功、0 失败、0 移除，耗时约 19 分钟；粉丝 `follower=2767270`；日备份为 `data/backup/app-20261031.db`。 + ## 配置 复制样例配置，并填写 B 站账号信息： @@ -198,8 +224,11 @@ uv run python tests/test_config.py # 配置加载、校验、ENV 覆盖 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 契约、分块、缺失与重试 +uv run python tests/test_bilibili_client.py # ids/infos/view/fans/nav 契约与重试 uv run python tests/test_folder_sync.py # 首次导入、差集、失效识别、容量预警 +uv run python tests/test_snapshot.py # 七项快照、断点续跑、失败占位、粉丝 +uv run python tests/test_maintenance.py # 日备份与保留期 +uv run python tests/test_timeutil.py # 上海时区与每日任务时间 uv run python tests/test_scheduler.py # 内置调度器到期/失败隔离 uv run python tests/test_logging.py # JSON 日志脱敏 uv run python tests/test_cli.py # CLI 离线路径 @@ -211,8 +240,9 @@ uv run python tests/test_cli.py # CLI 离线路径 config.json.bak 最近一次 auth 更新前的配置备份 data/plan-.json 排序计划、顺序快照和元数据缓存 data/journal-.jsonl 排序执行日志 -data/app.db P0 起的 SQLite 数据库（收藏夹、视频、黑名单） -data/backup/ 数据库迁移前的自动备份 +data/app.db SQLite 数据库（收藏夹、视频、快照、粉丝、黑名单） +data/backup/app-YYYYMMDD.db 每日快照后的 SQLite 备份 +data/backup/app-pre-v*.db 数据库迁移前的自动备份 ``` ## 文档索引 diff --git a/app.py b/app.py index e42112b..f1ba40f 100644 --- a/app.py +++ b/app.py @@ -1,8 +1,8 @@ # -*- coding: utf-8 -*- """看板服务入口。 -P0 提供 ``init`` / ``sync`` / ``run`` / ``status`` 子命令；后续阶段会在同一入口启动 -本地 Web 服务。 +P1 提供 ``init`` / ``sync`` / ``snapshot`` / ``auth-check`` / ``run`` / ``status`` +子命令；后续阶段会在同一入口启动本地 Web 服务。 """ import sys diff --git a/project.md b/project.md index 2d22d38..7d88d4e 100644 --- a/project.md +++ b/project.md @@ -1,7 +1,7 @@ # 咻咻满 Reaction 视频数据看板 —— 可行性分析与技术路线 -> 状态：**设计定稿；P0（数据底座与收藏夹同步）已实施并完成只读联调** -> 最后更新：2026-10-31（第 5 轮：P0 落地） +> 状态：**设计定稿；P0（数据底座与收藏夹同步）、P1（每日快照与粉丝追踪）已实施** +> 最后更新：2026-10-31（第 6 轮：P1 落地） > 全部探测均为只读（GET），未调用 `fav/resource/deal`、`fav/resource/sort` --- @@ -770,6 +770,11 @@ storage：SQLite WAL，单写线程 | 21 | **搜索 item 的 duration 格式** | `分钟:秒`（`105:5`、`336:31`），超 60 分钟不转小时 | | 22 | **搜索 item 自带统计字段** | `play/like/favorites/danmaku/review/duration/cover/pubdate/mid/uname` 齐全，通知可用索引值 | | 23 | **初始黑名单 9 个 mid 核对** | `x/web-interface/card` 全部 `code=0`，昵称/粉丝数已记录（§1.6） | +| 24 | `x/web-interface/view` 七项 stat | `code=0`，view/like/favorite/coin/share/reply/danmaku 齐全 | +| 25 | `view` 不可访问返回码 | 实测为 `62012`；代码同时兼容 `-404`、`62002`、HTTP 404 | +| 26 | `x/relation/stat` 粉丝数 | `code=0`，匿名可用，实测 `follower=2767268` | +| 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 条快照 | **未验证（需授权或外部配合）**： diff --git a/reaction_board/bilibili/client.py b/reaction_board/bilibili/client.py index 5a2b9bd..0c3747c 100644 --- a/reaction_board/bilibili/client.py +++ b/reaction_board/bilibili/client.py @@ -18,11 +18,21 @@ from dataclasses import dataclass from typing import Any, Mapping, Protocol, Sequence from .auth import make_headers -from .models import FavId, FavResource, FolderInfo, InfosResult +from .models import ( + FanStat, + FavId, + FavResource, + FolderInfo, + InfosResult, + LoginState, + ViewStat, +) BASE = "https://api.bilibili.com" DEFAULT_CHUNK = 200 MAX_CHUNK = 400 +# 视频不可访问/已删除：-404 为旧口径，62002/62012 为当前实测返回。 +VIDEO_UNAVAILABLE_CODES = frozenset({-404, 62002, 62012}) class BilibiliError(RuntimeError): @@ -360,3 +370,64 @@ class BilibiliClient: media_count=count, )) return out + + # ------------------------------------------------------------ view / fan + + def view(self, aid: int) -> ViewStat: + url = f"{BASE}/x/web-interface/view?aid={int(aid)}" + payload = self.get_json(url) + data = payload.get("data") + if not isinstance(data, Mapping): + raise BilibiliProtocolError("view data 不是对象") + try: + return ViewStat.from_api(data) + except (TypeError, ValueError) as exc: + raise BilibiliProtocolError(f"view 解析失败: {exc}") from exc + + def relation_stat(self, mid: int) -> FanStat: + url = f"{BASE}/x/relation/stat?vmid={int(mid)}" + payload = self.get_json(url) + data = payload.get("data") + if not isinstance(data, Mapping): + raise BilibiliProtocolError("relation/stat data 不是对象") + try: + return FanStat.from_api(data) + except (TypeError, ValueError) as exc: + raise BilibiliProtocolError(f"relation/stat 解析失败: {exc}") from exc + + def check_login(self) -> LoginState: + """检查 Cookie 登录态；未配置 Cookie 时不发请求。""" + if not self.cookie: + return LoginState( + is_login=False, + mid=None, + uname=None, + code=0, + message="未配置 cookie", + ) + url = f"{BASE}/x/web-interface/nav" + try: + payload = self.get_json(url, auth=True) + except BilibiliAPIError as exc: + return LoginState( + is_login=False, mid=None, uname=None, + code=exc.code, message=exc.message, + ) + except BilibiliError as exc: + return LoginState( + is_login=False, mid=None, uname=None, + code=-1, message=str(exc), + ) + data = payload.get("data") or {} + if not isinstance(data, Mapping): + raise BilibiliProtocolError("nav data 不是对象") + raw_mid = data.get("mid") + mid = raw_mid if isinstance(raw_mid, int) and not isinstance(raw_mid, bool) else None + uname = data.get("uname") + return LoginState( + is_login=bool(data.get("isLogin")), + mid=mid, + uname=str(uname) if isinstance(uname, str) else None, + code=0, + message=str(payload.get("message") or ""), + ) diff --git a/reaction_board/bilibili/models.py b/reaction_board/bilibili/models.py index d55bc83..ec56f5f 100644 --- a/reaction_board/bilibili/models.py +++ b/reaction_board/bilibili/models.py @@ -85,3 +85,83 @@ class InfosResult: requested: tuple[int, ...] items: dict[int, FavResource] missing: tuple[int, ...] + + +@dataclass(frozen=True) +class ViewStat: + """``x/web-interface/view`` 的七项统计。""" + + aid: int + bvid: str | None + title: str + view: int + like: int + favorite: int + coin: int + share: int + reply: int + danmaku: int + + def metrics(self) -> dict[str, int]: + return { + "view": self.view, + "like": self.like, + "favorite": self.favorite, + "coin": self.coin, + "share": self.share, + "reply": self.reply, + "danmaku": self.danmaku, + } + + @classmethod + def from_api(cls, item: Mapping[str, Any]) -> "ViewStat": + aid = item.get("aid") + if isinstance(aid, bool) or not isinstance(aid, int): + raise ValueError("view data.aid 不是整数") + stat = item.get("stat") + if not isinstance(stat, Mapping): + raise ValueError("view data.stat 不是对象") + metrics: dict[str, int] = {} + for key in ("view", "like", "favorite", "coin", "share", "reply", "danmaku"): + value = stat.get(key) + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise ValueError(f"view stat.{key} 不是非负整数") + metrics[key] = value + bvid = item.get("bvid") + if not isinstance(bvid, str) or not bvid: + bvid = None + return cls( + aid=aid, + bvid=bvid, + title=str(item.get("title") or ""), + **metrics, + ) + + +@dataclass(frozen=True) +class FanStat: + mid: int + follower: int + following: int + + @classmethod + def from_api(cls, item: Mapping[str, Any]) -> "FanStat": + mid = item.get("mid") + follower = item.get("follower") + following = item.get("following", 0) + if isinstance(mid, bool) or not isinstance(mid, int): + raise ValueError("relation/stat data.mid 不是整数") + if isinstance(follower, bool) or not isinstance(follower, int) or follower < 0: + raise ValueError("relation/stat data.follower 不是非负整数") + if isinstance(following, bool) or not isinstance(following, int) or following < 0: + raise ValueError("relation/stat data.following 不是非负整数") + return cls(mid=mid, follower=follower, following=following) + + +@dataclass(frozen=True) +class LoginState: + is_login: bool + mid: int | None + uname: str | None + code: int + message: str diff --git a/reaction_board/cli.py b/reaction_board/cli.py index 3af4a1c..30fce28 100644 --- a/reaction_board/cli.py +++ b/reaction_board/cli.py @@ -1,10 +1,12 @@ # -*- coding: utf-8 -*- -"""P0 命令行入口。 +"""看板命令行入口。 - python app.py init # 建库 / 迁移 / 写入黑名单种子（离线） - python app.py sync # 收藏夹只读同步（首次导入或增量） - python app.py run # 按 folder_sync_minutes 常驻周期同步（--once 只跑一次） - python app.py status # 查看库内状态（离线） + python app.py init # 建库 / 迁移 / 写入黑名单种子（离线） + python app.py sync # 收藏夹只读同步（首次导入或增量） + python app.py snapshot # 采集当日七项快照与粉丝数 + python app.py auth-check # 检查 Cookie 登录态 + python app.py run # 按配置周期常驻运行（--once 只跑一次） + python app.py status # 查看库内状态（离线） P2 起 ``python app.py`` 将在此基础上启动本地看板服务。 """ @@ -17,13 +19,15 @@ import sys from pathlib import Path from typing import Sequence -from .bilibili.client import BilibiliClient +from .bilibili.client import BilibiliClient, Pacer from .config import ConfigError, ConfigStore from .database import Database from .logging_json import JsonLogger from .repository import Repository from .scheduler import ScheduledTask, Scheduler from .services.folder_sync import FolderSyncService +from .services.snapshot import SnapshotService +from .timeutil import next_daily_run, parse_hhmm, shanghai_date ROOT = Path(__file__).resolve().parent.parent DEFAULT_CONFIG = ROOT / "config.json" @@ -39,7 +43,7 @@ def _build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="app.py", - description="咻咻满 Reaction 数据看板 —— P0 数据底座", + description="咻咻满 Reaction 数据看板 —— 数据底座、同步与每日快照", ) sub = parser.add_subparsers(dest="command", required=True) @@ -48,7 +52,14 @@ def _build_parser() -> argparse.ArgumentParser: sync.add_argument( "--full", action="store_true", help="强制全量刷新元数据（忽略 6 小时周期）" ) - run = sub.add_parser("run", parents=[common], help="按配置周期常驻运行收藏夹同步") + snapshot = sub.add_parser( + "snapshot", parents=[common], help="采集当日七项快照与粉丝数" + ) + snapshot.add_argument( + "--force", action="store_true", help="忽略当日完成标记，重新采集" + ) + sub.add_parser("auth-check", parents=[common], help="检查 Cookie 登录态") + run = sub.add_parser("run", parents=[common], help="按配置周期常驻运行同步与快照") run.add_argument( "--once", action="store_true", help="只执行一次到期任务后退出" ) @@ -71,6 +82,17 @@ def _print(payload) -> None: print(json.dumps(payload, ensure_ascii=False, indent=2), flush=True) +def _client(cfg, logger: JsonLogger, *, pacer=None, retries: int = 3) -> BilibiliClient: + return BilibiliClient( + cookie=cfg.bilibili.cookie, + csrf=cfg.bilibili.csrf, + mid=cfg.bilibili.mid, + pacer=pacer, + retries=retries, + logger=logger, + ) + + def cmd_init(args, logger: JsonLogger) -> int: store = ConfigStore(args.config, require_auth=False) cfg = store.get() @@ -98,12 +120,7 @@ def cmd_sync(args, logger: JsonLogger) -> int: db = _open_db(cfg, logger) try: repo = Repository(db) - client = BilibiliClient( - cookie=cfg.bilibili.cookie, - csrf=cfg.bilibili.csrf, - mid=cfg.bilibili.mid, - logger=logger, - ) + client = _client(cfg, logger) service = FolderSyncService(store, repo, client, logger=logger) summary = service.run(full_refresh=args.full) payload = summary.as_dict() @@ -115,6 +132,46 @@ def cmd_sync(args, logger: JsonLogger) -> int: db.close() +def cmd_snapshot(args, logger: JsonLogger) -> int: + store = ConfigStore(args.config, require_auth=False) + cfg = store.get() + db = _open_db(cfg, logger) + try: + repo = Repository(db) + client = _client( + cfg, logger, pacer=Pacer(1.0, 1.6), retries=1 + ) + service = SnapshotService(store, repo, client, logger=logger) + summary = service.run_if_needed(force=args.force) + payload = summary.as_dict() + payload["database"] = str(cfg.storage.database_path) + payload["snapshot_counts"] = repo.snapshot_counts(summary.stat_date) + payload["counts"] = repo.counts() + _print(payload) + return 0 if summary.completed else 1 + finally: + db.close() + + +def cmd_auth_check(args, logger: JsonLogger) -> int: + store = ConfigStore(args.config, require_auth=False) + cfg = store.get() + if not cfg.bilibili.cookie: + _print({"ok": False, "is_login": False, "message": "未配置 cookie"}) + return 1 + client = _client(cfg, logger, retries=1) + state = client.check_login() + _print({ + "ok": state.is_login, + "is_login": state.is_login, + "mid": state.mid, + "uname": state.uname, + "code": state.code, + "message": state.message, + }) + return 0 if state.is_login else 1 + + def cmd_run(args, logger: JsonLogger) -> int: store = ConfigStore(args.config, require_auth=True) cfg = store.get() @@ -122,26 +179,40 @@ def cmd_run(args, logger: JsonLogger) -> int: last: dict[str, object] = {} try: repo = Repository(db) - client = BilibiliClient( - cookie=cfg.bilibili.cookie, - csrf=cfg.bilibili.csrf, - mid=cfg.bilibili.mid, - logger=logger, - ) - service = FolderSyncService(store, repo, client, logger=logger) + folder_client = _client(cfg, logger) + folder_service = FolderSyncService(store, repo, folder_client, logger=logger) + snapshot_client = _client(cfg, logger, pacer=Pacer(1.0, 1.6), retries=1) + snapshot_service = SnapshotService(store, repo, snapshot_client, logger=logger) def run_sync(): - summary = service.run() - last["summary"] = summary + summary = folder_service.run() + last["folder"] = summary logger.info("folder_sync.summary", **summary.as_dict()) return summary + def run_snapshot(): + summary = snapshot_service.run_if_needed() + last["snapshot"] = summary + logger.info("snapshot.summary", **summary.as_dict()) + return summary + + hh, mm = parse_hhmm(cfg.schedule.snapshot_at) + snapshot_at = f"{hh:02d}:{mm:02d}" scheduler = Scheduler( - [ScheduledTask( - "folder_sync", - float(cfg.schedule.folder_sync_minutes * 60), - run_sync, - )], + [ + ScheduledTask( + "folder_sync", + float(cfg.schedule.folder_sync_minutes * 60), + run_sync, + ), + ScheduledTask( + "snapshot", + 86400.0, + run_snapshot, + next_run=0.0, + next_run_fn=lambda started: next_daily_run(started, snapshot_at), + ), + ], logger=logger, ) if args.once: @@ -150,11 +221,16 @@ def cmd_run(args, logger: JsonLogger) -> int: else: logger.info( "scheduler.start", - interval_seconds=cfg.schedule.folder_sync_minutes * 60, + folder_sync_seconds=cfg.schedule.folder_sync_minutes * 60, + snapshot_at=snapshot_at, ) scheduler.run_forever() - summary = last.get("summary") - return 0 if summary is None or getattr(summary, "ok", True) else 1 + + folder_summary = last.get("folder") + snapshot_summary = last.get("snapshot") + 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 finally: db.close() @@ -165,6 +241,15 @@ def cmd_status(args, logger: JsonLogger) -> int: db = _open_db(cfg, logger) try: repo = Repository(db) + today = shanghai_date() + snapshot = repo.snapshot_counts(today) + snapshot.update({ + "date": today, + "last_attempt": repo.kv_get("snapshot:last_attempt"), + "last_completed": repo.kv_get("snapshot:last_completed"), + "auth_ok": repo.kv_get("auth:ok"), + "auth_message": repo.kv_get("auth:message"), + }) _print({ "ok": True, "schema_version": db.schema_version, @@ -181,6 +266,7 @@ def cmd_status(args, logger: JsonLogger) -> int: } for row in repo.list_folders() ], + "snapshot": snapshot, **repo.counts(), }) finally: @@ -196,6 +282,10 @@ def main(argv: Sequence[str] | None = None) -> int: return cmd_init(args, logger) if args.command == "sync": return cmd_sync(args, logger) + if args.command == "snapshot": + return cmd_snapshot(args, logger) + if args.command == "auth-check": + return cmd_auth_check(args, logger) if args.command == "run": return cmd_run(args, logger) if args.command == "status": diff --git a/reaction_board/database.py b/reaction_board/database.py index 43a06e7..3066fe2 100644 --- a/reaction_board/database.py +++ b/reaction_board/database.py @@ -7,6 +7,7 @@ from __future__ import annotations +import os import queue import sqlite3 import threading @@ -165,6 +166,25 @@ class Database: finally: conn.close() + def backup_to(self, target: str | Path) -> Path: + """用 SQLite backup API 生成一致性备份；先写临时文件再原子替换。""" + target = Path(target) + target.parent.mkdir(parents=True, exist_ok=True) + tmp = target.with_name(target.name + ".tmp") + if tmp.exists(): + tmp.unlink() + + def run(conn: sqlite3.Connection) -> None: + dst = sqlite3.connect(tmp) + try: + conn.backup(dst) + finally: + dst.close() + + self.write(run) + os.replace(tmp, target) + return target + def close(self) -> None: if self._closed: return diff --git a/reaction_board/repository.py b/reaction_board/repository.py index 8a5d04c..3ec4f61 100644 --- a/reaction_board/repository.py +++ b/reaction_board/repository.py @@ -10,7 +10,7 @@ from __future__ import annotations import sqlite3 import time from dataclasses import dataclass -from typing import Any, Iterable +from typing import Any, Iterable, Mapping from .config import FolderConfig from .database import Database @@ -368,6 +368,136 @@ class Repository: return [dict(row) for row in rows] return self.db.read(run) + # ------------------------------------------------------------ snapshot + + def video_aids_for_snapshot(self) -> list[int]: + def run(conn: sqlite3.Connection) -> list[int]: + rows = conn.execute( + "SELECT aid FROM video WHERE status IN ('active', 'unfaved') " + "ORDER BY aid" + ).fetchall() + return [int(row["aid"]) for row in rows] + return self.db.read(run) + + def get_snapshot(self, stat_date: str, aid: int) -> dict[str, Any] | None: + def run(conn: sqlite3.Connection) -> dict[str, Any] | None: + row = conn.execute( + "SELECT * FROM snapshot WHERE stat_date = ? AND aid = ?", + (stat_date, int(aid)), + ).fetchone() + return dict(row) if row else None + return self.db.read(run) + + def upsert_snapshot( + self, + stat_date: str, + aid: int, + *, + metrics: Mapping[str, int] | None, + ok: bool, + err: str | None, + captured_at: int, + src: str = "view", + ) -> None: + values = metrics or {} + def run(conn: sqlite3.Connection) -> None: + conn.execute( + """ + INSERT INTO snapshot(stat_date, aid, "view", "like", favorite, coin, + share, reply, danmaku, captured_at, ok, err, src) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(stat_date, aid) DO UPDATE SET + "view" = excluded."view", + "like" = excluded."like", + favorite = excluded.favorite, + coin = excluded.coin, + share = excluded.share, + reply = excluded.reply, + danmaku = excluded.danmaku, + captured_at = excluded.captured_at, + ok = excluded.ok, + err = excluded.err, + src = excluded.src + """, + ( + stat_date, + int(aid), + values.get("view"), + values.get("like"), + values.get("favorite"), + values.get("coin"), + values.get("share"), + values.get("reply"), + values.get("danmaku"), + int(captured_at), + 1 if ok else 0, + err, + src, + ), + ) + self.db.write(run) + + def snapshot_counts(self, stat_date: str) -> dict[str, int]: + def run(conn: sqlite3.Connection) -> dict[str, int]: + rows = conn.execute( + "SELECT ok, COUNT(*) AS n FROM snapshot WHERE stat_date = ? GROUP BY ok", + (stat_date,), + ).fetchall() + by_ok = {int(row["ok"]): int(row["n"]) for row in rows} + return { + "total": sum(by_ok.values()), + "ok": by_ok.get(1, 0), + "failed": by_ok.get(0, 0), + } + return self.db.read(run) + + def upsert_up_stat( + self, + stat_date: str, + mid: int, + *, + follower: int | None, + following: int | None, + ok: bool, + captured_at: int, + ) -> None: + self.db.write(lambda conn: conn.execute( + """ + INSERT INTO up_stat(stat_date, mid, follower, following, captured_at, ok) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(stat_date, mid) DO UPDATE SET + follower = excluded.follower, + following = excluded.following, + captured_at = excluded.captured_at, + ok = excluded.ok + """, + (stat_date, int(mid), follower, following, int(captured_at), 1 if ok else 0), + )) + + def get_up_stat(self, stat_date: str, mid: int) -> dict[str, Any] | None: + def run(conn: sqlite3.Connection) -> dict[str, Any] | None: + row = conn.execute( + "SELECT * FROM up_stat WHERE stat_date = ? AND mid = ?", + (stat_date, int(mid)), + ).fetchone() + return dict(row) if row else None + return self.db.read(run) + + # ------------------------------------------------------------ kv + + def kv_get(self, key: str) -> str | None: + def run(conn: sqlite3.Connection) -> str | None: + row = conn.execute("SELECT v FROM kv WHERE k = ?", (key,)).fetchone() + return None if row is None else str(row["v"]) + return self.db.read(run) + + def kv_set(self, key: str, value: str) -> None: + self.db.write(lambda conn: conn.execute( + "INSERT INTO kv(k, v) VALUES (?, ?) " + "ON CONFLICT(k) DO UPDATE SET v = excluded.v", + (key, str(value)), + )) + # ------------------------------------------------------------ blacklist def seed_blacklist(self, mids: Iterable[int], *, now: int | None = None) -> int: diff --git a/reaction_board/scheduler.py b/reaction_board/scheduler.py index 5acfb43..03a2b1c 100644 --- a/reaction_board/scheduler.py +++ b/reaction_board/scheduler.py @@ -18,6 +18,7 @@ class ScheduledTask: interval: float fn: Callable[[], Any] next_run: float = 0.0 + next_run_fn: Callable[[float], float] | None = None @dataclass @@ -77,7 +78,10 @@ class Scheduler: else: self._log("info", "scheduler.task_ok", task=task.name) finished = self._clock() - task.next_run = started + task.interval + if task.next_run_fn is not None: + task.next_run = task.next_run_fn(started) + else: + task.next_run = started + task.interval result = TaskResult( name=task.name, started_at=started, diff --git a/reaction_board/services/maintenance.py b/reaction_board/services/maintenance.py new file mode 100644 index 0000000..7004407 --- /dev/null +++ b/reaction_board/services/maintenance.py @@ -0,0 +1,50 @@ +# -*- coding: utf-8 -*- +"""备份与保留期清理。""" + +from __future__ import annotations + +import time +from pathlib import Path + +from ..database import Database + + +def daily_backup_path(backup_dir: str | Path, stat_date: str) -> Path: + compact = stat_date.replace("-", "") + return Path(backup_dir) / f"app-{compact}.db" + + +def backup_daily( + db: Database, + backup_dir: str | Path, + stat_date: str, + *, + logger=None, +) -> Path: + target = daily_backup_path(backup_dir, stat_date) + db.backup_to(target) + if logger is not None: + logger.info("backup.daily", path=str(target), stat_date=stat_date) + return target + + +def prune_daily_backups( + backup_dir: str | Path, + *, + keep_days: int, + now: float | None = None, +) -> list[Path]: + """只清理 ``app-YYYYMMDD.db`` 日备份，不动迁移备份。""" + target_dir = Path(backup_dir) + if not target_dir.is_dir(): + return [] + cutoff = (time.time() if now is None else now) - keep_days * 86400 + removed: list[Path] = [] + for path in target_dir.glob("app-[0-9][0-9][0-9][0-9][0-9][0-9][0-9][0-9].db"): + try: + if path.stat().st_mtime < cutoff: + path.unlink() + removed.append(path) + except OSError: + continue + return removed diff --git a/reaction_board/services/snapshot.py b/reaction_board/services/snapshot.py new file mode 100644 index 0000000..2289fa5 --- /dev/null +++ b/reaction_board/services/snapshot.py @@ -0,0 +1,313 @@ +# -*- coding: utf-8 -*- +"""每日视频七项快照与粉丝数采集。 + +规则（DEVELOPMENT.md §7.4）： + +* 对 ``active`` / ``unfaved`` 视频逐条请求 ``view``； +* 成功写入七项指标；``ok=1`` 的同日快照直接跳过（断点续跑）； +* 视频不可访问时标记 ``removed``； +* 其他错误最多重试 3 次，仍失败写 ``ok=0`` 占位，不写假零值； +* 对 ``track_fans`` 逐个请求 ``relation/stat``； +* 完成后写入当日完成标记并生成日备份。 +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass, field +from typing import Any + +from ..bilibili.client import ( + VIDEO_UNAVAILABLE_CODES, + BilibiliAPIError, + BilibiliClient, + BilibiliError, + BilibiliHTTPError, +) +from ..config import AppConfig, ConfigStore +from ..repository import Repository +from ..timeutil import is_date, shanghai_date +from .maintenance import backup_daily, prune_daily_backups + + +@dataclass +class SnapshotSummary: + stat_date: str + started_at: int + total: int = 0 + success: int = 0 + failed: int = 0 + skipped: int = 0 + removed: int = 0 + fans_ok: int = 0 + fans_failed: int = 0 + auth_ok: bool | None = None + auth_message: str = "" + backup_path: str | None = None + backup_error: str | None = None + completed: bool = False + already_completed: bool = False + finished_at: int = 0 + errors: list[str] = field(default_factory=list) + + def as_dict(self) -> dict[str, Any]: + return { + "stat_date": self.stat_date, + "started_at": self.started_at, + "finished_at": self.finished_at, + "elapsed": max(0, self.finished_at - self.started_at), + "total": self.total, + "success": self.success, + "failed": self.failed, + "skipped": self.skipped, + "removed": self.removed, + "fans_ok": self.fans_ok, + "fans_failed": self.fans_failed, + "auth_ok": self.auth_ok, + "auth_message": self.auth_message, + "backup_path": self.backup_path, + "backup_error": self.backup_error, + "completed": self.completed, + "already_completed": self.already_completed, + "errors": self.errors, + } + + +class SnapshotService: + def __init__( + self, + config_store: ConfigStore, + repository: Repository, + client: BilibiliClient, + *, + clock=time.time, + sleeper=time.sleep, + logger=None, + retries: int = 3, + backoff: float = 5.0, + progress_every: int = 50, + ) -> None: + if retries < 1: + raise ValueError("retries 必须 >= 1") + self.config_store = config_store + self.repo = repository + self.client = client + self._clock = clock + self._sleep = sleeper + self.logger = logger + self.retries = retries + self.backoff = backoff + self.progress_every = progress_every + + 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) + + @staticmethod + def completed_key(stat_date: str) -> str: + return f"snapshot:completed:{stat_date}" + + def run_if_needed( + self, *, date: str | None = None, force: bool = False + ) -> SnapshotSummary: + now = self._now() + stat_date = date or shanghai_date(now) + if not is_date(stat_date): + raise ValueError(f"日期格式必须是 YYYY-MM-DD: {stat_date!r}") + if not force and self.repo.kv_get(self.completed_key(stat_date)) == "1": + self._log("info", "snapshot.skip", stat_date=stat_date, reason="already_completed") + return SnapshotSummary( + stat_date=stat_date, + started_at=now, + finished_at=now, + completed=True, + already_completed=True, + ) + return self.run(date=stat_date, force=force) + + def run(self, *, date: str | None = None, force: bool = False) -> SnapshotSummary: + cfg = self.config_store.get() + now = self._now() + stat_date = date or shanghai_date(now) + if not is_date(stat_date): + raise ValueError(f"日期格式必须是 YYYY-MM-DD: {stat_date!r}") + + summary = SnapshotSummary(stat_date=stat_date, started_at=now) + self._log("info", "snapshot.start", stat_date=stat_date, force=force) + + summary.auth_ok, summary.auth_message = self._check_auth(cfg, now) + + videos = self.repo.video_aids_for_snapshot() + summary.total = len(videos) + for index, aid in enumerate(videos, 1): + existing = self.repo.get_snapshot(stat_date, aid) + if existing is not None and int(existing.get("ok") or 0) == 1 and not force: + summary.skipped += 1 + continue + + stat, error, unavailable = self._fetch_view(aid) + if unavailable: + summary.removed += 1 + self.repo.set_video_status(aid, "removed", now=now) + self._log( + "warning", "snapshot.removed", + stat_date=stat_date, aid=aid, error=error, + ) + elif stat is not None: + self.repo.upsert_snapshot( + stat_date, + aid, + metrics=stat.metrics(), + ok=True, + err=None, + captured_at=self._now(), + ) + summary.success += 1 + else: + summary.failed += 1 + self.repo.upsert_snapshot( + stat_date, + aid, + metrics=None, + ok=False, + err=error, + captured_at=self._now(), + ) + summary.errors.append(f"aid {aid}: {error}") + self._log( + "error", "snapshot.failed", + stat_date=stat_date, aid=aid, error=error, + ) + + if self.progress_every > 0 and ( + index % self.progress_every == 0 or index == summary.total + ): + self._log( + "info", "snapshot.progress", + stat_date=stat_date, done=index, total=summary.total, + success=summary.success, failed=summary.failed, + skipped=summary.skipped, removed=summary.removed, + ) + + for mid in cfg.track_fans: + stat, error = self._fetch_fan(mid) + if stat is not None: + self.repo.upsert_up_stat( + stat_date, + mid, + follower=stat.follower, + following=stat.following, + ok=True, + captured_at=self._now(), + ) + summary.fans_ok += 1 + else: + summary.fans_failed += 1 + self.repo.upsert_up_stat( + stat_date, + mid, + follower=None, + following=None, + ok=False, + captured_at=self._now(), + ) + summary.errors.append(f"fans mid {mid}: {error}") + self._log( + "error", "snapshot.fan_failed", + stat_date=stat_date, mid=mid, error=error, + ) + + summary.completed = summary.failed == 0 and summary.fans_failed == 0 + self.repo.kv_set(self.completed_key(stat_date), "1" if summary.completed else "0") + self.repo.kv_set("snapshot:last_attempt", stat_date) + if summary.completed: + self.repo.kv_set("snapshot:last_completed", stat_date) + elif self.repo.kv_get("snapshot:last_completed") == stat_date: + self.repo.kv_set("snapshot:last_completed", "") + + try: + path = backup_daily( + self.repo.db, + cfg.storage.backup_dir, + stat_date, + logger=self.logger, + ) + prune_daily_backups( + cfg.storage.backup_dir, + keep_days=cfg.storage.backup_keep_days, + ) + summary.backup_path = str(path) + except Exception as exc: # noqa: BLE001 备份失败不推翻已采集数据 + summary.backup_error = str(exc) + self._log("error", "backup.failed", error=str(exc), stat_date=stat_date) + + summary.finished_at = self._now() + self._log( + "info" if summary.completed else "warning", + "snapshot.done", + **summary.as_dict(), + ) + return summary + + # ------------------------------------------------------------ internals + + def _check_auth(self, cfg: AppConfig, now: int) -> tuple[bool, str]: + if not cfg.bilibili.cookie: + self.repo.kv_set("auth:last_check", str(now)) + self.repo.kv_set("auth:ok", "0") + self.repo.kv_set("auth:message", "未配置 cookie") + return False, "未配置 cookie" + try: + state = self.client.check_login() + except BilibiliError as exc: + message = str(exc) + self.repo.kv_set("auth:last_check", str(now)) + self.repo.kv_set("auth:ok", "0") + self.repo.kv_set("auth:message", message) + self._log("warning", "auth.invalid", message=message) + return False, message + + self.repo.kv_set("auth:last_check", str(now)) + self.repo.kv_set("auth:ok", "1" if state.is_login else "0") + self.repo.kv_set("auth:message", state.message) + if state.mid is not None: + self.repo.kv_set("auth:mid", str(state.mid)) + if not state.is_login: + self._log( + "warning", "auth.invalid", + code=state.code, message=state.message, + ) + return state.is_login, state.message + + def _fetch_view(self, aid: int): + last_error = "" + for attempt in range(1, self.retries + 1): + try: + return self.client.view(aid), None, False + except BilibiliAPIError as exc: + last_error = str(exc) + if exc.code in VIDEO_UNAVAILABLE_CODES: + return None, last_error, True + except BilibiliHTTPError as exc: + last_error = str(exc) + if exc.status == 404: + return None, last_error, True + except BilibiliError as exc: + last_error = str(exc) + if attempt < self.retries: + self._sleep(self.backoff * (2 ** (attempt - 1))) + return None, last_error or "未知错误", False + + def _fetch_fan(self, mid: int): + last_error = "" + for attempt in range(1, self.retries + 1): + try: + return self.client.relation_stat(mid), None + except BilibiliError as exc: + last_error = str(exc) + if attempt < self.retries: + self._sleep(self.backoff * (2 ** (attempt - 1))) + return None, last_error or "未知错误" diff --git a/reaction_board/timeutil.py b/reaction_board/timeutil.py new file mode 100644 index 0000000..74e02e9 --- /dev/null +++ b/reaction_board/timeutil.py @@ -0,0 +1,59 @@ +# -*- coding: utf-8 -*- +"""上海时区与每日任务时间计算。 + +B 站数据口径使用 ``Asia/Shanghai``。中国自 1991 年起不再使用夏令时，因此这里 +用固定 UTC+8，避免 Windows 上依赖未安装的 ``tzdata`` 包。 +""" + +from __future__ import annotations + +import re +import time +from datetime import datetime, timedelta, timezone + +SHANGHAI_TZ = timezone(timedelta(hours=8)) +_HHMM_RE = re.compile(r"^([01]\d|2[0-3]):([0-5]\d)$") +_DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$") + + +def shanghai_datetime(ts: float | None = None) -> datetime: + return datetime.fromtimestamp(time.time() if ts is None else ts, SHANGHAI_TZ) + + +def shanghai_date(ts: float | None = None) -> str: + return shanghai_datetime(ts).strftime("%Y-%m-%d") + + +def parse_hhmm(value: str) -> tuple[int, int]: + match = _HHMM_RE.match(value or "") + if not match: + raise ValueError(f"时间必须是 HH:MM（24 小时制）: {value!r}") + return int(match.group(1)), int(match.group(2)) + + +def next_daily_run(now: float, hhmm: str) -> float: + """返回 ``now`` 之后（严格大于）最近的 ``hh:mm`` 上海时间时间戳。""" + hour, minute = parse_hhmm(hhmm) + current = shanghai_datetime(now) + target = current.replace(hour=hour, minute=minute, second=0, microsecond=0) + if target.timestamp() <= now: + target += timedelta(days=1) + return target.timestamp() + + +def is_date(value: str) -> bool: + if not _DATE_RE.match(value or ""): + return False + try: + datetime.strptime(value, "%Y-%m-%d") + except ValueError: + return False + return True + + +def day_bounds(date: str) -> tuple[float, float]: + """返回上海时区某一天的 [start, end) 时间戳。""" + if not is_date(date): + raise ValueError(f"日期格式必须是 YYYY-MM-DD: {date!r}") + start = datetime.strptime(date, "%Y-%m-%d").replace(tzinfo=SHANGHAI_TZ) + return start.timestamp(), (start + timedelta(days=1)).timestamp() diff --git a/tests/fixtures.py b/tests/fixtures.py index f835d61..12cd1b0 100644 --- a/tests/fixtures.py +++ b/tests/fixtures.py @@ -161,3 +161,108 @@ def info_for( f"bilibili://music/playlist/playpage/3728825651?page_type=3&oid={aid}&otype=2" ) return item + + +# GET x/web-interface/view?aid=117530286889638（真实七项 stat） +VIEW_PAYLOAD: dict[str, Any] = { + "code": 0, + "message": "0", + "ttl": 1, + "data": { + "aid": 117530286889638, + "bvid": "BV1xvxi6LEZD", + "cid": 43184885058, + "title": "【咻咻满充电专场①】百变音色机能怪，还有什么风格是你驾驭不了的？！", + "pubdate": 1793370433, + "ctime": 1793370433, + "duration": 9504, + "owner": {"mid": 434442072, "name": "_-抱抱-_", "face": ""}, + "stat": { + "aid": 117530286889638, + "view": 819, + "danmaku": 7, + "reply": 16, + "favorite": 43, + "coin": 9, + "share": 1, + "now_rank": 0, + "his_rank": 0, + "like": 44, + "dislike": 0, + "evaluation": "", + "vt": 0, + }, + }, +} + +# 当前实测：收藏夹里已不可访问的 aid 返回 62012，而不是旧文档里的 -404 +VIEW_UNAVAILABLE: dict[str, Any] = { + "code": 62012, + "message": "62012", + "ttl": 1, + "data": None, +} + +# GET x/relation/stat?vmid=37754047 +RELATION_PAYLOAD: dict[str, Any] = { + "code": 0, + "message": "0", + "ttl": 1, + "data": { + "mid": 37754047, + "following": 95, + "whisper": 0, + "black": 0, + "follower": 2767268, + "fans_medal_toast": None, + "fans_effect": None, + }, +} + +# GET x/web-interface/nav（已登录） +NAV_PAYLOAD: dict[str, Any] = { + "code": 0, + "message": "0", + "ttl": 1, + "data": { + "isLogin": True, + "mid": 28802351, + "uname": "冷萃编译器", + "face": "https://i0.hdslb.com/bfs/face/example.jpg", + "level_info": {"current_level": 5}, + "vipStatus": 0, + "wbi_img": { + "img_url": "https://i0.hdslb.com/bfs/wbi/example.png", + "sub_url": "https://i0.hdslb.com/bfs/wbi/example-sub.png", + }, + }, +} + + +def view_for( + aid: int, + *, + view: int = 819, + like: int = 44, + coin: int = 9, + favorite: int = 43, + share: int = 1, + reply: int = 16, + danmaku: int = 7, + bvid: str | None = None, +) -> dict[str, Any]: + item = copy.deepcopy(VIEW_PAYLOAD) + data = item["data"] + data["aid"] = aid + data["bvid"] = bvid or f"BV{aid}" + data["stat"].update({ + "aid": aid, + "view": view, + "like": like, + "coin": coin, + "favorite": favorite, + "share": share, + "reply": reply, + "danmaku": danmaku, + }) + return item diff --git a/tests/test_bilibili_client.py b/tests/test_bilibili_client.py index b0ecaf5..798a49d 100644 --- a/tests/test_bilibili_client.py +++ b/tests/test_bilibili_client.py @@ -25,8 +25,13 @@ from fixtures import ( # noqa: E402 FOLDERS_PAYLOAD, IDS_PAYLOAD, INVALID_INFO, + NAV_PAYLOAD, + RELATION_PAYLOAD, VALID_INFO, + VIEW_PAYLOAD, + VIEW_UNAVAILABLE, info_for, + view_for, ) @@ -274,6 +279,88 @@ def test_timeout_and_method_forwarded(): assert call["timeout"] == 30 +def test_view_seven_metrics(): + transport = QueueTransport([http_json(VIEW_PAYLOAD)]) + client, _ = make_client(transport) + stat = client.view(117530286889638) + assert stat.aid == 117530286889638 + assert stat.bvid == "BV1xvxi6LEZD" + assert stat.metrics() == { + "view": 819, "like": 44, "favorite": 43, "coin": 9, + "share": 1, "reply": 16, "danmaku": 7, + } + assert "aid=117530286889638" in transport.calls[0]["url"] + + +def test_view_unavailable_and_malformed(): + transport = QueueTransport([ + http_json(VIEW_UNAVAILABLE), http_json(VIEW_UNAVAILABLE), + ]) + client, _ = make_client(transport) + try: + client.view(1) + except BilibiliAPIError as exc: + assert exc.code == 62012 + else: + raise AssertionError("不可访问视频应抛业务错误") + + payload = view_for(1) + payload["data"]["stat"].pop("coin") + bad = http_json(payload) + transport = QueueTransport([bad, bad]) + client, _ = make_client(transport) + try: + client.view(1) + except BilibiliProtocolError as exc: + assert "coin" in str(exc) + else: + raise AssertionError("缺少 stat 字段应报协议错误") + + +def test_relation_stat(): + transport = QueueTransport([http_json(RELATION_PAYLOAD)]) + client, _ = make_client(transport) + fan = client.relation_stat(37754047) + assert fan.mid == 37754047 + assert fan.follower == 2767268 + assert fan.following == 95 + assert "vmid=37754047" in transport.calls[0]["url"] + + +def test_check_login(): + transport = QueueTransport([http_json(NAV_PAYLOAD)]) + client, _ = make_client(transport) + state = client.check_login() + assert state.is_login is True + assert state.mid == 28802351 + assert state.uname == "冷萃编译器" + assert "SESSDATA=secret" in transport.calls[0]["headers"]["cookie"] + + transport = QueueTransport([http_json({ + "code": 0, "message": "0", "ttl": 1, + "data": {"isLogin": False, "mid": 0, "uname": ""}, + })]) + client, _ = make_client(transport) + state = client.check_login() + assert state.is_login is False and state.mid == 0 + + transport = QueueTransport([http_json({"code": -101, "message": "账号未登录", "data": None})]) + client, _ = make_client(transport) + state = client.check_login() + assert state.is_login is False + assert state.code == -101 + assert "账号未登录" in state.message + + +def test_check_login_without_cookie_makes_no_request(): + transport = QueueTransport([]) + client, _ = make_client(transport, cookie="") + state = client.check_login() + assert state.is_login is False + assert state.message == "未配置 cookie" + assert transport.calls == [] + + def main(): test_fav_ids_parse_and_anonymous_header() test_anonymous_failure_falls_back_to_full_cookie() @@ -288,6 +375,11 @@ def main(): test_fav_infos_chunk_size_validation() test_pacer_delays_after_first_request() test_timeout_and_method_forwarded() + test_view_seven_metrics() + test_view_unavailable_and_malformed() + test_relation_stat() + test_check_login() + test_check_login_without_cookie_makes_no_request() print("OK: B 站只读客户端契约与重试测试全部通过") diff --git a/tests/test_cli.py b/tests/test_cli.py index 9d04cbf..458ba42 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -12,9 +12,11 @@ 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 LoginState # noqa: E402 from reaction_board.cli import main # noqa: E402 from reaction_board.database import Database # noqa: E402 from reaction_board.repository import Repository # noqa: E402 +import reaction_board.cli as cli # noqa: E402 def write_config(tmp, *, cookie="SESSDATA=x; buvid3=y", csrf="csrf"): @@ -86,24 +88,61 @@ def test_sync_requires_auth_without_network(): assert not (Path(tmp) / "app.db").exists() -def test_run_once_without_folders(): +def test_snapshot_offline_empty_db(): + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, cookie="", csrf="") + data = json.loads(path.read_text(encoding="utf-8")) + data["folders"] = [] + data["track_fans"] = [] + path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") + code, output = run_cli(["snapshot", "--config", str(path)]) + assert code == 0 + assert '"completed": true' in output + assert '"total": 0' in output + assert len(list((Path(tmp) / "backup").glob("app-*.db"))) == 1 + + +def test_auth_check_without_cookie(): + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, cookie="", csrf="") + code, output = run_cli(["auth-check", "--config", str(path)]) + assert code == 1 + payload = json.loads(output) + assert payload["is_login"] is False + assert "未配置 cookie" in payload["message"] + + +def test_run_once_offline(): + class FakeClient: + def check_login(self): + return LoginState(True, 1, "tester", 0, "OK") + with tempfile.TemporaryDirectory() as tmp: path = write_config(tmp) data = json.loads(path.read_text(encoding="utf-8")) data["folders"] = [] + data["track_fans"] = [] path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") - code, output = run_cli(["run", "--once", "--config", str(path)]) + + original = cli._client + cli._client = lambda *args, **kwargs: FakeClient() + try: + code, output = run_cli(["run", "--once", "--config", str(path)]) + finally: + cli._client = original assert code == 0 assert '"name": "folder_sync"' in output - assert '"ok": true' in output + assert '"name": "snapshot"' in output def main_test(): test_init_and_status() test_missing_config() test_sync_requires_auth_without_network() - test_run_once_without_folders() - print("OK: CLI init/sync/run/status/错误路径测试全部通过") + test_snapshot_offline_empty_db() + test_auth_check_without_cookie() + test_run_once_offline() + print("OK: CLI init/sync/snapshot/auth-check/run/status/错误路径测试全部通过") if __name__ == "__main__": diff --git a/tests/test_maintenance.py b/tests/test_maintenance.py new file mode 100644 index 0000000..630c4fc --- /dev/null +++ b/tests/test_maintenance.py @@ -0,0 +1,78 @@ +# -*- coding: utf-8 -*- +"""每日备份与保留期测试: uv run python tests/test_maintenance.py""" + +import os +import sqlite3 +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.database import Database # noqa: E402 +from reaction_board.services.maintenance import ( # noqa: E402 + backup_daily, + daily_backup_path, + prune_daily_backups, +) + + +def test_daily_backup_and_restore(): + with tempfile.TemporaryDirectory() as tmp: + db = Database(Path(tmp) / "app.db") + db.initialize() + try: + db.write(lambda c: c.execute("INSERT INTO kv(k, v) VALUES ('x', '42')")) + target = backup_daily(db, Path(tmp) / "backup", "2026-10-31") + assert target == daily_backup_path(Path(tmp) / "backup", "2026-10-31") + assert target.name == "app-20261031.db" + assert target.exists() + + check = sqlite3.connect(target) + try: + assert check.execute("SELECT v FROM kv WHERE k='x'").fetchone()[0] == "42" + finally: + check.close() + + # 覆盖同日备份也安全 + db.write(lambda c: c.execute("UPDATE kv SET v='43' WHERE k='x'")) + backup_daily(db, Path(tmp) / "backup", "2026-10-31") + check = sqlite3.connect(target) + try: + assert check.execute("SELECT v FROM kv WHERE k='x'").fetchone()[0] == "43" + finally: + check.close() + finally: + db.close() + + +def test_prune_daily_backups_only(): + with tempfile.TemporaryDirectory() as tmp: + backup_dir = Path(tmp) + old_daily = backup_dir / "app-20260101.db" + fresh_daily = backup_dir / "app-20261031.db" + old_migration = backup_dir / "app-pre-v1-20260101.db" + other = backup_dir / "keep.txt" + for path in (old_daily, fresh_daily, old_migration, other): + path.write_bytes(b"x") + now = 1_800_000_000 + old_time = now - 20 * 86400 + for path in (old_daily, old_migration): + os.utime(path, (old_time, old_time)) + os.utime(fresh_daily, (now, now)) + removed = prune_daily_backups(backup_dir, keep_days=14, now=now) + assert removed == [old_daily] + assert fresh_daily.exists() + assert old_migration.exists() # 迁移备份不属于日备份清理范围 + assert other.exists() + + +def main(): + test_daily_backup_and_restore() + test_prune_daily_backups_only() + print("OK: 每日备份与保留期测试全部通过") + + +if __name__ == "__main__": + main() diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py index f6669d3..6376a64 100644 --- a/tests/test_scheduler.py +++ b/tests/test_scheduler.py @@ -110,8 +110,27 @@ def test_max_iterations_and_validation(): raise AssertionError(f"应当失败: {tasks}, {kwargs}") +def test_next_run_fn_controls_reschedule(): + clock = [1_000.0] + calls = [] + task = ScheduledTask( + "daily", 86400, lambda: calls.append(clock[0]), + next_run_fn=lambda started: started + 123, + ) + scheduler = Scheduler([task], clock=lambda: clock[0]) + scheduler.run_due() + assert calls == [1_000.0] + assert task.next_run == 1_123.0 + clock[0] = 1_122.0 + assert scheduler.run_due() == [] + clock[0] = 1_123.0 + scheduler.run_due() + assert calls == [1_000.0, 1_123.0] + + def main(): test_run_when_due_and_reschedule() + test_next_run_fn_controls_reschedule() test_task_error_does_not_block_other_tasks() test_run_forever_with_stop_event() test_max_iterations_and_validation() diff --git a/tests/test_snapshot.py b/tests/test_snapshot.py new file mode 100644 index 0000000..def3630 --- /dev/null +++ b/tests/test_snapshot.py @@ -0,0 +1,331 @@ +# -*- coding: utf-8 -*- +"""每日快照与粉丝采集测试（transport 边界 Mock，不联网）: +uv run python tests/test_snapshot.py +""" + +import io +import json +import os +import sys +import tempfile +import urllib.parse +from pathlib import Path + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +sys.stdout.reconfigure(encoding="utf-8") +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from fixtures import ( # noqa: E402 + NAV_PAYLOAD, + RELATION_PAYLOAD, + VIEW_UNAVAILABLE, + view_for, +) +from reaction_board.bilibili.client import ( # noqa: E402 + BilibiliClient, + HTTPResponse, + Pacer, +) +from reaction_board.bilibili.models import FavResource # noqa: E402 +from reaction_board.config import ConfigStore, FolderConfig # noqa: E402 +from reaction_board.database import Database # noqa: E402 +from reaction_board.logging_json import JsonLogger # noqa: E402 +from reaction_board.repository import Repository # noqa: E402 +from reaction_board.services.snapshot import SnapshotService # noqa: E402 + +DATE = "2026-10-31" +FOLDER = FolderConfig(year=2026, media_id=3700023351, title="2026") + + +def http_json(payload): + return HTTPResponse( + 200, {"content-type": "application/json"}, + json.dumps(payload, ensure_ascii=False).encode("utf-8"), + ) + + +class SnapshotTransport: + def __init__(self, views, *, nav=None, fan=None): + self.views = views + self.nav = NAV_PAYLOAD if nav is None else nav + self.fan = RELATION_PAYLOAD if fan is None else fan + self.calls = [] + + def request(self, method, url, headers, data=None, timeout=30): + self.calls.append(url) + if "/x/web-interface/nav" in url: + return http_json(self.nav) + if "/x/web-interface/view" in url: + aid = int(urllib.parse.parse_qs( + urllib.parse.urlparse(url).query + )["aid"][0]) + payload = self.views[aid] + if isinstance(payload, HTTPResponse): + return payload + return http_json(payload) + if "/x/relation/stat" in url: + return http_json(self.fan) + raise AssertionError(f"未预期的 URL: {url}") + + def count(self, part): + return sum(1 for url in self.calls if part in url) + + +def make_service(tmp, transport, *, track_fans=(37754047,), logger=None, + retries=3, cookie="SESSDATA=secret; buvid3=device"): + path = Path(tmp) / "config.json" + path.write_text(json.dumps({ + "mid": 28802351, + "csrf": "csrf", + "cookie": cookie, + "storage": { + "database_path": str(Path(tmp) / "app.db"), + "backup_dir": str(Path(tmp) / "backup"), + "backup_keep_days": 14, + }, + "folders": [], + "track_fans": list(track_fans), + }, ensure_ascii=False), encoding="utf-8") + store = ConfigStore(path, env={}, require_auth=False) + 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) + client = BilibiliClient( + cookie=cookie, + csrf="csrf", + mid=28802351, + transport=transport, + pacer=Pacer(0, 0), + retries=1, + ) + service = SnapshotService( + store, repo, client, + clock=lambda: 1_800_000_000, + sleeper=lambda seconds: None, + logger=logger, + retries=retries, + backoff=0, + ) + return db, repo, service + + +def add_video(repo, aid, *, status="active"): + repo.upsert_from_fav( + FavResource( + aid=aid, bvid=f"BV{aid}", title=f"视频{aid}", up_mid=100, + up_name="UP", pubdate=1_700_000_000, duration=600, + cover=None, attr=0, fav_time=None, + ), + FOLDER.media_id, + was_added=True, + ) + if status != "active": + repo.set_video_status(aid, status) + + +def test_snapshot_success_resume_and_force(): + with tempfile.TemporaryDirectory() as tmp: + transport = SnapshotTransport({ + 1: view_for(1, view=100, like=10, coin=2, favorite=3, + share=1, reply=4, danmaku=5), + 2: view_for(2, view=200, like=20, coin=4, favorite=6, + share=2, reply=8, danmaku=10), + }) + db, repo, service = make_service(tmp, transport) + try: + add_video(repo, 1) + add_video(repo, 2) + result = service.run(date=DATE) + assert result.completed is True + assert result.total == 2 + assert result.success == 2 + assert result.failed == 0 + assert result.fans_ok == 1 + assert result.fans_failed == 0 + assert result.auth_ok is True + assert Path(result.backup_path or "").exists() + + row = repo.get_snapshot(DATE, 1) + assert (row["view"], row["like"], row["favorite"], row["coin"], + row["share"], row["reply"], row["danmaku"]) == (100, 10, 3, 2, 1, 4, 5) + assert row["ok"] == 1 and row["src"] == "view" + fan = repo.get_up_stat(DATE, 37754047) + assert fan["follower"] == 2767268 and fan["ok"] == 1 + assert repo.kv_get(service.completed_key(DATE)) == "1" + assert repo.kv_get("snapshot:last_completed") == DATE + assert repo.snapshot_counts(DATE) == {"total": 2, "ok": 2, "failed": 0} + + # 已完成 -> 不再请求 + view_before = transport.count("/web-interface/view") + again = service.run_if_needed(date=DATE) + assert again.already_completed is True + assert transport.count("/web-interface/view") == view_before + + # force 重新采集 + forced = service.run(date=DATE, force=True) + assert forced.success == 2 + assert transport.count("/web-interface/view") == view_before + 2 + finally: + db.close() + + +def test_resume_skips_ok_and_retries_failed(): + with tempfile.TemporaryDirectory() as tmp: + transport = SnapshotTransport({ + 1: view_for(1, view=111), + 2: view_for(2, view=222), + }) + db, repo, service = make_service(tmp, transport) + try: + add_video(repo, 1) + add_video(repo, 2) + repo.upsert_snapshot( + DATE, 1, metrics={"view": 1, "like": 1, "favorite": 1, + "coin": 1, "share": 1, "reply": 1, "danmaku": 1}, + ok=True, err=None, captured_at=1, + ) + repo.upsert_snapshot( + DATE, 2, metrics=None, ok=False, err="previous error", + captured_at=1, + ) + result = service.run(date=DATE) + assert result.skipped == 1 + assert result.success == 1 + assert result.completed is True + assert transport.count("/web-interface/view") == 1 + assert repo.get_snapshot(DATE, 1)["view"] == 1 # 未被覆盖 + assert repo.get_snapshot(DATE, 2)["view"] == 222 + finally: + db.close() + + +def test_unavailable_marks_removed(): + with tempfile.TemporaryDirectory() as tmp: + transport = SnapshotTransport({ + 1: VIEW_UNAVAILABLE, + 2: HTTPResponse(404, {}, b"not found"), + }) + db, repo, service = make_service(tmp, transport, track_fans=()) + try: + add_video(repo, 1) + add_video(repo, 2) + result = service.run(date=DATE) + assert result.removed == 2 + assert result.failed == 0 + assert result.completed is True + assert repo.get_video(1)["status"] == "removed" + assert repo.get_video(2)["status"] == "removed" + assert repo.get_snapshot(DATE, 1) is None + assert repo.get_snapshot(DATE, 2) is None + finally: + db.close() + + +def test_other_error_writes_failed_placeholder(): + with tempfile.TemporaryDirectory() as tmp: + transport = SnapshotTransport({ + 1: {"code": -500, "message": "服务异常", "data": None}, + }) + db, repo, service = make_service(tmp, transport, track_fans=(), retries=3) + try: + add_video(repo, 1) + result = service.run(date=DATE) + assert result.failed == 1 + assert result.success == 0 + assert result.completed is False + row = repo.get_snapshot(DATE, 1) + assert row["ok"] == 0 + assert row["view"] is None and row["like"] is None + assert "code=-500" in row["err"] + assert repo.kv_get(service.completed_key(DATE)) == "0" + # 每次业务失败会走匿名+完整 Cookie 两条请求，服务层重试 3 次 + assert transport.count("/web-interface/view") == 6 + finally: + db.close() + + +def test_fan_failure_placeholder(): + with tempfile.TemporaryDirectory() as tmp: + transport = SnapshotTransport({}, fan={"code": -500, "message": "x", "data": None}) + db, repo, service = make_service(tmp, transport, retries=2) + try: + result = service.run(date=DATE) + assert result.total == 0 + assert result.fans_failed == 1 + assert result.completed is False + row = repo.get_up_stat(DATE, 37754047) + assert row["ok"] == 0 + assert row["follower"] is None and row["following"] is None + finally: + db.close() + + +def test_auth_invalid_is_warned_but_snapshot_continues(): + with tempfile.TemporaryDirectory() as tmp: + stream = io.StringIO() + logger = JsonLogger(stream) + transport = SnapshotTransport( + {}, nav={"code": 0, "message": "0", "ttl": 1, + "data": {"isLogin": False, "mid": 0, "uname": ""}}, + fan=RELATION_PAYLOAD, + ) + db, repo, service = make_service(tmp, transport, logger=logger) + try: + result = service.run(date=DATE) + assert result.auth_ok is False + assert result.completed is True + assert repo.kv_get("auth:ok") == "0" + events = [json.loads(line) for line in stream.getvalue().splitlines()] + assert any(e["event"] == "auth.invalid" for e in events) + finally: + db.close() + + +def test_no_cookie_skips_nav_request(): + with tempfile.TemporaryDirectory() as tmp: + transport = SnapshotTransport({}, fan=RELATION_PAYLOAD) + db, repo, service = make_service(tmp, transport, cookie="") + try: + result = service.run(date=DATE) + assert result.auth_ok is False + assert result.auth_message == "未配置 cookie" + assert transport.count("/web-interface/nav") == 0 + finally: + db.close() + + +def test_force_failure_clears_last_completed(): + with tempfile.TemporaryDirectory() as tmp: + transport = SnapshotTransport({1: view_for(1, view=100)}) + db, repo, service = make_service(tmp, transport, track_fans=(), retries=1) + try: + add_video(repo, 1) + first = service.run(date=DATE) + assert first.completed is True + assert repo.kv_get("snapshot:last_completed") == DATE + + transport.views[1] = {"code": -500, "message": "x", "data": None} + second = service.run(date=DATE, force=True) + assert second.failed == 1 + assert second.completed is False + assert repo.kv_get("snapshot:last_completed") == "" + assert repo.get_snapshot(DATE, 1)["ok"] == 0 + finally: + db.close() + + +def main(): + test_snapshot_success_resume_and_force() + test_resume_skips_ok_and_retries_failed() + test_unavailable_marks_removed() + test_other_error_writes_failed_placeholder() + test_fan_failure_placeholder() + test_auth_invalid_is_warned_but_snapshot_continues() + test_no_cookie_skips_nav_request() + test_force_failure_clears_last_completed() + print("OK: 每日快照、断点续跑、失败占位与粉丝采集测试全部通过") + + +if __name__ == "__main__": + main() diff --git a/tests/test_timeutil.py b/tests/test_timeutil.py new file mode 100644 index 0000000..bfe4060 --- /dev/null +++ b/tests/test_timeutil.py @@ -0,0 +1,84 @@ +# -*- coding: utf-8 -*- +"""上海时区与每日时间计算测试: uv run python tests/test_timeutil.py""" + +import os +import sys +from datetime import datetime + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +sys.stdout.reconfigure(encoding="utf-8") + +from reaction_board.timeutil import ( # noqa: E402 + SHANGHAI_TZ, + day_bounds, + is_date, + next_daily_run, + parse_hhmm, + shanghai_date, + shanghai_datetime, +) + + +def ts(text): + return datetime.strptime(text, "%Y-%m-%d %H:%M:%S").replace(tzinfo=SHANGHAI_TZ).timestamp() + + +def test_shanghai_date_boundary(): + # UTC 2026-10-31 15:59:59 = 上海 2026-10-31 23:59:59 + assert shanghai_date(ts("2026-10-31 23:59:59")) == "2026-10-31" + # UTC 2026-10-31 16:00:00 = 上海 2026-11-01 00:00:00 + assert shanghai_date(ts("2026-11-01 00:00:00")) == "2026-11-01" + assert shanghai_datetime().tzinfo is not None + + +def test_parse_hhmm(): + assert parse_hhmm("00:01") == (0, 1) + assert parse_hhmm("23:59") == (23, 59) + for bad in ("24:00", "1:01", "00:60", "", "abc"): + try: + parse_hhmm(bad) + except ValueError: + pass + else: + raise AssertionError(f"{bad!r} 应失败") + + +def test_next_daily_run(): + # 当天 00:01 之前 -> 当天 00:01 + now = ts("2026-11-01 00:00:10") + nxt = next_daily_run(now, "00:01") + assert nxt == ts("2026-11-01 00:01:00") + # 正好 00:01:00 -> 下一天（严格大于） + assert next_daily_run(ts("2026-11-01 00:01:00"), "00:01") == ts("2026-11-02 00:01:00") + # 当天 00:01 之后 -> 次日 + assert next_daily_run(ts("2026-11-01 23:59:59"), "00:01") == ts("2026-11-02 00:01:00") + # 跨月 + assert next_daily_run(ts("2026-10-31 12:00:00"), "00:01") == ts("2026-11-01 00:01:00") + + +def test_day_bounds_and_date_validation(): + start, end = day_bounds("2026-11-01") + assert start == ts("2026-11-01 00:00:00") + assert end == ts("2026-11-02 00:00:00") + assert end - start == 86400 + assert is_date("2026-11-01") + for bad in ("2026-13-01", "2026-02-30", "2026/11/01", "20261101", ""): + assert not is_date(bad), bad + try: + day_bounds("bad") + except ValueError: + pass + else: + raise AssertionError("非法日期应失败") + + +def main(): + test_shanghai_date_boundary() + test_parse_hhmm() + test_next_daily_run() + test_day_bounds_and_date_validation() + print("OK: 上海时区与每日时间计算测试全部通过") + + +if __name__ == "__main__": + main()