diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index 766f429..bbde903 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -456,6 +456,9 @@ 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:completed:` 时 seed 记 `already_completed` 成功（即使 `last_completed` 已等于当日），**不得**误判 `no_source`； +- 若全站 `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` 恢复的视频会立即补采当天快照； @@ -466,6 +469,15 @@ P1 实现说明： - Cookie 登录态通过 `x/web-interface/nav` 检查，失效只告警，不阻塞匿名 `view`； - Telegram 摘要推迟到 P3，P1 先写结构化 JSON 日志和 `kv` 状态。 +调度避让与补跑： + +- 常驻启动时日更任务（`snapshot_seed` / `snapshot` / `fans`）按墙钟下一档 `seed_at` / `snapshot_at` 排期（可参考上次成功/失败记录），**不得** `next_run=0` 重启即跑； +- 仅到达当日 `snapshot_at` 后（或 CLI `--force`）才对 seed 壳 `force` 打穿；白天重启不因 `snapshot:seeded` 全量强采； +- 全站 `snapshot` **不因**其它任务 running 而 skip；到期/强制必须执行； +- 上海时区 **23:00–01:00** 为全站快照窗：`discovery` / `snapshot_recent` / `candidate_cleanup` 不执行，延后到 01:00 后；`fans` 短耗时，不避开任何任务； +- 任务 skip（含静默窗延后）写入 `ok=false` + `skipped=true`，**不得**记成成功或推进「当日已完成」； +- skip / fail 后短间隔补跑（约 5 分钟；`snapshot_seed` 避让/无源为约 10 分钟）；日更成功才推到次日 `snapshot_at` / `seed_at`。 + --- ## 8. Telegram 与入库操作 diff --git a/project.md b/project.md index 75f5548..1a7d7cc 100644 --- a/project.md +++ b/project.md @@ -800,7 +800,7 @@ secrets.example.json 凭据样例（auth 回写目标） **进程结构** ``` -scheduler 线程：discovery(每 60min) / snapshot_seed(每日 00:00) / snapshot+fans(每日 23:15) / snapshot_recent(小时) / candidate_cleanup(每日) +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 90d996c..514dc48 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 登录态 @@ -33,13 +34,23 @@ from .database import Database from .logging_json import JsonLogger from .notify.telegram import TelegramClient from .repository import Repository -from .scheduler import ScheduledTask, Scheduler +from .scheduler import ScheduledTask, Scheduler, TaskControl from .services.discovery import DiscoveryService from .services.library import LibraryService from .services.snapshot import SnapshotService from .services.telegram_poller import TelegramPoller from .services.up_profile import UpProfileService -from .timeutil import next_daily_run, parse_hhmm, shanghai_date +from . import task_state as task_state_mod +from .timeutil import ( + in_snapshot_quiet_window, + interval_next_run, + is_date, + next_after_snapshot_quiet_window, + next_daily_run, + parse_hhmm, + shanghai_date, + shanghai_datetime, +) ROOT = Path(__file__).resolve().parent.parent DEFAULT_CONFIG = default_config_path(ROOT) @@ -71,6 +82,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="执行一次搜索发现并写入候选池" ) @@ -143,17 +165,29 @@ def _task_running_key(name: str) -> str: return f"task:running:{name}" -def _mark_task_running(repo: Repository, name: str, *, started_at: int | None = None) -> None: - """任务开始时写入 running 标记，供 /system 显示「运行中」。""" - now = int(time.time()) +def _mark_task_running( + repo: Repository, + name: str, + *, + started_at: int | None = None, + stat_date: str | None = None, +) -> None: + """任务开始：running 标记 + 持久化 status=running（跨重启可续跑）。""" + now = int(started_at if started_at is not None else time.time()) record = { "name": name, - "started_at": int(started_at if started_at is not None else now), + "started_at": now, } try: repo.kv_set(_task_running_key(name), json.dumps(record, ensure_ascii=False)) except Exception: # noqa: BLE001 pass + try: + task_state_mod.mark_running( + repo, name, stat_date=stat_date, now=float(now), + ) + except Exception: # noqa: BLE001 + pass def _clear_task_running(repo: Repository, name: str) -> None: @@ -164,6 +198,7 @@ def _clear_task_running(repo: Repository, name: str) -> None: def _is_task_running(repo: Repository, name: str) -> bool: + """供 seed 等任务检测全站快照是否在跑（避让，而非互斥跳过成功）。""" try: return bool(repo.kv_get(_task_running_key(name))) except Exception: # noqa: BLE001 @@ -179,14 +214,27 @@ def _record_task( finished_at: int | None = None, payload=None, error: str | None = None, + skipped: bool = False, + next_due_at: float | None = None, + stat_date: str | None = None, + progress=None, ) -> None: - """把最近一次任务结果写入 kv，并清除 running 标记。""" + """把最近一次任务结果写入 kv，并清除 running 标记。 + + ``skipped=True`` 表示主动延后/避让，不得记成成功，也不推进「当日已完成」。 + 同步写入 ``task:state``：有 ``next_due_at`` 用其排程；否则结束残留 + ``running`` 并保留原 ``due_at``（CLI 一锤子路径）。 + """ now = int(time.time()) + if skipped: + ok = False + finished = int(finished_at if finished_at is not None else now) record = { "name": name, "ok": bool(ok), + "skipped": bool(skipped), "started_at": int(started_at if started_at is not None else now), - "finished_at": int(finished_at if finished_at is not None else now), + "finished_at": finished, "error": error, "summary": payload, } @@ -195,6 +243,255 @@ def _record_task( except Exception: # noqa: BLE001 任务状态记录失败不影响主流程 pass _clear_task_running(repo, name) + try: + if next_due_at is not None: + due = float(next_due_at) + else: + # CLI 等非调度路径：结束 running，保留原 due_at，避免残留 running。 + prev = task_state_mod.load_state(repo, name) + if prev is None or prev.status != task_state_mod.STATUS_RUNNING: + return + due = ( + float(prev.due_at) + if prev.due_at and float(prev.due_at) > 0 + else float(finished) + ) + if stat_date is None: + stat_date = prev.stat_date + if skipped: + task_state_mod.mark_deferred( + repo, + name, + due_at=due, + now=float(finished), + stat_date=stat_date, + progress=progress, + error=error, + ) + elif ok: + task_state_mod.mark_done( + repo, + name, + due_at=due, + now=float(finished), + stat_date=stat_date, + progress=progress, + ) + else: + task_state_mod.mark_failed( + repo, + name, + due_at=due, + now=float(finished), + stat_date=stat_date, + progress=progress, + error=error, + ) + except Exception: # noqa: BLE001 + pass + + +# 全站快照 / 粉丝失败后的补跑间隔（秒） +_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( + repo: Repository, + logger: JsonLogger, + name: str, + *, + started_at: int, +) -> TaskControl: + """静默窗内不执行：记 skip（非成功），并排到 01:00 之后。""" + finished = int(time.time()) + logger.info(f"{name}.skip", reason=_QUIET_SKIP_REASON) + next_run = next_after_snapshot_quiet_window(float(finished)) + _record_task( + repo, + name, + ok=False, + skipped=True, + started_at=started_at, + finished_at=finished, + payload={"skipped": True, "reason": _QUIET_SKIP_REASON}, + error=_QUIET_SKIP_REASON, + next_due_at=next_run, + ) + return TaskControl( + ok=False, + next_run=next_run, + error=_QUIET_SKIP_REASON, + ) + + +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) + next_run = _seed_retry_soon(float(finished)) + _record_task( + repo, + "snapshot_seed", + ok=False, + skipped=True, + started_at=started_at, + finished_at=finished, + payload={"skipped": True, "reason": reason}, + error=reason, + next_due_at=next_run, + ) + return TaskControl( + ok=False, + next_run=next_run, + error=reason, + ) + + +def _initial_daily_next_run( + repo: Repository, + name: str, + at: str, + *, + now: float | None = None, + retry_seconds: float | None = None, +) -> float: + """进程启动时日更任务的下次运行：优先 ``task:state.due_at``。 + + 无状态时从 ``task:last`` / seed 标记冷启动回填；不得返回 ``0`` 误伤。 + 失败且 ``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, + slot_hhmm=at, + retry_seconds=retry, + ) + + +def _reached_daily_slot(at: str, now: float | None = None) -> bool: + """当前上海时间是否已到达（含）今日 ``at`` 墙钟点。""" + current = shanghai_datetime(now) + hour, minute = parse_hhmm(at) + slot = current.replace(hour=hour, minute=minute, second=0, microsecond=0) + 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) + dt = shanghai_datetime(now) + slot = dt.replace(hour=hour, minute=minute, second=0, microsecond=0).timestamp() + if slot > now: + return float(slot) + return float(next_daily_run(now, snapshot_at)) + + +def _enqueue_evening_seal( + repo: Repository, + *, + stat_date: str, + snapshot_at: str, + now: float, +) -> None: + """seed 成功后：当晚 snapshot / fans 记 pending，due_at=当日 snapshot_at。""" + current = shanghai_datetime(now) + hour, minute = parse_hhmm(snapshot_at) + slot_ts = current.replace( + hour=hour, minute=minute, second=0, microsecond=0, + ).timestamp() + due = slot_ts if slot_ts > now else now + task_state_mod.ensure_pending( + repo, "snapshot", due_at=due, stat_date=stat_date, now=now, + ) + task_state_mod.ensure_pending( + repo, "fans", due_at=due, stat_date=stat_date, now=now, + ) def _client(credentials, logger: JsonLogger, *, pacer=None, retries: int = 3) -> BilibiliClient: @@ -297,51 +594,124 @@ def cmd_snapshot(args, logger: JsonLogger) -> int: ) service = SnapshotService(store, repo, client, logger=logger) started = int(time.time()) - _mark_task_running(repo, "snapshot", started_at=started) + hh, mm = parse_hhmm(cfg.schedule.snapshot_at) + snapshot_at = f"{hh:02d}:{mm:02d}" + _mark_task_running( + repo, "snapshot", started_at=started, + 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( repo, "snapshot", ok=False, started_at=started, - finished_at=int(time.time()), error=str(exc), + finished_at=finished, error=str(exc), + next_due_at=_retry_soon(float(finished)), + stat_date=shanghai_date(float(started)), ) raise video_payload = video_summary.as_dict() video_payload["database"] = str(cfg.storage.database_path) video_payload["snapshot_counts"] = repo.snapshot_counts(video_summary.stat_date) video_payload["counts"] = repo.counts() + video_finished = int(video_summary.finished_at or time.time()) video_ok = bool(video_summary.completed and not video_summary.backup_error) - _record_task( - repo, "snapshot", ok=video_ok, - started_at=int(video_summary.started_at or started), - finished_at=int(video_summary.finished_at or time.time()), - payload=video_payload, - error=video_summary.backup_error, - ) + if video_summary.reason == "seeded_pending_seal": + evening = _evening_due_at(float(started), snapshot_at) + _enqueue_evening_seal( + repo, + stat_date=video_summary.stat_date, + snapshot_at=snapshot_at, + now=float(video_finished), + ) + _record_task( + repo, "snapshot", ok=False, skipped=True, + started_at=int(video_summary.started_at or started), + finished_at=video_finished, + payload=video_payload, + error="seeded_pending_seal", + next_due_at=evening, + stat_date=video_summary.stat_date, + ) + elif video_ok: + _record_task( + repo, "snapshot", ok=True, + started_at=int(video_summary.started_at or started), + finished_at=video_finished, + payload=video_payload, + error=video_summary.backup_error, + next_due_at=next_daily_run(float(video_finished), snapshot_at), + stat_date=video_summary.stat_date, + ) + else: + _record_task( + repo, "snapshot", ok=False, + started_at=int(video_summary.started_at or started), + finished_at=video_finished, + payload=video_payload, + error=video_summary.backup_error or "snapshot_incomplete", + next_due_at=_retry_soon(float(video_finished)), + stat_date=video_summary.stat_date, + ) fans_started = int(time.time()) - _mark_task_running(repo, "fans", started_at=fans_started) + _mark_task_running( + repo, "fans", started_at=fans_started, + stat_date=video_summary.stat_date, + ) try: fans_summary = service.run_fans_if_needed( date=video_summary.stat_date, force=args.force, ) except Exception as exc: # noqa: BLE001 + finished = int(time.time()) _record_task( repo, "fans", ok=False, started_at=fans_started, - finished_at=int(time.time()), error=str(exc), + finished_at=finished, error=str(exc), + next_due_at=_retry_soon(float(finished)), + stat_date=video_summary.stat_date, ) raise fans_payload = fans_summary.as_dict() + fans_finished = int(fans_summary.finished_at or time.time()) fans_ok = fans_summary.fans_failed == 0 - _record_task( - repo, "fans", ok=fans_ok, - started_at=int(fans_summary.started_at or fans_started), - finished_at=int(fans_summary.finished_at or time.time()), - payload=fans_payload, - error="；".join( - err for err in fans_summary.errors if str(err).startswith("fans ") - ) or None, - ) + if fans_summary.reason == "seeded_pending_seal": + evening = _evening_due_at(float(fans_started), snapshot_at) + _record_task( + repo, "fans", ok=False, skipped=True, + started_at=int(fans_summary.started_at or fans_started), + finished_at=fans_finished, + payload=fans_payload, + error="seeded_pending_seal", + next_due_at=evening, + stat_date=video_summary.stat_date, + ) + elif fans_ok: + _record_task( + repo, "fans", ok=True, + started_at=int(fans_summary.started_at or fans_started), + finished_at=fans_finished, + payload=fans_payload, + next_due_at=next_daily_run(float(fans_finished), snapshot_at), + stat_date=video_summary.stat_date, + ) + else: + _record_task( + repo, "fans", ok=False, + started_at=int(fans_summary.started_at or fans_started), + finished_at=fans_finished, + payload=fans_payload, + error="；".join( + err for err in fans_summary.errors if str(err).startswith("fans ") + ) or None, + next_due_at=_retry_soon(float(fans_finished)), + stat_date=video_summary.stat_date, + ) payload = { **video_payload, "fans_ok": fans_summary.fans_ok, @@ -354,6 +724,107 @@ 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), + } + finished = int(time.time()) + _record_task( + repo, + "snapshot_seed", + ok=False, + skipped=True, + started_at=started, + finished_at=finished, + payload=payload, + error=_SEED_SKIP_SNAPSHOT_RUNNING, + next_due_at=_seed_retry_soon(float(finished)), + stat_date=payload["stat_date"], + ) + _print(payload) + return 1 + # seed 只读本地库，不调 B 站；占位 client 满足构造约束。 + service = SnapshotService(store, repo, object(), logger=logger) + seed_date = target or shanghai_date() + _mark_task_running( + repo, "snapshot_seed", started_at=started, stat_date=seed_date, + ) + try: + summary = service.run_seed_from_last_completed(date=target) + except Exception as exc: # noqa: BLE001 + finished = int(time.time()) + _record_task( + repo, "snapshot_seed", ok=False, started_at=started, + finished_at=finished, error=str(exc), + next_due_at=_seed_retry_soon(float(finished)), + stat_date=seed_date, + ) + raise + payload = summary.as_dict() + payload["database"] = str(cfg.storage.database_path) + finished = int(summary.finished_at or time.time()) + hh, mm = parse_hhmm(cfg.schedule.seed_at) + seed_at = f"{hh:02d}:{mm:02d}" + sh, sm = parse_hhmm(cfg.schedule.snapshot_at) + snapshot_at = f"{sh:02d}:{sm:02d}" + 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, + next_due_at=_seed_retry_soon(float(finished)), + stat_date=summary.stat_date or seed_date, + ) + _print(payload) + return 1 + next_due = next_daily_run(float(finished), seed_at) + _record_task( + repo, + "snapshot_seed", + ok=True, + started_at=int(summary.started_at or started), + finished_at=finished, + payload=payload, + next_due_at=next_due, + stat_date=summary.stat_date or seed_date, + ) + if summary.reason == "seeded": + _enqueue_evening_seal( + repo, + stat_date=summary.stat_date or seed_date, + snapshot_at=snapshot_at, + now=float(finished), + ) + _print(payload) + return 0 + finally: + db.close() + + def _telegram(cfg, logger: JsonLogger): if not cfg.telegram.enabled: return None @@ -385,24 +856,34 @@ def cmd_discover(args, logger: JsonLogger) -> int: logger=logger, ) started = int(time.time()) + interval = float(cfg.schedule.discover_minutes * 60) _mark_task_running(repo, "discovery", started_at=started) try: summary = service.run() except Exception as exc: # noqa: BLE001 + finished = int(time.time()) _record_task( repo, "discovery", ok=False, started_at=started, - finished_at=int(time.time()), error=str(exc), + finished_at=finished, error=str(exc), + next_due_at=_retry_soon(float(finished)), ) raise payload = summary.as_dict() payload["candidate_total"] = len(repo.list_candidates()) payload["database"] = str(cfg.storage.database_path) + finished = int(summary.finished_at or time.time()) + ok = not summary.errors + next_due = ( + interval_next_run(float(finished), interval) if ok + else _retry_soon(float(finished)) + ) _record_task( - repo, "discovery", ok=not summary.errors, + repo, "discovery", ok=ok, started_at=int(summary.started_at or started), - finished_at=int(summary.finished_at or time.time()), + finished_at=finished, payload=payload, error="；".join(summary.errors) if summary.errors else None, + next_due_at=next_due, ) _print(payload) return 0 if not summary.errors else 1 @@ -504,173 +985,377 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int: logger=logger, ) + hh, mm = parse_hhmm(cfg.schedule.snapshot_at) + snapshot_at = f"{hh:02d}:{mm:02d}" + sh, sm = parse_hhmm(cfg.schedule.seed_at) + seed_at = f"{sh:02d}:{sm:02d}" + recent_minutes = int(store.get().schedule.recent_snapshot_minutes) + discover_minutes = int(store.get().schedule.discover_minutes) + def run_discover(): started = int(time.time()) + if in_snapshot_quiet_window(float(started)): + return _quiet_window_defer( + repo, logger, "discovery", started_at=started, + ) _mark_task_running(repo, "discovery", started_at=started) + interval = float(store.get().schedule.discover_minutes * 60) try: summary = discovery_service.run() - except Exception as exc: # noqa: BLE001 记录后交给调度器隔离 + except Exception as exc: # noqa: BLE001 记录后补跑，不抛给次日 + finished = int(time.time()) + next_run = _retry_soon(float(finished)) _record_task( repo, "discovery", ok=False, started_at=started, - finished_at=int(time.time()), error=str(exc), + finished_at=finished, error=str(exc), + next_due_at=next_run, + ) + return TaskControl( + ok=False, next_run=next_run, error=str(exc), ) - raise payload = summary.as_dict() + ok = not summary.errors + finished = int(summary.finished_at or time.time()) + next_run = ( + interval_next_run(float(finished), interval) if ok + else _retry_soon(float(finished)) + ) _record_task( - repo, "discovery", ok=not summary.errors, + repo, "discovery", ok=ok, started_at=summary.started_at, - finished_at=summary.finished_at or int(time.time()), + finished_at=finished, payload=payload, error="；".join(summary.errors) if summary.errors else None, + next_due_at=next_run, ) last["discover"] = summary logger.info("discovery.summary", **payload) - return summary + return TaskControl(ok=ok, next_run=next_run) def run_cleanup(): started = int(time.time()) - removed = repo.delete_expired_candidates( - ttl_days=cfg.schedule.candidate_ttl_days - ) - expired = repo.expire_fav_actions() + if in_snapshot_quiet_window(float(started)): + return _quiet_window_defer( + repo, logger, "candidate_cleanup", started_at=started, + ) + try: + removed = repo.delete_expired_candidates( + ttl_days=cfg.schedule.candidate_ttl_days + ) + expired = repo.expire_fav_actions() + except Exception as exc: # noqa: BLE001 记录后约 5 分钟补跑 + finished = int(time.time()) + next_run = _retry_soon(float(finished)) + _record_task( + repo, "candidate_cleanup", ok=False, started_at=started, + finished_at=finished, error=str(exc), + next_due_at=next_run, + ) + return TaskControl( + ok=False, next_run=next_run, error=str(exc), + ) if removed or expired: repo.bump_cache_version("videos") + finished = int(time.time()) logger.info( "cleanup.done", candidates_removed=removed, fav_actions_expired=expired, ) + next_run = interval_next_run(float(finished), 86400.0) _record_task( repo, "candidate_cleanup", ok=True, started_at=started, - finished_at=int(time.time()), + finished_at=finished, payload={ "candidates_removed": removed, "fav_actions_expired": expired, }, + next_due_at=next_run, ) - return { - "candidates_removed": removed, - "fav_actions_expired": expired, - } + return TaskControl(ok=True, next_run=next_run) def run_snapshot(): + """全站快照：不因其它任务 running 而 skip；失败/未完成则短间隔补跑。""" started = int(time.time()) - if _is_task_running(repo, "snapshot_recent"): - logger.info( - "snapshot.skip", reason="snapshot_recent_running", - ) - _record_task( - repo, "snapshot", ok=True, started_at=started, - finished_at=int(time.time()), - payload={"skipped": True, "reason": "snapshot_recent_running"}, - ) - return None - _mark_task_running(repo, "snapshot", started_at=started) + today = shanghai_date(float(started)) + stat_date = task_state_mod.resume_stat_date( + repo, "snapshot", fallback=today, + ) + _mark_task_running( + repo, "snapshot", started_at=started, stat_date=stat_date, + ) try: - # 晚间全量：若当日是午夜 seed 壳，必须 force 打穿 ok=1 克隆行。 - today = shanghai_date() - force_seed = repo.kv_get( - SnapshotService.seeded_key(today) - ) == "1" - summary = snapshot_service.run_if_needed(force=force_seed) + # seed 壳 force：跨天未完成（stat_date < today）立刻打穿； + # 同日仅到达 snapshot_at 后才 force。due_at 管「会不会跑」。 + seeded = repo.kv_get(SnapshotService.seeded_key(stat_date)) == "1" + 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, + 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)) _record_task( repo, "snapshot", ok=False, started_at=started, - finished_at=int(time.time()), error=str(exc), + finished_at=finished, error=str(exc), + next_due_at=next_run, stat_date=stat_date, + ) + return TaskControl( + ok=False, next_run=next_run, error=str(exc), ) - raise payload = summary.as_dict() ok = bool(summary.completed and not summary.backup_error) + finished = int(summary.finished_at or time.time()) + # 白天误碰到 seed 壳：延到当晚 seal，禁止 5 分钟空转。 + if ( + not ok + and getattr(summary, "reason", None) == "seeded_pending_seal" + ): + current = shanghai_datetime(float(finished)) + hour, minute = parse_hhmm(snapshot_at) + slot_ts = current.replace( + hour=hour, minute=minute, second=0, microsecond=0, + ).timestamp() + next_run = slot_ts if slot_ts > float(finished) else _retry_soon( + float(finished), + ) + _record_task( + repo, "snapshot", ok=False, skipped=True, + started_at=summary.started_at, finished_at=finished, + payload=payload, error="seeded_pending_seal", + next_due_at=next_run, + stat_date=summary.stat_date or stat_date, + ) + last["snapshot"] = summary + logger.info("snapshot.summary", **payload) + return TaskControl( + ok=False, next_run=next_run, error="seeded_pending_seal", + ) + next_run = ( + next_daily_run(float(finished), snapshot_at) if ok + else _retry_soon(float(finished)) + ) + progress = { + "total": summary.total, + "success": summary.success, + "failed": summary.failed, + "skipped": summary.skipped, + } _record_task( repo, "snapshot", ok=ok, started_at=summary.started_at, - finished_at=summary.finished_at or int(time.time()), + finished_at=finished, payload=payload, error=summary.backup_error, + next_due_at=next_run, + stat_date=summary.stat_date or stat_date, + progress=progress, ) last["snapshot"] = summary logger.info("snapshot.summary", **payload) - return summary + return TaskControl(ok=ok, next_run=next_run) def run_snapshot_recent(): started = int(time.time()) - if _is_task_running(repo, "snapshot"): - logger.info( - "snapshot.recent_skip", reason="snapshot_running", + if in_snapshot_quiet_window(float(started)): + return _quiet_window_defer( + repo, logger, "snapshot_recent", started_at=started, ) - _record_task( - repo, "snapshot_recent", ok=True, started_at=started, - finished_at=int(time.time()), - payload={"skipped": True, "reason": "snapshot_running"}, - ) - return None _mark_task_running(repo, "snapshot_recent", started_at=started) + interval = float(store.get().schedule.recent_snapshot_minutes * 60) try: summary = snapshot_service.run_recent_videos() except Exception as exc: # noqa: BLE001 + finished = int(time.time()) + next_run = _retry_soon(float(finished)) _record_task( repo, "snapshot_recent", ok=False, started_at=started, - finished_at=int(time.time()), error=str(exc), + finished_at=finished, error=str(exc), + next_due_at=next_run, + ) + return TaskControl( + ok=False, next_run=next_run, error=str(exc), ) - raise payload = summary.as_dict() ok = summary.failed == 0 error = "；".join(summary.errors) if summary.errors else None + finished = int(summary.finished_at or time.time()) + next_run = ( + interval_next_run(float(finished), interval) if ok + else _retry_soon(float(finished)) + ) _record_task( repo, "snapshot_recent", ok=ok, started_at=summary.started_at, - finished_at=summary.finished_at or int(time.time()), + finished_at=finished, payload=payload, error=error, + next_due_at=next_run, ) last["snapshot_recent"] = summary logger.info("snapshot.recent_summary", **payload) - return summary + return TaskControl(ok=ok, next_run=next_run) def run_fans(): + """粉丝更新短耗时：不避开任何任务，可与全量并行。""" started = int(time.time()) - _mark_task_running(repo, "fans", started_at=started) + today = shanghai_date(float(started)) + stat_date = task_state_mod.resume_stat_date( + repo, "fans", fallback=today, + ) + _mark_task_running( + repo, "fans", started_at=started, stat_date=stat_date, + ) try: - today = shanghai_date() - force_seed = repo.kv_get( - SnapshotService.fans_seeded_key(today) - ) == "1" - summary = snapshot_service.run_fans_if_needed(force=force_seed) + seeded = ( + repo.kv_get(SnapshotService.fans_seeded_key(stat_date)) == "1" + ) + 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)) _record_task( repo, "fans", ok=False, started_at=started, - finished_at=int(time.time()), error=str(exc), + finished_at=finished, error=str(exc), + next_due_at=next_run, stat_date=stat_date, + ) + return TaskControl( + ok=False, next_run=next_run, error=str(exc), ) - raise payload = summary.as_dict() + finished = int(summary.finished_at or time.time()) + if getattr(summary, "reason", None) == "seeded_pending_seal": + current = shanghai_datetime(float(finished)) + hour, minute = parse_hhmm(snapshot_at) + slot_ts = current.replace( + hour=hour, minute=minute, second=0, microsecond=0, + ).timestamp() + next_run = slot_ts if slot_ts > float(finished) else _retry_soon( + float(finished), + ) + _record_task( + repo, "fans", ok=False, skipped=True, + started_at=summary.started_at, finished_at=finished, + payload=payload, error="seeded_pending_seal", + next_due_at=next_run, + stat_date=summary.stat_date or stat_date, + ) + last["fans"] = summary + logger.info("fans.summary", **payload) + return TaskControl( + ok=False, next_run=next_run, error="seeded_pending_seal", + ) ok = summary.fans_failed == 0 error = "；".join( err for err in summary.errors if str(err).startswith("fans ") ) or None + next_run = ( + next_daily_run(float(finished), snapshot_at) if ok + else _retry_soon(float(finished)) + ) _record_task( repo, "fans", ok=ok, started_at=summary.started_at, - finished_at=summary.finished_at or int(time.time()), + finished_at=finished, payload=payload, error=error, + next_due_at=next_run, + stat_date=summary.stat_date or stat_date, ) last["fans"] = summary logger.info("fans.summary", **payload) - return summary + return TaskControl(ok=ok, next_run=next_run) def run_snapshot_seed(): started = int(time.time()) - _mark_task_running(repo, "snapshot_seed", started_at=started) + today = shanghai_date(float(started)) + # 全站快照占用时不执行、不记成功；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, stat_date=today, + ) try: - summary = snapshot_service.run_seed_from_last_completed() + summary = snapshot_service.run_seed_from_last_completed(date=today) except Exception as exc: # noqa: BLE001 + finished = int(time.time()) + next_run = _seed_retry_soon(float(finished)) _record_task( repo, "snapshot_seed", ok=False, started_at=started, - finished_at=int(time.time()), error=str(exc), + finished_at=finished, error=str(exc), + next_due_at=next_run, stat_date=today, + ) + return TaskControl( + ok=False, + next_run=next_run, + error=str(exc), ) - raise payload = summary.as_dict() - _record_task( - repo, "snapshot_seed", ok=True, started_at=summary.started_at, - finished_at=summary.finished_at or int(time.time()), - payload=payload, - ) + finished = int(summary.finished_at or time.time()) last["snapshot_seed"] = summary logger.info("snapshot.seed_summary", **payload) - return summary + 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 + next_run = next_daily_run(float(finished), seed_at) + _record_task( + repo, "snapshot_seed", ok=True, started_at=summary.started_at, + finished_at=finished, + payload=payload, + next_due_at=next_run, + stat_date=summary.stat_date or today, + ) + if summary.reason == "seeded": + _enqueue_evening_seal( + repo, + stat_date=summary.stat_date or today, + snapshot_at=snapshot_at, + now=float(finished), + ) + return TaskControl(ok=True, next_run=next_run) def run_config_watch(): before = store.revision @@ -679,49 +1364,84 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int: logger.info("config.reloaded", revision=store.revision) return {"revision": store.revision} - hh, mm = parse_hhmm(cfg.schedule.snapshot_at) - snapshot_at = f"{hh:02d}:{mm:02d}" - sh, sm = parse_hhmm(cfg.schedule.seed_at) - seed_at = f"{sh:02d}:{sm:02d}" - recent_minutes = int(store.get().schedule.recent_snapshot_minutes) + # --once：next_run=0 立刻跑；常驻：一律从 task:state.due_at 恢复，禁止日更 next_run=0。 + now_ts = time.time() + if args.once: + seed_next = snapshot_next = fans_next = 0.0 + recent_next = discover_next = cleanup_next = 0.0 + else: + seed_next = _initial_daily_next_run( + repo, "snapshot_seed", seed_at, now=now_ts, + ) + snapshot_next = _initial_daily_next_run( + repo, "snapshot", snapshot_at, now=now_ts, + ) + fans_next = _initial_daily_next_run( + repo, "fans", snapshot_at, now=now_ts, + ) + recent_next = task_state_mod.startup_next_run( + repo, "snapshot_recent", wall_due_at=now_ts, now=now_ts, + ) + discover_next = task_state_mod.startup_next_run( + repo, "discovery", wall_due_at=now_ts, now=now_ts, + ) + 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( "snapshot_seed", 86400.0, run_snapshot_seed, - next_run=0.0, - next_run_fn=lambda started: next_daily_run(started, seed_at), + next_run=seed_next, ), ScheduledTask( "snapshot", 86400.0, run_snapshot, - next_run=0.0, - next_run_fn=lambda started: next_daily_run(started, snapshot_at), + next_run=snapshot_next, ), ScheduledTask( "fans", 86400.0, run_fans, - next_run=0.0, - next_run_fn=lambda started: next_daily_run(started, snapshot_at), + next_run=fans_next, ), ScheduledTask( "snapshot_recent", float(recent_minutes * 60), run_snapshot_recent, + next_run=recent_next, ), ScheduledTask( "discovery", - float(cfg.schedule.discover_minutes * 60), + float(discover_minutes * 60), run_discover, + next_run=discover_next, + ), + ScheduledTask( + "candidate_cleanup", + 86400.0, + run_cleanup, + next_run=cleanup_next, ), - ScheduledTask("candidate_cleanup", 86400.0, run_cleanup), ScheduledTask("config_watch", 60.0, run_config_watch), ], logger=logger, + due_reader=_due_reader, ) + if serve_web and not args.once: from .web.server import create_server, serve_forever @@ -863,6 +1583,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/scheduler.py b/reaction_board/scheduler.py index 03a2b1c..516571e 100644 --- a/reaction_board/scheduler.py +++ b/reaction_board/scheduler.py @@ -3,6 +3,9 @@ P0 只注册收藏夹同步任务；P1/P3 会在同一调度器上追加快照、粉丝和发现任务。 调度只负责“什么时候运行”，具体任务保持幂等，并在自己的服务层处理重试。 + +任务 ``fn`` 可返回 :class:`TaskControl` 以覆盖 ``ok`` / ``next_run`` +（用于静默窗延后、失败补跑等）；未返回时仍按 ``next_run_fn`` / ``interval`` 排期。 """ from __future__ import annotations @@ -12,6 +15,15 @@ from dataclasses import dataclass from typing import Any, Callable, Sequence +@dataclass +class TaskControl: + """任务可选返回值：覆盖成功标记与下次运行时间。""" + + ok: bool = True + next_run: float | None = None + error: str | None = None + + @dataclass class ScheduledTask: name: str @@ -47,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("至少需要一个调度任务") @@ -57,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: @@ -66,27 +81,56 @@ 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 error: str | None = None + control: TaskControl | None = None try: - task.fn() + raw = task.fn() except Exception as exc: # noqa: BLE001 单任务失败不终止调度 error = str(exc) - self._log("error", "scheduler.task_error", task=task.name, error=error) else: - self._log("info", "scheduler.task_ok", task=task.name) + if isinstance(raw, TaskControl): + control = raw finished = self._clock() - if task.next_run_fn is not None: + if control is not None and control.next_run is not None: + task.next_run = float(control.next_run) + elif task.next_run_fn is not None: task.next_run = task.next_run_fn(started) else: task.next_run = started + task.interval + ok = error is None + if control is not None: + ok = bool(control.ok) and error is None + if control.error and error is None: + error = control.error + # Skip/fail ≠ success：先定 ok/error，再记 scheduler 日志。 + if ok: + self._log("info", "scheduler.task_ok", task=task.name) + elif control is not None: + self._log( + "warning", + "scheduler.task_fail", + task=task.name, + error=error, + ) + else: + self._log( + "error", + "scheduler.task_error", + task=task.name, + error=error, + ) result = TaskResult( name=task.name, started_at=started, finished_at=finished, - ok=error is None, + ok=ok, error=error, ) results.append(result) diff --git a/reaction_board/services/snapshot.py b/reaction_board/services/snapshot.py index da8bc8e..be8204e 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, } @@ -140,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) @@ -153,7 +163,10 @@ class SnapshotService: raise ValueError(f"日期格式必须是 YYYY-MM-DD: {stat_date!r}") seeded = self.repo.kv_get(self.seeded_key(stat_date)) == "1" if force: - return self.run_videos(date=stat_date, force=True) + ks = True if keep_scraped is None else bool(keep_scraped) + return self.run_videos( + date=stat_date, force=True, keep_scraped=ks, + ) completed = self.repo.kv_get(self.completed_key(stat_date)) == "1" if seeded and not completed: @@ -167,6 +180,7 @@ class SnapshotService: finished_at=now, completed=False, already_completed=False, + reason="seeded_pending_seal", ) pending = self.repo.pending_snapshot_aids(stat_date) if completed and not pending: @@ -203,8 +217,19 @@ class SnapshotService: summary = self.run_videos(date=date, force=force) return self.run_fans(date=summary.stat_date, force=force, summary=summary) - def run_videos(self, *, date: str | None = None, force: bool = False) -> SnapshotSummary: - """采集视频七项指标并在全部成功后标记完成日 / 备份。""" + def run_videos( + self, + *, + date: str | None = None, + force: bool = False, + keep_scraped: bool = False, + ) -> SnapshotSummary: + """采集视频七项指标并在全部成功后标记完成日 / 备份。 + + ``force``：忽略完成标记，重跑。 + ``keep_scraped``：force 时仍跳过非 seed 的 ok=1（seal 断点续跑）； + CLI ``--force`` 不传此标志，会重采全部成功行。 + """ cfg = self.config_store.get() client = self._client() now = self._now() @@ -225,11 +250,19 @@ class SnapshotService: 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: - # 断点续跑：同日已成功则跳过。小时任务刷新过的行也是 ok=1， - # 自然覆盖「1 小时内 captured_at 已更新则不再次更新」的要求。 - summary.skipped += 1 - continue + if existing is not None and int(existing.get("ok") or 0) == 1: + src = str(existing.get("src") or "") + if keep_scraped and src == "seed": + # seal 续跑：必须打穿克隆壳 + pass + elif keep_scraped and src != "seed": + summary.skipped += 1 + continue + elif not force: + # 普通续跑：同日已成功则跳过（含小时任务刚刷新的） + summary.skipped += 1 + continue + # force 且非 keep_scraped：CLI 全量重采，不 skip stat, error, unavailable = self._fetch_view(client, aid) if unavailable: @@ -423,6 +456,7 @@ class SnapshotService: completed=False, already_completed=False, fans_skipped=len(cfg.track_fans), + reason="seeded_pending_seal", ) return self.run_fans(date=stat_date, force=True) if not self._pending_fan_mids(stat_date, cfg.track_fans): @@ -536,15 +570,8 @@ class SnapshotService: raise ValueError(f"日期格式必须是 YYYY-MM-DD: {stat_date!r}") summary = SnapshotSummary(stat_date=stat_date, started_at=now) - source = self.repo.kv_get("snapshot:last_completed") - if not source or source >= stat_date: - self._log( - "info", "snapshot.seed_skip", - stat_date=stat_date, reason="no_source", source=source, - ) - summary.finished_at = self._now() - summary.completed = True - return summary + # 目标日已完成：先于 no_source。全站快照当天完成后 + # last_completed == stat_date，不可误判为无源再 10 分钟重试。 if self.repo.kv_get(self.completed_key(stat_date)) == "1": self._log( "info", "snapshot.seed_skip", @@ -553,6 +580,17 @@ class SnapshotService: summary.finished_at = self._now() summary.completed = True summary.already_completed = True + summary.reason = "already_completed" + return summary + source = self.repo.kv_get("snapshot:last_completed") + if not source or source >= stat_date: + self._log( + "info", "snapshot.seed_skip", + stat_date=stat_date, reason="no_source", source=source, + ) + summary.finished_at = self._now() + summary.completed = False + summary.reason = "no_source" return summary if self.repo.kv_get(self.seeded_key(stat_date)) == "1": self._log( @@ -562,6 +600,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 +622,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/task_state.py b/reaction_board/task_state.py new file mode 100644 index 0000000..75b13cc --- /dev/null +++ b/reaction_board/task_state.py @@ -0,0 +1,438 @@ +# -*- coding: utf-8 -*- +"""调度任务状态持久化（跨进程重启 / 容器重建）。 + +每条任务一条 kv 记录 ``task:state:{name}``，字段见 :class:`TaskState`。 +调度器只认落盘的 ``due_at`` / ``status`` / ``stat_date`` / ``progress``， +不再靠启动时把 ``next_run=0`` 猜「要不要立刻跑」。 +""" + +from __future__ import annotations + +import json +import time +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" +STATUS_DONE = "done" +STATUS_FAILED = "failed" +STATUS_DEFERRED = "deferred" + +_ACTIVE = frozenset({STATUS_PENDING, STATUS_RUNNING, STATUS_FAILED, STATUS_DEFERRED}) + + +@dataclass +class TaskState: + name: str + status: str + due_at: float + stat_date: str | None = None + progress: dict[str, Any] | None = None + updated_at: float = 0.0 + error: str | None = None + + def as_dict(self) -> dict[str, Any]: + payload = asdict(self) + # JSON 友好：去掉 None progress 也可保留 + return payload + + +def state_key(name: str) -> str: + return f"task:state:{name}" + + +def load_state(repo: Repository, name: str) -> TaskState | None: + raw = repo.kv_get(state_key(name)) + if not raw: + return None + try: + data = json.loads(raw) + except (TypeError, ValueError, json.JSONDecodeError): + return None + if not isinstance(data, dict) or not data.get("name"): + return None + try: + due_at = float(data.get("due_at") or 0) + updated_at = float(data.get("updated_at") or 0) + except (TypeError, ValueError): + return None + progress = data.get("progress") + if progress is not None and not isinstance(progress, dict): + progress = None + return TaskState( + name=str(data.get("name") or name), + status=str(data.get("status") or STATUS_PENDING), + due_at=due_at, + stat_date=(str(data["stat_date"]) if data.get("stat_date") else None), + progress=progress, + updated_at=updated_at, + error=(str(data["error"]) if data.get("error") else None), + ) + + +def save_state(repo: Repository, state: TaskState) -> None: + repo.kv_set(state_key(state.name), json.dumps(state.as_dict(), ensure_ascii=False)) + + +def _now(now: float | None) -> float: + return time.time() if now is None else float(now) + + +def recover_stale_running( + repo: Repository, + name: str, + *, + now: float | None = None, +) -> TaskState | None: + """进程已死但仍为 ``running``：改为 ``pending``，保留 progress / stat_date。 + + ``due_at`` 取 ``min(原 due_at, now)``，使过期/中断任务可立即续跑， + 未到期的（极少见）仍尊重原 due_at。 + """ + current = _now(now) + state = load_state(repo, name) + if state is None or state.status != STATUS_RUNNING: + return state + state.status = STATUS_PENDING + state.due_at = min(float(state.due_at or current), current) + state.updated_at = current + save_state(repo, state) + return state + + +def mark_running( + repo: Repository, + name: str, + *, + stat_date: str | None, + now: float | None = None, + progress: Mapping[str, Any] | None = None, + due_at: float | None = None, +) -> TaskState: + current = _now(now) + prev = load_state(repo, name) + state = TaskState( + name=name, + status=STATUS_RUNNING, + due_at=float(due_at if due_at is not None else (prev.due_at if prev else current)), + stat_date=stat_date if stat_date is not None else (prev.stat_date if prev else None), + progress=dict(progress) if progress is not None else (prev.progress if prev else None), + updated_at=current, + error=None, + ) + save_state(repo, state) + return state + + +def mark_done( + repo: Repository, + name: str, + *, + due_at: float, + now: float | None = None, + stat_date: str | None = None, + progress: Mapping[str, Any] | None = None, +) -> TaskState: + """成功：记 done，并写入下一次 ``due_at``（下一条墙钟/间隔）。""" + current = _now(now) + state = TaskState( + name=name, + status=STATUS_DONE, + due_at=float(due_at), + stat_date=stat_date, + progress=dict(progress) if progress is not None else None, + updated_at=current, + error=None, + ) + save_state(repo, state) + return state + + +def mark_failed( + repo: Repository, + name: str, + *, + due_at: float, + now: float | None = None, + stat_date: str | None = None, + progress: Mapping[str, Any] | None = None, + error: str | None = None, +) -> TaskState: + current = _now(now) + prev = load_state(repo, name) + state = TaskState( + name=name, + status=STATUS_FAILED, + due_at=float(due_at), + stat_date=stat_date if stat_date is not None else (prev.stat_date if prev else None), + progress=( + dict(progress) if progress is not None + else (prev.progress if prev else None) + ), + updated_at=current, + error=error, + ) + save_state(repo, state) + return state + + +def mark_deferred( + repo: Repository, + name: str, + *, + due_at: float, + now: float | None = None, + stat_date: str | None = None, + progress: Mapping[str, Any] | None = None, + error: str | None = None, +) -> TaskState: + """主动延后（静默窗 / seed 避让等）：保留进度，写未来 due_at。""" + current = _now(now) + prev = load_state(repo, name) + state = TaskState( + name=name, + status=STATUS_DEFERRED, + due_at=float(due_at), + stat_date=stat_date if stat_date is not None else (prev.stat_date if prev else None), + progress=( + dict(progress) if progress is not None + else (prev.progress if prev else None) + ), + updated_at=current, + error=error, + ) + save_state(repo, state) + return state + + +def ensure_pending( + repo: Repository, + name: str, + *, + due_at: float, + stat_date: str | None, + now: float | None = None, + overwrite_done: bool = True, +) -> TaskState: + """保证存在一条 pending（如 seed 成功后挂上当晚 seal）。 + + 同日未完成：取更早的 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: + 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: + return prev + prev.due_at = due + prev.stat_date = sd + prev.status = STATUS_PENDING + prev.updated_at = current + save_state(repo, prev) + return prev + if prev is not None and prev.status == STATUS_RUNNING: + return prev + if prev is not None and prev.status == STATUS_DONE and not overwrite_done: + return 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 + + +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`` 是唯一时间门闩（失败延时不会被提前） + """ + 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) + due = float(state.due_at or 0) + if due <= 0: + return float(wall_due_at) + # 无论 status：due_at 是唯一时间门闩 + if due > current: + return due + # 已到期：未完成要续跑；done 也表示下一轮已到期 + return due + + +def resume_stat_date(repo: Repository, name: str, *, fallback: str) -> str: + """续跑时优先用状态里的归属日，避免跨日重启跑错天。""" + state = load_state(repo, name) + if state is not None and state.stat_date and state.status in _ACTIVE: + return state.stat_date + return fallback diff --git a/reaction_board/timeutil.py b/reaction_board/timeutil.py index 74e02e9..30c234a 100644 --- a/reaction_board/timeutil.py +++ b/reaction_board/timeutil.py @@ -57,3 +57,39 @@ def day_bounds(date: str) -> tuple[float, float]: 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() + + +# 全站快照静默窗：上海时区 [23:00, 01:00)。此间 discovery / recent / cleanup 不执行。 +SNAPSHOT_QUIET_START_HOUR = 23 +SNAPSHOT_QUIET_END_HOUR = 1 + + +def in_snapshot_quiet_window(ts: float | None = None) -> bool: + """是否处于全站快照静默窗 [23:00, 01:00)（上海时区，跨日）。""" + dt = shanghai_datetime(ts) + return dt.hour >= SNAPSHOT_QUIET_START_HOUR or dt.hour < SNAPSHOT_QUIET_END_HOUR + + +def next_after_snapshot_quiet_window(ts: float) -> float: + """若 ``ts`` 落在静默窗内，推到窗结束后的 01:00；否则原样返回。""" + if not in_snapshot_quiet_window(ts): + return ts + dt = shanghai_datetime(ts) + if dt.hour >= SNAPSHOT_QUIET_START_HOUR: + target = (dt + timedelta(days=1)).replace( + hour=SNAPSHOT_QUIET_END_HOUR, minute=0, second=0, microsecond=0, + ) + else: + target = dt.replace( + hour=SNAPSHOT_QUIET_END_HOUR, minute=0, second=0, microsecond=0, + ) + if target.timestamp() <= ts: + target += timedelta(days=1) + return target.timestamp() + + +def interval_next_run(now: float, interval: float) -> float: + """``now + interval``，若落入静默窗则推到 01:00 之后。""" + if interval <= 0: + raise ValueError("interval 必须大于 0") + return next_after_snapshot_quiet_window(now + interval) diff --git a/reaction_board/web/queries.py b/reaction_board/web/queries.py index 67a5c13..14b055d 100644 --- a/reaction_board/web/queries.py +++ b/reaction_board/web/queries.py @@ -34,7 +34,14 @@ from ..analytics import ( ) from ..config import ConfigStore from ..database import Database -from ..timeutil import day_bounds, next_daily_run, shanghai_date, shanghai_datetime +from ..timeutil import ( + day_bounds, + interval_next_run, + next_after_snapshot_quiet_window, + next_daily_run, + shanghai_date, + shanghai_datetime, +) from .cache import DASHBOARD_CACHE, TTL_FANS, TTL_VIDEOS, build_cache_key TRACKED_STATUSES = ("active", "unfaved") @@ -1971,9 +1978,27 @@ class DashboardQueries: "fans": int(next_daily_run(now, cfg.schedule.snapshot_at)), "snapshot": int(next_daily_run(now, cfg.schedule.snapshot_at)), "snapshot_seed": int(next_daily_run(now, cfg.schedule.seed_at)), - "discover": int(now + discover_interval), - "snapshot_recent": int(now + recent_interval), + "discover": int(interval_next_run(now, float(discover_interval))), + "snapshot_recent": int(interval_next_run(now, float(recent_interval))), } + retry_seconds = 300 + seed_retry_seconds = 600 + # 优先 task:state.due_at（跨重启唯一时间门闩）；无则回退 task:last 推导。 + state_rows = conn.execute( + "SELECT k, v FROM kv WHERE k LIKE 'task:state:%'" + ).fetchall() + for row in state_rows: + key = str(row["k"]).removeprefix("task:state:") + ui_name = "discover" if key == "discovery" else key + if ui_name not in next_runs: + continue + try: + state = json.loads(str(row["v"])) + due = float(state.get("due_at") or 0) + except (TypeError, ValueError, json.JSONDecodeError): + continue + if due > 0: + next_runs[ui_name] = int(due) for name, key, at in ( ("fans", "fans", cfg.schedule.snapshot_at), ("snapshot", "snapshot", cfg.schedule.snapshot_at), @@ -1981,23 +2006,58 @@ class DashboardQueries: ("discover", "discovery", None), ("snapshot_recent", "snapshot_recent", None), ): + # 已有落盘 due_at 则不再用 last 覆盖 + state_raw = self._kv(conn, f"task:state:{key}") + if state_raw: + try: + if float(json.loads(state_raw).get("due_at") or 0) > 0: + continue + except (TypeError, ValueError, json.JSONDecodeError): + pass record = tasks.get(key) if not (isinstance(record, dict) and record.get("finished_at")): continue finished_at = int(record["finished_at"]) + summary = record.get("summary") if isinstance(record.get("summary"), dict) else {} + skipped = bool(record.get("skipped") or summary.get("skipped")) + failed = record.get("ok") is False if name in {"discover", "snapshot_recent"}: - # 只按上次完成推下一档；进行中由 running 展示，不按墙钟抢跑。 + # 上次完成推下一档；skip/失败短补跑；静默窗推到 01:00 后。 interval = ( recent_interval if name == "snapshot_recent" else discover_interval ) - next_runs[name] = finished_at + interval + if skipped: + next_runs[name] = int( + next_after_snapshot_quiet_window(float(finished_at)) + ) + elif failed: + next_runs[name] = int(finished_at + retry_seconds) + else: + next_runs[name] = int( + interval_next_run(float(finished_at), float(interval)) + ) else: - next_runs[name] = int(next_daily_run(finished_at, at)) + if skipped or failed: + # 日更任务失败/skip：短间隔补跑，绝不按成功推到次日。 + # 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 = [ {"task": name, **record} for name, record in tasks.items() - if isinstance(record, dict) and record.get("ok") is False + if isinstance(record, dict) + and record.get("ok") is False + and not record.get("skipped") + and not ( + isinstance(record.get("summary"), dict) + and record["summary"].get("skipped") + ) ] counts = {str(row["status"]): int(row["n"]) for row in conn.execute( "SELECT status, COUNT(*) AS n FROM video GROUP BY status" diff --git a/reaction_board/web/static/views/system.js b/reaction_board/web/static/views/system.js index f4920d7..1145244 100644 --- a/reaction_board/web/static/views/system.js +++ b/reaction_board/web/static/views/system.js @@ -5,6 +5,15 @@ import { escapeHtml, formatCount, formatDateTime, formatNumber, } from "../format.js"; +function taskResultLabel(record) { + const summary = record.summary && typeof record.summary === "object" ? record.summary : {}; + if (record.skipped || summary.skipped) { + return `跳过`; + } + if (record.ok) return "成功"; + return `失败`; +} + function taskRows(tasks) { const entries = Object.entries(tasks || {}); if (!entries.length) return emptyState("暂无任务记录", "运行一次同步、快照或发现后会写入最近结果。"); @@ -15,12 +24,16 @@ function taskRows(tasks) { ${entries.map(([name, record]) => { const summary = record.summary && typeof record.summary === "object" ? Object.entries(record.summary) - .filter(([key]) => ["added", "removed", "candidates", "notified", "ok", "failed", "total", "completed", "backup_error"].includes(key)) + .filter(([key]) => [ + "added", "removed", "candidates", "notified", "ok", "failed", + "total", "completed", "backup_error", "skipped", "already_completed", + "reason", "elapsed", "success", + ].includes(key)) .map(([key, value]) => `${key}=${value}`).join(" ") : ""; return ` ${escapeHtml(name)} - ${record.ok ? "成功" : `失败`} + ${taskResultLabel(record)} ${escapeHtml(formatDateTime(record.finished_at))} ${escapeHtml(summary || record.error || "—")} `; @@ -100,7 +113,7 @@ export async function renderSystem(root) {
新视频小时同步${running.snapshot_recent ? "运行中" : escapeHtml(formatDateTime(next.snapshot_recent))}
新视频搜索${running.discover || running.discovery ? "运行中" : escapeHtml(formatDateTime(next.discover))}
-

进行中显示「运行中」；空闲时按该任务上次完成时间推算下次运行，不按墙钟抢跑。新视频小时同步与全站快照互斥。

+

进行中显示「运行中」；空闲时按该任务上次完成时间推算下次运行。23:00–01:00（上海）为全站快照窗：发现 / 小时同步 / 候选清理会延后到 01:00 后；粉丝更新不避开。跳过或失败不会记成成功，会短间隔补跑。

diff --git a/tests/test_cli.py b/tests/test_cli.py index b5098cb..b6f369d 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1,352 +1,968 @@ -# -*- coding: utf-8 -*- -"""CLI 入口测试（离线）: uv run python tests/test_cli.py""" - -import contextlib -import io -import json -import os -import sys -import tempfile -from pathlib import Path - -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) -sys.stdout.reconfigure(encoding="utf-8") - -from reaction_board.bilibili.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"): - path = Path(tmp) / "config.json" - path.write_text(json.dumps({ - "bilibili": { - "read": {"mid": 1, "csrf": csrf, "cookie": cookie}, - "write": {"mid": 2, "csrf": "write-csrf", "cookie": "write-cookie"}, - }, - "storage": { - "database_path": str(Path(tmp) / "app.db"), - "backup_dir": str(Path(tmp) / "backup"), - }, - "folders": [ - {"year": 2025, "media_id": 11, "title": "2025"}, - {"year": 2026, "media_id": 22, "title": "2026"}, - ], - "up_blacklist_seed": [101, 202], - }, ensure_ascii=False), encoding="utf-8") - return path - - -def run_cli(argv): - out = io.StringIO() - with contextlib.redirect_stdout(out): - code = main(argv) - return code, out.getvalue() - - -def test_init_and_status(): - with tempfile.TemporaryDirectory() as tmp: - path = write_config(tmp) - code, output = run_cli(["init", "--config", str(path)]) - assert code == 0 - assert '"schema_version": 8' in output - - db = Database(Path(tmp) / "app.db") - db.initialize() - try: - repo = Repository(db) - assert len(repo.list_folders()) == 2 - assert [r["mid"] for r in repo.blacklist()] == [101, 202] - assert repo.counts()["video_total"] == 0 - finally: - db.close() - - code, output = run_cli(["status", "--config", str(path)]) - assert code == 0 - assert '"configured_folders"' in output - assert '"media_count": null' in output - assert '"video_total": 0' in output - - -def test_missing_config(): - with tempfile.TemporaryDirectory() as tmp: - code, output = run_cli(["init", "--config", str(Path(tmp) / "nope.json")]) - assert code == 2 - payload = json.loads(output.strip().splitlines()[-1]) - assert payload["level"] == "error" - assert "缺少配置文件" in payload["error"] - - -def test_ingest_without_refs_is_offline(): - with tempfile.TemporaryDirectory() as tmp: - path = write_config(tmp, cookie="", csrf="") - code, output = run_cli(["ingest", "--config", str(path), "没有视频"]) - assert code == 1 - data = json.loads(output[output.rindex("{\n"):]) - assert data["refs"] == 0 - assert data["results"] == [] - - -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_auth_check_all_and_auth_dispatch(): - with tempfile.TemporaryDirectory() as tmp: - path = write_config(tmp, cookie="", csrf="") - code, output = run_cli(["auth-check", "--account", "all", "--config", str(path)]) - assert code == 1 - payload = json.loads(output) - assert payload["ok"] is False - assert [item["account"] for item in payload["accounts"]] == ["read", "write"] - - original = cli.AuthService - calls = [] - - class FakeAuthService: - def __init__(self, path, *, account, logger): - calls.append((str(path), account)) - - def run(self, *, timeout, no_browser): - assert timeout == 9 and no_browser is True - return 7 - - cli.AuthService = FakeAuthService - try: - code, _ = run_cli([ - "auth", "--account", "write", "--timeout", "9", "--no-browser", "--config", str(path), - ]) - finally: - cli.AuthService = original - assert code == 7 - assert calls == [(str(path), "write")] - - -def test_run_once_offline(): - class FakeClient: - def check_login(self): - return LoginState(True, 1, "tester", 0, "OK") - - class FakeDiscovery: - def __init__(self, *args, **kwargs): - pass - - def run(self): - from reaction_board.services.discovery import DiscoverySummary - return DiscoverySummary(started_at=1, finished_at=1) - - with tempfile.TemporaryDirectory() as tmp: - path = write_config(tmp) - data = json.loads(path.read_text(encoding="utf-8")) - data["folders"] = [] - data["track_fans"] = [] - path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") - - original = cli._client - original_discovery = cli.DiscoveryService - selected = [] - - def fake_client(credentials, *args, **kwargs): - selected.append(credentials) - return FakeClient() - - cli._client = fake_client - cli.DiscoveryService = FakeDiscovery - try: - code, output = run_cli(["run", "--once", "--config", str(path)]) - finally: - cli._client = original - cli.DiscoveryService = original_discovery - assert code == 0 - assert '"name": "snapshot"' in output - assert '"name": "folder_sync"' not in output - assert selected and all(credentials.mid == 1 for credentials in selected) - - db = Database(Path(tmp) / "app.db") - db.initialize() - try: - repo = Repository(db) - for name in ("snapshot", "discovery", "candidate_cleanup"): - assert repo.kv_get(f"task:last:{name}"), name - finally: - db.close() - - -def test_serve_preview_guard(): - with tempfile.TemporaryDirectory() as tmp: - path = write_config(tmp) - assert run_cli(["init", "--config", str(path)])[0] == 0 - db_path = Path(tmp) / "app.db" - try: - cli._assert_preview_database(db_path) - except cli.ConfigError as exc: - assert "preview:seeded" in str(exc) - else: - raise AssertionError("普通数据库不应通过 --preview 校验") - - db = Database(db_path) - db.initialize() - try: - Repository(db).kv_set("preview:seeded", "1") - finally: - db.close() - cli._assert_preview_database(db_path) - - try: - cli._assert_preview_database(Path(tmp) / "missing.db") - except cli.ConfigError as exc: - assert "不存在" in str(exc) - else: - raise AssertionError("缺失数据库应报错") - - -def test_default_command_starts_web_and_scheduler(): - with tempfile.TemporaryDirectory() as tmp: - path = write_config(tmp) - calls = [] - original = cli.cmd_run - - def fake_run(args, logger, *, serve_web=False): - calls.append((args.command, serve_web, args.once)) - return 0 - - cli.cmd_run = fake_run - try: - code = main(["--config", str(path)]) - finally: - cli.cmd_run = original - assert code == 0 - assert calls == [("run", True, False)] - - -def test_snapshot_backup_failure_returns_error_and_retries(): - import reaction_board.services.snapshot as snapshot_module - - 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") - - original = snapshot_module.backup_daily - - def fail_backup(*args, **kwargs): - raise OSError("disk full") - - snapshot_module.backup_daily = fail_backup - try: - code, output = run_cli(["snapshot", "--config", str(path)]) - finally: - snapshot_module.backup_daily = original - assert code == 1 - marker = output.find("{\n") - assert marker >= 0, output - payload = json.loads(output[marker:]) - assert payload["completed"] is True - assert "disk full" in payload["backup_error"] - - code, output = run_cli(["snapshot", "--config", str(path)]) - assert code == 0 - marker = output.find("{\n") - payload = json.loads(output[marker:]) - assert payload["backup_error"] is None - assert Path(payload["backup_path"]).exists() - - -def test_candidates_and_discover_commands(): - from reaction_board.services.discovery import DiscoverySummary - - with tempfile.TemporaryDirectory() as tmp: - path = write_config(tmp) - assert run_cli(["init", "--config", str(path)])[0] == 0 - db = Database(Path(tmp) / "app.db") - db.initialize() - try: - repo = Repository(db) - repo.upsert_candidate( - aid=123, bvid="BV123", title="咻咻满 reaction", - up_mid=100, up_name="UP", pubdate=1, - cover=None, score=3, channel="search:test#p1", - ) - finally: - db.close() - - code, output = run_cli(["candidates", "--config", str(path)]) - assert code == 0 - assert '"count": 1' in output - assert '"aid": 123' in output - - class FakeDiscovery: - def __init__(self, *args, **kwargs): - pass - - def run(self): - return DiscoverySummary( - started_at=1, finished_at=2, candidates=3, notified=1 - ) - - original = cli.DiscoveryService - cli.DiscoveryService = FakeDiscovery - try: - code, output = run_cli(["discover", "--config", str(path)]) - finally: - cli.DiscoveryService = original - assert code == 0 - assert '"candidates": 3' in output - assert '"notified": 1' in output - - -def test_backfill_followers_dry_run(): - with tempfile.TemporaryDirectory() as tmp: - path = write_config(tmp, cookie="", csrf="") - code, output = run_cli([ - "backfill-followers", "--dry-run", "--config", str(path), - ]) - assert code == 0 - assert '"requested": 0' in output - assert '"failed": 0' in output - assert '"dry_run": true' in output - assert '"total": 0' in output - - -def main_test(): - test_init_and_status() - test_missing_config() - test_ingest_without_refs_is_offline() - test_snapshot_offline_empty_db() - test_auth_check_without_cookie() - test_auth_check_all_and_auth_dispatch() - test_run_once_offline() - test_serve_preview_guard() - test_default_command_starts_web_and_scheduler() - 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/错误路径测试全部通过") - - -if __name__ == "__main__": - main_test() +# -*- coding: utf-8 -*- +"""CLI 入口测试（离线）: uv run python tests/test_cli.py""" + +import contextlib +import io +import json +import os +import sys +import tempfile +from pathlib import Path + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +sys.stdout.reconfigure(encoding="utf-8") + +from reaction_board.bilibili.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"): + path = Path(tmp) / "config.json" + path.write_text(json.dumps({ + "bilibili": { + "read": {"mid": 1, "csrf": csrf, "cookie": cookie}, + "write": {"mid": 2, "csrf": "write-csrf", "cookie": "write-cookie"}, + }, + "storage": { + "database_path": str(Path(tmp) / "app.db"), + "backup_dir": str(Path(tmp) / "backup"), + }, + "folders": [ + {"year": 2025, "media_id": 11, "title": "2025"}, + {"year": 2026, "media_id": 22, "title": "2026"}, + ], + "up_blacklist_seed": [101, 202], + }, ensure_ascii=False), encoding="utf-8") + return path + + +def run_cli(argv): + out = io.StringIO() + with contextlib.redirect_stdout(out): + code = main(argv) + return code, out.getvalue() + + +def test_init_and_status(): + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp) + code, output = run_cli(["init", "--config", str(path)]) + assert code == 0 + assert '"schema_version": 8' in output + + db = Database(Path(tmp) / "app.db") + db.initialize() + try: + repo = Repository(db) + assert len(repo.list_folders()) == 2 + assert [r["mid"] for r in repo.blacklist()] == [101, 202] + assert repo.counts()["video_total"] == 0 + finally: + db.close() + + code, output = run_cli(["status", "--config", str(path)]) + assert code == 0 + assert '"configured_folders"' in output + assert '"media_count": null' in output + assert '"video_total": 0' in output + + +def test_missing_config(): + with tempfile.TemporaryDirectory() as tmp: + code, output = run_cli(["init", "--config", str(Path(tmp) / "nope.json")]) + assert code == 2 + payload = json.loads(output.strip().splitlines()[-1]) + assert payload["level"] == "error" + assert "缺少配置文件" in payload["error"] + + +def test_ingest_without_refs_is_offline(): + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, cookie="", csrf="") + code, output = run_cli(["ingest", "--config", str(path), "没有视频"]) + assert code == 1 + data = json.loads(output[output.rindex("{\n"):]) + assert data["refs"] == 0 + assert data["results"] == [] + + +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_auth_check_all_and_auth_dispatch(): + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, cookie="", csrf="") + code, output = run_cli(["auth-check", "--account", "all", "--config", str(path)]) + assert code == 1 + payload = json.loads(output) + assert payload["ok"] is False + assert [item["account"] for item in payload["accounts"]] == ["read", "write"] + + original = cli.AuthService + calls = [] + + class FakeAuthService: + def __init__(self, path, *, account, logger): + calls.append((str(path), account)) + + def run(self, *, timeout, no_browser): + assert timeout == 9 and no_browser is True + return 7 + + cli.AuthService = FakeAuthService + try: + code, _ = run_cli([ + "auth", "--account", "write", "--timeout", "9", "--no-browser", "--config", str(path), + ]) + finally: + cli.AuthService = original + assert code == 7 + assert calls == [(str(path), "write")] + + +def test_run_once_offline(): + class FakeClient: + def check_login(self): + return LoginState(True, 1, "tester", 0, "OK") + + class FakeDiscovery: + def __init__(self, *args, **kwargs): + pass + + def run(self): + from reaction_board.services.discovery import DiscoverySummary + return DiscoverySummary(started_at=1, finished_at=1) + + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp) + data = json.loads(path.read_text(encoding="utf-8")) + data["folders"] = [] + data["track_fans"] = [] + path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") + + original = cli._client + original_discovery = cli.DiscoveryService + selected = [] + + def fake_client(credentials, *args, **kwargs): + selected.append(credentials) + return FakeClient() + + cli._client = fake_client + cli.DiscoveryService = FakeDiscovery + try: + code, output = run_cli(["run", "--once", "--config", str(path)]) + finally: + cli._client = original + cli.DiscoveryService = original_discovery + assert code == 0 + assert '"name": "snapshot"' in output + assert '"name": "folder_sync"' not in output + assert selected and all(credentials.mid == 1 for credentials in selected) + + db = Database(Path(tmp) / "app.db") + db.initialize() + try: + repo = Repository(db) + for name in ("snapshot", "discovery", "candidate_cleanup"): + assert repo.kv_get(f"task:last:{name}"), name + finally: + db.close() + + +def test_serve_preview_guard(): + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp) + assert run_cli(["init", "--config", str(path)])[0] == 0 + db_path = Path(tmp) / "app.db" + try: + cli._assert_preview_database(db_path) + except cli.ConfigError as exc: + assert "preview:seeded" in str(exc) + else: + raise AssertionError("普通数据库不应通过 --preview 校验") + + db = Database(db_path) + db.initialize() + try: + Repository(db).kv_set("preview:seeded", "1") + finally: + db.close() + cli._assert_preview_database(db_path) + + try: + cli._assert_preview_database(Path(tmp) / "missing.db") + except cli.ConfigError as exc: + assert "不存在" in str(exc) + else: + raise AssertionError("缺失数据库应报错") + + +def test_default_command_starts_web_and_scheduler(): + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp) + calls = [] + original = cli.cmd_run + + def fake_run(args, logger, *, serve_web=False): + calls.append((args.command, serve_web, args.once)) + return 0 + + cli.cmd_run = fake_run + try: + code = main(["--config", str(path)]) + finally: + cli.cmd_run = original + assert code == 0 + assert calls == [("run", True, False)] + + +def test_snapshot_backup_failure_returns_error_and_retries(): + import reaction_board.services.snapshot as snapshot_module + + 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") + + original = snapshot_module.backup_daily + + def fail_backup(*args, **kwargs): + raise OSError("disk full") + + snapshot_module.backup_daily = fail_backup + try: + code, output = run_cli(["snapshot", "--config", str(path)]) + finally: + snapshot_module.backup_daily = original + assert code == 1 + marker = output.find("{\n") + assert marker >= 0, output + payload = json.loads(output[marker:]) + assert payload["completed"] is True + assert "disk full" in payload["backup_error"] + + code, output = run_cli(["snapshot", "--config", str(path)]) + assert code == 0 + marker = output.find("{\n") + payload = json.loads(output[marker:]) + assert payload["backup_error"] is None + assert Path(payload["backup_path"]).exists() + + +def test_candidates_and_discover_commands(): + from reaction_board.services.discovery import DiscoverySummary + + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp) + assert run_cli(["init", "--config", str(path)])[0] == 0 + db = Database(Path(tmp) / "app.db") + db.initialize() + try: + repo = Repository(db) + repo.upsert_candidate( + aid=123, bvid="BV123", title="咻咻满 reaction", + up_mid=100, up_name="UP", pubdate=1, + cover=None, score=3, channel="search:test#p1", + ) + finally: + db.close() + + code, output = run_cli(["candidates", "--config", str(path)]) + assert code == 0 + assert '"count": 1' in output + assert '"aid": 123' in output + + class FakeDiscovery: + def __init__(self, *args, **kwargs): + pass + + def run(self): + return DiscoverySummary( + started_at=1, finished_at=2, candidates=3, notified=1 + ) + + original = cli.DiscoveryService + cli.DiscoveryService = FakeDiscovery + try: + code, output = run_cli(["discover", "--config", str(path)]) + finally: + cli.DiscoveryService = original + assert code == 0 + assert '"candidates": 3' in output + assert '"notified": 1' in output + + +def test_backfill_followers_dry_run(): + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, cookie="", csrf="") + code, output = run_cli([ + "backfill-followers", "--dry-run", "--config", str(path), + ]) + assert code == 0 + assert '"requested": 0' in output + assert '"failed": 0' in output + assert '"dry_run": true' in output + 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；落盘 task:state=deferred。""" + from reaction_board.logging_json import JsonLogger + from reaction_board.task_state import STATUS_DEFERRED, load_state + + 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" + expected_due = finished + cli._SEED_RETRY_SECONDS + assert abs(ctrl.next_run - expected_due) < 0.01 + last = json.loads(repo.kv_get("task:last:snapshot_seed")) + assert last["skipped"] is True and last["ok"] is False + state = load_state(repo, "snapshot_seed") + assert state is not None + assert state.status == STATUS_DEFERRED + assert abs(state.due_at - expected_due) < 0.01 + assert state.due_at > finished + assert state.error == "snapshot_running" + finally: + db.close() + + +def test_quiet_window_defer_persists_deferred_state(): + """H8: 静默窗 defer 落盘 status=deferred + due_at=窗后 01:00，不得记成功。""" + from datetime import datetime + from unittest import mock + + from reaction_board.logging_json import JsonLogger + from reaction_board.task_state import STATUS_DEFERRED, load_state + from reaction_board.timeutil import ( + SHANGHAI_TZ, + next_after_snapshot_quiet_window, + ) + + 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() + # 静默窗内：2026-11-03 23:30 上海 + quiet = datetime(2026, 11, 3, 23, 30, tzinfo=SHANGHAI_TZ).timestamp() + expected_due = next_after_snapshot_quiet_window(quiet) + assert expected_due > quiet + with mock.patch("reaction_board.cli.time.time", return_value=quiet): + ctrl = cli._quiet_window_defer( + repo, logger, "discovery", started_at=int(quiet) - 1, + ) + assert ctrl.ok is False + assert ctrl.error == cli._QUIET_SKIP_REASON + assert abs(ctrl.next_run - expected_due) < 0.01 + last = json.loads(repo.kv_get("task:last:discovery")) + assert last["ok"] is False and last["skipped"] is True + assert last["error"] == cli._QUIET_SKIP_REASON + # 不得推成成功完成类 kv + assert repo.kv_get("snapshot:last_completed") is None + assert repo.kv_get("discovery:last_completed") is None + state = load_state(repo, "discovery") + assert state is not None + assert state.status == STATUS_DEFERRED + assert abs(state.due_at - expected_due) < 0.01 + assert state.due_at > quiet + assert state.error == cli._QUIET_SKIP_REASON + finally: + db.close() + + + + +def test_snapshot_seed_same_day_completed_succeeds(): + """当日已完成时手动 snapshot-seed 应成功 already_completed，不走 no_source。""" + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, cookie="", csrf="") + code, _ = run_cli(["init", "--config", str(path)]) + assert code == 0 + today = "2026-10-31" + db = Database(Path(tmp) / "app.db") + db.initialize() + try: + repo = Repository(db) + repo.kv_set("snapshot:last_completed", today) + repo.kv_set("snapshot:completed:" + today, "1") + finally: + db.close() + code, output = run_cli([ + "snapshot-seed", "--config", str(path), "--date", today, + ]) + assert code == 0, output + assert '"reason": "already_completed"' in output + assert "no_source" not 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 True + finally: + db.close() + + +def test_candidate_cleanup_failure_retries_soon(): + """candidate_cleanup 异常应记失败任务并约 5 分钟补跑，而非落到 86400 间隔。""" + class FakeClient: + def check_login(self): + return LoginState(True, 1, "tester", 0, "OK") + + class FakeDiscovery: + def __init__(self, *args, **kwargs): + pass + + def run(self): + from reaction_board.services.discovery import DiscoverySummary + return DiscoverySummary(started_at=1, finished_at=1) + + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp) + data = json.loads(path.read_text(encoding="utf-8")) + data["folders"] = [] + data["track_fans"] = [] + path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") + + original_client = cli._client + original_discovery = cli.DiscoveryService + original_delete = Repository.delete_expired_candidates + captured = {} + + def boom(self, *args, **kwargs): + raise RuntimeError("cleanup db boom") + + def fake_client(credentials, *args, **kwargs): + return FakeClient() + + # 只跑 cleanup：把其它任务 next_run 推远；通过 Scheduler 包装拦截。 + original_scheduler = cli.Scheduler + + class CaptureScheduler(original_scheduler): + def __init__(self, tasks, **kwargs): + far = 1_000_000_000_000.0 + for task in tasks: + if task.name != "candidate_cleanup": + task.next_run = far + else: + task.next_run = 0.0 + super().__init__(tasks, **kwargs) + captured["scheduler"] = self + + cli._client = fake_client + cli.DiscoveryService = FakeDiscovery + cli.Scheduler = CaptureScheduler + Repository.delete_expired_candidates = boom + try: + code, output = run_cli(["run", "--once", "--config", str(path)]) + finally: + cli._client = original_client + cli.DiscoveryService = original_discovery + cli.Scheduler = original_scheduler + Repository.delete_expired_candidates = original_delete + + assert code == 0, output + from reaction_board.task_state import STATUS_FAILED, load_state + + db = Database(Path(tmp) / "app.db") + db.initialize() + try: + repo = Repository(db) + last = json.loads(repo.kv_get("task:last:candidate_cleanup")) + assert last["ok"] is False + assert "cleanup db boom" in last["error"] + state = load_state(repo, "candidate_cleanup") + assert state is not None + assert state.status == STATUS_FAILED + assert state.due_at > 0 + delta_state = state.due_at - __import__("time").time() + assert 250 <= delta_state <= 360, delta_state + assert "cleanup db boom" in (state.error or "") + finally: + db.close() + scheduler = captured["scheduler"] + cleanup_task = next(t for t in scheduler.tasks if t.name == "candidate_cleanup") + # next_run ≈ finished + 300；允许几秒时钟误差 + assert cleanup_task.next_run > 0 + delta = cleanup_task.next_run - __import__("time").time() + assert 250 <= delta <= 360, delta + + + + +def test_initial_daily_next_run_uses_wall_clock_not_zero(): + """常驻启动：无状态时对齐墙钟；有 task:state.due_at 时认落盘。""" + from reaction_board.timeutil import next_daily_run, SHANGHAI_TZ + from reaction_board.task_state import mark_done + from datetime import datetime + + 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) + # 2026-11-03 10:00 上海 + now = datetime(2026, 11, 3, 10, 0, tzinfo=SHANGHAI_TZ).timestamp() + nxt = cli._initial_daily_next_run( + repo, "snapshot", "23:15", now=now, + ) + assert nxt == next_daily_run(now, "23:15") + assert nxt > now + # 昨晚成功 → state.due_at=今晚 23:15 + yday_done = datetime(2026, 11, 2, 23, 20, tzinfo=SHANGHAI_TZ).timestamp() + due = next_daily_run(yday_done, "23:15") + mark_done(repo, "snapshot", due_at=due, now=yday_done, stat_date="2026-11-02") + nxt2 = cli._initial_daily_next_run( + repo, "snapshot", "23:15", now=now, + ) + assert nxt2 == due + assert nxt2 > now + finally: + db.close() + + +def test_reached_daily_slot_gates_force_seed(): + from reaction_board.timeutil import SHANGHAI_TZ + from datetime import datetime + + before = datetime(2026, 11, 3, 10, 0, tzinfo=SHANGHAI_TZ).timestamp() + at = datetime(2026, 11, 3, 23, 15, tzinfo=SHANGHAI_TZ).timestamp() + after = datetime(2026, 11, 3, 23, 16, tzinfo=SHANGHAI_TZ).timestamp() + assert cli._reached_daily_slot("23:15", before) is False + assert cli._reached_daily_slot("23:15", at) is True + assert cli._reached_daily_slot("23:15", after) is True + + +def test_restart_after_seed_schedules_snapshot_at_wall_clock(): + """午夜 seed 后重启：认 task:state pending due_at=当晚，首轮不得跑/force。""" + from reaction_board.scheduler import ScheduledTask, Scheduler + from reaction_board.timeutil import next_daily_run, SHANGHAI_TZ, shanghai_date + from reaction_board.task_state import ensure_pending, mark_done + from datetime import datetime + + 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, 10, 30, tzinfo=SHANGHAI_TZ).timestamp() + today = shanghai_date(now) + repo.kv_set(f"snapshot:seeded:{today}", "1") + slot = datetime(2026, 11, 3, 23, 15, tzinfo=SHANGHAI_TZ).timestamp() + seed_done_at = datetime(2026, 11, 3, 0, 1, tzinfo=SHANGHAI_TZ).timestamp() + mark_done( + repo, "snapshot_seed", + due_at=next_daily_run(seed_done_at, "00:00"), + now=seed_done_at, + stat_date=today, + ) + ensure_pending( + repo, "snapshot", due_at=slot, stat_date=today, now=seed_done_at, + ) + ensure_pending( + repo, "fans", due_at=slot, stat_date=today, now=seed_done_at, + ) + snapshot_at = "23:15" + seed_at = "00:00" + snapshot_next = cli._initial_daily_next_run( + repo, "snapshot", snapshot_at, now=now, + ) + seed_next = cli._initial_daily_next_run( + repo, "snapshot_seed", seed_at, now=now, + ) + fans_next = cli._initial_daily_next_run( + repo, "fans", snapshot_at, now=now, + ) + assert snapshot_next == slot + assert fans_next == slot + assert seed_next == next_daily_run(seed_done_at, seed_at) + assert seed_next > now + + force_calls = [] + + def run_snapshot(): + force = ( + repo.kv_get(f"snapshot:seeded:{today}") == "1" + and cli._reached_daily_slot(snapshot_at, clock[0]) + ) + force_calls.append(force) + if force and clock[0] < slot: + raise AssertionError("daytime restart must not force-seal seed") + return None + + clock = [now] + scheduler = Scheduler( + [ + ScheduledTask("snapshot_seed", 86400.0, lambda: None, next_run=seed_next), + ScheduledTask("snapshot", 86400.0, run_snapshot, next_run=snapshot_next), + ScheduledTask("fans", 86400.0, lambda: None, next_run=fans_next), + ], + clock=lambda: clock[0], + ) + assert scheduler.run_due() == [] + assert force_calls == [] + clock[0] = slot + results = scheduler.run_due() + assert len(results) >= 1 + assert force_calls == [True] + finally: + db.close() + + slot = datetime(2026, 11, 3, 23, 15, tzinfo=SHANGHAI_TZ).timestamp() + assert cli._reached_daily_slot("23:15", slot) is True + + +def test_daytime_seed_marker_does_not_force_without_slot(): + """即使 seeded=1，未到 snapshot_at 也不得 force（防守：即便误触发 run_snapshot）。""" + from reaction_board.timeutil import SHANGHAI_TZ + from datetime import datetime + + 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, 12, 0, tzinfo=SHANGHAI_TZ).timestamp() + today = "2026-11-03" + repo.kv_set(f"snapshot:seeded:{today}", "1") + force_seed = ( + repo.kv_get(f"snapshot:seeded:{today}") == "1" + and cli._reached_daily_slot("23:15", now) + ) + assert force_seed is False + finally: + 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() + test_ingest_without_refs_is_offline() + test_snapshot_offline_empty_db() + test_auth_check_without_cookie() + test_auth_check_all_and_auth_dispatch() + test_run_once_offline() + test_serve_preview_guard() + test_default_command_starts_web_and_scheduler() + test_snapshot_backup_failure_returns_error_and_retries() + test_candidates_and_discover_commands() + test_backfill_followers_dry_run() + 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_snapshot_seed_same_day_completed_succeeds() + test_defer_snapshot_seed_schedules_ten_minutes() + test_quiet_window_defer_persists_deferred_state() + test_candidate_cleanup_failure_retries_soon() + test_initial_daily_next_run_uses_wall_clock_not_zero() + 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/错误路径测试全部通过") + + +if __name__ == "__main__": + main_test() diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py index 6376a64..a06e817 100644 --- a/tests/test_scheduler.py +++ b/tests/test_scheduler.py @@ -8,7 +8,7 @@ import threading sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) sys.stdout.reconfigure(encoding="utf-8") -from reaction_board.scheduler import ScheduledTask, Scheduler # noqa: E402 +from reaction_board.scheduler import ScheduledTask, Scheduler, TaskControl # noqa: E402 def test_run_when_due_and_reschedule(): @@ -128,9 +128,111 @@ def test_next_run_fn_controls_reschedule(): assert calls == [1_000.0, 1_123.0] + +def test_task_control_overrides_ok_and_next_run(): + clock = [1_000.0] + task = ScheduledTask( + "defer", + 60, + lambda: TaskControl(ok=False, next_run=1_500.0, error="quiet"), + ) + scheduler = Scheduler([task], clock=lambda: clock[0]) + results = scheduler.run_due() + assert len(results) == 1 + assert results[0].ok is False + assert results[0].error == "quiet" + assert task.next_run == 1_500.0 + # 未到 next_run 不重跑 + clock[0] = 1_499.0 + assert scheduler.run_due() == [] + clock[0] = 1_500.0 + task.fn = lambda: TaskControl(ok=True, next_run=2_000.0) + results = scheduler.run_due() + assert results[0].ok is True + assert task.next_run == 2_000.0 + + + + +class _CaptureLogger: + def __init__(self): + self.events = [] + + def info(self, event, **fields): + self.events.append(("info", event, fields)) + + def warning(self, event, **fields): + self.events.append(("warning", event, fields)) + + def error(self, event, **fields): + self.events.append(("error", event, fields)) + + +def test_task_control_ok_false_logs_fail_not_ok(): + """TaskControl(ok=False) 不得记 scheduler.task_ok（Skip ≠ success）。""" + clock = [1_000.0] + logger = _CaptureLogger() + task = ScheduledTask( + "quiet", + 60, + lambda: TaskControl(ok=False, next_run=1_500.0, error="snapshot_quiet_window"), + ) + scheduler = Scheduler([task], clock=lambda: clock[0], logger=logger) + results = scheduler.run_due() + assert results[0].ok is False + assert results[0].error == "snapshot_quiet_window" + assert ("info", "scheduler.task_ok", {"task": "quiet"}) not in logger.events + fail_events = [e for e in logger.events if e[1] == "scheduler.task_fail"] + assert len(fail_events) == 1 + assert fail_events[0][0] == "warning" + assert fail_events[0][2]["task"] == "quiet" + assert fail_events[0][2]["error"] == "snapshot_quiet_window" + + +def test_task_control_ok_true_still_logs_task_ok(): + clock = [1_000.0] + logger = _CaptureLogger() + task = ScheduledTask( + "ok_task", + 60, + lambda: TaskControl(ok=True, next_run=1_200.0), + ) + scheduler = Scheduler([task], clock=lambda: clock[0], logger=logger) + results = scheduler.run_due() + assert results[0].ok is True + assert ("info", "scheduler.task_ok", {"task": "ok_task"}) in logger.events + assert not any(e[1] == "scheduler.task_fail" for e in logger.events) + + +def test_raised_exception_logs_task_fail(): + clock = [100.0] + logger = _CaptureLogger() + + def boom(): + raise RuntimeError("db down") + + scheduler = Scheduler( + [ScheduledTask("bad", 10, boom)], + clock=lambda: clock[0], + logger=logger, + ) + results = scheduler.run_due() + assert results[0].ok is False + assert results[0].error == "db down" + assert not any(e[1] == "scheduler.task_ok" for e in logger.events) + assert not any(e[1] == "scheduler.task_fail" for e in logger.events) + err_events = [e for e in logger.events if e[1] == "scheduler.task_error"] + assert len(err_events) == 1 + assert err_events[0][0] == "error" + assert err_events[0][2]["error"] == "db down" + def main(): test_run_when_due_and_reschedule() test_next_run_fn_controls_reschedule() + test_task_control_overrides_ok_and_next_run() + test_task_control_ok_false_logs_fail_not_ok() + test_task_control_ok_true_still_logs_task_ok() + test_raised_exception_logs_task_fail() 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 index 7769bfe..79da107 100644 --- a/tests/test_snapshot.py +++ b/tests/test_snapshot.py @@ -537,6 +537,36 @@ def test_full_snapshot_skips_ok_rows_including_hourly_refresh(): +def test_seed_same_day_completed_is_already_completed_not_no_source(): + """全站完成后 last_completed==target：应 already_completed，不误判 no_source。""" + clock = 1_800_000_000 + with tempfile.TemporaryDirectory() as tmp: + transport = SnapshotTransport({1: view_for(1)}) + db, repo, service = make_service(tmp, transport, track_fans=()) + try: + add_video(repo, 1) + today = "2026-10-31" + repo.upsert_snapshot( + today, 1, + metrics={"view": 10, "like": 1, "coin": 0, "favorite": 0, + "share": 0, "reply": 0, "danmaku": 0}, + ok=True, err=None, captured_at=clock, + ) + # 正常收尾态：last_completed 已推进到当日，且 completed 标记在位 + repo.kv_set("snapshot:last_completed", today) + repo.kv_set(service.completed_key(today), "1") + + summary = service.run_seed_from_last_completed(date=today) + assert summary.reason == "already_completed" + assert summary.already_completed is True + assert summary.completed is True + assert summary.reason != "no_source" + assert transport.count("/web-interface/view") == 0 + finally: + db.close() + + + def test_seed_from_last_completed_does_not_mark_completed(): clock = 1_800_000_000 with tempfile.TemporaryDirectory() as tmp: @@ -631,6 +661,117 @@ def test_seeded_day_skips_without_force_and_seals_with_force(): db.close() +def test_seal_keep_scraped_skips_view_progress_punches_seed(): + """晚间 seal（keep_scraped）：已 view 的 ok=1 作进度跳过，seed 壳仍打穿。""" + clock = 1_800_000_000 + 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) + today = DATE + # 两条 seed 壳 + for aid, view in ((1, 10), (2, 20)): + 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="seed", + ) + # 模拟 seal 中途：aid=1 已真实采集 + 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 + 1, src="view", + ) + repo.kv_set(service.seeded_key(today), "1") + + sealed = service.run_if_needed(date=today, force=True) + assert sealed.completed is True + assert sealed.skipped >= 1 # aid=1 进度跳过 + assert sealed.success >= 1 # aid=2 seed 打穿 + # 只应请求 aid=2 的 view（aid=1 已是 view 进度） + assert transport.count("/web-interface/view") == 1 + assert int(repo.get_snapshot(today, 1)["view"]) == 111 + assert repo.get_snapshot(today, 1)["src"] == "view" + assert int(repo.get_snapshot(today, 2)["view"]) == 222 + assert repo.get_snapshot(today, 2)["src"] == "view" + finally: + 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() @@ -646,8 +787,12 @@ def main(): test_backup_failure_retries_without_api() test_run_recent_videos_refreshes_without_completion_markers() test_full_snapshot_skips_ok_rows_including_hourly_refresh() + test_seed_same_day_completed_is_already_completed_not_no_source() 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 new file mode 100644 index 0000000..117bc3d --- /dev/null +++ b/tests/test_task_state.py @@ -0,0 +1,398 @@ +# -*- coding: utf-8 -*- +"""task:state 持久化与启动续跑: uv run python tests/test_task_state.py""" + +import json +import os +import sys +import tempfile +from datetime import datetime +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.repository import Repository # noqa: E402 +from reaction_board.scheduler import ScheduledTask, Scheduler # noqa: E402 +from reaction_board.task_state import ( # noqa: E402 + STATUS_DEFERRED, + STATUS_DONE, + STATUS_FAILED, + STATUS_PENDING, + STATUS_RUNNING, + bootstrap_from_last, + ensure_pending, + load_state, + mark_deferred, + mark_done, + mark_failed, + mark_running, + recover_stale_running, + resume_stat_date, + startup_next_run, +) +from reaction_board.timeutil import SHANGHAI_TZ, next_daily_run # noqa: E402 + + +def _repo(tmp: str) -> tuple[Database, Repository]: + db = Database(Path(tmp) / "app.db") + db.initialize() + return db, Repository(db) + + +def test_restart_mid_run_resumes_same_stat_date(): + with tempfile.TemporaryDirectory() as tmp: + db, repo = _repo(tmp) + try: + now = datetime(2026, 11, 3, 23, 40, tzinfo=SHANGHAI_TZ).timestamp() + mark_running( + repo, "snapshot", stat_date="2026-11-03", now=now, + progress={"done": 10, "total": 100}, + ) + # 模拟进程崩溃：running 残留 + recovered = recover_stale_running(repo, "snapshot", now=now + 60) + assert recovered is not None + assert recovered.status == STATUS_PENDING + assert recovered.stat_date == "2026-11-03" + assert recovered.progress == {"done": 10, "total": 100} + assert recovered.due_at <= now + 60 + assert resume_stat_date(repo, "snapshot", fallback="2026-11-04") == "2026-11-03" + nxt = startup_next_run( + repo, "snapshot", + wall_due_at=next_daily_run(now + 60, "23:15"), + now=now + 60, + ) + assert nxt <= now + 60 + finally: + db.close() + + +def test_seeded_before_snapshot_at_does_not_force_run(): + """seed 后挂 pending；白天重启认落盘 due_at（刻意 ≠ 墙钟）且未到不跑。""" + with tempfile.TemporaryDirectory() as tmp: + db, repo = _repo(tmp) + try: + now = datetime(2026, 11, 3, 10, 30, tzinfo=SHANGHAI_TZ).timestamp() + # 刻意偏离墙钟 23:15，证明认落盘而非 fallback + slot = datetime(2026, 11, 3, 22, 0, tzinfo=SHANGHAI_TZ).timestamp() + wall = next_daily_run(now, "23:15") + assert slot != wall + ensure_pending( + repo, "snapshot", due_at=slot, stat_date="2026-11-03", now=now, + ) + repo.kv_set("snapshot:seeded:2026-11-03", "1") + nxt = startup_next_run( + repo, "snapshot", + wall_due_at=wall, + now=now, + ) + assert nxt == slot + assert nxt != wall + assert nxt > now + calls = [] + + def run_snapshot(): + calls.append("ran") + return None + + clock = [now] + sched = Scheduler( + [ScheduledTask("snapshot", 86400.0, run_snapshot, next_run=nxt)], + clock=lambda: clock[0], + ) + assert sched.run_due() == [] + assert calls == [] + finally: + db.close() + + +def test_failed_future_due_at_does_not_run_early(): + with tempfile.TemporaryDirectory() as tmp: + db, repo = _repo(tmp) + try: + now = datetime(2026, 11, 3, 12, 0, tzinfo=SHANGHAI_TZ).timestamp() + due = now + 600 + mark_failed( + repo, "snapshot", due_at=due, now=now, + stat_date="2026-11-03", error="boom", + ) + nxt = startup_next_run( + repo, "snapshot", + wall_due_at=next_daily_run(now, "23:15"), + now=now, + ) + assert nxt == due + clock = [now] + calls = [] + sched = Scheduler( + [ + ScheduledTask( + "snapshot", 86400.0, lambda: calls.append(1), next_run=nxt, + ) + ], + clock=lambda: clock[0], + ) + assert sched.run_due() == [] + assert calls == [] + finally: + db.close() + + +def test_due_at_passed_runs_including_failed_retry(): + with tempfile.TemporaryDirectory() as tmp: + db, repo = _repo(tmp) + try: + now = datetime(2026, 11, 3, 12, 10, tzinfo=SHANGHAI_TZ).timestamp() + due = now - 30 # 已过重试点 + mark_failed( + repo, "snapshot", due_at=due, now=due - 300, + stat_date="2026-11-03", error="boom", + ) + nxt = startup_next_run( + repo, "snapshot", + wall_due_at=next_daily_run(now, "23:15"), + now=now, + ) + # 不得退回今晚墙钟；应立刻可跑 + assert nxt == due + assert nxt <= now + wall = next_daily_run(now, "23:15") + assert nxt != wall + calls = [] + clock = [now] + sched = Scheduler( + [ + ScheduledTask( + "snapshot", 86400.0, lambda: calls.append(1), next_run=nxt, + ) + ], + clock=lambda: clock[0], + ) + results = sched.run_due() + assert len(results) == 1 + assert calls == [1] + finally: + db.close() + + +def test_done_writes_next_due_and_blocks_until_then(): + with tempfile.TemporaryDirectory() as tmp: + db, repo = _repo(tmp) + try: + now = datetime(2026, 11, 3, 23, 40, tzinfo=SHANGHAI_TZ).timestamp() + next_due = next_daily_run(now, "23:15") + mark_done(repo, "snapshot", due_at=next_due, now=now, stat_date="2026-11-03") + state = load_state(repo, "snapshot") + assert state is not None + assert state.status == STATUS_DONE + nxt = startup_next_run( + repo, "snapshot", wall_due_at=next_due, now=now + 60, + ) + assert nxt == next_due + assert nxt > now + 60 + finally: + db.close() + + +def test_no_state_falls_back_to_wall_clock(): + with tempfile.TemporaryDirectory() as tmp: + db, repo = _repo(tmp) + try: + now = datetime(2026, 11, 3, 10, 0, tzinfo=SHANGHAI_TZ).timestamp() + wall = next_daily_run(now, "23:15") + nxt = startup_next_run(repo, "snapshot", wall_due_at=wall, now=now) + assert nxt == wall + finally: + db.close() + + + +def test_fail_or_defer_preserves_progress(): + """H7: mark_failed / mark_deferred 不传 progress 时保留已有 progress 与 stat_date。""" + with tempfile.TemporaryDirectory() as tmp: + db, repo = _repo(tmp) + try: + now = datetime(2026, 11, 3, 12, 0, tzinfo=SHANGHAI_TZ).timestamp() + progress = {"done": 42, "total": 100, "cursor": "aid:99"} + mark_running( + repo, "snapshot", stat_date="2026-11-03", now=now, + progress=progress, + ) + before = load_state(repo, "snapshot") + assert before is not None + assert before.progress == progress + assert before.stat_date == "2026-11-03" + + due_fail = now + 300 + mark_failed( + repo, "snapshot", due_at=due_fail, now=now + 1, + progress=None, error="boom", + ) + failed = load_state(repo, "snapshot") + assert failed is not None + assert failed.status == STATUS_FAILED + assert failed.due_at == due_fail + assert failed.progress == progress + assert failed.stat_date == "2026-11-03" + assert failed.error == "boom" + + due_defer = now + 600 + mark_deferred( + repo, "snapshot", due_at=due_defer, now=now + 2, + progress=None, error="snapshot_quiet_window", + ) + deferred = load_state(repo, "snapshot") + assert deferred is not None + assert deferred.status == STATUS_DEFERRED + assert deferred.due_at == due_defer + assert deferred.progress == progress + assert deferred.stat_date == "2026-11-03" + assert deferred.error == "snapshot_quiet_window" + finally: + 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() + test_failed_future_due_at_does_not_run_early() + test_due_at_passed_runs_including_failed_retry() + 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") + + +if __name__ == "__main__": + main_test() diff --git a/tests/test_timeutil.py b/tests/test_timeutil.py index bfe4060..7a5bd20 100644 --- a/tests/test_timeutil.py +++ b/tests/test_timeutil.py @@ -11,7 +11,10 @@ sys.stdout.reconfigure(encoding="utf-8") from reaction_board.timeutil import ( # noqa: E402 SHANGHAI_TZ, day_bounds, + in_snapshot_quiet_window, + interval_next_run, is_date, + next_after_snapshot_quiet_window, next_daily_run, parse_hhmm, shanghai_date, @@ -72,11 +75,40 @@ def test_day_bounds_and_date_validation(): raise AssertionError("非法日期应失败") + +def test_snapshot_quiet_window(): + assert in_snapshot_quiet_window(ts("2026-11-02 23:00:00")) + assert in_snapshot_quiet_window(ts("2026-11-02 23:59:59")) + assert in_snapshot_quiet_window(ts("2026-11-03 00:00:00")) + assert in_snapshot_quiet_window(ts("2026-11-03 00:59:59")) + assert not in_snapshot_quiet_window(ts("2026-11-03 01:00:00")) + assert not in_snapshot_quiet_window(ts("2026-11-02 22:59:59")) + assert not in_snapshot_quiet_window(ts("2026-11-02 12:00:00")) + + assert next_after_snapshot_quiet_window(ts("2026-11-02 22:00:00")) == ts( + "2026-11-02 22:00:00" + ) + assert next_after_snapshot_quiet_window(ts("2026-11-02 23:15:00")) == ts( + "2026-11-03 01:00:00" + ) + assert next_after_snapshot_quiet_window(ts("2026-11-03 00:30:00")) == ts( + "2026-11-03 01:00:00" + ) + # 22:50 + 60min = 23:50 → 推到次日 01:00 + assert interval_next_run(ts("2026-11-02 22:50:00"), 3600) == ts( + "2026-11-03 01:00:00" + ) + assert interval_next_run(ts("2026-11-02 12:00:00"), 3600) == ts( + "2026-11-02 13:00:00" + ) + + def main(): test_shanghai_date_boundary() test_parse_hhmm() test_next_daily_run() test_day_bounds_and_date_validation() + test_snapshot_quiet_window() print("OK: 上海时区与每日时间计算测试全部通过") diff --git a/tests/test_web_queries.py b/tests/test_web_queries.py index fbb185c..48d9538 100644 --- a/tests/test_web_queries.py +++ b/tests/test_web_queries.py @@ -610,12 +610,15 @@ def test_growth_auto_falls_back_to_one_day(): def test_snapshot_recent_next_run_follows_completion(): import json + from datetime import datetime + from reaction_board.timeutil import SHANGHAI_TZ, interval_next_run with tempfile.TemporaryDirectory() as tmp: db, q = setup(tmp) try: repo = Repository(db) - finished = FIXED_NOW - 10 + # 午后完成：间隔后仍在静默窗外，应严格 finished+3600 + finished = int(datetime(2026, 10, 31, 15, 0, 0, tzinfo=SHANGHAI_TZ).timestamp()) repo.kv_set( "task:last:snapshot_recent", json.dumps({ @@ -627,6 +630,128 @@ def test_snapshot_recent_next_run_follows_completion(): DASHBOARD_CACHE.clear() system = q.system() assert system["next_runs"]["snapshot_recent"] == finished + 3600 + assert system["next_runs"]["snapshot_recent"] == int( + interval_next_run(float(finished), 3600) + ) + finally: + db.close() + + +def test_snapshot_recent_next_run_defers_quiet_window_and_retry_on_skip(): + import json + from datetime import datetime + from reaction_board.timeutil import SHANGHAI_TZ, next_after_snapshot_quiet_window + + with tempfile.TemporaryDirectory() as tmp: + db, q = setup(tmp) + try: + repo = Repository(db) + # 22:50 成功 → 下一档 23:50 落入静默窗 → 推到 01:00 + finished = int(datetime(2026, 10, 31, 22, 50, 0, tzinfo=SHANGHAI_TZ).timestamp()) + repo.kv_set( + "task:last:snapshot_recent", + json.dumps({ + "name": "snapshot_recent", "ok": True, + "started_at": finished - 5, "finished_at": finished, + "error": None, "summary": {}, + }), + ) + DASHBOARD_CACHE.clear() + system = q.system() + expected = int(datetime(2026, 11, 1, 1, 0, 0, tzinfo=SHANGHAI_TZ).timestamp()) + assert system["next_runs"]["snapshot_recent"] == expected + + # skip 记录：补跑到窗后 01:00，不得当成功推进 + quiet = int(datetime(2026, 10, 31, 23, 20, 0, tzinfo=SHANGHAI_TZ).timestamp()) + repo.kv_set( + "task:last:snapshot_recent", + json.dumps({ + "name": "snapshot_recent", "ok": False, "skipped": True, + "started_at": quiet, "finished_at": quiet, + "error": "snapshot_quiet_window", + "summary": {"skipped": True, "reason": "snapshot_quiet_window"}, + }), + ) + DASHBOARD_CACHE.clear() + system = q.system() + assert system["next_runs"]["snapshot_recent"] == int( + next_after_snapshot_quiet_window(float(quiet)) + ) + assert not any(e["task"] == "snapshot_recent" for e in system["errors"]) + finally: + db.close() + + +def test_snapshot_failed_next_run_retries_soon_not_next_day(): + 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, 10, 31, 23, 15, 0, tzinfo=SHANGHAI_TZ).timestamp()) + repo.kv_set( + "task:last:snapshot", + json.dumps({ + "name": "snapshot", "ok": False, + "started_at": finished - 1, "finished_at": finished, + "error": "boom", "summary": {"completed": False}, + }), + ) + DASHBOARD_CACHE.clear() + system = q.system() + assert system["next_runs"]["snapshot"] == finished + 300 + assert system["next_runs"]["snapshot"] != int( + next_daily_run(finished, "23:15") + ) + assert any(e["task"] == "snapshot" for e in system["errors"]) + finally: + 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() @@ -882,6 +1007,9 @@ def main_test(): test_system_tasks_sorted_by_started_at_desc() test_growth_auto_falls_back_to_one_day() 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() @@ -948,13 +1076,15 @@ def test_system_status_uses_completed_date_and_running_flags(): def test_system_next_runs_follow_last_completion_not_wall_clock(): import json + from datetime import datetime + from reaction_board.timeutil import SHANGHAI_TZ with tempfile.TemporaryDirectory() as tmp: db, q = setup(tmp) try: repo = Repository(db) - # 发现任务刚完成不久：下次 = finished + 60min，即使 now 已超过也不跳档 - finished = FIXED_NOW - 30 + # 午后完成：下次 = finished + 60min，不跳档、不误入静默窗推算 + finished = int(datetime(2026, 10, 31, 15, 0, 0, tzinfo=SHANGHAI_TZ).timestamp()) repo.kv_set( "task:last:discovery", json.dumps({ @@ -966,7 +1096,6 @@ def test_system_next_runs_follow_last_completion_not_wall_clock(): DASHBOARD_CACHE.clear() system = q.system() assert system["next_runs"]["discover"] == finished + 3600 - # 仍小于/等于 now 也可以（表示本轮后的下一档已到期或未到），但绝不能跳到 +7200 assert system["next_runs"]["discover"] != finished + 7200 finally: db.close()