fix(system): three next-runs, completed-date data_date, running mark

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

@@ -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: