tools: add split_config_secrets for config.json → jsonc + secrets

DEVELOPMENT.md

@@ -267,6 +267,7 @@ PRAGMA synchronous = NORMAL;
 业务参数放在 `config.jsonc`（JSONC：`//` / `/* */` 注释与尾逗号；实现为自研轻量剥离后再 `json.loads`，无额外依赖）。
 敏感凭据放同目录 `secrets.json`（纯 JSON，供 `auth` 原子回写）。二者都必须被 Git 忽略，日志中不得输出敏感值。
 默认路径优先 `config.jsonc`，否则回落 `config.json`；旧单体 `config.json`（凭据内嵌）仍可加载，若存在 `secrets.json` 则凭据以它为准。
+从旧单体迁到拆分布局：`uv run python tools/split_config_secrets.py [--dry-run] [--force] [--keep-src]`（默认写出同目录 `config.jsonc` + `secrets.json`，并把源改名为 `.bak`）。
 
 `config.jsonc` 示例：
 
@@ -456,9 +457,6 @@ P1 实现说明：
 
 - 时区使用固定 UTC+8（中国无夏令时），避免 Windows 缺少 `tzdata`；
 - 午夜 `seed_at`（默认 `00:00`）把 `snapshot:last_completed` 的 ok=1 行克隆为当日壳，写入 `snapshot:seeded:<date>=1` / `fans:seeded:<date>=1`，**不**推进完成日；
-- 也可手动：`python app.py snapshot-seed [--date YYYY-MM-DD]`（Debian / `docker compose exec re-xxm python app.py snapshot-seed`）；
-- 目标日已有 `snapshot:completed:<date>` 时 seed 记 `already_completed` 成功（即使 `last_completed` 已等于当日），**不得**误判 `no_source`；
-- 若全站 `snapshot` 正在跑，或目标日未完成且 `last_completed` 暂不可用（`no_source`），seed **不记成功**，约 10 分钟后重试直到真正克隆成功；
 - 晚间 `snapshot_at`（默认 `23:15`）对 seeded 日 force 重采；成功后写 `snapshot:completed:<date>=1`、推进 `last_completed` 并清除 seeded；
 - 当日全部视频与粉丝采集成功后在 `kv` 写入 `snapshot:completed:<date>=1`；启动时未完成会补跑（纯 seed 日不误标完成）；
 - 完成标记不是唯一门槛：每次检查 `pending_snapshot_aids(date)`，新入库或从 `dropped` 恢复的视频会立即补采当天快照；
@@ -469,15 +467,6 @@ 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 与入库操作
@@ -620,7 +609,7 @@ interaction_rate = (like + coin + favorite) / max(view, 1)
 
 - 查询前一有效快照而不是填补缺失日期；
 - 缺失日期在图表中断开；
-- 增速榜要求播放基数至少 1000；
+- 增速榜要求播放基数至少 200；
 - 互动率榜要求播放至少 100；
 - 样本不足两天的视频明确标注，不伪造增长数据。
 
@@ -768,7 +757,7 @@ uv run python tests/test_moves.py
 - 粉丝采集对每条 `relation/stat` 加 45s 硬超时看门狗（防 DNS/传输层挂起），每 25 条输出 `up_profile.progress` 进度；单条失败不阻塞整批；
 - `app.py backfill-followers`：一次性回填历史视频粉丝数，支持 `--limit` / `--dry-run`；
 - 新增内容效率/互动效率榜与 UP 内容效率（Gamma–Poisson EB 收缩），过滤 < 600 秒与未满 2 天的视频；
-- `analytics` 配置段：`min_duration_seconds`、`maturity_days`、`min_view_for_ranking`、`video_shrink_minutes`、`shrink_minutes`、`growth_rate_prior`；
+- `analytics` 配置段：`min_duration_seconds`、`maturity_days`、`min_view_for_ranking`、`video_shrink_minutes`、`shrink_minutes`；
 - 主导航拆分：`/videos`（视频）与 `/ups`（UP主）独立；旧 `/rank` 保留兼容；
 - 新增 `/about` 数据说明页与 `/api/about`；
 - 卡片精简：列表只保留主指标与互动率，明细移到详情页；

README.md

@@ -314,6 +314,13 @@ Copy-Item secrets.example.json secrets.json
 默认启动优先读 `config.jsonc`，若不存在则回落 `config.json`（旧单体文件仍可用）。
 `python app.py auth` **只回写** `secrets.json`，不会冲掉 JSONC 注释。
 
+若部署目录仍是旧版单体 `config.json`，可用拆分脚本一次迁出：
+
+```powershell
+uv run python tools/split_config_secrets.py --dry-run
+uv run python tools/split_config_secrets.py  # 写出 config.jsonc + secrets.json，源改名为 config.json.bak
+```
+
 凭据文件双账号格式：
 
 ```json

config.docker.example.jsonc

@@ -108,6 +108,5 @@
     "video_shrink_minutes": 60,
     "shrink_minutes": null,
     "home_growth_days": "auto",
-    "growth_rate_prior": 1000,
   },
 }

config.example.jsonc

@@ -123,7 +123,5 @@
     "shrink_minutes": null,
     // "auto" = 7 日有数据用 7，否则 1
     "home_growth_days": "auto",
-    // 增速榜排序：Δ / (基线 + prior)。卡片上的百分比仍是 Δ / 基线。
-    "growth_rate_prior": 1000,
   },
 }

project.md

@@ -380,7 +380,7 @@ for folder in 配置的年度夹（2025、2026、…）:
 - 两个夹共 831 条，一次性导入 `video` 表；
 - **历史快照无法回填**：B 站不提供任意日期历史数据，只能从部署当天开始积累；
 - **开发联调阶段产生的数据一律保留**（不清库、不重建），正式上线直接续用同一个 `app.db`；
-- 新发现/新入库的视频从"入库当天"开始积累快照。近 1 日：当日发布按 0 起算（一次成功采样即可）；更早发布的新入库按入库时播放量起算，第二次采样走晚间全量，不进小时快照。近 7/30 日仍要跨日快照。
+- 新发现/新入库的视频从"入库当天"开始积累快照，第一天没有增量是正常的，看板对"仅 1 个数据点"的视频标注"样本不足"。
 
 ---
 
@@ -473,8 +473,6 @@ CREATE TABLE snapshot (
   ok          INTEGER NOT NULL DEFAULT 1,
   err         TEXT,
   src         TEXT DEFAULT 'view',      -- view | infos（降级时标记）
-  view_open   INTEGER,                  -- 同日第一次成功采样的播放量
-  open_captured_at INTEGER,             -- 开盘采样时刻
   PRIMARY KEY (stat_date, aid)
 );
 CREATE INDEX idx_snap_aid ON snapshot(aid, stat_date);
@@ -674,26 +672,23 @@ w = {view:0.30, like:0.20, coin:0.20, favorite:0.15,
 
 ### 8.2 增长榜
 
-> 看板 UI 中“增长”标注为**播放增量**（区间内绝对增量），“增速”标注为**播放增速**（增量 / 基线播放）。近 7/30 日基线取不早于 `最新快照 - N 天` 的最近一条成功快照。近 1 日：有更早日期快照则用前一日；否则当日发布基准为 0，更早发布的新入库用当天第一次成功采样（`view_open`）。
+> 看板 UI 中“增长”标注为**播放增量**（区间内绝对增量），“增速”标注为**播放增速**（增量 / 基线播放）。两者基线都取不早于 `最新快照 - N 天` 的最近一条成功快照。
 
 ```
 d7 = (view_today - view_{today-7}) / 实际间隔天数
 r7 = (view_today - view_{today-7}) / max(view_{today-7}, 1)
-d1_published_today = view_today - 0          # 只进增量榜，不计算增速
-d1_imported_today  = view_today - view_open   # 需同日第二次采样
-sort_rate = Δ / (基线 + analytics.growth_rate_prior)  # 默认 prior=1000，只用于增速榜排序
 ```
 
 | 榜 | 口径 | 基数门槛 |
 |---|---|---|
 | 累计热度 | §8.1 | — |
 | 近 7 天播放增量 | `d7(view)` | — |
-| 近 7 天增速 | `r7(view)` | 基线播放 `>= 1000` |
+| 近 7 天增速 | `r7(view)` | `view >= 200` |
 | 互动率 | `(like+coin+fav)/view` | `view >= 100` |
 | 单项榜 | 7 项直接排序 | — |
 | UP 榜 | 按 `up_mid` 聚合 Σ/AVG | — |
 
-**冷启动**：近 1 日不要求跨日快照。当日发布相对 0，只进播放增量、不进播放增速；更早发布且当天入库的片子相对 `view_open`（第二次采样走晚间全量，不进小时快照）。近 7/30 日仍要跨日快照。缺失日不插值。
+**冷启动**：不足 2 天数据的视频用"发布后 24h 增量"，并标注"样本不足"；缺失日不插值。
 
 ### 8.3 趋势
 
@@ -800,7 +795,7 @@ secrets.example.json            凭据样例（auth 回写目标）
 **进程结构**
 
 ```
-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)
+scheduler 线程：discovery(每 60min) / snapshot_seed(每日 00:00) / snapshot+fans(每日 23:15) / snapshot_recent(小时) / candidate_cleanup(每日)
 web 线程：ThreadingHTTPServer :8080
 telegram：getUpdates long polling（回调 + 私聊文本入库）
 storage：SQLite WAL，单写线程

reaction_board/analytics.py

@@ -34,7 +34,7 @@ DEFAULT_HEAT_WEIGHTS: dict[str, float] = {
 }
 
 MIN_VIEW_FOR_INTERACTION = 100
-MIN_VIEW_FOR_GROWTH_RATE = 1000
+MIN_VIEW_FOR_GROWTH_RATE = 200
 MIN_EFFICIENCY_DURATION = 600
 DEFAULT_MATURITY_DAYS = 2
 DEFAULT_SHRINK_MINUTES = 120.0
@@ -313,20 +313,15 @@ def compute_growth(
     current: int | float | None,
     previous: int | float | None,
     gap_days: int | float | None,
-    *,
-    allow_zero_gap: bool = False,
 ) -> Growth:
     """计算累计差值、日均差值和相对增速。
 
-    缺失基线或间隔为负时返回不可用。默认间隔不足 1 天也不可用，不把 0 当成真实增长。
-    ``allow_zero_gap=True`` 时允许同一天的两次采样（或相对 0）计算增量，``per_day`` 为 None。
+    缺失基线或间隔不足 1 天时返回不可用，不把 0 当作真实增长。
     """
     if current is None or previous is None or gap_days is None:
         return Growth(available=False)
     gap = int(gap_days)
-    if gap < 0:
-        return Growth(available=False)
-    if gap == 0 and not allow_zero_gap:
+    if gap <= 0:
         return Growth(available=False)
     current_value = max(0, int(current))
     previous_value = max(0, int(previous))
@@ -334,35 +329,12 @@ def compute_growth(
     return Growth(
         available=True,
         delta=delta,
-        per_day=None if gap == 0 else delta / gap,
+        per_day=delta / gap,
         rate=delta / max(previous_value, 1),
         gap_days=gap,
     )
 
 
-def smoothed_growth_rate(
-    delta: int | float | None,
-    previous: int | float | None,
-    prior: int | float,
-) -> float | None:
-    """增速榜排序分：``Δ / (基线 + prior)``。
-
-    ``rate`` 仍是真实相对增速。``prior`` 只削弱小基数在排序里的优势。
-    """
-    if delta is None or previous is None or isinstance(delta, bool) or isinstance(previous, bool):
-        return None
-    if isinstance(prior, bool):
-        return None
-    try:
-        strength = float(prior)
-        base = max(0, int(previous)) + strength
-    except (TypeError, ValueError):
-        return None
-    if strength < 0 or base <= 0:
-        return None
-    return int(delta) / base
-
-
 def normalize_cover(url: object) -> str | None:
     """把 B 站封面统一为 https；空值返回 None。"""
     if not isinstance(url, str):

reaction_board/cli.py

@@ -4,7 +4,6 @@
     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 登录态
@@ -34,22 +33,13 @@ from .database import Database
 from .logging_json import JsonLogger
 from .notify.telegram import TelegramClient
 from .repository import Repository
-from .scheduler import ScheduledTask, Scheduler, TaskControl
+from .scheduler import ScheduledTask, Scheduler
 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 (
-    in_snapshot_quiet_window,
-    interval_next_run,
-    is_date,
-    next_after_snapshot_quiet_window,
-    next_daily_run,
-    parse_hhmm,
-    shanghai_date,
-    shanghai_datetime,
-)
+from .timeutil import next_daily_run, parse_hhmm, shanghai_date
 
 ROOT = Path(__file__).resolve().parent.parent
 DEFAULT_CONFIG = default_config_path(ROOT)
@@ -81,17 +71,6 @@ 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="执行一次搜索发现并写入候选池"
     )
@@ -185,7 +164,6 @@ 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
@@ -201,19 +179,12 @@ def _record_task(
     finished_at: int | None = None,
     payload=None,
     error: str | None = None,
-    skipped: bool = False,
 ) -> None:
-    """把最近一次任务结果写入 kv，并清除 running 标记。
-
-    ``skipped=True`` 表示主动延后/避让，不得记成成功，也不推进「当日已完成」。
-    """
+    """把最近一次任务结果写入 kv，并清除 running 标记。"""
     now = int(time.time())
-    if skipped:
-        ok = False
     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),
         "error": error,
@@ -226,125 +197,6 @@ def _record_task(
     _clear_task_running(repo, name)
 
 
-# 全站快照 / 粉丝失败后的补跑间隔（秒）
-_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)
-    _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,
-    )
-    return TaskControl(
-        ok=False,
-        next_run=next_after_snapshot_quiet_window(float(finished)),
-        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)
-    _record_task(
-        repo,
-        "snapshot_seed",
-        ok=False,
-        skipped=True,
-        started_at=started_at,
-        finished_at=finished,
-        payload={"skipped": True, "reason": reason},
-        error=reason,
-    )
-    return TaskControl(
-        ok=False,
-        next_run=_seed_retry_soon(float(finished)),
-        error=reason,
-    )
-
-
-def _initial_daily_next_run(
-    repo: Repository,
-    name: str,
-    at: str,
-    *,
-    now: float | None = None,
-    retry_seconds: float | None = None,
-) -> float:
-    """进程启动时日更任务的下次运行：墙钟下一档，或与系统页一致的上次记录推导。
-
-    不得返回 ``0`` / 过去时间来“立刻跑一遍”——午夜 seed 后重启否则会白天全量强采。
-    """
-    current = time.time() if now is None else float(now)
-    wall = next_daily_run(current, at)
-    raw = repo.kv_get(f"task:last:{name}")
-    if not raw:
-        return wall
-    try:
-        record = json.loads(raw)
-    except (TypeError, ValueError, json.JSONDecodeError):
-        return wall
-    if not isinstance(record, dict) or record.get("finished_at") is None:
-        return wall
-    finished_at = float(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 skipped or failed:
-        gap = (
-            _SEED_RETRY_SECONDS if name == "snapshot_seed"
-            else float(_TASK_RETRY_SECONDS if retry_seconds is None else retry_seconds)
-        )
-        retry_at = finished_at + gap
-        return retry_at if retry_at > current else wall
-    scheduled = next_daily_run(finished_at, at)
-    # 成功记录推到的下一档若已过，退回墙钟下一档（严格 > now）
-    return scheduled if scheduled > current else wall
-
-
-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 _client(credentials, logger: JsonLogger, *, pacer=None, retries: int = 3) -> BilibiliClient:
     return BilibiliClient(
         cookie=credentials.cookie,
@@ -502,82 +354,6 @@ def cmd_snapshot(args, logger: JsonLogger) -> int:
         db.close()
 
 
-def cmd_snapshot_seed(args, logger: JsonLogger) -> int:
-    """离线克隆：``python app.py snapshot-seed [--date YYYY-MM-DD]``。
-
-    与调度器同一套规则：全站 ``snapshot`` 运行中或 ``no_source`` 时不记成功，
-    返回码非 0，便于 ``docker compose exec re-xxm python app.py snapshot-seed`` 排查。
-    """
-    store = ConfigStore(args.config, require_auth=False)
-    cfg = store.get()
-    db = _open_db(cfg, logger)
-    try:
-        repo = Repository(db)
-        target = args.date
-        if target is not None and not is_date(target):
-            raise ConfigError(f"日期格式必须是 YYYY-MM-DD: {target!r}")
-        started = int(time.time())
-        if _is_task_running(repo, "snapshot"):
-            payload = {
-                "ok": False,
-                "skipped": True,
-                "reason": _SEED_SKIP_SNAPSHOT_RUNNING,
-                "stat_date": target or shanghai_date(),
-                "database": str(cfg.storage.database_path),
-            }
-            _record_task(
-                repo,
-                "snapshot_seed",
-                ok=False,
-                skipped=True,
-                started_at=started,
-                finished_at=int(time.time()),
-                payload=payload,
-                error=_SEED_SKIP_SNAPSHOT_RUNNING,
-            )
-            _print(payload)
-            return 1
-        # seed 只读本地库，不调 B 站；占位 client 满足构造约束。
-        service = SnapshotService(store, repo, object(), logger=logger)
-        _mark_task_running(repo, "snapshot_seed", started_at=started)
-        try:
-            summary = service.run_seed_from_last_completed(date=target)
-        except Exception as exc:  # noqa: BLE001
-            _record_task(
-                repo, "snapshot_seed", ok=False, started_at=started,
-                finished_at=int(time.time()), error=str(exc),
-            )
-            raise
-        payload = summary.as_dict()
-        payload["database"] = str(cfg.storage.database_path)
-        finished = int(summary.finished_at or time.time())
-        if summary.reason == _SEED_SKIP_NO_SOURCE:
-            _record_task(
-                repo,
-                "snapshot_seed",
-                ok=False,
-                skipped=True,
-                started_at=int(summary.started_at or started),
-                finished_at=finished,
-                payload={**payload, "skipped": True, "reason": _SEED_SKIP_NO_SOURCE},
-                error=_SEED_SKIP_NO_SOURCE,
-            )
-            _print(payload)
-            return 1
-        _record_task(
-            repo,
-            "snapshot_seed",
-            ok=True,
-            started_at=int(summary.started_at or started),
-            finished_at=finished,
-            payload=payload,
-        )
-        _print(payload)
-        return 0
-    finally:
-        db.close()
-
-
 def _telegram(cfg, logger: JsonLogger):
     if not cfg.telegram.enabled:
         return None
@@ -728,248 +504,173 @@ 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  记录后补跑，不抛给次日
-                finished = int(time.time())
+            except Exception as exc:  # noqa: BLE001  记录后交给调度器隔离
                 _record_task(
                     repo, "discovery", ok=False, started_at=started,
-                    finished_at=finished, error=str(exc),
-                )
-                return TaskControl(
-                    ok=False, next_run=_retry_soon(float(finished)), error=str(exc),
+                    finished_at=int(time.time()), error=str(exc),
                 )
+                raise
             payload = summary.as_dict()
-            ok = not summary.errors
-            finished = int(summary.finished_at or time.time())
             _record_task(
-                repo, "discovery", ok=ok,
+                repo, "discovery", ok=not summary.errors,
                 started_at=summary.started_at,
-                finished_at=finished,
+                finished_at=summary.finished_at or int(time.time()),
                 payload=payload,
                 error="；".join(summary.errors) if summary.errors else None,
             )
             last["discover"] = summary
             logger.info("discovery.summary", **payload)
-            next_run = (
-                interval_next_run(float(finished), interval) if ok
-                else _retry_soon(float(finished))
-            )
-            return TaskControl(ok=ok, next_run=next_run)
+            return summary
 
         def run_cleanup():
             started = int(time.time())
-            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())
-                _record_task(
-                    repo, "candidate_cleanup", ok=False, started_at=started,
-                    finished_at=finished, error=str(exc),
-                )
-                return TaskControl(
-                    ok=False, next_run=_retry_soon(float(finished)), error=str(exc),
-                )
+            removed = repo.delete_expired_candidates(
+                ttl_days=cfg.schedule.candidate_ttl_days
+            )
+            expired = repo.expire_fav_actions()
             if removed or expired:
                 repo.bump_cache_version("videos")
-            finished = int(time.time())
             logger.info(
                 "cleanup.done",
                 candidates_removed=removed, fav_actions_expired=expired,
             )
             _record_task(
                 repo, "candidate_cleanup", ok=True, started_at=started,
-                finished_at=finished,
+                finished_at=int(time.time()),
                 payload={
                     "candidates_removed": removed,
                     "fav_actions_expired": expired,
                 },
             )
-            return TaskControl(
-                ok=True,
-                next_run=interval_next_run(float(finished), 86400.0),
-            )
+            return {
+                "candidates_removed": removed,
+                "fav_actions_expired": expired,
+            }
 
         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)
             try:
-                # 仅晚间 snapshot_at 及之后：seed 壳才 force 打穿 ok=1 克隆行。
-                # 白天/重启不得因 seeded 标记立刻全量强采。
+                # 晚间全量：若当日是午夜 seed 壳，必须 force 打穿 ok=1 克隆行。
                 today = shanghai_date()
-                force_seed = (
-                    repo.kv_get(SnapshotService.seeded_key(today)) == "1"
-                    and _reached_daily_slot(snapshot_at, float(started))
-                )
+                force_seed = repo.kv_get(
+                    SnapshotService.seeded_key(today)
+                ) == "1"
                 summary = snapshot_service.run_if_needed(force=force_seed)
             except Exception as exc:  # noqa: BLE001
-                finished = int(time.time())
                 _record_task(
                     repo, "snapshot", ok=False, started_at=started,
-                    finished_at=finished, error=str(exc),
-                )
-                return TaskControl(
-                    ok=False, next_run=_retry_soon(float(finished)), error=str(exc),
+                    finished_at=int(time.time()), 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())
             _record_task(
                 repo, "snapshot", ok=ok, started_at=summary.started_at,
-                finished_at=finished,
+                finished_at=summary.finished_at or int(time.time()),
                 payload=payload, error=summary.backup_error,
             )
             last["snapshot"] = summary
             logger.info("snapshot.summary", **payload)
-            next_run = (
-                next_daily_run(float(finished), snapshot_at) if ok
-                else _retry_soon(float(finished))
-            )
-            return TaskControl(ok=ok, next_run=next_run)
+            return summary
 
         def run_snapshot_recent():
             started = int(time.time())
-            if in_snapshot_quiet_window(float(started)):
-                return _quiet_window_defer(
-                    repo, logger, "snapshot_recent", started_at=started,
+            if _is_task_running(repo, "snapshot"):
+                logger.info(
+                    "snapshot.recent_skip", reason="snapshot_running",
                 )
+                _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())
                 _record_task(
                     repo, "snapshot_recent", ok=False, started_at=started,
-                    finished_at=finished, error=str(exc),
-                )
-                return TaskControl(
-                    ok=False, next_run=_retry_soon(float(finished)), error=str(exc),
+                    finished_at=int(time.time()), 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())
             _record_task(
                 repo, "snapshot_recent", ok=ok, started_at=summary.started_at,
-                finished_at=finished,
+                finished_at=summary.finished_at or int(time.time()),
                 payload=payload, error=error,
             )
             last["snapshot_recent"] = summary
             logger.info("snapshot.recent_summary", **payload)
-            next_run = (
-                interval_next_run(float(finished), interval) if ok
-                else _retry_soon(float(finished))
-            )
-            return TaskControl(ok=ok, next_run=next_run)
+            return summary
 
         def run_fans():
-            """粉丝更新短耗时：不避开任何任务，可与全量并行。"""
             started = int(time.time())
             _mark_task_running(repo, "fans", started_at=started)
             try:
                 today = shanghai_date()
-                force_seed = (
-                    repo.kv_get(SnapshotService.fans_seeded_key(today)) == "1"
-                    and _reached_daily_slot(snapshot_at, float(started))
-                )
+                force_seed = repo.kv_get(
+                    SnapshotService.fans_seeded_key(today)
+                ) == "1"
                 summary = snapshot_service.run_fans_if_needed(force=force_seed)
             except Exception as exc:  # noqa: BLE001
-                finished = int(time.time())
                 _record_task(
                     repo, "fans", ok=False, started_at=started,
-                    finished_at=finished, error=str(exc),
-                )
-                return TaskControl(
-                    ok=False, next_run=_retry_soon(float(finished)), error=str(exc),
+                    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
-            finished = int(summary.finished_at or time.time())
             _record_task(
                 repo, "fans", ok=ok, started_at=summary.started_at,
-                finished_at=finished,
+                finished_at=summary.finished_at or int(time.time()),
                 payload=payload, error=error,
             )
             last["fans"] = summary
             logger.info("fans.summary", **payload)
-            next_run = (
-                next_daily_run(float(finished), snapshot_at) if ok
-                else _retry_soon(float(finished))
-            )
-            return TaskControl(ok=ok, next_run=next_run)
+            return summary
 
         def run_snapshot_seed():
             started = int(time.time())
-            # 全站快照占用时不执行、不记成功；10 分钟后重试直到真正 seed 成功。
-            if _is_task_running(repo, "snapshot"):
-                return _defer_snapshot_seed(
-                    repo,
-                    logger,
-                    started_at=started,
-                    reason=_SEED_SKIP_SNAPSHOT_RUNNING,
-                )
             _mark_task_running(repo, "snapshot_seed", started_at=started)
             try:
                 summary = snapshot_service.run_seed_from_last_completed()
             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),
-                )
-                return TaskControl(
-                    ok=False,
-                    next_run=_seed_retry_soon(float(finished)),
-                    error=str(exc),
+                    finished_at=int(time.time()), error=str(exc),
                 )
+                raise
             payload = summary.as_dict()
-            finished = int(summary.finished_at or time.time())
-            last["snapshot_seed"] = summary
-            logger.info("snapshot.seed_summary", **payload)
-            if summary.reason == _SEED_SKIP_NO_SOURCE:
-                return _defer_snapshot_seed(
-                    repo,
-                    logger,
-                    started_at=int(summary.started_at or started),
-                    reason=_SEED_SKIP_NO_SOURCE,
-                    finished_at=finished,
-                )
-            # already_completed / already_seeded / seeded → 成功，推到次日 seed_at
             _record_task(
                 repo, "snapshot_seed", ok=True, started_at=summary.started_at,
-                finished_at=finished,
+                finished_at=summary.finished_at or int(time.time()),
                 payload=payload,
             )
-            return TaskControl(
-                ok=True, next_run=next_daily_run(float(finished), seed_at),
-            )
+            last["snapshot_seed"] = summary
+            logger.info("snapshot.seed_summary", **payload)
+            return summary
 
         def run_config_watch():
             before = store.revision
@@ -978,39 +679,33 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
                 logger.info("config.reloaded", revision=store.revision)
             return {"revision": store.revision}
 
-        # --once：保持 next_run=0 立刻跑到期逻辑；常驻：日更对齐墙钟，避免重启即强采。
-        now_ts = time.time()
-        if args.once:
-            seed_next = snapshot_next = fans_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,
-            )
+        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)
         scheduler = Scheduler(
             [
                 ScheduledTask(
                     "snapshot_seed",
                     86400.0,
                     run_snapshot_seed,
-                    next_run=seed_next,
+                    next_run=0.0,
+                    next_run_fn=lambda started: next_daily_run(started, seed_at),
                 ),
                 ScheduledTask(
                     "snapshot",
                     86400.0,
                     run_snapshot,
-                    next_run=snapshot_next,
+                    next_run=0.0,
+                    next_run_fn=lambda started: next_daily_run(started, snapshot_at),
                 ),
                 ScheduledTask(
                     "fans",
                     86400.0,
                     run_fans,
-                    next_run=fans_next,
+                    next_run=0.0,
+                    next_run_fn=lambda started: next_daily_run(started, snapshot_at),
                 ),
                 ScheduledTask(
                     "snapshot_recent",
@@ -1019,7 +714,7 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
                 ),
                 ScheduledTask(
                     "discovery",
-                    float(discover_minutes * 60),
+                    float(cfg.schedule.discover_minutes * 60),
                     run_discover,
                 ),
                 ScheduledTask("candidate_cleanup", 86400.0, run_cleanup),
@@ -1027,7 +722,6 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
             ],
             logger=logger,
         )
-
         if serve_web and not args.once:
             from .web.server import create_server, serve_forever
 
@@ -1169,8 +863,6 @@ 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":

reaction_board/config.py

@@ -52,7 +52,6 @@ DEFAULT_ANALYTICS: dict[str, Any] = {
     "video_shrink_minutes": 60,
     "shrink_minutes": None,
     "home_growth_days": "auto",
-    "growth_rate_prior": 1000,
 }
 
 DEFAULT_CONFIG: dict[str, Any] = {
@@ -211,7 +210,6 @@ class AnalyticsConfig:
     video_shrink_minutes: int
     shrink_minutes: int | None
     home_growth_days: str | int
-    growth_rate_prior: int
 
 
 @dataclass(frozen=True)
@@ -594,11 +592,6 @@ def _build_config(data: Mapping[str, Any], path: Path, *, secrets_path: Path) ->
             )
         ),
         home_growth_days=home_growth_days,
-        growth_rate_prior=_as_int(
-            analytics_raw.get("growth_rate_prior", 1000),
-            "analytics.growth_rate_prior",
-            minimum=0,
-        ),
     )
 
     return AppConfig(

reaction_board/migrations.py

@@ -244,18 +244,6 @@ MIGRATIONS: tuple[Migration, ...] = (
             "CREATE INDEX idx_visit_date ON visit(stat_date)",
         ),
     ),
-    Migration(
-        version=8,
-        name="snapshot open view",
-        statements=(
-            # 同日第一次成功采样的播放量，供近 1 日「入库后增量」做基准。
-            # 已有成功行无法找回被覆盖的入库值，只能把当前 view 当作开盘。
-            'ALTER TABLE snapshot ADD COLUMN view_open INTEGER',
-            "ALTER TABLE snapshot ADD COLUMN open_captured_at INTEGER",
-            'UPDATE snapshot SET view_open = "view", open_captured_at = captured_at '
-            "WHERE ok = 1 AND view_open IS NULL",
-        ),
-    ),
 )
 
 

reaction_board/repository.py

@@ -690,17 +690,12 @@ class Repository:
         src: str = "view",
     ) -> None:
         values = metrics or {}
-        ok_flag = 1 if ok else 0
-        view_value = values.get("view")
-        open_view = view_value if ok else None
-        open_at = int(captured_at) if ok else None
         def run(conn: sqlite3.Connection) -> None:
             conn.execute(
                 """
                 INSERT INTO snapshot(stat_date, aid, "view", "like", favorite, coin,
-                                     share, reply, danmaku, captured_at, ok, err, src,
-                                     view_open, open_captured_at)
-                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                                     share, reply, danmaku, captured_at, ok, err, src)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                 ON CONFLICT(stat_date, aid) DO UPDATE SET
                   "view" = excluded."view",
                   "like" = excluded."like",
@@ -712,22 +707,12 @@ class Repository:
                   captured_at = excluded.captured_at,
                   ok = excluded.ok,
                   err = excluded.err,
-                  src = excluded.src,
-                  view_open = CASE
-                    WHEN excluded.ok = 1 AND snapshot.view_open IS NULL
-                    THEN excluded."view"
-                    ELSE snapshot.view_open
-                  END,
-                  open_captured_at = CASE
-                    WHEN excluded.ok = 1 AND snapshot.open_captured_at IS NULL
-                    THEN excluded.captured_at
-                    ELSE snapshot.open_captured_at
-                  END
+                  src = excluded.src
                 """,
                 (
                     stat_date,
                     int(aid),
-                    view_value,
+                    values.get("view"),
                     values.get("like"),
                     values.get("favorite"),
                     values.get("coin"),
@@ -735,11 +720,9 @@ class Repository:
                     values.get("reply"),
                     values.get("danmaku"),
                     int(captured_at),
-                    ok_flag,
+                    1 if ok else 0,
                     err,
                     src,
-                    open_view,
-                    open_at,
                 ),
             )
         self.db.write(run)
@@ -758,23 +741,18 @@ class Repository:
                 """
                 INSERT INTO snapshot(
                   stat_date, aid, "view", "like", favorite, coin,
-                  share, reply, danmaku, captured_at, ok, err, src,
-                  view_open, open_captured_at
+                  share, reply, danmaku, captured_at, ok, err, src
                 )
                 SELECT
                   ?, aid, "view", "like", favorite, coin,
-                  share, reply, danmaku, ?, 1, NULL, ?,
-                  "view", ?
+                  share, reply, danmaku, ?, 1, NULL, ?
                 FROM snapshot
                 WHERE stat_date = ? AND ok = 1
                   AND aid NOT IN (
                     SELECT aid FROM snapshot WHERE stat_date = ?
                   )
                 """,
-                (
-                    target_date, int(captured_at), src, int(captured_at),
-                    source_date, target_date,
-                ),
+                (target_date, int(captured_at), src, source_date, target_date),
             )
             return int(cur.rowcount or 0)
         return self.db.write(run)

reaction_board/scheduler.py

@@ -3,9 +3,6 @@
 
 P0 只注册收藏夹同步任务；P1/P3 会在同一调度器上追加快照、粉丝和发现任务。
 调度只负责“什么时候运行”，具体任务保持幂等，并在自己的服务层处理重试。
-
-任务 ``fn`` 可返回 :class:`TaskControl` 以覆盖 ``ok`` / ``next_run``
-（用于静默窗延后、失败补跑等）；未返回时仍按 ``next_run_fn`` / ``interval`` 排期。
 """
 
 from __future__ import annotations
@@ -15,15 +12,6 @@ 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
@@ -82,48 +70,23 @@ class Scheduler:
                 continue
             started = current
             error: str | None = None
-            control: TaskControl | None = None
             try:
-                raw = task.fn()
+                task.fn()
             except Exception as exc:  # noqa: BLE001  单任务失败不终止调度
                 error = str(exc)
+                self._log("error", "scheduler.task_error", task=task.name, error=error)
             else:
-                if isinstance(raw, TaskControl):
-                    control = raw
+                self._log("info", "scheduler.task_ok", task=task.name)
             finished = self._clock()
-            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:
+            if 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=ok,
+                ok=error is None,
                 error=error,
             )
             results.append(result)

reaction_board/services/snapshot.py

@@ -51,7 +51,6 @@ 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]:
@@ -74,7 +73,6 @@ class SnapshotSummary:
             "backup_error": self.backup_error,
             "completed": self.completed,
             "already_completed": self.already_completed,
-            "reason": self.reason,
             "errors": self.errors,
         }
 
@@ -312,8 +310,7 @@ class SnapshotService:
         """刷新近 N 天新视频的当日快照；不碰完成日标记与日备份。
 
         「近 N 天」按 pubdate 墙钟窗口：``now - recent_video_days * 86400``，
-        不是上海日历日差。当天入库但发布更早的旧片不进小时任务。
-        允许覆盖同日已有 ``ok=1`` 行以便小时级刷新。
+        不是上海日历日差。允许覆盖同日已有 ``ok=1`` 行以便小时级刷新。
         """
         cfg = self.config_store.get()
         client = self._client()
@@ -538,8 +535,15 @@ class SnapshotService:
             raise ValueError(f"日期格式必须是 YYYY-MM-DD: {stat_date!r}")
 
         summary = SnapshotSummary(stat_date=stat_date, started_at=now)
-        # 目标日已完成：先于 no_source。全站快照当天完成后
-        # last_completed == stat_date，不可误判为无源再 10 分钟重试。
+        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
         if self.repo.kv_get(self.completed_key(stat_date)) == "1":
             self._log(
                 "info", "snapshot.seed_skip",
@@ -548,17 +552,6 @@ 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(
@@ -568,7 +561,6 @@ 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(
@@ -590,7 +582,6 @@ 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,

reaction_board/timeutil.py

@@ -57,39 +57,3 @@ 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)

reaction_board/web/queries.py

@@ -20,7 +20,6 @@ from ..analytics import (
     METRICS,
     MIN_VIEW_FOR_GROWTH_RATE,
     compute_growth,
-    smoothed_growth_rate,
     content_hours,
     eb_rate,
     effective_shrink_minutes,
@@ -34,14 +33,7 @@ from ..analytics import (
 )
 from ..config import ConfigStore
 from ..database import Database
-from ..timeutil import (
-    day_bounds,
-    interval_next_run,
-    next_after_snapshot_quiet_window,
-    next_daily_run,
-    shanghai_date,
-    shanghai_datetime,
-)
+from ..timeutil import day_bounds, next_daily_run, shanghai_date, shanghai_datetime
 from .cache import DASHBOARD_CACHE, TTL_FANS, TTL_VIDEOS, build_cache_key
 
 TRACKED_STATUSES = ("active", "unfaved")
@@ -66,19 +58,6 @@ def _chunks(values: Sequence[int], size: int = 400) -> Iterable[Sequence[int]]:
         yield values[index:index + size]
 
 
-def _growth_baseline_view(item: Mapping[str, Any]) -> int | None:
-    """增速分母：当前播放减去区间增量。zero 基线的增量等于当前播放，结果为 0。"""
-    growth = item.get("growth") or {}
-    current = (item.get("metrics") or {}).get("view")
-    delta = growth.get("delta")
-    if current is None or delta is None:
-        return None
-    try:
-        return int(current) - int(delta)
-    except (TypeError, ValueError):
-        return None
-
-
 def _date_minus(day: str, days: int) -> str:
     return (date.fromisoformat(day) - timedelta(days=int(days))).isoformat()
 
@@ -406,18 +385,6 @@ class DashboardQueries:
         percentiles = score_percentiles(scores)
         return videos, snapshots, scores, percentiles
 
-    @staticmethod
-    def _publish_date(pubdate: object) -> str | None:
-        if pubdate is None or isinstance(pubdate, bool):
-            return None
-        try:
-            ts = int(pubdate)
-        except (TypeError, ValueError):
-            return None
-        if ts <= 0:
-            return None
-        return shanghai_date(ts)
-
     @staticmethod
     def _growth_for(
         snapshot: Mapping[str, Any] | None,
@@ -439,70 +406,6 @@ class DashboardQueries:
                 gap = None
         return compute_growth(current, before, gap)
 
-    @classmethod
-    def _same_day_growth(
-        cls,
-        snapshot: Mapping[str, Any] | None,
-        video: Mapping[str, Any],
-        metric: str = "view",
-    ) -> dict[str, Any] | None:
-        """近 1 日、没有更早快照时：当日发布相对 0；否则用 view_open。"""
-        if snapshot is None:
-            return None
-        current = snapshot.get(metric)
-        if current is None:
-            return None
-        stat_date = str(snapshot.get("stat_date") or "")
-        published_on = cls._publish_date(video.get("pubdate"))
-        if published_on is not None and published_on == stat_date:
-            growth = compute_growth(current, 0, 0, allow_zero_gap=True)
-            payload = growth.as_dict()
-            payload["baseline"] = "zero"
-            # previous=0 时 rate = view/1，会按原始播放碾压真日环比。
-            payload["rate"] = None
-            return payload
-        try:
-            open_at = snapshot.get("open_captured_at")
-            captured_at = snapshot.get("captured_at")
-            view_open = snapshot.get("view_open")
-            open_ready = (
-                open_at is not None
-                and captured_at is not None
-                and int(open_at) < int(captured_at)
-                and view_open is not None
-            )
-        except (TypeError, ValueError):
-            open_ready = False
-        if not open_ready:
-            return None
-        growth = compute_growth(current, view_open, 0, allow_zero_gap=True)
-        payload = growth.as_dict()
-        payload["baseline"] = "open"
-        return payload
-
-    @classmethod
-    def _growth_result(
-        cls,
-        video: Mapping[str, Any],
-        snapshot: Mapping[str, Any] | None,
-        previous: Mapping[str, Any] | None,
-        *,
-        days_back: int,
-        metric: str = "view",
-    ) -> dict[str, Any]:
-        if previous is not None:
-            growth = cls._growth_for(snapshot, previous, metric=metric)
-            payload = growth.as_dict()
-            payload["baseline"] = "snapshot" if growth.available else None
-            return payload
-        if days_back == 1:
-            fallback = cls._same_day_growth(snapshot, video, metric=metric)
-            if fallback is not None:
-                return fallback
-        payload = compute_growth(None, None, None).as_dict()
-        payload["baseline"] = None
-        return payload
-
     def _item(
         self,
         video: Mapping[str, Any],
@@ -512,7 +415,6 @@ class DashboardQueries:
         percentile: float | None = None,
         previous: Mapping[str, Any] | None = None,
         include_growth: bool = False,
-        growth_days: int | None = None,
     ) -> dict[str, Any]:
         metrics = _metric_map(snapshot)
         duration = video.get("duration")
@@ -553,9 +455,7 @@ class DashboardQueries:
             ),
         }
         if include_growth:
-            item["growth"] = self._growth_result(
-                video, snapshot, previous, days_back=int(growth_days or 1),
-            )
+            item["growth"] = self._growth_for(snapshot, previous).as_dict()
         return item
 
     # ------------------------------------------------------------ overview
@@ -574,7 +474,7 @@ class DashboardQueries:
             videos = self._videos(conn)
             aids = [int(video["aid"]) for video in videos]
             snapshots = self._snapshot_map(conn, aids)
-            # 播放增量：近 1 日。有更早日用相邻成功日；当日发布相对 0；否则用 view_open。
+            # 播放增量：按片相对当前点的相邻成功日，避免 seed/recent 拉到 T-2。
             previous = self._snapshot_previous_relative(conn, snapshots, days_back=1)
             totals = {metric: 0 for metric in METRICS}
             for video in videos:
@@ -589,13 +489,12 @@ class DashboardQueries:
             for video in videos:
                 aid = int(video["aid"])
                 snapshot = snapshots.get(aid)
-                if snapshot is None:
+                prior = previous.get(aid)
+                if snapshot is None or prior is None:
                     continue
-                growth = self._growth_result(
-                    video, snapshot, previous.get(aid), days_back=1,
-                )
-                if growth.get("available") and growth.get("delta") is not None:
-                    view_delta_sum += int(growth["delta"])
+                growth = self._growth_for(snapshot, prior)
+                if growth.available and growth.delta is not None:
+                    view_delta_sum += growth.delta
                     available_deltas += 1
             if available_deltas:
                 view_delta = view_delta_sum
@@ -723,32 +622,19 @@ class DashboardQueries:
                 continue
         return None
 
-    @staticmethod
-    def _task_started_sort_key(value: Any) -> tuple[int, int]:
-        """缺 started_at 的排末尾；有值的按开始时间倒序（含进行中）。"""
-        if isinstance(value, Mapping):
-            raw = value.get("started_at")
-            try:
-                return (0, -int(raw))
-            except (TypeError, ValueError):
-                pass
-        return (1, 0)
-
     def _task_records(self, conn: sqlite3.Connection) -> dict[str, Any]:
         rows = conn.execute(
-            "SELECT k, v FROM kv WHERE k LIKE 'task:last:%'"
+            "SELECT k, v FROM kv WHERE k LIKE 'task:last:%' ORDER BY k"
         ).fetchall()
-        items: list[tuple[str, Any]] = []
+        result: dict[str, Any] = {}
         for row in rows:
             name = str(row["k"]).removeprefix("task:last:")
             try:
                 value = json.loads(str(row["v"]))
             except (TypeError, json.JSONDecodeError):
                 value = {"raw": row["v"]}
-            items.append((name, value))
-        items.sort(key=lambda pair: self._task_started_sort_key(pair[1]))
-        # 插入序即开始时间倒序；前端 Object.entries 保留该序，勿再排序。
-        return dict(items)
+            result[name] = value
+        return result
 
     def _task_running(self, conn: sqlite3.Connection) -> dict[str, Any]:
         rows = conn.execute(
@@ -982,7 +868,6 @@ class DashboardQueries:
                     percentile=percentiles.get(int(video["aid"])),
                     previous=previous.get(int(video["aid"])),
                     include_growth=sort in {"growth", "growth_rate"},
-                    growth_days=growth_days,
                 )
                 for video in videos
             ]
@@ -1054,13 +939,9 @@ class DashboardQueries:
                 if sort == "growth_rate":
                     items = [
                         item for item in items
-                        if item.get("growth", {}).get("baseline") != "zero"
-                        and item.get("growth", {}).get("rate") is not None
-                        and (_growth_baseline_view(item) or 0) >= MIN_VIEW_FOR_GROWTH_RATE
+                        if (item["metrics"] or {}).get("view", 0) >= MIN_VIEW_FOR_GROWTH_RATE
                     ]
 
-            rate_prior = max(0, int(analytics.growth_rate_prior))
-
             def sort_key(item: Mapping[str, Any]):
                 if sort == "added":
                     return (int(item["first_seen"]), int(item["pubdate"]))
@@ -1077,18 +958,9 @@ class DashboardQueries:
                     )
                     value = item.get(field)
                     return value if value is not None else float("-inf")
-                if sort == "growth":
-                    value = item.get("growth", {}).get("delta")
-                    return value if value is not None else float("-inf")
-                if sort == "growth_rate":
-                    growth = item.get("growth") or {}
-                    delta = growth.get("delta")
-                    current = (item.get("metrics") or {}).get("view")
-                    if delta is None or current is None:
-                        return float("-inf")
-                    value = smoothed_growth_rate(
-                        delta, int(current) - int(delta), rate_prior,
-                    )
+                if sort in {"growth", "growth_rate"}:
+                    field = "delta" if sort == "growth" else "rate"
+                    value = item.get("growth", {}).get(field)
                     return value if value is not None else float("-inf")
                 value = (item["metrics"] or {}).get(sort)
                 return value if value is not None else -1
@@ -1105,20 +977,9 @@ class DashboardQueries:
             reason = None
             if sort in {"growth", "growth_rate"} and not growth_available:
                 available = False
-                if growth_days == 1 and sort == "growth_rate":
-                    reason = (
-                        "近 1 日增速需要基线播放至少 "
-                        f"{MIN_VIEW_FOR_GROWTH_RATE}；当日发布不计入增速"
-                    )
-                elif growth_days == 1:
-                    reason = (
-                        "近 1 日增长需要前一日快照；当日发布有一次成功采样即可；"
-                        "更早发布的新入库需要当天至少两次成功采样"
-                    )
-                else:
-                    reason = (
-                        f"近 {growth_days} 日增长需要至少 {growth_days + 1} 天快照"
-                    )
+                reason = (
+                    f"近 {growth_days} 日增长需要至少 {growth_days + 1} 天快照"
+                )
             duration_filter = None
             if sort in duration_sorts:
                 duration_filter = {
@@ -1149,7 +1010,6 @@ class DashboardQueries:
                 "available": available,
                 "reason": reason,
                 "duration_filter": duration_filter,
-                "growth_rate_prior": rate_prior if sort == "growth_rate" else None,
             }
         return self.db.read(run)
 
@@ -1171,19 +1031,16 @@ class DashboardQueries:
                 video = dict(row)
             snapshots = self._snapshot_map(conn, [aid])
             snapshot = snapshots.get(aid)
-            previous = self._snapshot_map(
-                conn, [aid], before=snapshot["stat_date"] if snapshot else None,
-            ).get(aid)
+            previous = self._snapshot_map(conn, [aid], before=snapshot["stat_date"] if snapshot else None).get(aid)
             _, _, heat, percentiles = self._global_heat(conn)
+            growth = self._growth_for(snapshot, previous)
             item = self._item(
                 video,
                 snapshot,
                 heat=heat.get(aid),
                 percentile=percentiles.get(aid),
             )
-            item["growth"] = self._growth_result(
-                video, snapshot, previous, days_back=1,
-            )
+            item["growth"] = growth.as_dict()
             item["snapshot_count"] = int(conn.execute(
                 "SELECT COUNT(*) FROM snapshot WHERE aid = ?", (aid,)
             ).fetchone()[0])
@@ -1796,7 +1653,6 @@ class DashboardQueries:
                     "video_shrink_minutes": cfg.analytics.video_shrink_minutes,
                     "shrink_minutes": cfg.analytics.shrink_minutes,
                     "home_growth_days": cfg.analytics.home_growth_days,
-                    "growth_rate_prior": cfg.analytics.growth_rate_prior,
                 },
                 "heat_weights": {
                     metric: float(value)
@@ -1978,11 +1834,9 @@ 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(interval_next_run(now, float(discover_interval))),
-                "snapshot_recent": int(interval_next_run(now, float(recent_interval))),
+                "discover": int(now + discover_interval),
+                "snapshot_recent": int(now + recent_interval),
             }
-            retry_seconds = 300
-            seed_retry_seconds = 600
             for name, key, at in (
                 ("fans", "fans", cfg.schedule.snapshot_at),
                 ("snapshot", "snapshot", cfg.schedule.snapshot_at),
@@ -1994,46 +1848,19 @@ class DashboardQueries:
                 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"}:
-                    # 上次完成推下一档；skip/失败短补跑；静默窗推到 01:00 后。
+                    # 只按上次完成推下一档；进行中由 running 展示，不按墙钟抢跑。
                     interval = (
                         recent_interval if name == "snapshot_recent"
                         else discover_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))
-                        )
+                    next_runs[name] = finished_at + interval
                 else:
-                    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))
+                    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
-                and not record.get("skipped")
-                and not (
-                    isinstance(record.get("summary"), dict)
-                    and record["summary"].get("skipped")
-                )
+                if isinstance(record, dict) and record.get("ok") is False
             ]
             counts = {str(row["status"]): int(row["n"]) for row in conn.execute(
                 "SELECT status, COUNT(*) AS n FROM video GROUP BY status"

reaction_board/web/static/components.js

@@ -1,6 +1,6 @@
 import {
   coverUrl, escapeAttr, escapeHtml, formatCount, formatDate,
-  formatDelta, formatDuration, formatPercent, formatSignedPercent, metricLabel, statusLabel,
+  formatDelta, formatDuration, formatPercent, metricLabel, statusLabel,
 } from "./format.js";
 
 export function cover(item, className = "thumb") {
@@ -30,15 +30,9 @@ export function avatar(item, className = "avatar") {
     loading="lazy" referrerpolicy="no-referrer"></div>`;
 }
 
-function growthBaselineTag(growth) {
-  if (growth?.baseline === "zero") return "<small>今日</small>";
-  if (growth?.baseline === "open") return "<small>入库后</small>";
-  return "";
-}
-
 export function videoCard(item, {
   rank = null, primary = "view", primaryLabel = "", primaryText = null,
-  showStatus = false, growthField = "delta",
+  showStatus = false,
 } = {}) {
   const metrics = item.metrics || {};
   const hasMetric = Object.prototype.hasOwnProperty.call(metrics, primary);
@@ -49,13 +43,7 @@ export function videoCard(item, {
   const primaryValue = typeof raw === "number" ? formatCount(raw) : raw;
   const label = primaryLabel || metricLabel(primary);
   const growth = item.growth;
-  const showRate = growthField === "rate";
-  const rate = showRate && growth?.available ? growth.rate : null;
-  const delta = !showRate && growth?.available ? growth.delta : null;
-  const change = showRate
-    ? (rate === null || rate === undefined ? null : formatSignedPercent(rate))
-    : (delta === null || delta === undefined ? null : formatDelta(delta));
-  const changeUp = showRate ? Number(rate) >= 0 : Number(delta) >= 0;
+  const delta = growth?.available ? growth.delta : null;
   const title = escapeHtml(item.title || "(无标题)");
   const up = escapeHtml(item.up_name || "未知 UP");
   const upMid = item.up_mid;
@@ -85,7 +73,7 @@ export function videoCard(item, {
           <span class="primary-stat">
             <b>${escapeHtml(primaryValue ?? "—")}</b><small>${escapeHtml(label)}</small>
           </span>
-          ${change !== null ? `<span class="delta ${changeUp ? "up" : "down"}">${change}${growthBaselineTag(growth)}</span>` : ""}
+          ${delta !== null ? `<span class="delta ${delta >= 0 ? "up" : "down"}">${formatDelta(delta)}</span>` : ""}
           <span class="secondary-stat">互动 ${formatPercent(item.interaction_rate)}</span>
         </div>
       </div>
@@ -130,7 +118,7 @@ export function metricGrid(item) {
   return `<div class="metric-grid">
     ${["view", "like", "coin", "favorite", "share", "reply", "danmaku"].map((metric) => {
       const extra = metric === "view" && growth?.available
-        ? `<span class="metric-delta ${growth.delta >= 0 ? "up" : "down"}">${formatDelta(growth.delta)}${growthBaselineTag(growth)}</span>`
+        ? `<span class="metric-delta ${growth.delta >= 0 ? "up" : "down"}">${formatDelta(growth.delta)}</span>`
         : "";
       return `<div class="metric-cell">
         <span class="metric-label">${escapeHtml(metricLabel(metric))}</span>

reaction_board/web/static/format.js

@@ -46,12 +46,6 @@ export function formatPercent(value, digits = 2) {
   return `${(Number(value) * 100).toFixed(digits)}%`;
 }
 
-export function formatSignedPercent(value, digits = 2) {
-  const text = formatPercent(value, digits);
-  if (text === "—" || Number(value) <= 0) return text;
-  return `+${text}`;
-}
-
 export function formatDate(timestamp) {
   if (!timestamp) return "—";
   const date = new Date(Number(timestamp) * 1000);

reaction_board/web/static/style.css

@@ -353,7 +353,7 @@ a.live-chip:hover { border-color: color-mix(in srgb, var(--accent) 55%, var(--bo
   transition: border-color 0.15s cubic-bezier(0.22, 1, 0.36, 1),
     background-color 0.15s cubic-bezier(0.22, 1, 0.36, 1);
 }
-.video-card { grid-template-columns: 160px minmax(0, 1fr); }
+.video-card { grid-template-columns: 128px minmax(0, 1fr); }
 .up-card { grid-template-columns: 64px minmax(0, 1fr); align-items: center; }
 .video-card:hover, .up-card:hover {
   border-color: color-mix(in srgb, var(--accent) 45%, var(--border));
@@ -995,7 +995,7 @@ a.live-chip:hover { border-color: color-mix(in srgb, var(--accent) 55%, var(--bo
   .up-stats { grid-template-columns: repeat(5, minmax(0, 1fr)); }
   .detail-hero { flex-direction: row; align-items: flex-start; }
   .hero-cover { width: 300px; flex: 0 0 300px; }
-  .video-card { grid-template-columns: 160px minmax(0, 1fr); }
+  .video-card { grid-template-columns: 144px minmax(0, 1fr); }
   .secondary-stat.optional { display: inline; }
 }
 
@@ -1048,7 +1048,7 @@ a.live-chip:hover { border-color: color-mix(in srgb, var(--accent) 55%, var(--bo
 }
 
 @media (max-width: 380px) {
-  .video-card { gap: 8px; }
+  .video-card { grid-template-columns: 104px minmax(0, 1fr); gap: 8px; }
   .metric-grid { grid-template-columns: repeat(2, minmax(0, 1fr)); }
   .candidate-card { grid-template-columns: 80px minmax(0, 1fr); }
   .candidate-thumb { width: 80px; }

reaction_board/web/static/views/about.js

@@ -50,7 +50,6 @@ export async function renderAbout(root) {
         <li><strong>播放 / 投币 / 收藏 / 评论</strong>：直接取当日快照；弹幕与转发不单列，已并入“互动”。</li>
         <li><strong>互动</strong>：点赞 + 投币 + 收藏。综合热度也使用全部七项指标，按配置权重加权。</li>
         <li><strong>互动率</strong>：(点赞 + 投币 + 收藏) / 播放；播放不足 100 不计算。</li>
-        <li><strong>播放增速</strong>：卡片上的百分比是增量 / 基线播放。基线播放不足 1000 不进增速榜。排序用增量 / (基线 + ${analytics.growth_rate_prior})。当日发布没有基线，不进增速榜。</li>
         <li><strong>综合热度</strong>：每项先 <code>log1p</code>，再按中位数与 IQR 做稳健标准化，最后加权求和。分数只用于同一批数据内排序，不能跨日期比较。</li>
         <li><strong>内容效率</strong>：播放 / 内容小时。内容小时 = 视频总时长（多 P 为各 P 之和）÷ 3600。榜单里的单条视频用收缩后的估计：<code>(播放 + κ·μ₀) / (小时 + κ)</code>，κ 默认 ${analytics.video_shrink_minutes} 分钟、μ₀ 为全库播放/小时；详情页保留原始播放/小时。</li>
         <li><strong>互动效率</strong>：互动 / 内容小时，互动口径与收缩方式同上。</li>
@@ -93,7 +92,7 @@ export async function renderAbout(root) {
         <li>B 站不公开完播率与单 P 播放，内容小时是“供给时长”，不是“实际观看时长”。</li>
         <li>多 P 可能让同一用户产生多次播放计数，本站不单独校正，属于已知偏差。</li>
         <li>粉丝数可能包含不活跃账号；粉丝快照暂不参与任何榜单。</li>
-        <li>近 7/30 日增长仍要跨日快照。近 1 日：当日发布按 0 起算；更早发布的新入库按入库时播放量起算，需当天至少两次采样。样本不足时显示提示，不伪造 0。</li>
+        <li>样本不足（少于 2 天快照、单视频 UP）会显示提示，不伪造 0。</li>
       </ul>
     </section>
 

reaction_board/web/static/views/home.js

@@ -69,7 +69,7 @@ export async function renderHome(root) {
   const growthEmptyTitle = growth.reason
     || `近 ${growthDays} 日增长暂不可用`;
   const growthEmptyHint = growthDays <= 1
-    ? "当日发布按 0 起算；更早发布的新入库需要当天至少两次采样。"
+    ? "当前快照不足，完成至少 2 天快照后即可显示播放增量。"
     : `需要至少 ${growthDays + 1} 天快照才能计算近 ${growthDays} 日增长。`;
   const updatedLabel = formatClock(overview.updated_at);
   const dataDateLabel = formatDataDate(overview.data_date);

reaction_board/web/static/views/rankings.js

@@ -166,7 +166,6 @@ function listHtml(data, scope, sort, startRank) {
     return videoCard(item, {
       rank: startRank + index, primary: primary.primary,
       primaryLabel: primary.label, primaryText: primary.text,
-      growthField: sort === "growth_rate" ? "rate" : "delta",
     });
   }).join("")}</div>`;
 }
@@ -180,21 +179,6 @@ function durationNote(data) {
   return `<div class="section-note">内容效率与互动效率：时长 ≥ ${filter.min_duration_seconds} 秒、发布满 ${filter.maturity_days} 天、播放 ≥ ${filter.min_view}${shrink}；时长缺失不参与。</div>`;
 }
 
-function growthNote(sort, growthDays, prior) {
-  if (sort === "growth_rate") {
-    const strength = Number(prior);
-    const smooth = Number.isFinite(strength)
-      ? `百分比是真实相对增量；排序按平滑增速（基线 + ${strength}）。`
-      : "百分比是真实相对增量；排序按平滑增速。";
-    const cold = Number(growthDays) === 1
-      ? "基线播放不足 1000、以及当日发布，不进增速榜。"
-      : "基线播放不足 1000 不进增速榜。";
-    return `<div class="section-note">${smooth}${cold}</div>`;
-  }
-  if (sort !== "growth" || Number(growthDays) !== 1) return "";
-  return `<div class="section-note">近 1 日：当日发布按 0 起算；更早发布的新入库按入库时播放量起算，需要当天至少两次采样。</div>`;
-}
-
 function bindControls(root) {
   root.querySelector("#rank-sort")?.addEventListener("change", (event) => {
     replaceQuery({ sort: event.target.value, page: null });
@@ -261,7 +245,6 @@ async function renderScope(root, { query }, scope) {
     </header>
     ${controls({ scope, sort, within, year, growthDays, q, years })}
     ${durationNote(data)}
-    ${growthNote(sort, growthDays, data.growth_rate_prior)}
     ${data.available === false ? `<div class="notice-row warning">
       <strong>${sort.startsWith("growth") ? "增长数据尚不可用" : "榜单筛选后为空"}</strong>
       <span>${escapeHtml(data.reason || "")}</span>

reaction_board/web/static/views/system.js

@@ -5,15 +5,6 @@ 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 `<span class="muted">跳过</span>`;
-  }
-  if (record.ok) return "成功";
-  return `<span class="danger">失败</span>`;
-}
-
 function taskRows(tasks) {
   const entries = Object.entries(tasks || {});
   if (!entries.length) return emptyState("暂无任务记录", "运行一次同步、快照或发现后会写入最近结果。");
@@ -24,16 +15,12 @@ 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", "skipped", "already_completed",
-                "reason", "elapsed", "success",
-              ].includes(key))
+              .filter(([key]) => ["added", "removed", "candidates", "notified", "ok", "failed", "total", "completed", "backup_error"].includes(key))
               .map(([key, value]) => `${key}=${value}`).join(" ")
             : "";
           return `<tr>
             <td>${escapeHtml(name)}</td>
-            <td>${taskResultLabel(record)}</td>
+            <td>${record.ok ? "成功" : `<span class="danger">失败</span>`}</td>
             <td>${escapeHtml(formatDateTime(record.finished_at))}</td>
             <td class="summary-cell">${escapeHtml(summary || record.error || "—")}</td>
           </tr>`;
@@ -113,7 +100,7 @@ export async function renderSystem(root) {
         <div><span>新视频小时同步</span><strong>${running.snapshot_recent ? "运行中" : escapeHtml(formatDateTime(next.snapshot_recent))}</strong></div>
         <div><span>新视频搜索</span><strong>${running.discover || running.discovery ? "运行中" : escapeHtml(formatDateTime(next.discover))}</strong></div>
       </div>
-      <p class="section-note">进行中显示「运行中」；空闲时按该任务上次完成时间推算下次运行。23:00–01:00（上海）为全站快照窗：发现 / 小时同步 / 候选清理会延后到 01:00 后；粉丝更新不避开。跳过或失败不会记成成功，会短间隔补跑。</p>
+      <p class="section-note">进行中显示「运行中」；空闲时按该任务上次完成时间推算下次运行，不按墙钟抢跑。新视频小时同步与全站快照互斥。</p>
     </section>
 
     <section class="panel">

reaction_board/web/static/views/video.js

@@ -62,7 +62,7 @@ export async function renderVideo(root, { params, query }) {
           <span>${escapeHtml(formatDuration(detail.duration))}</span>
           ${detail.pages ? `<span>${formatNumber(detail.pages)} P</span>` : ""}
           ${badge(statusLabel(detail.status), `status-${detail.status}`)}
-          ${samples < 2 && !detail.growth?.available ? badge("样本不足", "warning") : ""}
+          ${samples < 2 ? badge("样本不足", "warning") : ""}
         </div>
         ${detail.blacklisted ? `<p class="section-note warning-text">该 UP 在黑名单中；黑名单只影响未来发现，不影响当前视频追踪。</p>` : ""}
         ${detail.up_follower_backfilled ? `<p class="section-note">UP 粉丝数为回填值，不是真实的入库时点。</p>` : ""}

tests/test_analytics.py

@@ -12,7 +12,6 @@ from reaction_board.analytics import (  # noqa: E402
     METRICS,
     age_days,
     compute_growth,
-    smoothed_growth_rate,
     content_hours,
     eb_rate,
     effective_shrink_minutes,
@@ -132,18 +131,6 @@ def test_compute_growth():
     no_gap = compute_growth(100, 90, 0)
     assert no_gap.available is False
 
-    same_day = compute_growth(100, 90, 0, allow_zero_gap=True)
-    assert same_day.available is True
-    assert same_day.delta == 10
-    assert same_day.per_day is None
-    assert same_day.gap_days == 0
-    assert abs(same_day.rate - 10 / 90) < 1e-12
-
-    from_zero_same_day = compute_growth(50, 0, 0, allow_zero_gap=True)
-    assert from_zero_same_day.delta == 50
-    assert from_zero_same_day.rate == 50.0
-    assert from_zero_same_day.per_day is None
-
     growth = compute_growth(1000, 800, 7)
     assert growth.available is True
     assert growth.delta == 200
@@ -162,19 +149,6 @@ def test_compute_growth():
     assert negative.per_day == -5.0
 
 
-def test_smoothed_growth_rate():
-    # 80→512：真实增速 5.4；prior=500 时排序分 432/580
-    assert abs(smoothed_growth_rate(432, 80, 500) - 432 / 580) < 1e-12
-    assert abs(smoothed_growth_rate(432, 5000, 500) - 432 / 5500) < 1e-12
-    assert abs(smoothed_growth_rate(432, 80, 1000) - 432 / 1080) < 1e-12
-    # prior=0 时排序分退回真实相对增速（基线 ≥ 1）
-    assert smoothed_growth_rate(432, 80, 0) == 432 / 80
-    assert smoothed_growth_rate(None, 80, 500) is None
-    assert smoothed_growth_rate(432, None, 500) is None
-    assert smoothed_growth_rate(10, 0, 0) is None
-    assert smoothed_growth_rate(10, 80, -1) is None
-
-
 def test_normalize_cover_and_hosts():
     assert normalize_cover(None) is None
     assert normalize_cover("") is None
@@ -265,7 +239,6 @@ def main_test():
     test_score_percentiles()
     test_interaction_rate_thresholds()
     test_compute_growth()
-    test_smoothed_growth_rate()
     test_normalize_cover_and_hosts()
     test_content_efficiency_helpers()
     test_age_and_maturity()

tests/test_cli.py

@@ -51,7 +51,7 @@ def test_init_and_status():
         path = write_config(tmp)
         code, output = run_cli(["init", "--config", str(path)])
         assert code == 0
-        assert '"schema_version": 8' in output
+        assert '"schema_version": 7' in output
 
         db = Database(Path(tmp) / "app.db")
         db.initialize()
@@ -332,422 +332,6 @@ def test_backfill_followers_dry_run():
         assert '"total": 0' in output
 
 
-
-def test_help_lists_snapshot_seed():
-    from reaction_board.cli import _build_parser
-
-    parser = _build_parser()
-    help_text = parser.format_help()
-    assert "snapshot-seed" in help_text
-    args = parser.parse_args(["snapshot-seed", "--date", "2026-10-31"])
-    assert args.command == "snapshot-seed"
-    assert args.date == "2026-10-31"
-
-
-def test_snapshot_seed_cli_clones_offline():
-    from reaction_board.bilibili.models import FavResource
-
-    with tempfile.TemporaryDirectory() as tmp:
-        path = write_config(tmp, cookie="", csrf="")
-        code, _ = run_cli(["init", "--config", str(path)])
-        assert code == 0
-        db = Database(Path(tmp) / "app.db")
-        db.initialize()
-        try:
-            repo = Repository(db)
-            media_id = 11
-            repo.upsert_from_fav(
-                FavResource(
-                    aid=1, bvid="BV1xx411c7mD", title="t", up_mid=9,
-                    up_name="u", pubdate=1, duration=10, cover=None,
-                    attr=0, fav_time=None,
-                ),
-                media_id,
-                was_added=True,
-            )
-            yday, today = "2026-10-30", "2026-10-31"
-            repo.upsert_snapshot(
-                yday, 1,
-                metrics={"view": 10, "like": 1, "coin": 0, "favorite": 0,
-                         "share": 0, "reply": 0, "danmaku": 0},
-                ok=True, err=None, captured_at=1_800_000_000,
-            )
-            repo.kv_set("snapshot:last_completed", yday)
-            repo.kv_set("snapshot:completed:" + yday, "1")
-        finally:
-            db.close()
-
-        code, output = run_cli([
-            "snapshot-seed", "--config", str(path), "--date", today,
-        ])
-        assert code == 0, output
-        assert '"reason": "seeded"' in output
-        db = Database(Path(tmp) / "app.db")
-        db.initialize()
-        try:
-            repo = Repository(db)
-            assert repo.kv_get("snapshot:seeded:" + today) == "1"
-            row = repo.get_snapshot(today, 1)
-            assert row is not None and int(row["view"]) == 10
-            assert row["src"] == "seed"
-        finally:
-            db.close()
-
-
-def test_snapshot_seed_defers_when_snapshot_running():
-    with tempfile.TemporaryDirectory() as tmp:
-        path = write_config(tmp, cookie="", csrf="")
-        code, _ = run_cli(["init", "--config", str(path)])
-        assert code == 0
-        db = Database(Path(tmp) / "app.db")
-        db.initialize()
-        try:
-            repo = Repository(db)
-            repo.kv_set(
-                "task:running:snapshot",
-                json.dumps({"name": "snapshot", "started_at": 1}),
-            )
-            repo.kv_set("snapshot:last_completed", "2026-10-30")
-        finally:
-            db.close()
-
-        code, output = run_cli([
-            "snapshot-seed", "--config", str(path), "--date", "2026-10-31",
-        ])
-        assert code == 1, output
-        assert "snapshot_running" in output
-        db = Database(Path(tmp) / "app.db")
-        db.initialize()
-        try:
-            repo = Repository(db)
-            last = json.loads(repo.kv_get("task:last:snapshot_seed"))
-            assert last["ok"] is False
-            assert last["skipped"] is True
-            assert last["error"] == "snapshot_running"
-            assert repo.kv_get("task:running:snapshot_seed") is None
-            assert repo.kv_get("task:running:snapshot") is not None
-        finally:
-            db.close()
-
-
-def test_snapshot_seed_no_source_is_non_success():
-    with tempfile.TemporaryDirectory() as tmp:
-        path = write_config(tmp, cookie="", csrf="")
-        code, _ = run_cli(["init", "--config", str(path)])
-        assert code == 0
-        code, output = run_cli([
-            "snapshot-seed", "--config", str(path), "--date", "2026-10-31",
-        ])
-        assert code == 1, output
-        assert "no_source" in output
-        db = Database(Path(tmp) / "app.db")
-        db.initialize()
-        try:
-            repo = Repository(db)
-            last = json.loads(repo.kv_get("task:last:snapshot_seed"))
-            assert last["ok"] is False and last["skipped"] is True
-            assert last["error"] == "no_source"
-        finally:
-            db.close()
-
-
-def test_defer_snapshot_seed_schedules_ten_minutes():
-    """调度辅助：skip 记非成功，next_run = finished + 600。"""
-    from reaction_board.logging_json import JsonLogger
-
-    with tempfile.TemporaryDirectory() as tmp:
-        path = write_config(tmp, cookie="", csrf="")
-        code, _ = run_cli(["init", "--config", str(path)])
-        assert code == 0
-        db = Database(Path(tmp) / "app.db")
-        db.initialize()
-        try:
-            repo = Repository(db)
-            logger = JsonLogger()
-            started = 1_800_000_000
-            finished = started + 5
-            ctrl = cli._defer_snapshot_seed(
-                repo, logger, started_at=started, reason="snapshot_running",
-                finished_at=finished,
-            )
-            assert ctrl.ok is False
-            assert ctrl.error == "snapshot_running"
-            assert abs(ctrl.next_run - (finished + cli._SEED_RETRY_SECONDS)) < 0.01
-            last = json.loads(repo.kv_get("task:last:snapshot_seed"))
-            assert last["skipped"] is True and last["ok"] is False
-        finally:
-            db.close()
-
-
-
-
-def 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
-        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"]
-        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():
-    """常驻启动：无上次记录时应对齐下一档墙钟，不得为 0。"""
-    from reaction_board.timeutil import next_daily_run, 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)
-            # 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
-            # 成功记录在昨晚 → 仍是今晚 23:15
-            yday_done = datetime(2026, 11, 2, 23, 20, tzinfo=SHANGHAI_TZ).timestamp()
-            repo.kv_set(
-                "task:last:snapshot",
-                json.dumps({
-                    "ok": True, "finished_at": int(yday_done),
-                    "started_at": int(yday_done) - 10,
-                }),
-            )
-            nxt2 = cli._initial_daily_next_run(
-                repo, "snapshot", "23:15", now=now,
-            )
-            assert nxt2 == next_daily_run(yday_done, "23:15")
-            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 后重启：snapshot 应排到当晚 snapshot_at，首轮 run_due 不得 force 全量。"""
-    from reaction_board.scheduler import ScheduledTask, Scheduler
-    from reaction_board.timeutil import next_daily_run, SHANGHAI_TZ, shanghai_date
-    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)
-            # 白天：假定今日已 seed
-            now = datetime(2026, 11, 3, 10, 30, tzinfo=SHANGHAI_TZ).timestamp()
-            today = shanghai_date(now)
-            repo.kv_set(f"snapshot:seeded:{today}", "1")
-            repo.kv_set(
-                "task:last:snapshot_seed",
-                json.dumps({
-                    "ok": True,
-                    "finished_at": int(
-                        datetime(2026, 11, 3, 0, 1, tzinfo=SHANGHAI_TZ).timestamp()
-                    ),
-                    "started_at": int(
-                        datetime(2026, 11, 3, 0, 0, tzinfo=SHANGHAI_TZ).timestamp()
-                    ),
-                    "summary": {"reason": "seeded"},
-                }),
-            )
-            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 == next_daily_run(now, snapshot_at)
-            assert fans_next == next_daily_run(now, snapshot_at)
-            # seed 已成功 → 下一档是次日 00:00
-            assert seed_next == next_daily_run(
-                float(json.loads(repo.kv_get("task:last:snapshot_seed"))["finished_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, now)
-                )
-                force_calls.append(force)
-                if force:
-                    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 == []
-            # 跳到 snapshot_at：才允许 force
-            clock[0] = next_daily_run(now, snapshot_at)
-            # 任务 next_run 已是该点，到期应执行；此时 reached_daily_slot 为真
-            # 但 run_snapshot 闭包捕获的是旧 now——改用 clock
-        finally:
-            db.close()
-
-    # 单独验证：到达 slot 后 force 门打开
-    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 main_test():
     test_init_and_status()
     test_missing_config()
@@ -761,18 +345,7 @@ def main_test():
     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_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()
-    print("OK: CLI init/sync/snapshot/snapshot-seed/auth-check/run/status/backfill/错误路径测试全部通过")
+    print("OK: CLI init/sync/snapshot/auth-check/run/status/backfill/错误路径测试全部通过")
 
 
 if __name__ == "__main__":

tests/test_config.py

@@ -146,7 +146,6 @@ def test_analytics_defaults_and_validation():
         assert cfg.analytics.video_shrink_minutes == 60
         assert cfg.analytics.shrink_minutes is None
         assert cfg.analytics.home_growth_days == "auto"
-        assert cfg.analytics.growth_rate_prior == 1000
         assert cfg.schedule.recent_snapshot_minutes == 60
         assert cfg.schedule.recent_video_days == 2
         assert cfg.schedule.snapshot_at == "23:15"
@@ -160,7 +159,6 @@ def test_analytics_defaults_and_validation():
             "video_shrink_minutes": 30,
             "shrink_minutes": 90,
             "home_growth_days": 1,
-            "growth_rate_prior": 2000,
         }
         data["schedule"] = {
             "discover_minutes": 30,
@@ -178,7 +176,6 @@ def test_analytics_defaults_and_validation():
         assert cfg.analytics.video_shrink_minutes == 30
         assert cfg.analytics.shrink_minutes == 90
         assert cfg.analytics.home_growth_days == 1
-        assert cfg.analytics.growth_rate_prior == 2000
         assert cfg.schedule.recent_snapshot_minutes == 15
         assert cfg.schedule.recent_video_days == 3
 
@@ -197,7 +194,6 @@ def test_analytics_defaults_and_validation():
             ({"home_growth_days": 3}, "home_growth_days"),
             ({"home_growth_days": "7"}, "home_growth_days"),
             ({"home_growth_days": 0}, "home_growth_days"),
-            ({"growth_rate_prior": -1}, "growth_rate_prior"),
         ):
             data = base_config()
             data["analytics"] = bad

tests/test_database.py

@@ -24,7 +24,7 @@ def test_write_read_transaction():
     with tempfile.TemporaryDirectory() as tmp:
         db = make_db(tmp)
         try:
-            assert db.schema_version == 8
+            assert db.schema_version == 7
             db.write(lambda c: c.execute("INSERT INTO kv(k, v) VALUES ('a', '1')"))
             assert db.read(lambda c: c.execute("SELECT v FROM kv WHERE k='a'").fetchone()[0]) == "1"
 
@@ -94,8 +94,8 @@ def test_close_and_reinitialize():
 
         db2 = make_db(tmp)
         try:
-            assert db2.schema_version == 8
-            assert db2.read(lambda c: c.execute("SELECT MAX(version) FROM schema_version").fetchone()[0]) == 8
+            assert db2.schema_version == 7
+            assert db2.read(lambda c: c.execute("SELECT MAX(version) FROM schema_version").fetchone()[0]) == 7
         finally:
             db2.close()
 
@@ -123,7 +123,7 @@ def test_migration_backup_on_existing_db():
         db = Database(path, backup_dir=Path(tmp) / "backup")
         try:
             applied = db.initialize()
-            assert applied == [1, 2, 3, 4, 5, 6, 7, 8]
+            assert applied == [1, 2, 3, 4, 5, 6, 7]
             assert len(list((Path(tmp) / "backup").glob("app-pre-v0-*.db"))) == 1
         finally:
             db.close()

tests/test_migrations.py

@@ -38,8 +38,8 @@ def test_fresh_and_idempotent():
         conn = sqlite3.connect(Path(tmp) / "app.db")
         try:
             applied = run_migrations(conn)
-            assert applied == [1, 2, 3, 4, 5, 6, 7, 8]
-            assert current_version(conn) == 8
+            assert applied == [1, 2, 3, 4, 5, 6, 7]
+            assert current_version(conn) == 7
             assert EXPECTED_TABLES <= table_names(conn)
             assert "schema_version" in table_names(conn)
             columns = {
@@ -70,14 +70,10 @@ def test_fresh_and_idempotent():
             assert {
                 "stat_date", "ip", "path", "hits", "first_at", "last_at",
             } <= visit_columns
-            snapshot_columns = {
-                row[1] for row in conn.execute("PRAGMA table_info(snapshot)")
-            }
-            assert {"view_open", "open_captured_at"} <= snapshot_columns
             # 再跑一次不应重复应用
             assert run_migrations(conn) == []
             row = conn.execute("SELECT COUNT(*) FROM schema_version").fetchone()
-            assert row[0] == 8
+            assert row[0] == 7
         finally:
             conn.close()
 
@@ -92,15 +88,15 @@ def test_backup_before_upgrade():
             conn.execute("INSERT INTO kv(k, v) VALUES ('hello', 'world')")
             conn.commit()
 
-            v9 = Migration(9, "add probe", ("CREATE TABLE probe (id INTEGER PRIMARY KEY)",))
+            v8 = Migration(8, "add probe", ("CREATE TABLE probe (id INTEGER PRIMARY KEY)",))
             applied = run_migrations(
                 conn,
                 backup_dir=backup_dir,
                 now=1_800_000_000,
-                migrations=MIGRATIONS + (v9,),
+                migrations=MIGRATIONS + (v8,),
             )
-            assert applied == [9]
-            backups = list(backup_dir.glob("app-pre-v8-*.db"))
+            assert applied == [8]
+            backups = list(backup_dir.glob("app-pre-v7-*.db"))
             assert len(backups) == 1
             check = sqlite3.connect(backups[0])
             try:
@@ -185,46 +181,13 @@ def test_v5_normalizes_unfaved():
             )
             conn.commit()
             applied = run_migrations(conn)
-            assert applied == [5, 6, 7, 8]
+            assert applied == [5, 6, 7]
             rows = dict(conn.execute("SELECT aid, status FROM video").fetchall())
             assert rows == {1: "active", 2: "dropped"}
         finally:
             conn.close()
 
 
-def test_v8_backfills_view_open():
-    with tempfile.TemporaryDirectory() as tmp:
-        conn = sqlite3.connect(Path(tmp) / "app.db")
-        try:
-            run_migrations(conn, migrations=MIGRATIONS[:7])
-            conn.execute(
-                "INSERT INTO video(aid, title, pubdate, status, first_seen) "
-                "VALUES (1, 't', 1, 'active', 1)"
-            )
-            conn.execute(
-                'INSERT INTO snapshot(stat_date, aid, "view", captured_at, ok) '
-                "VALUES ('2026-10-31', 1, 500, 100, 1)"
-            )
-            conn.execute(
-                'INSERT INTO snapshot(stat_date, aid, "view", captured_at, ok) '
-                "VALUES ('2026-10-30', 1, NULL, 90, 0)"
-            )
-            conn.commit()
-            assert run_migrations(conn) == [8]
-            ok_row = conn.execute(
-                "SELECT view_open, open_captured_at FROM snapshot "
-                "WHERE stat_date='2026-10-31'"
-            ).fetchone()
-            assert ok_row[0] == 500 and ok_row[1] == 100
-            fail_row = conn.execute(
-                "SELECT view_open, open_captured_at FROM snapshot "
-                "WHERE stat_date='2026-10-30'"
-            ).fetchone()
-            assert fail_row[0] is None and fail_row[1] is None
-        finally:
-            conn.close()
-
-
 def main():
     test_fresh_and_idempotent()
     test_backup_before_upgrade()
@@ -232,7 +195,6 @@ def main():
     test_prune_backups()
     test_validate_migration_order()
     test_v5_normalizes_unfaved()
-    test_v8_backfills_view_open()
     print("OK: SQLite 迁移、备份与回滚测试全部通过")
 
 

tests/test_repository.py

@@ -296,59 +296,6 @@ def test_pending_snapshot_aids():
             db.close()
 
 
-def test_upsert_snapshot_preserves_view_open():
-    with tempfile.TemporaryDirectory() as tmp:
-        db, repo = setup(tmp)
-        try:
-            metrics = {
-                "view": 10, "like": 1, "favorite": 0, "coin": 0,
-                "share": 0, "reply": 0, "danmaku": 0,
-            }
-            repo.upsert_snapshot(
-                "2026-10-31", 1, metrics=metrics, ok=True, err=None, captured_at=10,
-            )
-            first = repo.get_snapshot("2026-10-31", 1)
-            assert first["view"] == 10
-            assert first["view_open"] == 10
-            assert first["open_captured_at"] == 10
-
-            metrics["view"] = 40
-            repo.upsert_snapshot(
-                "2026-10-31", 1, metrics=metrics, ok=True, err=None, captured_at=20,
-            )
-            second = repo.get_snapshot("2026-10-31", 1)
-            assert second["view"] == 40
-            assert second["view_open"] == 10
-            assert second["open_captured_at"] == 10
-            assert second["captured_at"] == 20
-
-            repo.upsert_snapshot(
-                "2026-10-31", 2, metrics=None, ok=False, err="x", captured_at=1,
-            )
-            failed = repo.get_snapshot("2026-10-31", 2)
-            assert failed["view_open"] is None
-            metrics["view"] = 7
-            repo.upsert_snapshot(
-                "2026-10-31", 2, metrics=metrics, ok=True, err=None, captured_at=5,
-            )
-            recovered = repo.get_snapshot("2026-10-31", 2)
-            assert recovered["view"] == 7
-            assert recovered["view_open"] == 7
-            assert recovered["open_captured_at"] == 5
-
-            copied = repo.copy_ok_snapshots(
-                "2026-10-31", "2026-11-01", captured_at=30, src="seed",
-            )
-            assert copied == 2
-            seeded = repo.get_snapshot("2026-11-01", 1)
-            assert seeded["view"] == 40
-            assert seeded["view_open"] == 40
-            assert seeded["open_captured_at"] == 30
-            assert seeded["src"] == "seed"
-        finally:
-            db.close()
-
-
 def test_up_follower_snapshot_and_backfill():
     with tempfile.TemporaryDirectory() as tmp:
         db, repo = setup(tmp)
@@ -420,7 +367,6 @@ def main():
     test_primary_folder_is_order_independent()
     test_dropped_invalid_preserves_original_title()
     test_pending_snapshot_aids()
-    test_upsert_snapshot_preserves_view_open()
     test_up_follower_snapshot_and_backfill()
     print("OK: 收藏夹/视频状态机与黑名单测试全部通过")
 

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, TaskControl  # noqa: E402
+from reaction_board.scheduler import ScheduledTask, Scheduler  # noqa: E402
 
 
 def test_run_when_due_and_reschedule():
@@ -128,111 +128,9 @@ 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()

tests/test_snapshot.py

@@ -444,22 +444,19 @@ def test_run_recent_videos_refreshes_without_completion_markers():
             1: view_for(1, view=111),
             2: view_for(2, view=222),
             3: view_for(3, view=333),
-            4: view_for(4, view=444),
         })
         db, repo, service = make_service(tmp, transport, track_fans=())
         try:
-            # 1/2 在墙钟近 2 天内；3 为旧片；4 发布已久但当天入库，不进小时任务
+            # 1/2 在墙钟近 2 天内；3 为旧片
             add_video(repo, 1)
             add_video(repo, 2)
             add_video(repo, 3)
-            add_video(repo, 4)
             repo.db.write(lambda conn: conn.executemany(
                 "UPDATE video SET pubdate = ? WHERE aid = ?",
                 [
                     (clock - 3600, 1),
                     (clock - 86400, 2),
                     (clock - 10 * 86400, 3),
-                    (clock - 10 * 86400, 4),
                 ],
             ))
             # 预置同日 ok=1，验证小时任务可覆盖刷新
@@ -480,27 +477,23 @@ def test_run_recent_videos_refreshes_without_completion_markers():
             )) == {1, 2}
 
             row = repo.db.read(lambda conn: dict(conn.execute(
-                "SELECT view, ok, captured_at, view_open FROM snapshot "
-                "WHERE stat_date=? AND aid=1",
+                "SELECT view, ok, captured_at FROM snapshot WHERE stat_date=? AND aid=1",
                 (DATE,),
             ).fetchone()))
             assert row["view"] == 111
             assert row["ok"] == 1
             assert row["captured_at"] == clock
-            assert row["view_open"] == 1
 
             # 完成日 / 备份标记保持原样
             assert repo.kv_get(service.completed_key(DATE)) is None
             assert repo.kv_get("snapshot:last_completed") is None
             assert repo.kv_get(service.backup_key(DATE)) is None
             assert transport.count("/web-interface/view") == 2
-            assert repo.get_snapshot(DATE, 4) is None
 
             # 全量任务随后才写完成日；且已 ok=1 的跳过
             full = service.run_videos(date=DATE)
             assert full.skipped >= 2  # 1、2 已被小时任务写过
-            assert full.success >= 1  # 至少旧片 3、4
-            assert repo.get_snapshot(DATE, 4) is not None
+            assert full.success >= 1  # 至少旧片 3
             assert repo.kv_get(service.completed_key(DATE)) == "1"
             assert repo.kv_get("snapshot:last_completed") == DATE
         finally:
@@ -537,36 +530,6 @@ 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:
@@ -599,7 +562,6 @@ def test_seed_from_last_completed_does_not_mark_completed():
             row = repo.get_snapshot(today, 1)
             assert row is not None
             assert int(row["view"]) == 10
-            assert int(row["view_open"]) == 10
             assert row["src"] == "seed"
             fan = repo.get_up_stat(today, 37754047)
             assert fan is not None and int(fan["follower"]) == 100
@@ -676,7 +638,6 @@ 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()
     print("OK: 每日快照、断点续跑、失败占位与粉丝采集测试全部通过")

tests/test_split_config_secrets.py

@@ -0,0 +1,194 @@
+# -*- coding: utf-8 -*-
+"""config.json → jsonc + secrets 拆分脚本测试（离线）:
+uv run python tests/test_split_config_secrets.py"""
+
+from __future__ import annotations
+
+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 import jsonc  # noqa: E402
+from reaction_board.config import load_config  # noqa: E402
+from tools.split_config_secrets import (  # noqa: E402
+    split_config_file,
+    split_raw,
+)
+
+
+def _monolith() -> dict:
+    return {
+        "bilibili": {
+            "read": {"mid": 11, "csrf": "read-csrf", "cookie": "SESSDATA=secret-read"},
+            "write": {"mid": 22, "csrf": "write-csrf", "cookie": "SESSDATA=secret-write"},
+        },
+        "telegram": {
+            "enabled": True,
+            "bot_token": "tg-secret-token",
+            "chat_id": 12345,
+        },
+        "web": {
+            "host": "127.0.0.1",
+            "port": 8080,
+            "auth_token": "web-secret-token",
+            "public_base_url": "",
+        },
+        "folders": [{"year": 2026, "media_id": 1, "title": "t"}],
+        "track_fans": [37754047],
+        "up_blacklist_seed": [],
+    }
+
+
+def test_split_raw_extracts_secrets_keeps_business():
+    business, secrets, stats = split_raw(_monolith())
+    assert "bilibili" not in business
+    assert business["telegram"] == {"enabled": True}
+    assert business["web"]["host"] == "127.0.0.1"
+    assert business["web"]["port"] == 8080
+    assert "auth_token" not in business["web"]
+    assert secrets["bilibili"]["read"]["mid"] == 11
+    assert secrets["telegram"] == {"bot_token": "tg-secret-token", "chat_id": 12345}
+    assert secrets["web"] == {"auth_token": "web-secret-token"}
+    assert "bilibili" in stats["moved_paths"]
+    assert "telegram.bot_token" in stats["moved_paths"]
+    assert "web.auth_token" in stats["moved_paths"]
+
+
+def test_split_raw_omits_absent_secret_stubs():
+    raw = {
+        "telegram": {"enabled": False},
+        "web": {"host": "0.0.0.0", "port": 9},
+        "folders": [],
+    }
+    business, secrets, stats = split_raw(raw)
+    assert secrets == {}
+    assert business["telegram"] == {"enabled": False}
+    assert business["web"]["host"] == "0.0.0.0"
+    assert stats["moved_count"] == 0
+
+
+def test_dry_run_writes_nothing():
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        src = root / "config.json"
+        src.write_text(json.dumps(_monolith(), ensure_ascii=False), encoding="utf-8")
+        summary = split_config_file(src, dry_run=True)
+        assert summary["dry_run"] is True
+        assert summary["moved_count"] >= 3
+        assert not (root / "config.jsonc").exists()
+        assert not (root / "secrets.json").exists()
+        assert src.exists()
+        # 摘要不得泄露 cookie/token
+        blob = json.dumps(summary, ensure_ascii=False)
+        assert "SESSDATA" not in blob
+        assert "tg-secret-token" not in blob
+        assert "web-secret-token" not in blob
+
+
+def test_refuses_overwrite_without_force():
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        src = root / "config.json"
+        src.write_text(json.dumps(_monolith(), ensure_ascii=False), encoding="utf-8")
+        (root / "config.jsonc").write_text("// existing\n{}\n", encoding="utf-8")
+        try:
+            split_config_file(src, keep_src=True)
+        except FileExistsError as exc:
+            assert "config.jsonc" in str(exc)
+            assert "--force" in str(exc)
+        else:
+            raise AssertionError("已有输出时无 --force 必须失败")
+        assert src.exists()
+        assert "existing" in (root / "config.jsonc").read_text(encoding="utf-8")
+
+
+def test_writes_and_renames_src_by_default():
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        src = root / "config.json"
+        src.write_text(json.dumps(_monolith(), ensure_ascii=False), encoding="utf-8")
+        summary = split_config_file(src)
+        cfg_path = root / "config.jsonc"
+        sec_path = root / "secrets.json"
+        bak = root / "config.json.bak"
+        assert summary["ok"] is True
+        assert not src.exists()
+        assert bak.exists()
+        assert cfg_path.exists()
+        assert sec_path.exists()
+
+        text = cfg_path.read_text(encoding="utf-8")
+        assert text.startswith("//")
+        assert "secrets.json" in text
+        business = jsonc.load(cfg_path)
+        secrets = json.loads(sec_path.read_text(encoding="utf-8"))
+        assert "bilibili" not in business
+        assert business["telegram"]["enabled"] is True
+        assert "bot_token" not in business["telegram"]
+        assert business["web"]["host"] == "127.0.0.1"
+        assert "auth_token" not in business["web"]
+        assert secrets["bilibili"]["read"]["mid"] == 11
+        assert secrets["telegram"]["chat_id"] == 12345
+        assert secrets["web"]["auth_token"] == "web-secret-token"
+
+        loaded = load_config(cfg_path, env={}, require_auth=False)
+        assert loaded.bilibili.read.mid == 11
+        assert loaded.bilibili.write.mid == 22
+        assert loaded.telegram.enabled is True
+        assert loaded.telegram.bot_token == "tg-secret-token"
+        assert loaded.web.host == "127.0.0.1"
+        assert loaded.web.auth_token == "web-secret-token"
+        assert summary["verify"]["checked"] is True
+        assert summary["verify"]["read_mid"] == 11
+
+
+def test_force_overwrite_and_keep_src():
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        src = root / "config.json"
+        src.write_text(json.dumps(_monolith(), ensure_ascii=False), encoding="utf-8")
+        (root / "config.jsonc").write_text("{}\n", encoding="utf-8")
+        (root / "secrets.json").write_text("{}\n", encoding="utf-8")
+        summary = split_config_file(src, force=True, keep_src=True)
+        assert src.exists()
+        assert summary["src_renamed"] is False
+        secrets = json.loads((root / "secrets.json").read_text(encoding="utf-8"))
+        assert secrets["bilibili"]["read"]["mid"] == 11
+
+
+def test_honors_secrets_path_relative():
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        src = root / "config.json"
+        data = _monolith()
+        data["secrets_path"] = "creds/my-secrets.json"
+        src.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
+        summary = split_config_file(src, keep_src=True)
+        sec = root / "creds" / "my-secrets.json"
+        assert Path(summary["secrets_out"]) == sec
+        assert sec.exists()
+        business = jsonc.load(root / "config.jsonc")
+        assert business.get("secrets_path") == "creds/my-secrets.json"
+
+
+def main():
+    test_split_raw_extracts_secrets_keeps_business()
+    test_split_raw_omits_absent_secret_stubs()
+    test_dry_run_writes_nothing()
+    test_refuses_overwrite_without_force()
+    test_writes_and_renames_src_by_default()
+    test_force_overwrite_and_keep_src()
+    test_honors_secrets_path_relative()
+    print(
+        "OK: split_config_secrets 抽取 bilibili/telegram/web 凭据、"
+        "dry-run、拒绝覆盖、改名与 secrets_path 测试全部通过"
+    )
+
+
+if __name__ == "__main__":
+    main()

tests/test_timeutil.py

@@ -11,10 +11,7 @@ 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,
@@ -75,40 +72,11 @@ 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: 上海时区与每日时间计算测试全部通过")
 
 

tests/test_web_queries.py

@@ -15,7 +15,7 @@ from reaction_board.repository import Repository  # noqa: E402
 from reaction_board.timeutil import day_bounds  # noqa: E402
 from reaction_board.web.cache import DASHBOARD_CACHE  # noqa: E402
 from reaction_board.web.queries import DashboardQueries  # noqa: E402
-from tests.web_fixtures import DATES, FOLDER, build_db, resource, write_config  # noqa: E402
+from tests.web_fixtures import DATES, build_db, write_config  # noqa: E402
 
 FIXED_NOW = 1_793_459_000  # 2026-10-31 前后
 
@@ -46,7 +46,7 @@ def test_overview():
             assert overview["counts"]["candidates_pending"] == 1
             assert overview["totals"]["view"] == 1200 + 5200 + 120
             assert overview["interactions"] > 0
-            assert overview["view_delta"] == (100 + 100 + 120)  # 101/102 跨日 + 103 当日发布
+            assert overview["view_delta"] == (100 + 100 + 0)  # 101/102 的日增量
             assert overview["snapshot_health"]["total"] == 3
             assert overview["folders"][0]["media_count"] == 3
             assert overview["folders"][0]["ratio"] < 1
@@ -484,7 +484,7 @@ def test_about_payload():
             assert about["snapshot_days"] == 3
             assert about["snapshot_first_date"] == DATES[0]
             assert about["fans_start_date"] == DATES[0]
-            assert about["schema_version"] == 8
+            assert about["schema_version"] == 7
             assert about["analytics"] == {
                 "min_duration_seconds": 600,
                 "maturity_days": 2,
@@ -492,7 +492,6 @@ def test_about_payload():
                 "video_shrink_minutes": 60,
                 "shrink_minutes": None,
                 "home_growth_days": "auto",
-                "growth_rate_prior": 1000,
             }
             assert about["heat_weights"]["view"] == 0.30
             assert about["followers"]["videos_total"] == 3
@@ -578,13 +577,9 @@ def test_growth_auto_falls_back_to_one_day():
             assert auto["available"] is True
             assert auto["growth_days"] == 1
             assert auto["days"] == 1
-            assert {item["aid"] for item in auto["items"]} == {101, 102, 103}
+            assert {item["aid"] for item in auto["items"]} == {101, 102}
             first = next(item for item in auto["items"] if item["aid"] == 101)
             assert first["growth"]["delta"] == 100  # 相邻日
-            assert first["growth"]["baseline"] == "snapshot"
-            today = next(item for item in auto["items"] if item["aid"] == 103)
-            assert today["growth"]["baseline"] == "zero"
-            assert today["growth"]["delta"] == 120
 
             forced7 = q.list_videos(sort="growth", growth_days=7)
             assert forced7["available"] is False
@@ -610,15 +605,12 @@ 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+3600
-            finished = int(datetime(2026, 10, 31, 15, 0, 0, tzinfo=SHANGHAI_TZ).timestamp())
+            finished = FIXED_NOW - 10
             repo.kv_set(
                 "task:last:snapshot_recent",
                 json.dumps({
@@ -630,128 +622,6 @@ 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()
 
@@ -797,10 +667,9 @@ def test_overview_adjacent_previous_with_seed_and_dual_fields():
                 + overview["counts"]["unfaved"]
                 + overview["counts"]["removed"]
             )
-            # 101/102：当前=seed(=完成日数值)，相邻 previous=完成日本身 → 增量 0
-            # 103：当日发布，近 1 日相对 0，计入当前播放
+            # 当前=seed(=完成日数值)，相邻 previous=完成日本身 → 增量 0
             assert overview["view_delta_available"] is True
-            assert overview["view_delta"] == 120
+            assert overview["view_delta"] == 0
 
             growth = q.list_videos(sort="growth", growth_days=1)
             assert growth["available"] is True
@@ -809,186 +678,9 @@ def test_overview_adjacent_previous_with_seed_and_dual_fields():
                     assert item["growth"]["available"] is True
                     assert item["growth"]["delta"] == 0
                     assert item["growth"]["gap_days"] == 1
-                    assert item["growth"]["baseline"] == "snapshot"
         finally:
             db.close()
 
-
-def _view_metrics(view: int) -> dict[str, int]:
-    return {
-        "view": view, "like": 0, "coin": 0, "favorite": 0,
-        "share": 0, "reply": 0, "danmaku": 0,
-    }
-
-
-def test_same_day_growth_baselines():
-    """近 1 日：当日发布相对 0；更早发布用 view_open；7 日不用这两条退路。"""
-    with tempfile.TemporaryDirectory() as tmp:
-        db, q = setup(tmp)
-        try:
-            repo = Repository(db)
-            today = DATES[-1]
-            start = int(day_bounds(today)[0])
-            old_pub = int(day_bounds("2026-09-01")[0])
-
-            repo.upsert_snapshot(
-                today, 103, metrics=_view_metrics(200),
-                ok=True, err=None, captured_at=start + 3600,
-            )
-            repo.upsert_from_fav(
-                resource(105, up_mid=1, up_name="UP A", pubdate=old_pub),
-                FOLDER.media_id, was_added=True, now=start + 10,
-            )
-            repo.upsert_snapshot(
-                today, 105, metrics=_view_metrics(500),
-                ok=True, err=None, captured_at=start + 10,
-            )
-            repo.upsert_snapshot(
-                today, 105, metrics=_view_metrics(800),
-                ok=True, err=None, captured_at=start + 20,
-            )
-            repo.upsert_from_fav(
-                resource(106, up_mid=1, up_name="UP A", pubdate=old_pub),
-                FOLDER.media_id, was_added=True, now=start + 30,
-            )
-            repo.upsert_snapshot(
-                today, 106, metrics=_view_metrics(900),
-                ok=True, err=None, captured_at=start + 30,
-            )
-            repo.upsert_from_fav(
-                resource(107, up_mid=1, up_name="UP A", pubdate=old_pub),
-                FOLDER.media_id, was_added=True, now=start + 40,
-            )
-            repo.upsert_snapshot(
-                today, 107, metrics=_view_metrics(50),
-                ok=True, err=None, captured_at=start + 40,
-            )
-            db.write(lambda conn: conn.execute(
-                "UPDATE video SET pubdate = 0 WHERE aid = 107"
-            ))
-            DASHBOARD_CACHE.clear()
-
-            one = q.list_videos(sort="growth", growth_days=1, page_size=20)
-            by_aid = {item["aid"]: item for item in one["items"]}
-            assert 103 in by_aid and 105 in by_aid
-            assert 106 not in by_aid and 107 not in by_aid
-            assert by_aid[103]["growth"]["baseline"] == "zero"
-            assert by_aid[103]["growth"]["delta"] == 200
-            assert by_aid[103]["growth"]["rate"] is None
-            assert by_aid[103]["growth"]["gap_days"] == 0
-            assert by_aid[105]["growth"]["baseline"] == "open"
-            assert by_aid[105]["growth"]["delta"] == 300
-            assert by_aid[101]["growth"]["baseline"] == "snapshot"
-            assert by_aid[101]["growth"]["delta"] == 100
-
-            rate = q.list_videos(sort="growth_rate", growth_days=1, page_size=20)
-            rate_aids = {item["aid"] for item in rate["items"]}
-            assert 103 not in rate_aids and 105 not in rate_aids
-            assert 101 in rate_aids
-
-            seven = q.list_videos(sort="growth", growth_days=7)
-            seven_aids = {item["aid"] for item in seven["items"]}
-            assert 103 not in seven_aids
-            assert 105 not in seven_aids
-
-            detail103 = q.video_detail(103)
-            assert detail103["growth"]["baseline"] == "zero"
-            assert detail103["growth"]["delta"] == 200
-            detail105 = q.video_detail(105)
-            assert detail105["growth"]["baseline"] == "open"
-            assert detail105["growth"]["delta"] == 300
-            assert q.video_detail(106)["growth"]["available"] is False
-            assert q.video_detail(107)["growth"]["available"] is False
-
-            overview = q.overview()
-            assert overview["view_delta"] == 100 + 100 + 200 + 300
-        finally:
-            db.close()
-
-
-def test_growth_rate_sort_uses_prior():
-    """基线不足 1000 不进增速榜；达标后按基线 + prior 排序，不按 raw rate。"""
-    with tempfile.TemporaryDirectory() as tmp:
-        db, q = setup(tmp)
-        try:
-            repo = Repository(db)
-            latest = DATES[-1]
-            previous_day = DATES[-2]
-            start = int(day_bounds(latest)[0])
-            pub = int(day_bounds("2026-09-01")[0])
-            # 301：80→512，真实 +540%，基线低于 1000，只留在增量榜。
-            # 304：1000→2500，真实增速高于 302，平滑后应排在 302 后面。
-            for aid, views in ((301, (80, 512)), (302, (5000, 10000)), (304, (1000, 2500))):
-                repo.upsert_from_fav(
-                    resource(aid, up_mid=1, up_name="UP A", pubdate=pub),
-                    FOLDER.media_id, was_added=True, now=start,
-                )
-                repo.upsert_snapshot(
-                    previous_day, aid, metrics=_view_metrics(views[0]),
-                    ok=True, err=None, captured_at=start - 86400,
-                )
-                repo.upsert_snapshot(
-                    latest, aid, metrics=_view_metrics(views[1]),
-                    ok=True, err=None, captured_at=start + 60,
-                )
-            DASHBOARD_CACHE.clear()
-            ranked = q.list_videos(sort="growth_rate", growth_days=1, page_size=20)
-            order = [item["aid"] for item in ranked["items"]]
-            assert ranked["growth_rate_prior"] == 1000
-            assert 301 not in order
-            assert order.index(302) < order.index(304)
-            modest = next(item for item in ranked["items"] if item["aid"] == 304)
-            assert abs(modest["growth"]["rate"] - 1500 / 1000) < 1e-12
-
-            growth = q.list_videos(sort="growth", growth_days=1, page_size=20)
-            grown = next(item for item in growth["items"] if item["aid"] == 301)
-            assert grown["growth"]["delta"] == 432
-            assert abs(grown["growth"]["rate"] - 432 / 80) < 1e-12
-        finally:
-            db.close()
-
-
-def test_system_tasks_sorted_by_started_at_desc():
-    """最近任务按 started_at 倒序；缺开始时间的排末尾（含进行中无 finished）。"""
-    import json
-
-    with tempfile.TemporaryDirectory() as tmp:
-        db, q = setup(tmp)
-        try:
-            repo = Repository(db)
-            # 故意打乱写入/键名字母序；finished_at 与 started_at 错开，避免误用完成时间
-            samples = (
-                # name, started_at, finished_at
-                ("zzz_old", FIXED_NOW - 300, FIXED_NOW - 10),   # 开始最早，完成却很晚
-                ("aaa_mid", FIXED_NOW - 100, FIXED_NOW - 90),
-                ("mmm_new", FIXED_NOW - 10, None),              # 进行中：只有 started
-                ("no_time", None, FIXED_NOW),                  # 缺 started，应末尾
-            )
-            for name, started, finished in samples:
-                payload = {
-                    "name": name,
-                    "ok": True if finished is not None else True,
-                    "error": None,
-                    "summary": {},
-                }
-                if started is not None:
-                    payload["started_at"] = started
-                if finished is not None:
-                    payload["finished_at"] = finished
-                repo.kv_set(f"task:last:{name}", json.dumps(payload))
-            for legacy in ("snapshot", "fans", "discovery"):
-                repo.kv_delete(f"task:last:{legacy}")
-            DASHBOARD_CACHE.clear()
-            system = q.system()
-            assert list(system["tasks"].keys()) == [
-                "mmm_new", "aaa_mid", "zzz_old", "no_time",
-            ]
-            overview = q.overview()
-            assert list(overview["tasks"].keys()) == list(system["tasks"].keys())
-        finally:
-            db.close()
-
-
 def main_test():
     test_overview()
     test_video_listing_filters_and_sorting()
@@ -1004,15 +696,9 @@ def main_test():
     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()
-    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()
     print("OK: Web 查询层总览/榜单/详情/UP/时长效率/粉丝/候选/系统/访问统计测试全部通过")
 
 
@@ -1076,15 +762,13 @@ 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，不跳档、不误入静默窗推算
-            finished = int(datetime(2026, 10, 31, 15, 0, 0, tzinfo=SHANGHAI_TZ).timestamp())
+            # 发现任务刚完成不久：下次 = finished + 60min，即使 now 已超过也不跳档
+            finished = FIXED_NOW - 30
             repo.kv_set(
                 "task:last:discovery",
                 json.dumps({
@@ -1096,6 +780,7 @@ 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()

tests/test_web_server.py

@@ -75,7 +75,7 @@ def test_readonly_routes_and_static():
             assert status == 200 and payload["ok"] is True
             status, _, payload = running.json("/api/about")
             assert status == 200
-            assert payload["data"]["schema_version"] == 8
+            assert payload["data"]["schema_version"] == 7
             assert payload["data"]["analytics"]["min_duration_seconds"] == 600
 
             status, _, overview = running.json("/api/overview")

tests/web_fixtures.py

@@ -138,7 +138,7 @@ def build_db(tmp: str | Path) -> Path:
                 day, 37754047, follower=1000 + index * 10, following=5,
                 ok=True, captured_at=ts(day) + 60,
             )
-        # 103 只有最后一天快照（发布日=快照日，近 1 日相对 0）；104 另加一条失败快照
+        # 103 只有最后一天快照，用于验证样本不足；104 另加一条失败快照
         repo.upsert_snapshot(
             DATES[-1], 103,
             metrics={"view": 120, "like": 5, "coin": 1, "favorite": 2, "share": 0, "reply": 1, "danmaku": 1},

tools/seed_preview_db.py

@@ -237,8 +237,8 @@ def _insert_snapshot(
         """
         INSERT OR REPLACE INTO snapshot(
           stat_date, aid, "view", "like", favorite, coin, share, reply,
-          danmaku, captured_at, ok, err, src, view_open, open_captured_at
-        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'preview', ?, ?)
+          danmaku, captured_at, ok, err, src
+        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'preview')
         """,
         (
             stat_date,
@@ -253,8 +253,6 @@ def _insert_snapshot(
             int(captured_at),
             1 if ok else 0,
             err,
-            values.get("view") if ok else None,
-            int(captured_at) if ok else None,
         ),
     )
 

tools/split_config_secrets.py

@@ -0,0 +1,326 @@
+# -*- coding: utf-8 -*-
+"""将旧版单体 ``config.json`` 拆成 ``config.jsonc`` + ``secrets.json``。
+
+从源文件抽出凭据段：
+
+- 整段 ``bilibili``（若存在）
+- ``telegram.bot_token`` / ``telegram.chat_id``（``enabled`` 等非密钥留在业务文件）
+- ``web.auth_token``（``host`` / ``port`` / ``track`` 等留在业务文件）
+
+源里没有的密钥键不会写空壳。``telegram`` / ``web`` 抽空后若无剩余键，则不写入
+secrets；业务侧若还有非密钥键则保留对应段。``bilibili`` 整段移出业务文件。
+
+用法::
+
+    # 先看会挪哪些键（不打印 cookie/token 原文）
+    uv run python tools/split_config_secrets.py --dry-run
+
+    # 默认：写 config.jsonc + secrets.json，并把源文件改名为 config.json.bak
+    uv run python tools/split_config_secrets.py /path/to/config.json
+
+    # 已有输出时需显式覆盖；保留源文件不改名
+    uv run python tools/split_config_secrets.py --force --keep-src
+"""
+
+from __future__ import annotations
+
+import argparse
+import copy
+import json
+import os
+import sys
+from pathlib import Path
+from typing import Any, Mapping, Sequence
+
+ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(ROOT))
+
+from reaction_board import jsonc  # noqa: E402
+from reaction_board.config import load_config, resolve_secrets_path  # noqa: E402
+
+TELEGRAM_SECRET_KEYS = ("bot_token", "chat_id")
+WEB_SECRET_KEYS = ("auth_token",)
+
+CONFIG_HEADER = """\
+// 业务配置（可含注释与尾逗号）。敏感凭据在同目录 secrets.json。
+// `python app.py auth` 只回写 secrets.json，不会冲掉本文件注释。
+// 本文件由 tools/split_config_secrets.py 从旧版单体配置拆出。
+"""
+
+
+def _is_mapping(value: Any) -> bool:
+    return isinstance(value, Mapping) and not isinstance(value, (str, bytes))
+
+
+def split_raw(raw: Mapping[str, Any]) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]:
+    """拆分业务配置与凭据。
+
+    返回 ``(business, secrets, stats)``。``stats`` 仅含键路径计数，不含密钥值。
+    """
+    business: dict[str, Any] = copy.deepcopy(dict(raw))
+    secrets: dict[str, Any] = {}
+    moved: list[str] = []
+
+    bilibili = business.pop("bilibili", None)
+    if _is_mapping(bilibili):
+        secrets["bilibili"] = copy.deepcopy(dict(bilibili))
+        moved.append("bilibili")
+    elif bilibili is not None:
+        # 非法类型原样丢掉业务侧，避免残留；不写入 secrets
+        moved.append("bilibili(non-object-dropped)")
+
+    telegram = business.get("telegram")
+    if _is_mapping(telegram):
+        tg_biz = dict(telegram)
+        tg_sec: dict[str, Any] = {}
+        for key in TELEGRAM_SECRET_KEYS:
+            if key in tg_biz:
+                tg_sec[key] = tg_biz.pop(key)
+                moved.append(f"telegram.{key}")
+        if tg_sec:
+            secrets["telegram"] = tg_sec
+        if tg_biz:
+            business["telegram"] = tg_biz
+        else:
+            business.pop("telegram", None)
+
+    web = business.get("web")
+    if _is_mapping(web):
+        web_biz = dict(web)
+        web_sec: dict[str, Any] = {}
+        for key in WEB_SECRET_KEYS:
+            if key in web_biz:
+                web_sec[key] = web_biz.pop(key)
+                moved.append(f"web.{key}")
+        if web_sec:
+            secrets["web"] = web_sec
+        if web_biz:
+            business["web"] = web_biz
+        else:
+            business.pop("web", None)
+
+    stats = {
+        "moved_paths": moved,
+        "moved_count": len(moved),
+        "business_top_keys": sorted(business.keys()),
+        "secrets_top_keys": sorted(secrets.keys()),
+        "has_bilibili": "bilibili" in secrets,
+        "telegram_secret_keys": sorted(secrets.get("telegram", {}).keys())
+        if isinstance(secrets.get("telegram"), dict)
+        else [],
+        "web_secret_keys": sorted(secrets.get("web", {}).keys())
+        if isinstance(secrets.get("web"), dict)
+        else [],
+    }
+    return business, secrets, stats
+
+
+def _atomic_write_json(path: Path, data: Mapping[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp = path.with_name(path.name + ".tmp")
+    with tmp.open("w", encoding="utf-8") as fh:
+        json.dump(data, fh, ensure_ascii=False, indent=2)
+        fh.write("\n")
+    os.replace(tmp, path)
+
+
+def _atomic_write_jsonc(path: Path, data: Mapping[str, Any], *, header: str) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    body = json.dumps(data, ensure_ascii=False, indent=2)
+    if not body.endswith("\n"):
+        body += "\n"
+    text = header.rstrip() + "\n" + body
+    tmp = path.with_name(path.name + ".tmp")
+    with tmp.open("w", encoding="utf-8") as fh:
+        fh.write(text)
+    os.replace(tmp, path)
+
+
+def resolve_outputs(
+    src: Path,
+    raw: Mapping[str, Any],
+    *,
+    config_out: Path | None,
+    secrets_out: Path | None,
+) -> tuple[Path, Path]:
+    """解析输出路径；``secrets_path`` 相对源文件目录。"""
+    base = src.resolve().parent
+    cfg_path = config_out if config_out is not None else (base / "config.jsonc")
+    if secrets_out is not None:
+        sec_path = secrets_out
+    else:
+        sec_path = resolve_secrets_path(src, raw)
+    return cfg_path, sec_path
+
+
+def split_config_file(
+    src: Path,
+    *,
+    config_out: Path | None = None,
+    secrets_out: Path | None = None,
+    dry_run: bool = False,
+    force: bool = False,
+    keep_src: bool = False,
+) -> dict[str, Any]:
+    """执行拆分；返回可 JSON 序列化的摘要（不含密钥原文）。"""
+    src = Path(src)
+    if not src.exists():
+        raise FileNotFoundError(f"源配置不存在: {src}")
+
+    try:
+        raw = jsonc.load(src)
+    except ValueError as exc:
+        raise ValueError(f"源配置不是合法 JSON/JSONC: {src}: {exc}") from exc
+    if not isinstance(raw, Mapping):
+        raise ValueError(f"源配置顶层必须是对象: {src}")
+
+    business, secrets, stats = split_raw(raw)
+    cfg_path, sec_path = resolve_outputs(
+        src, raw, config_out=config_out, secrets_out=secrets_out
+    )
+
+    summary: dict[str, Any] = {
+        "ok": True,
+        "dry_run": bool(dry_run),
+        "src": str(src),
+        "config_out": str(cfg_path),
+        "secrets_out": str(sec_path),
+        "keep_src": bool(keep_src),
+        "src_bak": None if keep_src else str(src.with_name(src.name + ".bak")),
+        **stats,
+    }
+
+    if dry_run:
+        return summary
+
+    conflicts = [p for p in (cfg_path, sec_path) if p.exists()]
+    if conflicts and not force:
+        names = ", ".join(str(p) for p in conflicts)
+        raise FileExistsError(
+            f"输出已存在（需 --force 覆盖）: {names}"
+        )
+
+    _atomic_write_jsonc(cfg_path, business, header=CONFIG_HEADER)
+    _atomic_write_json(sec_path, secrets)
+
+    # 可选校验：合并加载后 mid 与源一致（不打印 cookie）
+    verify: dict[str, Any] = {"checked": False}
+    if secrets.get("bilibili") or raw.get("bilibili"):
+        cfg = load_config(cfg_path, env={}, require_auth=False)
+        src_bili = raw.get("bilibili") if _is_mapping(raw.get("bilibili")) else {}
+        read_mid = None
+        write_mid = None
+        if _is_mapping(src_bili.get("read")):
+            read_mid = src_bili["read"].get("mid")
+        if _is_mapping(src_bili.get("write")):
+            write_mid = src_bili["write"].get("mid")
+        if read_mid is not None and int(cfg.bilibili.read.mid) != int(read_mid):
+            raise RuntimeError(
+                f"校验失败: bilibili.read.mid {cfg.bilibili.read.mid} != {read_mid}"
+            )
+        if write_mid is not None and int(cfg.bilibili.write.mid) != int(write_mid):
+            raise RuntimeError(
+                f"校验失败: bilibili.write.mid {cfg.bilibili.write.mid} != {write_mid}"
+            )
+        verify = {
+            "checked": True,
+            "read_mid": int(cfg.bilibili.read.mid),
+            "write_mid": int(cfg.bilibili.write.mid),
+        }
+    summary["verify"] = verify
+
+    if not keep_src:
+        bak = src.with_name(src.name + ".bak")
+        if bak.exists() and not force:
+            raise FileExistsError(
+                f"备份已存在（需 --force 覆盖）: {bak}"
+            )
+        if bak.exists():
+            bak.unlink()
+        os.replace(src, bak)
+        summary["src_bak"] = str(bak)
+        summary["src_renamed"] = True
+    else:
+        summary["src_renamed"] = False
+
+    return summary
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(
+        description="拆分旧版 config.json 为 config.jsonc + secrets.json",
+    )
+    parser.add_argument(
+        "src_pos",
+        nargs="?",
+        default=None,
+        help="源配置路径（默认仓库根目录 config.json）",
+    )
+    parser.add_argument(
+        "--src",
+        default=None,
+        help="源配置路径（与位置参数二选一；默认 ROOT/config.json）",
+    )
+    parser.add_argument(
+        "--config-out",
+        default=None,
+        help="业务 JSONC 输出路径（默认与源同目录 config.jsonc）",
+    )
+    parser.add_argument(
+        "--secrets-out",
+        default=None,
+        help="凭据 JSON 输出路径（默认同目录 secrets.json，"
+        "或源内 secrets_path）",
+    )
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="只打印将挪动的键路径，不写文件、不改名",
+    )
+    parser.add_argument(
+        "--force",
+        action="store_true",
+        help="允许覆盖已存在的输出（及 .bak）",
+    )
+    parser.add_argument(
+        "--keep-src",
+        action="store_true",
+        help="成功写入后保留源文件；默认改名为 <src>.bak，"
+        "以便 default_config_path 优先读到新的 config.jsonc",
+    )
+    args = parser.parse_args(list(argv) if argv is not None else None)
+
+    if args.src_pos and args.src and args.src_pos != args.src:
+        print(
+            json.dumps(
+                {"ok": False, "error": "位置参数与 --src 冲突"},
+                ensure_ascii=False,
+            ),
+            file=sys.stderr,
+        )
+        return 2
+    src = Path(args.src_pos or args.src or (ROOT / "config.json"))
+    config_out = Path(args.config_out) if args.config_out else None
+    secrets_out = Path(args.secrets_out) if args.secrets_out else None
+
+    try:
+        summary = split_config_file(
+            src,
+            config_out=config_out,
+            secrets_out=secrets_out,
+            dry_run=args.dry_run,
+            force=args.force,
+            keep_src=args.keep_src,
+        )
+        print(json.dumps(summary, ensure_ascii=False, indent=2))
+        return 0
+    except (OSError, ValueError, RuntimeError) as exc:
+        print(
+            json.dumps({"ok": False, "error": str(exc)}, ensure_ascii=False),
+            file=sys.stderr,
+        )
+        return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())