From af90ce8f871c3db92c57dd8c880e0fe93ab6dc37 Mon Sep 17 00:00:00 2001 From: bot-ci Date: Mon, 2 Nov 2026 00:46:20 +0800 Subject: [PATCH] fix(system): three next-runs, completed-date data_date, running mark MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Split /system schedule into fans / snapshot+UP / discovery next-runs. Drive data_date and snapshot_days from fully completed video snapshots only, keep prior success date on failure, show task failure records, and mark in-progress tasks as 运行中 without wall-clock next-run jumps. --- reaction_board/cli.py | 140 +++++++++++++++++++--- reaction_board/repository.py | 5 + reaction_board/services/snapshot.py | 120 ++++++++++++++++--- reaction_board/web/queries.py | 56 +++++++-- reaction_board/web/static/views/system.js | 8 +- tests/test_snapshot.py | 8 +- tests/test_web_queries.py | 90 +++++++++++++- tests/web_fixtures.py | 18 +++ tools/seed_preview_db.py | 18 +++ 9 files changed, 411 insertions(+), 52 deletions(-) diff --git a/reaction_board/cli.py b/reaction_board/cli.py index c0f39ae..c580f69 100644 --- a/reaction_board/cli.py +++ b/reaction_board/cli.py @@ -139,6 +139,30 @@ def _print(payload) -> None: print(json.dumps(payload, ensure_ascii=False, indent=2), flush=True) +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()) + record = { + "name": name, + "started_at": int(started_at if started_at is not None else now), + } + try: + repo.kv_set(_task_running_key(name), json.dumps(record, ensure_ascii=False)) + except Exception: # noqa: BLE001 + pass + + +def _clear_task_running(repo: Repository, name: str) -> None: + try: + repo.kv_delete(_task_running_key(name)) + except Exception: # noqa: BLE001 + pass + + def _record_task( repo: Repository, name: str, @@ -149,7 +173,7 @@ def _record_task( payload=None, error: str | None = None, ) -> None: - """把最近一次任务结果写入 kv，供只读系统页展示。""" + """把最近一次任务结果写入 kv，并清除 running 标记。""" now = int(time.time()) record = { "name": name, @@ -163,6 +187,7 @@ def _record_task( repo.kv_set(f"task:last:{name}", json.dumps(record, ensure_ascii=False)) except Exception: # noqa: BLE001 任务状态记录失败不影响主流程 pass + _clear_task_running(repo, name) def _client(credentials, logger: JsonLogger, *, pacer=None, retries: int = 3) -> BilibiliClient: @@ -264,20 +289,60 @@ def cmd_snapshot(args, logger: JsonLogger) -> int: cfg.bilibili.read, logger, pacer=Pacer(1.0, 1.6), retries=1 ) service = SnapshotService(store, repo, client, logger=logger) - summary = service.run_if_needed(force=args.force) - payload = summary.as_dict() - payload["database"] = str(cfg.storage.database_path) - payload["snapshot_counts"] = repo.snapshot_counts(summary.stat_date) - payload["counts"] = repo.counts() - ok = bool(summary.completed and not summary.backup_error) + started = int(time.time()) + _mark_task_running(repo, "snapshot", started_at=started) + try: + video_summary = service.run_if_needed(force=args.force) + except Exception as exc: # noqa: BLE001 + _record_task( + repo, "snapshot", ok=False, started_at=started, + finished_at=int(time.time()), error=str(exc), + ) + 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_ok = bool(video_summary.completed and not video_summary.backup_error) _record_task( - repo, "snapshot", ok=ok, - started_at=int(summary.started_at or time.time()), - finished_at=int(time.time()), payload=payload, - error=summary.backup_error, + 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, ) + + fans_started = int(time.time()) + _mark_task_running(repo, "fans", started_at=fans_started) + try: + fans_summary = service.run_fans_if_needed( + date=video_summary.stat_date, force=args.force, + ) + except Exception as exc: # noqa: BLE001 + _record_task( + repo, "fans", ok=False, started_at=fans_started, + finished_at=int(time.time()), error=str(exc), + ) + raise + fans_payload = fans_summary.as_dict() + 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, + ) + payload = { + **video_payload, + "fans_ok": fans_summary.fans_ok, + "fans_failed": fans_summary.fans_failed, + "fans_skipped": fans_summary.fans_skipped, + } _print(payload) - return 0 if ok else 1 + return 0 if video_ok and fans_ok else 1 finally: db.close() @@ -312,13 +377,22 @@ def cmd_discover(args, logger: JsonLogger) -> int: notifier=None if args.no_notify else _telegram(cfg, logger), logger=logger, ) - summary = service.run() + started = int(time.time()) + _mark_task_running(repo, "discovery", started_at=started) + try: + summary = service.run() + except Exception as exc: # noqa: BLE001 + _record_task( + repo, "discovery", ok=False, started_at=started, + finished_at=int(time.time()), error=str(exc), + ) + raise payload = summary.as_dict() payload["candidate_total"] = len(repo.list_candidates()) payload["database"] = str(cfg.storage.database_path) _record_task( repo, "discovery", ok=not summary.errors, - started_at=int(summary.started_at), + started_at=int(summary.started_at or started), finished_at=int(summary.finished_at or time.time()), payload=payload, error="；".join(summary.errors) if summary.errors else None, @@ -425,6 +499,7 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int: def run_discover(): started = int(time.time()) + _mark_task_running(repo, "discovery", started_at=started) try: summary = discovery_service.run() except Exception as exc: # noqa: BLE001 记录后交给调度器隔离 @@ -472,6 +547,7 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int: def run_snapshot(): started = int(time.time()) + _mark_task_running(repo, "snapshot", started_at=started) try: summary = snapshot_service.run_if_needed() except Exception as exc: # noqa: BLE001 @@ -491,6 +567,31 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int: logger.info("snapshot.summary", **payload) return summary + def run_fans(): + started = int(time.time()) + _mark_task_running(repo, "fans", started_at=started) + try: + summary = snapshot_service.run_fans_if_needed() + except Exception as exc: # noqa: BLE001 + _record_task( + repo, "fans", ok=False, started_at=started, + finished_at=int(time.time()), error=str(exc), + ) + raise + payload = summary.as_dict() + ok = summary.fans_failed == 0 + error = "；".join( + err for err in summary.errors if str(err).startswith("fans ") + ) or None + _record_task( + repo, "fans", ok=ok, started_at=summary.started_at, + finished_at=summary.finished_at or int(time.time()), + payload=payload, error=error, + ) + last["fans"] = summary + logger.info("fans.summary", **payload) + return summary + def run_config_watch(): before = store.revision store.get() @@ -509,6 +610,13 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int: next_run=0.0, next_run_fn=lambda started: next_daily_run(started, snapshot_at), ), + ScheduledTask( + "fans", + 86400.0, + run_fans, + next_run=0.0, + next_run_fn=lambda started: next_daily_run(started, snapshot_at), + ), ScheduledTask( "discovery", float(cfg.schedule.discover_minutes * 60), @@ -547,10 +655,12 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int: scheduler.run_forever() snapshot_summary = last.get("snapshot") + fans_summary = last.get("fans") discover_summary = last.get("discover") snapshot_ok = snapshot_summary is None or getattr(snapshot_summary, "completed", True) + fans_ok = fans_summary is None or getattr(fans_summary, "fans_failed", 0) == 0 discover_ok = discover_summary is None or not getattr(discover_summary, "errors", []) - return 0 if snapshot_ok and discover_ok else 1 + return 0 if snapshot_ok and fans_ok and discover_ok else 1 finally: if web_server is not None: try: diff --git a/reaction_board/repository.py b/reaction_board/repository.py index 928815c..b4d2f3b 100644 --- a/reaction_board/repository.py +++ b/reaction_board/repository.py @@ -775,6 +775,11 @@ class Repository: (key, str(value)), )) + def kv_delete(self, key: str) -> None: + self.db.write(lambda conn: conn.execute( + "DELETE FROM kv WHERE k = ?", (key,), + )) + # ------------------------------------------------------------ cache version @staticmethod diff --git a/reaction_board/services/snapshot.py b/reaction_board/services/snapshot.py index 8736b02..362e71a 100644 --- a/reaction_board/services/snapshot.py +++ b/reaction_board/services/snapshot.py @@ -7,8 +7,8 @@ * 成功写入七项指标；``ok=1`` 的同日快照直接跳过（断点续跑）； * 视频不可访问时标记 ``removed``； * 其他错误最多重试 3 次，仍失败写 ``ok=0`` 占位，不写假零值； -* 对 ``track_fans`` 逐个请求 ``relation/stat``； -* 完成后写入当日完成标记并生成日备份。 +* 对 ``track_fans`` 逐个请求 ``relation/stat``（独立任务，不影响视频完成日）； +* 视频采集全部成功后写入当日完成标记并生成日备份。 """ from __future__ import annotations @@ -131,12 +131,13 @@ class SnapshotService: def run_if_needed( self, *, date: str | None = None, force: bool = False ) -> SnapshotSummary: + """视频快照（含 UP 头像补齐）；粉丝采集请用 ``run_fans_if_needed``。""" now = self._now() stat_date = date or shanghai_date(now) if not is_date(stat_date): raise ValueError(f"日期格式必须是 YYYY-MM-DD: {stat_date!r}") if force: - return self.run(date=stat_date, force=True) + return self.run_videos(date=stat_date, force=True) completed = self.repo.kv_get(self.completed_key(stat_date)) == "1" pending = self.repo.pending_snapshot_aids(stat_date) @@ -151,7 +152,7 @@ class SnapshotService: completed=True, already_completed=True, ) - return self.run(date=stat_date, force=False) + return self.run_videos(date=stat_date, force=False) def backup_only(self, stat_date: str) -> SnapshotSummary: """采集已完成但备份缺失时，只重试备份，不请求任何外部接口。""" @@ -170,6 +171,12 @@ class SnapshotService: return summary def run(self, *, date: str | None = None, force: bool = False) -> SnapshotSummary: + """兼容 CLI：先跑视频快照，再跑粉丝采集，合并进同一摘要。""" + 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: + """采集视频七项指标并在全部成功后标记完成日 / 备份。""" cfg = self.config_store.get() client = self._client() now = self._now() @@ -240,6 +247,77 @@ class SnapshotService: skipped=summary.skipped, removed=summary.removed, ) + # 完成日只看视频；粉丝是独立调度任务。 + summary.completed = summary.failed == 0 + self.repo.kv_set(self.completed_key(stat_date), "1" if summary.completed else "0") + self.repo.kv_set("snapshot:last_attempt", stat_date) + if summary.completed: + self.repo.kv_set("snapshot:last_completed", stat_date) + elif self.repo.kv_get("snapshot:last_completed") == stat_date: + self.repo.kv_set("snapshot:last_completed", "") + + self._run_backup(cfg, stat_date, summary) + + if summary.success or summary.removed or summary.failed: + self.repo.bump_cache_version("videos") + + summary.finished_at = self._now() + self._log( + "info" if summary.completed else "warning", + "snapshot.done", + **summary.as_dict(), + ) + return summary + + def run_fans_if_needed( + self, *, date: str | None = None, force: bool = False + ) -> SnapshotSummary: + now = self._now() + stat_date = date or shanghai_date(now) + if not is_date(stat_date): + raise ValueError(f"日期格式必须是 YYYY-MM-DD: {stat_date!r}") + cfg = self.config_store.get() + if not cfg.track_fans: + summary = SnapshotSummary( + stat_date=stat_date, started_at=now, finished_at=now, completed=True, + ) + self._log("info", "fans.skip", stat_date=stat_date, reason="no_track_fans") + return summary + if not force and not self._pending_fan_mids(stat_date, cfg.track_fans): + self._log("info", "fans.skip", stat_date=stat_date, reason="already_completed") + return SnapshotSummary( + stat_date=stat_date, + started_at=now, + finished_at=now, + completed=True, + already_completed=True, + fans_skipped=len(cfg.track_fans), + ) + return self.run_fans(date=stat_date, force=force) + + def run_fans( + self, + *, + date: str | None = None, + force: bool = False, + summary: SnapshotSummary | None = None, + ) -> SnapshotSummary: + """采集 track_fans 粉丝数；成败不影响视频完成日。""" + cfg = self.config_store.get() + client = self._client() + now = self._now() + stat_date = date or shanghai_date(now) + if not is_date(stat_date): + raise ValueError(f"日期格式必须是 YYYY-MM-DD: {stat_date!r}") + + standalone = summary is None + if summary is None: + summary = SnapshotSummary(stat_date=stat_date, started_at=now) + else: + # CLI 合并跑：保留视频摘要，只追加粉丝段耗时终点 + summary.stat_date = stat_date + + self._log("info", "fans.start", stat_date=stat_date, force=force, total=len(cfg.track_fans)) for mid in cfg.track_fans: existing_fan = self.repo.get_up_stat(stat_date, mid) if ( @@ -276,29 +354,33 @@ class SnapshotService: stat_date=stat_date, mid=mid, error=error, ) - summary.completed = summary.failed == 0 and summary.fans_failed == 0 - self.repo.kv_set(self.completed_key(stat_date), "1" if summary.completed else "0") - self.repo.kv_set("snapshot:last_attempt", stat_date) - if summary.completed: - self.repo.kv_set("snapshot:last_completed", stat_date) - elif self.repo.kv_get("snapshot:last_completed") == stat_date: - self.repo.kv_set("snapshot:last_completed", "") - - self._run_backup(cfg, stat_date, summary) - - if summary.success or summary.removed or summary.failed: - self.repo.bump_cache_version("videos") if summary.fans_ok or summary.fans_failed: self.repo.bump_cache_version("fans") + fans_ok = summary.fans_failed == 0 + # 独立粉丝任务用 completed 表示粉丝段成败；合并进视频摘要时不改动视频完成标记。 + if standalone: + summary.completed = fans_ok summary.finished_at = self._now() self._log( - "info" if summary.completed else "warning", - "snapshot.done", - **summary.as_dict(), + "info" if fans_ok else "warning", + "fans.done", + stat_date=stat_date, + fans_ok=summary.fans_ok, + fans_failed=summary.fans_failed, + fans_skipped=summary.fans_skipped, + completed=summary.completed, ) return summary + def _pending_fan_mids(self, stat_date: str, mids: tuple[int, ...] | list[int]) -> list[int]: + pending: list[int] = [] + for mid in mids: + existing = self.repo.get_up_stat(stat_date, int(mid)) + if existing is None or int(existing.get("ok") or 0) != 1: + pending.append(int(mid)) + return pending + # ------------------------------------------------------------ internals def _run_backup( diff --git a/reaction_board/web/queries.py b/reaction_board/web/queries.py index 6134548..4efd913 100644 --- a/reaction_board/web/queries.py +++ b/reaction_board/web/queries.py @@ -227,15 +227,20 @@ class DashboardQueries: @staticmethod def _latest_date(conn: sqlite3.Connection) -> str | None: + """数据日期：仅取视频快照任务完整成功后的完成日，避免采集中途抢跑。""" row = conn.execute( - "SELECT MAX(stat_date) AS d FROM snapshot WHERE ok = 1" + "SELECT v FROM kv WHERE k = 'snapshot:last_completed'" ).fetchone() - return None if row is None or row["d"] is None else str(row["d"]) + if row is not None and row["v"]: + return str(row["v"]) + return None @staticmethod def _snapshot_days(conn: sqlite3.Connection) -> int: + """快照天数：统计已完整完成的采集日（completed=1），与数据日期口径一致。""" row = conn.execute( - "SELECT COUNT(DISTINCT stat_date) AS n FROM snapshot WHERE ok = 1" + "SELECT COUNT(*) AS n FROM kv " + "WHERE k LIKE 'snapshot:completed:%' AND v = '1'" ).fetchone() return int(row["n"] or 0) @@ -580,6 +585,23 @@ class DashboardQueries: result[name] = value return result + def _task_running(self, conn: sqlite3.Connection) -> dict[str, Any]: + rows = conn.execute( + "SELECT k, v FROM kv WHERE k LIKE 'task:running:%' ORDER BY k" + ).fetchall() + result: dict[str, Any] = {} + for row in rows: + name = str(row["k"]).removeprefix("task:running:") + try: + value = json.loads(str(row["v"])) + except (TypeError, json.JSONDecodeError): + value = {"raw": row["v"]} + if not isinstance(value, dict): + value = {"raw": value} + value["running"] = True + result[name] = value + return result + def _folders_overview(self, conn: sqlite3.Connection) -> list[dict[str, Any]]: rows = conn.execute( """ @@ -1722,21 +1744,30 @@ class DashboardQueries: ).fetchall() ] tasks = self._task_records(conn) + running = self._task_running(conn) now = self._now() + discover_interval = int(cfg.schedule.discover_minutes) * 60 next_runs = { - "discover": int(now + cfg.schedule.discover_minutes * 60), + "fans": int(next_daily_run(now, cfg.schedule.snapshot_at)), "snapshot": int(next_daily_run(now, cfg.schedule.snapshot_at)), + "discover": int(now + discover_interval), } - for name, key, interval_minutes in ( - ("discover", "discovery", cfg.schedule.discover_minutes), + for name, key in ( + ("fans", "fans"), + ("snapshot", "snapshot"), + ("discover", "discovery"), ): record = tasks.get(key) - if isinstance(record, dict) and record.get("finished_at"): - interval = int(interval_minutes) * 60 - candidate = int(record["finished_at"]) + interval - while candidate <= now: - candidate += interval - next_runs[name] = candidate + if not (isinstance(record, dict) and record.get("finished_at")): + continue + finished_at = int(record["finished_at"]) + if name == "discover": + # 只按上次完成推下一档；进行中由 running 展示，不按墙钟抢跑。 + next_runs[name] = finished_at + discover_interval + else: + next_runs[name] = int( + next_daily_run(finished_at, cfg.schedule.snapshot_at) + ) errors = [ {"task": name, **record} for name, record in tasks.items() @@ -1754,6 +1785,7 @@ class DashboardQueries: "counts": counts, "blacklist": blacklist, "tasks": tasks, + "running": running, "errors": errors, "next_runs": next_runs, "auth": { diff --git a/reaction_board/web/static/views/system.js b/reaction_board/web/static/views/system.js index 9fd9767..85b7fd5 100644 --- a/reaction_board/web/static/views/system.js +++ b/reaction_board/web/static/views/system.js @@ -93,11 +93,11 @@ export async function renderSystem(root) {
${sectionHeader("预计下次运行")}
-
搜索发现${escapeHtml(formatDateTime(next.discover))}
-
每日快照${escapeHtml(formatDateTime(next.snapshot))}
-
发现周期${system.schedule.discover_minutes} 分钟
+
咻咻满粉丝数更新${running.fans ? "运行中" : escapeHtml(formatDateTime(next.fans))}
+
视频数据快照 + UP 信息${running.snapshot ? "运行中" : escapeHtml(formatDateTime(next.snapshot))}
+
新视频搜索${running.discover || running.discovery ? "运行中" : escapeHtml(formatDateTime(next.discover))}
-

预计时间按配置和最近任务完成时间推算；实际以常驻进程为准。

+

进行中显示「运行中」；空闲时按该任务上次完成时间推算下次运行，不按墙钟抢跑。

diff --git a/tests/test_snapshot.py b/tests/test_snapshot.py index 74720af..be18b35 100644 --- a/tests/test_snapshot.py +++ b/tests/test_snapshot.py @@ -256,13 +256,19 @@ def test_fan_failure_placeholder(): transport = SnapshotTransport({}, fan={"code": -500, "message": "x", "data": None}) db, repo, service = make_service(tmp, transport, retries=2) try: + # 合并跑：视频完成日不受粉丝失败影响 result = service.run(date=DATE) assert result.total == 0 assert result.fans_failed == 1 - assert result.completed is False + assert result.completed is True + assert repo.kv_get("snapshot:last_completed") == DATE row = repo.get_up_stat(DATE, 37754047) assert row["ok"] == 0 assert row["follower"] is None and row["following"] is None + + fans_only = service.run_fans(date=DATE, force=True) + assert fans_only.fans_failed == 1 + assert fans_only.completed is False finally: db.close() diff --git a/tests/test_web_queries.py b/tests/test_web_queries.py index d4fe29f..c905446 100644 --- a/tests/test_web_queries.py +++ b/tests/test_web_queries.py @@ -357,10 +357,14 @@ def test_fans_candidates_system_and_export(): assert system["snapshot_days"] == 3 assert system["blacklist"][0]["video_count"] == 1 assert system["tasks"]["snapshot"]["ok"] is True + assert system["tasks"]["fans"]["ok"] is True assert system["errors"] == [] assert system["preview"] is False + assert system["running"] == {} + assert set(system["next_runs"]) >= {"fans", "snapshot", "discover"} assert system["next_runs"]["snapshot"] > 0 - assert system["next_runs"]["discover"] > FIXED_NOW + assert system["next_runs"]["fans"] > 0 + assert system["next_runs"]["discover"] > 0 finally: db.close() @@ -562,8 +566,92 @@ def main_test(): test_stats_query() test_fans_custom_range_modes() test_fans_candidates_system_and_export() + test_system_status_uses_completed_date_and_running_flags() + test_system_next_runs_follow_last_completion_not_wall_clock() print("OK: Web 查询层总览/榜单/详情/UP/时长效率/粉丝/候选/系统/访问统计测试全部通过") + +def test_system_status_uses_completed_date_and_running_flags(): + """采集中途的 ok=1 不得抢跑数据日期；running 显示三任务。""" + import json + + with tempfile.TemporaryDirectory() as tmp: + db, q = setup(tmp) + try: + repo = Repository(db) + # 模拟今日部分写入成功，但完成日仍停在前一天 + today = "2026-11-01" + repo.upsert_snapshot( + today, 101, + metrics={"view": 1, "like": 0, "coin": 0, "favorite": 0, + "share": 0, "reply": 0, "danmaku": 0}, + ok=True, err=None, captured_at=FIXED_NOW, + ) + assert repo.kv_get("snapshot:last_completed") == DATES[-1] + + system = q.system() + overview = q.overview() + assert system["data_date"] == DATES[-1] + assert overview["data_date"] == DATES[-1] + assert system["snapshot_days"] == 3 + + # 三个下次运行字段；标记 snapshot 运行中 + repo.kv_set( + "task:running:snapshot", + json.dumps({"name": "snapshot", "started_at": FIXED_NOW}), + ) + DASHBOARD_CACHE.clear() + system = q.system() + assert "snapshot" in system["running"] + assert system["running"]["snapshot"]["running"] is True + assert set(system["next_runs"]) >= {"fans", "snapshot", "discover"} + + # 失败记录保留，完成日不变 + repo.kv_set( + "task:last:snapshot", + json.dumps({ + "name": "snapshot", "ok": False, + "started_at": FIXED_NOW - 10, "finished_at": FIXED_NOW, + "error": "boom", "summary": None, + }), + ) + repo.kv_delete("task:running:snapshot") + DASHBOARD_CACHE.clear() + system = q.system() + assert system["data_date"] == DATES[-1] + assert overview["data_date"] == DATES[-1] + assert any(err["task"] == "snapshot" for err in system["errors"]) + assert system["tasks"]["snapshot"]["ok"] is False + finally: + db.close() + + +def test_system_next_runs_follow_last_completion_not_wall_clock(): + import json + + with tempfile.TemporaryDirectory() as tmp: + db, q = setup(tmp) + try: + repo = Repository(db) + # 发现任务刚完成不久：下次 = finished + 60min，即使 now 已超过也不跳档 + finished = FIXED_NOW - 30 + repo.kv_set( + "task:last:discovery", + json.dumps({ + "name": "discovery", "ok": True, + "started_at": finished - 5, "finished_at": finished, + "error": None, "summary": {}, + }), + ) + 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() + + if __name__ == "__main__": main_test() diff --git a/tests/web_fixtures.py b/tests/web_fixtures.py index 41dcde8..9fcd9a8 100644 --- a/tests/web_fixtures.py +++ b/tests/web_fixtures.py @@ -166,6 +166,8 @@ def build_db(tmp: str | Path) -> Path: repo.mark_candidate_decision(202, "rejected", now=ts("2026-10-30")) repo.finish_folder_sync(FOLDER, media_count=3, full_refresh=True) repo.kv_set("snapshot:last_completed", DATES[-1]) + for day in DATES: + repo.kv_set(f"snapshot:completed:{day}", "1") repo.kv_set("auth:ok", "1") repo.kv_set("auth:message", "OK") repo.kv_set( @@ -176,6 +178,22 @@ def build_db(tmp: str | Path) -> Path: "error": None, "summary": {"completed": True}, }), ) + repo.kv_set( + "task:last:fans", + json.dumps({ + "name": "fans", "ok": True, + "started_at": ts(DATES[-1]) + 60, "finished_at": ts(DATES[-1]) + 90, + "error": None, "summary": {"fans_ok": 1, "fans_failed": 0}, + }), + ) + repo.kv_set( + "task:last:discovery", + json.dumps({ + "name": "discovery", "ok": True, + "started_at": ts(DATES[-1]) - 3600, "finished_at": ts(DATES[-1]) - 3500, + "error": None, "summary": {"candidates": 0}, + }), + ) finally: db.close() return path diff --git a/tools/seed_preview_db.py b/tools/seed_preview_db.py index 146602d..1a93994 100644 --- a/tools/seed_preview_db.py +++ b/tools/seed_preview_db.py @@ -278,6 +278,13 @@ def _seed_task_records( "error": None, "summary": {"refs": 2, "added": 2, "ok": True}, }, + "fans": { + "name": "fans", "ok": True, + "started_at": int(base + 1124), + "finished_at": int(base + 1130), + "error": None, + "summary": {"fans_ok": 1, "fans_failed": 0, "fans_skipped": 0}, + }, "discovery": { "name": "discovery", "ok": True, "started_at": int(base + 21 * 3600), @@ -309,6 +316,17 @@ def _seed_task_records( "INSERT OR REPLACE INTO kv(k, v) VALUES ('snapshot:last_completed', ?)", (latest_day.isoformat(),), ) + conn.execute( + "INSERT OR REPLACE INTO kv(k, v) VALUES (?, '1')", + (f"snapshot:completed:{latest_day.isoformat()}",), + ) + for (day,) in conn.execute( + "SELECT DISTINCT stat_date FROM snapshot WHERE ok = 1" + ).fetchall(): + conn.execute( + "INSERT OR REPLACE INTO kv(k, v) VALUES (?, '1')", + (f"snapshot:completed:{day}",), + ) conn.execute( "INSERT OR REPLACE INTO kv(k, v) VALUES (?, '1')", (f"snapshot:backup:{latest_day.isoformat()}",), -- 2.54.0