From f6bef59e073f096d7d112aac6537c6b56391bf51 Mon Sep 17 00:00:00 2001 From: bot-ci Date: Tue, 3 Nov 2026 02:35:13 +0800 Subject: [PATCH] fix(scheduler): daily slot off-plan + cross-midnight resume (#8) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Cross-midnight: unfinished yesterday seals ASAP with yesterday's stat_date; force seal ignores today's snapshot_at gate when stat_date < today. - Off-plan already_completed: compare completion started_at (not finished_at) to today's planned slot; earlier → still run at slot for all daily tasks. - Cold start: bootstrap task:state from task:last / seed markers. - ensure_pending: do not glue new due onto a different day's unfinished state. - run_if_needed(force): keep_scraped True for seal, False for CLI --force. - Scheduler due_reader: pick up disk due_at each loop (seed→evening seal). --- reaction_board/cli.py | 138 ++++++++++++++++++++--- reaction_board/scheduler.py | 7 ++ reaction_board/services/snapshot.py | 14 ++- reaction_board/task_state.py | 163 ++++++++++++++++++++++++++-- tests/test_cli.py | 131 ++++++++++++++++++++++ tests/test_snapshot.py | 69 ++++++++++++ tests/test_task_state.py | 132 ++++++++++++++++++++++ 7 files changed, 630 insertions(+), 24 deletions(-) diff --git a/reaction_board/cli.py b/reaction_board/cli.py index 5ba2eac..514dc48 100644 --- a/reaction_board/cli.py +++ b/reaction_board/cli.py @@ -377,14 +377,19 @@ def _initial_daily_next_run( ) -> float: """进程启动时日更任务的下次运行：优先 ``task:state.due_at``。 - 不得返回 ``0`` / 用「无状态」以外的过去时间去误伤——无状态时对齐墙钟。 - 失败且 ``due_at`` 已过：返回该 ``due_at``（调度器会立刻补跑），不退回次日墙钟。 + 无状态时从 ``task:last`` / seed 标记冷启动回填；不得返回 ``0`` 误伤。 + 失败且 ``due_at`` 已过：立刻补跑，不退回次日墙钟。 """ - del retry_seconds # 间隔已落在 task:state.due_at current = time.time() if now is None else float(now) wall = next_daily_run(current, at) + retry = float(retry_seconds) if retry_seconds is not None else _TASK_RETRY_SECONDS return task_state_mod.startup_next_run( - repo, name, wall_due_at=wall, now=current, + repo, + name, + wall_due_at=wall, + now=current, + slot_hhmm=at, + retry_seconds=retry, ) @@ -396,6 +401,67 @@ def _reached_daily_slot(at: str, now: float | None = None) -> bool: return current >= slot +def _today_slot_ts(at: str, now: float | None = None) -> float: + """今日 ``at`` 墙钟点的时间戳（不论是否已过）。""" + current = shanghai_datetime(now) + hour, minute = parse_hhmm(at) + return current.replace(hour=hour, minute=minute, second=0, microsecond=0).timestamp() + + +def _last_success_started_at(repo: Repository, name: str) -> float | None: + """最近一次成功完成记录的 ``started_at``（不用 finished_at）。""" + try: + raw = repo.kv_get(f"task:last:{name}") + except Exception: # noqa: BLE001 + return None + if not raw: + return None + try: + data = json.loads(raw) + except (TypeError, ValueError, json.JSONDecodeError): + return None + if not isinstance(data, dict): + return None + if not data.get("ok") or data.get("skipped"): + return None + try: + return float(data["started_at"]) + except (KeyError, TypeError, ValueError): + return None + + +def _off_plan_completion( + repo: Repository, + name: str, + at: str, + now: float | None = None, +) -> bool: + """成功完成的 started_at 早于当日计划点 → 非计划内，到点仍应再跑。 + + 产品规则：只用完成记录的 **started_at** 与今日 ``at`` 比较，禁止 finished_at。 + """ + started = _last_success_started_at(repo, name) + if started is None: + return False + return float(started) < _today_slot_ts(at, now) + + +def _should_force_daily_seal( + *, + seeded: bool, + stat_date: str, + today: str, + slot_hhmm: str, + now: float, +) -> bool: + """seed 壳是否 force 打穿：跨天未完成立即 force；同日须已到计划点。""" + if not seeded: + return False + if stat_date < today: + return True + return _reached_daily_slot(slot_hhmm, now) + + def _evening_due_at(now: float, snapshot_at: str) -> float: """尚未到当日 snapshot_at 则挂今晚，否则下一档墙钟。""" hour, minute = parse_hhmm(snapshot_at) @@ -535,7 +601,11 @@ def cmd_snapshot(args, logger: JsonLogger) -> int: stat_date=shanghai_date(float(started)), ) try: - video_summary = service.run_if_needed(force=args.force) + # CLI --force：全量重采（keep_scraped=False）；非 force 走常规门闩 + video_summary = service.run_if_needed( + force=args.force, + keep_scraped=False if args.force else None, + ) except Exception as exc: # noqa: BLE001 finished = int(time.time()) _record_task( @@ -1014,16 +1084,33 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int: repo, "snapshot", started_at=started, stat_date=stat_date, ) try: - # seed 壳 ≠ 白天强采：仅到达当日 snapshot_at（或显式续跑已到期的 - # seal pending）才 force。due_at 门闩负责「会不会跑」；这里是 - # force 的第二道闸。 + # seed 壳 force：跨天未完成（stat_date < today）立刻打穿； + # 同日仅到达 snapshot_at 后才 force。due_at 管「会不会跑」。 seeded = repo.kv_get(SnapshotService.seeded_key(stat_date)) == "1" - force_seed = seeded and _reached_daily_slot( - snapshot_at, float(started), + force_seed = _should_force_daily_seal( + seeded=seeded, + stat_date=stat_date, + today=today, + slot_hhmm=snapshot_at, + now=float(started), ) summary = snapshot_service.run_if_needed( - date=stat_date, force=force_seed, + date=stat_date, + force=force_seed, + keep_scraped=True if force_seed else None, ) + # 日更规则：already_completed 但完成 started_at 早于今日计划点 + # → 非计划内，到点仍按计划再跑（全量，不用 finished_at）。 + if ( + summary.already_completed + and _reached_daily_slot(snapshot_at, float(started)) + and _off_plan_completion( + repo, "snapshot", snapshot_at, float(started), + ) + ): + summary = snapshot_service.run_if_needed( + date=stat_date, force=True, keep_scraped=False, + ) except Exception as exc: # noqa: BLE001 finished = int(time.time()) next_run = _retry_soon(float(finished)) @@ -1138,12 +1225,26 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int: seeded = ( repo.kv_get(SnapshotService.fans_seeded_key(stat_date)) == "1" ) - force_seed = seeded and _reached_daily_slot( - snapshot_at, float(started), + force_seed = _should_force_daily_seal( + seeded=seeded, + stat_date=stat_date, + today=today, + slot_hhmm=snapshot_at, + now=float(started), ) summary = snapshot_service.run_fans_if_needed( date=stat_date, force=force_seed, ) + if ( + summary.already_completed + and _reached_daily_slot(snapshot_at, float(started)) + and _off_plan_completion( + repo, "fans", snapshot_at, float(started), + ) + ): + summary = snapshot_service.run_fans_if_needed( + date=stat_date, force=True, + ) except Exception as exc: # noqa: BLE001 finished = int(time.time()) next_run = _retry_soon(float(finished)) @@ -1287,6 +1388,16 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int: cleanup_next = task_state_mod.startup_next_run( repo, "candidate_cleanup", wall_due_at=now_ts, now=now_ts, ) + def _due_reader(task_name: str) -> float | None: + state = task_state_mod.load_state(repo, task_name) + if state is None: + return None + try: + due = float(state.due_at or 0) + except (TypeError, ValueError): + return None + return due if due > 0 else None + scheduler = Scheduler( [ ScheduledTask( @@ -1328,6 +1439,7 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int: ScheduledTask("config_watch", 60.0, run_config_watch), ], logger=logger, + due_reader=_due_reader, ) if serve_web and not args.once: diff --git a/reaction_board/scheduler.py b/reaction_board/scheduler.py index 2febe37..516571e 100644 --- a/reaction_board/scheduler.py +++ b/reaction_board/scheduler.py @@ -59,6 +59,7 @@ class Scheduler: clock=time.time, sleeper=time.sleep, logger=None, + due_reader: Callable[[str], float | None] | None = None, ) -> None: if not tasks: raise ValueError("至少需要一个调度任务") @@ -69,6 +70,8 @@ class Scheduler: self._clock = clock self._sleeper = sleeper self.logger = logger + # 每轮从落盘 due_at 刷新 next_run（seed 挂当晚 seal 等同进程可见） + self._due_reader = due_reader def _log(self, level: str, event: str, **fields) -> None: if self.logger is not None: @@ -78,6 +81,10 @@ class Scheduler: current = self._clock() if now is None else now results: list[TaskResult] = [] for task in self.tasks: + if self._due_reader is not None: + disk_due = self._due_reader(task.name) + if disk_due is not None: + task.next_run = float(disk_due) if task.next_run > current: continue started = current diff --git a/reaction_board/services/snapshot.py b/reaction_board/services/snapshot.py index b88df85..be8204e 100644 --- a/reaction_board/services/snapshot.py +++ b/reaction_board/services/snapshot.py @@ -142,12 +142,20 @@ class SnapshotService: return f"fans:seeded:{stat_date}" def run_if_needed( - self, *, date: str | None = None, force: bool = False + self, + *, + date: str | None = None, + force: bool = False, + keep_scraped: bool | None = None, ) -> SnapshotSummary: """视频快照（含 UP 头像补齐）；粉丝采集请用 ``run_fans_if_needed``。 当日若仅有午夜 seed 壳且未 ``force``，直接返回等待晚间落盘， 避免把「全员 skip」误标成 completed。 + + ``force`` + ``keep_scraped``： + - 晚间 seal / 断点续跑：``keep_scraped=True``（默认）保留已采 ok=1； + - CLI ``--force``：显式 ``keep_scraped=False`` 全量重采。 """ now = self._now() stat_date = date or shanghai_date(now) @@ -155,9 +163,9 @@ class SnapshotService: raise ValueError(f"日期格式必须是 YYYY-MM-DD: {stat_date!r}") seeded = self.repo.kv_get(self.seeded_key(stat_date)) == "1" if force: - # 晚间 seal / 续跑：打穿 seed，保留已真实采集的 ok=1 作为进度。 + ks = True if keep_scraped is None else bool(keep_scraped) return self.run_videos( - date=stat_date, force=True, keep_scraped=True, + date=stat_date, force=True, keep_scraped=ks, ) completed = self.repo.kv_get(self.completed_key(stat_date)) == "1" diff --git a/reaction_board/task_state.py b/reaction_board/task_state.py index 14f05cc..75b13cc 100644 --- a/reaction_board/task_state.py +++ b/reaction_board/task_state.py @@ -10,10 +10,12 @@ from __future__ import annotations import json import time -from dataclasses import asdict, dataclass, field +from dataclasses import asdict, dataclass +from datetime import timedelta from typing import Any, Mapping from .repository import Repository +from .timeutil import parse_hhmm, shanghai_date, shanghai_datetime STATUS_PENDING = "pending" STATUS_RUNNING = "running" @@ -219,13 +221,36 @@ def ensure_pending( ) -> TaskState: """保证存在一条 pending（如 seed 成功后挂上当晚 seal）。 - 若已有未完成状态且 due_at 更早，保留更早的 due_at； + 同日未完成：取更早的 due_at。 + 跨日：不得把新一天的 due 粘到昨天的 ``stat_date`` 上—— + 若已有更早日期的未完成记录，保留它（跨天续跑优先）； + 若已有更新日期的未完成，则重置为调用方的 due/stat_date。 ``overwrite_done`` 时覆盖已完成记录。 """ current = _now(now) prev = load_state(repo, name) if prev is not None and prev.status in _ACTIVE and prev.status != STATUS_RUNNING: - # 未完成：取更早的 due_at，补上缺失的 stat_date + if ( + stat_date is not None + and prev.stat_date + and prev.stat_date != stat_date + ): + # 未完成的更早归属日优先保留，避免 seed 挂今晚时冲掉昨天续跑。 + if prev.stat_date < stat_date: + return prev + # prev 归属日更新：按调用方重置，禁止跨日字段拼接。 + state = TaskState( + name=name, + status=STATUS_PENDING, + due_at=float(due_at), + stat_date=stat_date, + progress=None, + updated_at=current, + error=None, + ) + save_state(repo, state) + return state + # 同日（或缺一边 stat_date）：取更早 due_at，补上缺失的 stat_date due = min(float(prev.due_at or due_at), float(due_at)) sd = prev.stat_date or stat_date if due == prev.due_at and sd == prev.stat_date and prev.status == STATUS_PENDING: @@ -253,23 +278,145 @@ def ensure_pending( return state +def bootstrap_from_last( + repo: Repository, + name: str, + *, + wall_due_at: float, + now: float | None = None, + retry_seconds: float = 300.0, +) -> TaskState | None: + """升级/冷启动：无 ``task:state`` 时从 ``task:last`` 回填。 + + - 成功 → ``done`` + ``due_at=wall_due_at`` + - 失败/跳过 → ``failed``/``deferred``，``due_at`` 为补跑点（已过则立刻） + """ + current = _now(now) + existing = load_state(repo, name) + if existing is not None: + return existing + raw = repo.kv_get(f"task:last:{name}") + if not raw: + return None + try: + data = json.loads(raw) + except (TypeError, ValueError, json.JSONDecodeError): + return None + if not isinstance(data, dict): + return None + ok = bool(data.get("ok")) + skipped = bool(data.get("skipped")) + try: + finished = float(data.get("finished_at") or data.get("started_at") or current) + except (TypeError, ValueError): + finished = current + summary = data.get("summary") + stat_date = None + if isinstance(summary, dict) and summary.get("stat_date"): + stat_date = str(summary["stat_date"]) + error = data.get("error") + error_s = str(error) if error else None + if ok and not skipped: + state = TaskState( + name=name, + status=STATUS_DONE, + due_at=float(wall_due_at), + stat_date=stat_date, + updated_at=current, + error=None, + ) + save_state(repo, state) + return state + due = finished + float(retry_seconds) + if due > current: + pass + else: + due = current + state = TaskState( + name=name, + status=STATUS_DEFERRED if skipped else STATUS_FAILED, + due_at=float(due), + stat_date=stat_date, + updated_at=current, + error=error_s, + ) + save_state(repo, state) + return state + + +def bootstrap_seed_markers( + repo: Repository, + name: str, + *, + slot_hhmm: str, + wall_due_at: float, + now: float | None = None, +) -> TaskState | None: + """无 state 时：若存在未 seal 的 seed 壳，挂 pending（跨天优先昨天）。 + + 已过当日 ``slot_hhmm``（或归属日早于今天）→ ``due_at=now`` 尽快补跑。 + """ + del wall_due_at # 未完成 seed 不走纯墙钟 + current = _now(now) + existing = load_state(repo, name) + if existing is not None: + return existing + if name not in {"snapshot", "fans"}: + return None + today = shanghai_date(current) + yday = (shanghai_datetime(current) - timedelta(days=1)).strftime("%Y-%m-%d") + hour, minute = parse_hhmm(slot_hhmm) + today_slot = shanghai_datetime(current).replace( + hour=hour, minute=minute, second=0, microsecond=0, + ).timestamp() + + def _seeded_unfinished(day: str) -> bool: + if name == "snapshot": + if repo.kv_get(f"snapshot:seeded:{day}") != "1": + return False + return repo.kv_get(f"snapshot:completed:{day}") != "1" + # fans：seed 键在成功 seal 后删除，有键即未完成 + return repo.kv_get(f"fans:seeded:{day}") == "1" + + for day in (yday, today): + if not _seeded_unfinished(day): + continue + if day < today or current >= today_slot: + due = current + else: + due = float(today_slot) + return ensure_pending( + repo, name, due_at=due, stat_date=day, now=current, + ) + return None + + def startup_next_run( repo: Repository, name: str, *, wall_due_at: float, now: float | None = None, + slot_hhmm: str | None = None, + retry_seconds: float = 300.0, ) -> float: """常驻启动时的 ``ScheduledTask.next_run``：只认落盘状态。 + - 无状态时先 ``bootstrap_from_last`` / seed 标记回填 - 恢复残留 ``running`` → ``pending`` - - ``due_at > now`` → 原样返回（失败延时不会被提前） - - ``due_at <= now`` 且未完成 → 立刻可跑（返回 ``due_at``，可为过去时间） - - ``done`` 且 ``due_at > now`` → 等下一次 - - ``done`` 且 ``due_at <= now`` → 到期，返回 ``due_at`` - - 无状态 → 墙钟 ``wall_due_at`` + - ``due_at`` 是唯一时间门闩（失败延时不会被提前） """ current = _now(now) + if load_state(repo, name) is None: + bootstrap_from_last( + repo, name, wall_due_at=wall_due_at, now=current, + retry_seconds=retry_seconds, + ) + if slot_hhmm: + bootstrap_seed_markers( + repo, name, slot_hhmm=slot_hhmm, + wall_due_at=wall_due_at, now=current, + ) state = recover_stale_running(repo, name, now=current) if state is None: return float(wall_due_at) diff --git a/tests/test_cli.py b/tests/test_cli.py index f4ef469..b6f369d 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -805,6 +805,134 @@ def test_daytime_seed_marker_does_not_force_without_slot(): db.close() + +def test_off_plan_completion_uses_started_at_not_finished_at(): + """规则2：完成记录 started_at 早于今日计划点 → off-plan；禁用 finished_at。""" + from datetime import datetime + from reaction_board.timeutil import SHANGHAI_TZ + + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, cookie="", csrf="") + assert run_cli(["init", "--config", str(path)])[0] == 0 + db = Database(Path(tmp) / "app.db") + db.initialize() + try: + repo = Repository(db) + slot = datetime(2026, 11, 3, 23, 15, tzinfo=SHANGHAI_TZ).timestamp() + # 白天开始、晚上才结束：started_at < slot，finished_at > slot + day_start = datetime(2026, 11, 3, 10, 0, tzinfo=SHANGHAI_TZ).timestamp() + day_end = datetime(2026, 11, 3, 23, 30, tzinfo=SHANGHAI_TZ).timestamp() + repo.kv_set( + "task:last:snapshot", + json.dumps({ + "name": "snapshot", + "ok": True, + "skipped": False, + "started_at": int(day_start), + "finished_at": int(day_end), + "error": None, + "summary": {"stat_date": "2026-11-03"}, + }, ensure_ascii=False), + ) + assert cli._last_success_started_at(repo, "snapshot") == float(int(day_start)) + assert cli._off_plan_completion(repo, "snapshot", "23:15", slot) is True + # 若误用 finished_at 会判成计划内；started_at 必须仍为 off-plan + assert day_end >= slot + assert day_start < slot + + # 计划内：started_at == slot → 不可 off-plan + repo.kv_set( + "task:last:snapshot", + json.dumps({ + "name": "snapshot", + "ok": True, + "skipped": False, + "started_at": int(slot), + "finished_at": int(slot + 600), + "error": None, + "summary": {"stat_date": "2026-11-03"}, + }, ensure_ascii=False), + ) + assert cli._off_plan_completion(repo, "snapshot", "23:15", slot) is False + assert cli._off_plan_completion( + repo, "snapshot", "23:15", slot + 60, + ) is False + finally: + db.close() + + +def test_cross_midnight_force_seal_ignores_todays_slot_gate(): + """规则1：昨天 seed 未完成，跨日后即便未到今日 23:15 也 force。""" + from datetime import datetime + from reaction_board.timeutil import SHANGHAI_TZ + + now = datetime(2026, 11, 3, 0, 10, tzinfo=SHANGHAI_TZ).timestamp() + assert cli._reached_daily_slot("23:15", now) is False + assert cli._should_force_daily_seal( + seeded=True, + stat_date="2026-11-02", + today="2026-11-03", + slot_hhmm="23:15", + now=now, + ) is True + # 同日白天仍不得 force + day = datetime(2026, 11, 3, 12, 0, tzinfo=SHANGHAI_TZ).timestamp() + assert cli._should_force_daily_seal( + seeded=True, + stat_date="2026-11-03", + today="2026-11-03", + slot_hhmm="23:15", + now=day, + ) is False + night = datetime(2026, 11, 3, 23, 15, tzinfo=SHANGHAI_TZ).timestamp() + assert cli._should_force_daily_seal( + seeded=True, + stat_date="2026-11-03", + today="2026-11-03", + slot_hhmm="23:15", + now=night, + ) is True + + +def test_initial_daily_bootstraps_failed_last_without_state(): + """A: 无 state、仅 task:last 失败 → 启动 next_run 立刻可补，不退墙钟。""" + from datetime import datetime + from reaction_board.timeutil import SHANGHAI_TZ, next_daily_run + from reaction_board.task_state import STATUS_FAILED, load_state + + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, cookie="", csrf="") + assert run_cli(["init", "--config", str(path)])[0] == 0 + db = Database(Path(tmp) / "app.db") + db.initialize() + try: + repo = Repository(db) + now = datetime(2026, 11, 3, 18, 0, tzinfo=SHANGHAI_TZ).timestamp() + repo.kv_set( + "task:last:snapshot", + json.dumps({ + "name": "snapshot", + "ok": False, + "skipped": False, + "started_at": int(now - 900), + "finished_at": int(now - 800), + "error": "net", + "summary": {"stat_date": "2026-11-03"}, + }, ensure_ascii=False), + ) + wall = next_daily_run(now, "23:15") + nxt = cli._initial_daily_next_run( + repo, "snapshot", "23:15", now=now, + ) + assert nxt <= now + assert nxt != wall + state = load_state(repo, "snapshot") + assert state is not None + assert state.status == STATUS_FAILED + finally: + db.close() + + def main_test(): test_init_and_status() test_missing_config() @@ -830,6 +958,9 @@ def main_test(): test_reached_daily_slot_gates_force_seed() test_restart_after_seed_schedules_snapshot_at_wall_clock() test_daytime_seed_marker_does_not_force_without_slot() + test_off_plan_completion_uses_started_at_not_finished_at() + test_cross_midnight_force_seal_ignores_todays_slot_gate() + test_initial_daily_bootstraps_failed_last_without_state() print("OK: CLI init/sync/snapshot/snapshot-seed/auth-check/run/status/backfill/错误路径测试全部通过") diff --git a/tests/test_snapshot.py b/tests/test_snapshot.py index 8d8ac4c..79da107 100644 --- a/tests/test_snapshot.py +++ b/tests/test_snapshot.py @@ -705,6 +705,73 @@ def test_seal_keep_scraped_skips_view_progress_punches_seed(): db.close() + +def test_force_keep_scraped_false_rescrapes_view_progress(): + """C: CLI --force 路径 keep_scraped=False 须重采已有 view 进度行。""" + clock = 1_800_000_000 + with tempfile.TemporaryDirectory() as tmp: + transport = SnapshotTransport({ + 1: view_for(1, view=999), + 2: view_for(2, view=222), + }) + db, repo, service = make_service(tmp, transport) + try: + add_video(repo, 1) + add_video(repo, 2) + today = DATE + for aid, view, src in ((1, 111, "view"), (2, 20, "seed")): + repo.upsert_snapshot( + today, aid, + metrics={"view": view, "like": 1, "coin": 0, "favorite": 0, + "share": 0, "reply": 0, "danmaku": 0}, + ok=True, err=None, captured_at=clock, src=src, + ) + repo.kv_set(service.seeded_key(today), "1") + forced = service.run_if_needed( + date=today, force=True, keep_scraped=False, + ) + assert forced.completed is True + # 两条都重采 + assert transport.count("/web-interface/view") == 2 + assert int(repo.get_snapshot(today, 1)["view"]) == 999 + finally: + db.close() + + +def test_force_default_keep_scraped_skips_view_progress(): + """C: force 默认 keep_scraped=True（seal）跳过已 view 行。""" + clock = 1_800_000_000 + with tempfile.TemporaryDirectory() as tmp: + transport = SnapshotTransport({ + 1: view_for(1, view=999), + 2: view_for(2, view=222), + }) + db, repo, service = make_service(tmp, transport) + try: + add_video(repo, 1) + add_video(repo, 2) + today = DATE + repo.upsert_snapshot( + today, 1, + metrics={"view": 111, "like": 1, "coin": 0, "favorite": 0, + "share": 0, "reply": 0, "danmaku": 0}, + ok=True, err=None, captured_at=clock, src="view", + ) + repo.upsert_snapshot( + today, 2, + metrics={"view": 20, "like": 1, "coin": 0, "favorite": 0, + "share": 0, "reply": 0, "danmaku": 0}, + ok=True, err=None, captured_at=clock, src="seed", + ) + repo.kv_set(service.seeded_key(today), "1") + sealed = service.run_if_needed(date=today, force=True) + assert sealed.completed is True + assert transport.count("/web-interface/view") == 1 + assert int(repo.get_snapshot(today, 1)["view"]) == 111 + finally: + db.close() + + def main(): test_snapshot_success_resume_and_force() test_resume_skips_ok_and_retries_failed() @@ -724,6 +791,8 @@ def main(): test_seed_from_last_completed_does_not_mark_completed() test_seeded_day_skips_without_force_and_seals_with_force() test_seal_keep_scraped_skips_view_progress_punches_seed() + test_force_keep_scraped_false_rescrapes_view_progress() + test_force_default_keep_scraped_skips_view_progress() print("OK: 每日快照、断点续跑、失败占位与粉丝采集测试全部通过") diff --git a/tests/test_task_state.py b/tests/test_task_state.py index 3ca8b52..117bc3d 100644 --- a/tests/test_task_state.py +++ b/tests/test_task_state.py @@ -20,6 +20,7 @@ from reaction_board.task_state import ( # noqa: E402 STATUS_FAILED, STATUS_PENDING, STATUS_RUNNING, + bootstrap_from_last, ensure_pending, load_state, mark_deferred, @@ -251,6 +252,133 @@ def test_fail_or_defer_preserves_progress(): db.close() + +def test_bootstrap_from_last_fail_due_soon(): + """A: 仅有 task:last 失败、无 state → 回填 failed 且 due_at 可立刻补跑。""" + with tempfile.TemporaryDirectory() as tmp: + db, repo = _repo(tmp) + try: + now = datetime(2026, 11, 3, 12, 0, tzinfo=SHANGHAI_TZ).timestamp() + wall = next_daily_run(now, "23:15") + repo.kv_set( + "task:last:snapshot", + json.dumps({ + "name": "snapshot", + "ok": False, + "skipped": False, + "started_at": int(now - 600), + "finished_at": int(now - 500), + "error": "boom", + "summary": {"stat_date": "2026-11-03"}, + }, ensure_ascii=False), + ) + assert load_state(repo, "snapshot") is None + boot = bootstrap_from_last( + repo, "snapshot", wall_due_at=wall, now=now, retry_seconds=300.0, + ) + assert boot is not None + assert boot.status == STATUS_FAILED + assert boot.stat_date == "2026-11-03" + assert boot.due_at <= now + nxt = startup_next_run( + repo, "snapshot", wall_due_at=wall, now=now, slot_hhmm="23:15", + ) + assert nxt <= now + assert nxt != wall + finally: + db.close() + + +def test_ensure_pending_does_not_glue_yesterday_stat_date(): + """B: 昨天未完成保留；不得把今晚 due 粘到昨天的 stat_date 上。""" + with tempfile.TemporaryDirectory() as tmp: + db, repo = _repo(tmp) + try: + now = datetime(2026, 11, 3, 0, 10, tzinfo=SHANGHAI_TZ).timestamp() + yday_due = now - 60 + ensure_pending( + repo, "snapshot", due_at=yday_due, + stat_date="2026-11-02", now=now - 120, + ) + tonight = datetime(2026, 11, 3, 23, 15, tzinfo=SHANGHAI_TZ).timestamp() + # seed 想挂今晚：不得冲掉昨天未完成 + kept = ensure_pending( + repo, "snapshot", due_at=tonight, + stat_date="2026-11-03", now=now, + ) + assert kept.stat_date == "2026-11-02" + assert kept.due_at == yday_due + assert resume_stat_date( + repo, "snapshot", fallback="2026-11-03", + ) == "2026-11-02" + finally: + db.close() + + +def test_cross_midnight_mid_seal_resumes_yesterday(): + """规则1 / B: 23:15 崩溃 → 00:10 重启续跑昨天，due 立刻可跑。""" + with tempfile.TemporaryDirectory() as tmp: + db, repo = _repo(tmp) + try: + crash = datetime(2026, 11, 2, 23, 40, tzinfo=SHANGHAI_TZ).timestamp() + mark_running( + repo, "snapshot", stat_date="2026-11-02", now=crash, + progress={"done": 3, "total": 10}, + ) + repo.kv_set("snapshot:seeded:2026-11-02", "1") + restart = datetime(2026, 11, 3, 0, 10, tzinfo=SHANGHAI_TZ).timestamp() + wall = next_daily_run(restart, "23:15") + nxt = startup_next_run( + repo, "snapshot", wall_due_at=wall, now=restart, slot_hhmm="23:15", + ) + assert nxt <= restart + assert resume_stat_date( + repo, "snapshot", fallback="2026-11-03", + ) == "2026-11-02" + state = load_state(repo, "snapshot") + assert state is not None + assert state.status == STATUS_PENDING + assert state.progress == {"done": 3, "total": 10} + finally: + db.close() + + +def test_scheduler_due_reader_picks_up_enqueued_seal(): + """D: 同进程 seed 写盘 due_at=now 后，下一轮 run_due 须跑 seal。""" + with tempfile.TemporaryDirectory() as tmp: + db, repo = _repo(tmp) + try: + now = datetime(2026, 11, 3, 23, 20, tzinfo=SHANGHAI_TZ).timestamp() + # snapshot 原 next_run 挂次日；盘上被 enqueue 成 now + tomorrow = next_daily_run(now, "23:15") + calls = [] + + def run_snap(): + calls.append("snap") + return None + + def due_reader(name: str): + st = load_state(repo, name) + return float(st.due_at) if st and st.due_at else None + + clock = [now] + sched = Scheduler( + [ScheduledTask("snapshot", 86400.0, run_snap, next_run=tomorrow)], + clock=lambda: clock[0], + due_reader=due_reader, + ) + assert sched.run_due() == [] + assert calls == [] + ensure_pending( + repo, "snapshot", due_at=now, stat_date="2026-11-03", now=now, + ) + results = sched.run_due() + assert len(results) == 1 + assert calls == ["snap"] + finally: + db.close() + + def main_test(): test_restart_mid_run_resumes_same_stat_date() test_seeded_before_snapshot_at_does_not_force_run() @@ -259,6 +387,10 @@ def main_test(): test_done_writes_next_due_and_blocks_until_then() test_no_state_falls_back_to_wall_clock() test_fail_or_defer_preserves_progress() + test_bootstrap_from_last_fail_due_soon() + test_ensure_pending_does_not_glue_yesterday_stat_date() + test_cross_midnight_mid_seal_resumes_yesterday() + test_scheduler_due_reader_picks_up_enqueued_seal() print("OK: task_state persistence / resume / due_at gates")