From bb18d232f60b3258526f20ea7f4e16753d78695d Mon Sep 17 00:00:00 2001 From: bot-ci Date: Tue, 3 Nov 2026 00:22:11 +0800 Subject: [PATCH] feat(scheduler): add snapshot-seed CLI and 10m seed deferral Add `python app.py snapshot-seed [--date]` (offline clone, Debian / docker compose exec friendly). When seed fires while full-site snapshot is running, or last_completed is missing (no_source), record skip as non-success and retry in 10 minutes until a real seed succeeds; fans and quiet-window rules unchanged. --- DEVELOPMENT.md | 4 +- project.md | 2 +- reaction_board/cli.py | 160 +++++++++++++++++++++++++++- reaction_board/services/snapshot.py | 8 +- reaction_board/web/queries.py | 8 +- tests/test_cli.py | 154 +++++++++++++++++++++++++- tests/test_web_queries.py | 46 ++++++++ 7 files changed, 374 insertions(+), 8 deletions(-) diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index ddc07dd..6531d92 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -456,6 +456,8 @@ P1 实现说明： - 时区使用固定 UTC+8（中国无夏令时），避免 Windows 缺少 `tzdata`； - 午夜 `seed_at`（默认 `00:00`）把 `snapshot:last_completed` 的 ok=1 行克隆为当日壳，写入 `snapshot:seeded:=1` / `fans:seeded:=1`，**不**推进完成日； +- 也可手动：`python app.py snapshot-seed [--date YYYY-MM-DD]`（Debian / `docker compose exec re-xxm python app.py snapshot-seed`）； +- 若全站 `snapshot` 正在跑，或 `last_completed` 暂不可用（`no_source`），seed **不记成功**，约 10 分钟后重试直到真正克隆成功； - 晚间 `snapshot_at`（默认 `23:15`）对 seeded 日 force 重采；成功后写 `snapshot:completed:=1`、推进 `last_completed` 并清除 seeded； - 当日全部视频与粉丝采集成功后在 `kv` 写入 `snapshot:completed:=1`；启动时未完成会补跑（纯 seed 日不误标完成）； - 完成标记不是唯一门槛：每次检查 `pending_snapshot_aids(date)`，新入库或从 `dropped` 恢复的视频会立即补采当天快照； @@ -471,7 +473,7 @@ P1 实现说明： - 全站 `snapshot` **不因**其它任务 running 而 skip；到期/强制必须执行； - 上海时区 **23:00–01:00** 为全站快照窗：`discovery` / `snapshot_recent` / `candidate_cleanup` 不执行，延后到 01:00 后；`fans` 短耗时，不避开任何任务； - 任务 skip（含静默窗延后）写入 `ok=false` + `skipped=true`，**不得**记成成功或推进「当日已完成」； -- skip / fail 后短间隔补跑（约 5 分钟）；日更成功才推到次日 `snapshot_at` / `seed_at`。 +- skip / fail 后短间隔补跑（约 5 分钟；`snapshot_seed` 避让/无源为约 10 分钟）；日更成功才推到次日 `snapshot_at` / `seed_at`。 --- diff --git a/project.md b/project.md index c7ccfce..1a7d7cc 100644 --- a/project.md +++ b/project.md @@ -800,7 +800,7 @@ secrets.example.json 凭据样例（auth 回写目标） **进程结构** ``` -scheduler 线程：discovery(每 discover_minutes，避开 23:00–01:00) / snapshot_seed(每日 00:00) / snapshot(每日 23:15，不因其它任务 skip) / fans(每日 23:15，不避开) / snapshot_recent(小时，避开 23:00–01:00) / candidate_cleanup(每日，避开 23:00–01:00) +scheduler 线程：discovery(每 discover_minutes，避开 23:00–01:00) / snapshot_seed(每日 00:00；snapshot 占用或 no_source 则 10 分钟重试，CLI: snapshot-seed) / snapshot(每日 23:15，不因其它任务 skip) / fans(每日 23:15，不避开) / snapshot_recent(小时，避开 23:00–01:00) / candidate_cleanup(每日，避开 23:00–01:00) web 线程：ThreadingHTTPServer :8080 telegram：getUpdates long polling（回调 + 私聊文本入库） storage：SQLite WAL，单写线程 diff --git a/reaction_board/cli.py b/reaction_board/cli.py index 0f778c9..905f59b 100644 --- a/reaction_board/cli.py +++ b/reaction_board/cli.py @@ -4,6 +4,7 @@ python app.py init # 建库 / 迁移 / 写入黑名单种子（离线） python app.py ingest # 手动入库 B 站视频链接 / BV 号 / av 号（只写本地库） python app.py snapshot # 采集当日七项快照与粉丝数 + python app.py snapshot-seed # 用上一完成日克隆当日壳（离线，可 --date） python app.py discover # 执行一次搜索发现并写入候选池 python app.py candidates # 查看候选池（离线） python app.py auth-check # 检查 Cookie 登录态 @@ -42,6 +43,7 @@ from .services.up_profile import UpProfileService from .timeutil import ( in_snapshot_quiet_window, interval_next_run, + is_date, next_after_snapshot_quiet_window, next_daily_run, parse_hhmm, @@ -78,6 +80,17 @@ def _build_parser() -> argparse.ArgumentParser: snapshot.add_argument( "--force", action="store_true", help="忽略当日完成标记，重新采集" ) + snapshot_seed = sub.add_parser( + "snapshot-seed", + parents=[common], + help="用上一完成日克隆当日快照壳（离线，不调 B 站）", + ) + snapshot_seed.add_argument( + "--date", + default=None, + metavar="YYYY-MM-DD", + help="目标统计日（默认上海时区今天）", + ) discover = sub.add_parser( "discover", parents=[common], help="执行一次搜索发现并写入候选池" ) @@ -170,6 +183,14 @@ def _clear_task_running(repo: Repository, name: str) -> None: pass +def _is_task_running(repo: Repository, name: str) -> bool: + """供 seed 等任务检测全站快照是否在跑（避让，而非互斥跳过成功）。""" + try: + return bool(repo.kv_get(_task_running_key(name))) + except Exception: # noqa: BLE001 + return False + + def _record_task( repo: Repository, name: str, @@ -206,7 +227,11 @@ def _record_task( # 全站快照 / 粉丝失败后的补跑间隔（秒） _TASK_RETRY_SECONDS = 300.0 +# snapshot_seed：全站快照占用 / 尚无 last_completed 时的补跑间隔（秒） +_SEED_RETRY_SECONDS = 600.0 _QUIET_SKIP_REASON = "snapshot_quiet_window" +_SEED_SKIP_SNAPSHOT_RUNNING = "snapshot_running" +_SEED_SKIP_NO_SOURCE = "no_source" def _quiet_window_defer( @@ -240,6 +265,38 @@ def _retry_soon(now: float | None = None) -> float: return (time.time() if now is None else now) + _TASK_RETRY_SECONDS +def _seed_retry_soon(now: float | None = None) -> float: + return (time.time() if now is None else now) + _SEED_RETRY_SECONDS + + +def _defer_snapshot_seed( + repo: Repository, + logger: JsonLogger, + *, + started_at: int, + reason: str, + finished_at: int | None = None, +) -> TaskControl: + """seed 因 snapshot 占用 / no_source 延后：记 skip（非成功），10 分钟后再试。""" + finished = int(finished_at if finished_at is not None else time.time()) + logger.info("snapshot_seed.skip", reason=reason) + _record_task( + repo, + "snapshot_seed", + ok=False, + skipped=True, + started_at=started_at, + finished_at=finished, + payload={"skipped": True, "reason": reason}, + error=reason, + ) + return TaskControl( + ok=False, + next_run=_seed_retry_soon(float(finished)), + error=reason, + ) + + def _client(credentials, logger: JsonLogger, *, pacer=None, retries: int = 3) -> BilibiliClient: return BilibiliClient( cookie=credentials.cookie, @@ -397,6 +454,82 @@ def cmd_snapshot(args, logger: JsonLogger) -> int: db.close() +def cmd_snapshot_seed(args, logger: JsonLogger) -> int: + """离线克隆：``python app.py snapshot-seed [--date YYYY-MM-DD]``。 + + 与调度器同一套规则：全站 ``snapshot`` 运行中或 ``no_source`` 时不记成功， + 返回码非 0，便于 ``docker compose exec re-xxm python app.py snapshot-seed`` 排查。 + """ + store = ConfigStore(args.config, require_auth=False) + cfg = store.get() + db = _open_db(cfg, logger) + try: + repo = Repository(db) + target = args.date + if target is not None and not is_date(target): + raise ConfigError(f"日期格式必须是 YYYY-MM-DD: {target!r}") + started = int(time.time()) + if _is_task_running(repo, "snapshot"): + payload = { + "ok": False, + "skipped": True, + "reason": _SEED_SKIP_SNAPSHOT_RUNNING, + "stat_date": target or shanghai_date(), + "database": str(cfg.storage.database_path), + } + _record_task( + repo, + "snapshot_seed", + ok=False, + skipped=True, + started_at=started, + finished_at=int(time.time()), + payload=payload, + error=_SEED_SKIP_SNAPSHOT_RUNNING, + ) + _print(payload) + return 1 + # seed 只读本地库，不调 B 站；占位 client 满足构造约束。 + service = SnapshotService(store, repo, object(), logger=logger) + _mark_task_running(repo, "snapshot_seed", started_at=started) + try: + summary = service.run_seed_from_last_completed(date=target) + except Exception as exc: # noqa: BLE001 + _record_task( + repo, "snapshot_seed", ok=False, started_at=started, + finished_at=int(time.time()), error=str(exc), + ) + raise + payload = summary.as_dict() + payload["database"] = str(cfg.storage.database_path) + finished = int(summary.finished_at or time.time()) + if summary.reason == _SEED_SKIP_NO_SOURCE: + _record_task( + repo, + "snapshot_seed", + ok=False, + skipped=True, + started_at=int(summary.started_at or started), + finished_at=finished, + payload={**payload, "skipped": True, "reason": _SEED_SKIP_NO_SOURCE}, + error=_SEED_SKIP_NO_SOURCE, + ) + _print(payload) + return 1 + _record_task( + repo, + "snapshot_seed", + ok=True, + started_at=int(summary.started_at or started), + finished_at=finished, + payload=payload, + ) + _print(payload) + return 0 + finally: + db.close() + + def _telegram(cfg, logger: JsonLogger): if not cfg.telegram.enabled: return None @@ -733,6 +866,14 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int: def run_snapshot_seed(): started = int(time.time()) + # 全站快照占用时不执行、不记成功；10 分钟后重试直到真正 seed 成功。 + if _is_task_running(repo, "snapshot"): + return _defer_snapshot_seed( + repo, + logger, + started_at=started, + reason=_SEED_SKIP_SNAPSHOT_RUNNING, + ) _mark_task_running(repo, "snapshot_seed", started_at=started) try: summary = snapshot_service.run_seed_from_last_completed() @@ -743,17 +884,28 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int: finished_at=finished, error=str(exc), ) return TaskControl( - ok=False, next_run=_retry_soon(float(finished)), error=str(exc), + ok=False, + next_run=_seed_retry_soon(float(finished)), + error=str(exc), ) payload = summary.as_dict() finished = int(summary.finished_at or time.time()) + last["snapshot_seed"] = summary + logger.info("snapshot.seed_summary", **payload) + if summary.reason == _SEED_SKIP_NO_SOURCE: + return _defer_snapshot_seed( + repo, + logger, + started_at=int(summary.started_at or started), + reason=_SEED_SKIP_NO_SOURCE, + finished_at=finished, + ) + # already_completed / already_seeded / seeded → 成功，推到次日 seed_at _record_task( repo, "snapshot_seed", ok=True, started_at=summary.started_at, finished_at=finished, payload=payload, ) - last["snapshot_seed"] = summary - logger.info("snapshot.seed_summary", **payload) return TaskControl( ok=True, next_run=next_daily_run(float(finished), seed_at), ) @@ -942,6 +1094,8 @@ def main(argv: Sequence[str] | None = None) -> int: return cmd_init(args, logger) if args.command == "snapshot": return cmd_snapshot(args, logger) + if args.command == "snapshot-seed": + return cmd_snapshot_seed(args, logger) if args.command == "ingest": return cmd_ingest(args, logger) if args.command == "backfill-followers": diff --git a/reaction_board/services/snapshot.py b/reaction_board/services/snapshot.py index da8bc8e..97c578c 100644 --- a/reaction_board/services/snapshot.py +++ b/reaction_board/services/snapshot.py @@ -51,6 +51,7 @@ class SnapshotSummary: completed: bool = False already_completed: bool = False finished_at: int = 0 + reason: str | None = None errors: list[str] = field(default_factory=list) def as_dict(self) -> dict[str, Any]: @@ -73,6 +74,7 @@ class SnapshotSummary: "backup_error": self.backup_error, "completed": self.completed, "already_completed": self.already_completed, + "reason": self.reason, "errors": self.errors, } @@ -543,7 +545,8 @@ class SnapshotService: stat_date=stat_date, reason="no_source", source=source, ) summary.finished_at = self._now() - summary.completed = True + summary.completed = False + summary.reason = "no_source" return summary if self.repo.kv_get(self.completed_key(stat_date)) == "1": self._log( @@ -553,6 +556,7 @@ class SnapshotService: summary.finished_at = self._now() summary.completed = True summary.already_completed = True + summary.reason = "already_completed" return summary if self.repo.kv_get(self.seeded_key(stat_date)) == "1": self._log( @@ -562,6 +566,7 @@ class SnapshotService: summary.finished_at = self._now() summary.completed = True summary.already_completed = True + summary.reason = "already_seeded" return summary copied = self.repo.copy_ok_snapshots( @@ -583,6 +588,7 @@ class SnapshotService: self.repo.bump_cache_version("fans") summary.finished_at = self._now() summary.completed = True # seed 任务本身成功；不等于快照落盘完成 + summary.reason = "seeded" self._log( "info", "snapshot.seed_done", stat_date=stat_date, source=source, diff --git a/reaction_board/web/queries.py b/reaction_board/web/queries.py index baeea07..8ec9b38 100644 --- a/reaction_board/web/queries.py +++ b/reaction_board/web/queries.py @@ -1982,6 +1982,7 @@ class DashboardQueries: "snapshot_recent": int(interval_next_run(now, float(recent_interval))), } retry_seconds = 300 + seed_retry_seconds = 600 for name, key, at in ( ("fans", "fans", cfg.schedule.snapshot_at), ("snapshot", "snapshot", cfg.schedule.snapshot_at), @@ -2015,7 +2016,12 @@ class DashboardQueries: else: if skipped or failed: # 日更任务失败/skip：短间隔补跑，绝不按成功推到次日。 - next_runs[name] = int(finished_at + retry_seconds) + # snapshot_seed：全站快照占用 / no_source 时 10 分钟补跑。 + gap = ( + seed_retry_seconds if name == "snapshot_seed" + else retry_seconds + ) + next_runs[name] = int(finished_at + gap) else: next_runs[name] = int(next_daily_run(finished_at, at)) errors = [ diff --git a/tests/test_cli.py b/tests/test_cli.py index b5098cb..2fe7070 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -332,6 +332,153 @@ def test_backfill_followers_dry_run(): assert '"total": 0' in output + +def test_help_lists_snapshot_seed(): + from reaction_board.cli import _build_parser + + parser = _build_parser() + help_text = parser.format_help() + assert "snapshot-seed" in help_text + args = parser.parse_args(["snapshot-seed", "--date", "2026-10-31"]) + assert args.command == "snapshot-seed" + assert args.date == "2026-10-31" + + +def test_snapshot_seed_cli_clones_offline(): + from reaction_board.bilibili.models import FavResource + + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, cookie="", csrf="") + code, _ = run_cli(["init", "--config", str(path)]) + assert code == 0 + db = Database(Path(tmp) / "app.db") + db.initialize() + try: + repo = Repository(db) + media_id = 11 + repo.upsert_from_fav( + FavResource( + aid=1, bvid="BV1xx411c7mD", title="t", up_mid=9, + up_name="u", pubdate=1, duration=10, cover=None, + attr=0, fav_time=None, + ), + media_id, + was_added=True, + ) + yday, today = "2026-10-30", "2026-10-31" + repo.upsert_snapshot( + yday, 1, + metrics={"view": 10, "like": 1, "coin": 0, "favorite": 0, + "share": 0, "reply": 0, "danmaku": 0}, + ok=True, err=None, captured_at=1_800_000_000, + ) + repo.kv_set("snapshot:last_completed", yday) + repo.kv_set("snapshot:completed:" + yday, "1") + finally: + db.close() + + code, output = run_cli([ + "snapshot-seed", "--config", str(path), "--date", today, + ]) + assert code == 0, output + assert '"reason": "seeded"' in output + db = Database(Path(tmp) / "app.db") + db.initialize() + try: + repo = Repository(db) + assert repo.kv_get("snapshot:seeded:" + today) == "1" + row = repo.get_snapshot(today, 1) + assert row is not None and int(row["view"]) == 10 + assert row["src"] == "seed" + finally: + db.close() + + +def test_snapshot_seed_defers_when_snapshot_running(): + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, cookie="", csrf="") + code, _ = run_cli(["init", "--config", str(path)]) + assert code == 0 + db = Database(Path(tmp) / "app.db") + db.initialize() + try: + repo = Repository(db) + repo.kv_set( + "task:running:snapshot", + json.dumps({"name": "snapshot", "started_at": 1}), + ) + repo.kv_set("snapshot:last_completed", "2026-10-30") + finally: + db.close() + + code, output = run_cli([ + "snapshot-seed", "--config", str(path), "--date", "2026-10-31", + ]) + assert code == 1, output + assert "snapshot_running" in output + db = Database(Path(tmp) / "app.db") + db.initialize() + try: + repo = Repository(db) + last = json.loads(repo.kv_get("task:last:snapshot_seed")) + assert last["ok"] is False + assert last["skipped"] is True + assert last["error"] == "snapshot_running" + assert repo.kv_get("task:running:snapshot_seed") is None + assert repo.kv_get("task:running:snapshot") is not None + finally: + db.close() + + +def test_snapshot_seed_no_source_is_non_success(): + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, cookie="", csrf="") + code, _ = run_cli(["init", "--config", str(path)]) + assert code == 0 + code, output = run_cli([ + "snapshot-seed", "--config", str(path), "--date", "2026-10-31", + ]) + assert code == 1, output + assert "no_source" in output + db = Database(Path(tmp) / "app.db") + db.initialize() + try: + repo = Repository(db) + last = json.loads(repo.kv_get("task:last:snapshot_seed")) + assert last["ok"] is False and last["skipped"] is True + assert last["error"] == "no_source" + finally: + db.close() + + +def test_defer_snapshot_seed_schedules_ten_minutes(): + """调度辅助：skip 记非成功，next_run = finished + 600。""" + from reaction_board.logging_json import JsonLogger + + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, cookie="", csrf="") + code, _ = run_cli(["init", "--config", str(path)]) + assert code == 0 + db = Database(Path(tmp) / "app.db") + db.initialize() + try: + repo = Repository(db) + logger = JsonLogger() + started = 1_800_000_000 + finished = started + 5 + ctrl = cli._defer_snapshot_seed( + repo, logger, started_at=started, reason="snapshot_running", + finished_at=finished, + ) + assert ctrl.ok is False + assert ctrl.error == "snapshot_running" + assert abs(ctrl.next_run - (finished + cli._SEED_RETRY_SECONDS)) < 0.01 + last = json.loads(repo.kv_get("task:last:snapshot_seed")) + assert last["skipped"] is True and last["ok"] is False + finally: + db.close() + + def main_test(): test_init_and_status() test_missing_config() @@ -345,7 +492,12 @@ def main_test(): test_snapshot_backup_failure_returns_error_and_retries() test_candidates_and_discover_commands() test_backfill_followers_dry_run() - print("OK: CLI init/sync/snapshot/auth-check/run/status/backfill/错误路径测试全部通过") + test_help_lists_snapshot_seed() + test_snapshot_seed_cli_clones_offline() + test_snapshot_seed_defers_when_snapshot_running() + test_snapshot_seed_no_source_is_non_success() + test_defer_snapshot_seed_schedules_ten_minutes() + print("OK: CLI init/sync/snapshot/snapshot-seed/auth-check/run/status/backfill/错误路径测试全部通过") if __name__ == "__main__": diff --git a/tests/test_web_queries.py b/tests/test_web_queries.py index f0b6b05..48d9538 100644 --- a/tests/test_web_queries.py +++ b/tests/test_web_queries.py @@ -711,6 +711,51 @@ def test_snapshot_failed_next_run_retries_soon_not_next_day(): db.close() +def test_snapshot_seed_skip_next_run_retries_ten_minutes(): + import json + from datetime import datetime + from reaction_board.timeutil import SHANGHAI_TZ, next_daily_run + + with tempfile.TemporaryDirectory() as tmp: + db, q = setup(tmp) + try: + repo = Repository(db) + finished = int(datetime(2026, 11, 1, 0, 0, 0, tzinfo=SHANGHAI_TZ).timestamp()) + repo.kv_set( + "task:last:snapshot_seed", + json.dumps({ + "name": "snapshot_seed", "ok": False, "skipped": True, + "started_at": finished - 1, "finished_at": finished, + "error": "snapshot_running", + "summary": {"skipped": True, "reason": "snapshot_running"}, + }), + ) + DASHBOARD_CACHE.clear() + system = q.system() + assert system["next_runs"]["snapshot_seed"] == finished + 600 + assert system["next_runs"]["snapshot_seed"] != int( + next_daily_run(finished, "00:00") + ) + # skip 不算 errors + assert not any(e["task"] == "snapshot_seed" for e in system["errors"]) + + # no_source 同样 10 分钟 + repo.kv_set( + "task:last:snapshot_seed", + json.dumps({ + "name": "snapshot_seed", "ok": False, "skipped": True, + "started_at": finished - 1, "finished_at": finished, + "error": "no_source", + "summary": {"skipped": True, "reason": "no_source"}, + }), + ) + DASHBOARD_CACHE.clear() + system = q.system() + assert system["next_runs"]["snapshot_seed"] == finished + 600 + finally: + db.close() + + def test_overview_adjacent_previous_with_seed_and_dual_fields(): """seed 当日行存在时，播放增量按片相邻成功日；overview 暴露更新时刻与总数。""" @@ -964,6 +1009,7 @@ def main_test(): test_snapshot_recent_next_run_follows_completion() test_snapshot_recent_next_run_defers_quiet_window_and_retry_on_skip() test_snapshot_failed_next_run_retries_soon_not_next_day() + test_snapshot_seed_skip_next_run_retries_ten_minutes() test_overview_adjacent_previous_with_seed_and_dual_fields() test_same_day_growth_baselines() test_growth_rate_sort_uses_prior()