feat: home growth auto(7→1) and hourly recent-video snapshot

- Add analytics.home_growth_days (auto|1|7); API growth_days=auto tries 7 then 1
- Homepage shows active window and growth-board link; clearer empty reason
- New snapshot_recent scheduler task (schedule.recent_snapshot_minutes/days)
- run_recent_videos refreshes new aids only; never touches completed/backup
- Full 00:01 job keeps ok=1 skip (covers hourly refresh within 1h); mutex with recent

tests/test_config.py

@@ -139,6 +139,9 @@ def test_analytics_defaults_and_validation():
         assert cfg.analytics.min_view_for_ranking == 2000
         assert cfg.analytics.video_shrink_minutes == 60
         assert cfg.analytics.shrink_minutes is None
+        assert cfg.analytics.home_growth_days == "auto"
+        assert cfg.schedule.recent_snapshot_minutes == 60
+        assert cfg.schedule.recent_video_days == 2
 
         data = base_config()
         data["analytics"] = {
@@ -147,6 +150,14 @@ def test_analytics_defaults_and_validation():
             "min_view_for_ranking": 100,
             "video_shrink_minutes": 30,
             "shrink_minutes": 90,
+            "home_growth_days": 1,
+        }
+        data["schedule"] = {
+            "discover_minutes": 30,
+            "snapshot_at": "00:01",
+            "candidate_ttl_days": 7,
+            "recent_snapshot_minutes": 15,
+            "recent_video_days": 3,
         }
         write(path, data)
         cfg = load_config(path, env={})
@@ -155,6 +166,14 @@ def test_analytics_defaults_and_validation():
         assert cfg.analytics.min_view_for_ranking == 100
         assert cfg.analytics.video_shrink_minutes == 30
         assert cfg.analytics.shrink_minutes == 90
+        assert cfg.analytics.home_growth_days == 1
+        assert cfg.schedule.recent_snapshot_minutes == 15
+        assert cfg.schedule.recent_video_days == 3
+
+        data = base_config()
+        data["analytics"] = {"home_growth_days": 7}
+        write(path, data)
+        assert load_config(path, env={}).analytics.home_growth_days == 7
 
         for bad, hint in (
             ({"min_duration_seconds": -1}, "min_duration_seconds"),
@@ -163,6 +182,9 @@ def test_analytics_defaults_and_validation():
             ({"video_shrink_minutes": -1}, "video_shrink_minutes"),
             ({"shrink_minutes": 0}, "shrink_minutes"),
             ({"shrink_minutes": "auto"}, "shrink_minutes"),
+            ({"home_growth_days": 3}, "home_growth_days"),
+            ({"home_growth_days": "7"}, "home_growth_days"),
+            ({"home_growth_days": 0}, "home_growth_days"),
         ):
             data = base_config()
             data["analytics"] = bad
@@ -243,12 +265,38 @@ def test_web_track_defaults_and_validation():
             raise AssertionError("web.track 必须是对象")
 
 
+def test_schedule_recent_validation():
+    with tempfile.TemporaryDirectory() as tmp:
+        path = Path(tmp) / "config.json"
+        for bad, hint in (
+            ({"recent_snapshot_minutes": 0}, "recent_snapshot_minutes"),
+            ({"recent_video_days": 0}, "recent_video_days"),
+            ({"recent_snapshot_minutes": -5}, "recent_snapshot_minutes"),
+        ):
+            data = base_config()
+            data["schedule"] = {
+                "discover_minutes": 60,
+                "snapshot_at": "00:01",
+                "candidate_ttl_days": 7,
+                **bad,
+            }
+            write(path, data)
+            try:
+                load_config(path, env={})
+            except ConfigError as exc:
+                assert hint in str(exc), (bad, str(exc))
+            else:
+                raise AssertionError(f"非法 schedule 必须拒绝: {bad}")
+
+
+
 def main():
     test_dual_accounts_and_env_override()
     test_flat_and_missing_account_rejected()
     test_validation_require_read_auth_and_hot_reload()
     test_invalid_values()
     test_analytics_defaults_and_validation()
+    test_schedule_recent_validation()
     test_web_track_defaults_and_validation()
     print("OK: 双账号配置加载、拒绝 flat 格式、ENV 覆盖、热加载与 analytics/web.track 测试全部通过")
 

tests/test_snapshot.py

@@ -436,6 +436,100 @@ def test_backup_failure_retries_without_api():
         db.close()
 
 
+def test_run_recent_videos_refreshes_without_completion_markers():
+    """小时任务只刷新近窗新片，不写完成日 / 备份。"""
+    clock = 1_800_000_000
+    with tempfile.TemporaryDirectory() as tmp:
+        transport = SnapshotTransport({
+            1: view_for(1, view=111),
+            2: view_for(2, view=222),
+            3: view_for(3, view=333),
+        })
+        db, repo, service = make_service(tmp, transport, track_fans=())
+        try:
+            # 1/2 在墙钟近 2 天内；3 为旧片
+            add_video(repo, 1)
+            add_video(repo, 2)
+            add_video(repo, 3)
+            repo.db.write(lambda conn: conn.executemany(
+                "UPDATE video SET pubdate = ? WHERE aid = ?",
+                [
+                    (clock - 3600, 1),
+                    (clock - 86400, 2),
+                    (clock - 10 * 86400, 3),
+                ],
+            ))
+            # 预置同日 ok=1，验证小时任务可覆盖刷新
+            repo.upsert_snapshot(
+                DATE, 1,
+                metrics={"view": 1, "like": 0, "coin": 0, "favorite": 0,
+                         "share": 0, "reply": 0, "danmaku": 0},
+                ok=True, err=None, captured_at=clock - 100,
+            )
+            assert repo.kv_get(service.completed_key(DATE)) is None
+
+            summary = service.run_recent_videos(date=DATE, recent_days=2)
+            assert summary.total == 2
+            assert summary.success == 2
+            assert summary.failed == 0
+            assert set(repo.video_aids_recent_for_snapshot(
+                since_pubdate=clock - 2 * 86400
+            )) == {1, 2}
+
+            row = repo.db.read(lambda conn: dict(conn.execute(
+                "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 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
+
+            # 全量任务随后才写完成日；且已 ok=1 的跳过
+            full = service.run_videos(date=DATE)
+            assert full.skipped >= 2  # 1、2 已被小时任务写过
+            assert full.success >= 1  # 至少旧片 3
+            assert repo.kv_get(service.completed_key(DATE)) == "1"
+            assert repo.kv_get("snapshot:last_completed") == DATE
+        finally:
+            db.close()
+
+
+def test_full_snapshot_skips_ok_rows_including_hourly_refresh():
+    """00:01 全量：同日 ok=1（含小时任务刚刷）直接跳过。"""
+    clock = 1_800_000_000
+    with tempfile.TemporaryDirectory() as tmp:
+        transport = SnapshotTransport({1: view_for(1), 2: view_for(2)})
+        db, repo, service = make_service(tmp, transport, track_fans=())
+        try:
+            add_video(repo, 1)
+            add_video(repo, 2)
+            repo.upsert_snapshot(
+                DATE, 1,
+                metrics={"view": 9, "like": 0, "coin": 0, "favorite": 0,
+                         "share": 0, "reply": 0, "danmaku": 0},
+                ok=True, err=None, captured_at=clock - 30,  # 「1 小时内」
+            )
+            result = service.run_videos(date=DATE)
+            assert result.skipped == 1
+            assert result.success == 1
+            # aid=1 保持小时任务写入的 view=9，未被全量覆盖
+            row = repo.db.read(lambda conn: dict(conn.execute(
+                "SELECT view FROM snapshot WHERE stat_date=? AND aid=1",
+                (DATE,),
+            ).fetchone()))
+            assert row["view"] == 9
+            assert transport.count("/web-interface/view") == 1
+        finally:
+            db.close()
+
+
+
 def main():
     test_snapshot_success_resume_and_force()
     test_resume_skips_ok_and_retries_failed()
@@ -449,6 +543,8 @@ def main():
     test_fan_resume_preserves_successful_value()
     test_force_interrupt_leaves_incomplete_marker()
     test_backup_failure_retries_without_api()
+    test_run_recent_videos_refreshes_without_completion_markers()
+    test_full_snapshot_skips_ok_rows_including_hourly_refresh()
     print("OK: 每日快照、断点续跑、失败占位与粉丝采集测试全部通过")
 
 

tests/test_web_queries.py

@@ -124,7 +124,8 @@ def test_growth_and_cold_start():
             cold = q.list_videos(sort="growth", growth_days=2)
             assert cold["available"] is False
             assert cold["items"] == []
-            assert "两天" in cold["reason"]
+            assert "至少 3 天快照" in cold["reason"]
+            assert cold["growth_days"] == 2
         finally:
             db.close()
 
@@ -361,10 +362,15 @@ def test_fans_candidates_system_and_export():
             assert system["errors"] == []
             assert system["preview"] is False
             assert system["running"] == {}
-            assert set(system["next_runs"]) >= {"fans", "snapshot", "discover"}
+            assert set(system["next_runs"]) >= {
+                "fans", "snapshot", "discover", "snapshot_recent",
+            }
             assert system["next_runs"]["snapshot"] > 0
             assert system["next_runs"]["fans"] > 0
             assert system["next_runs"]["discover"] > 0
+            assert system["next_runs"]["snapshot_recent"] > 0
+            assert system["schedule"]["recent_snapshot_minutes"] == 60
+            assert system["schedule"]["recent_video_days"] == 2
         finally:
             db.close()
 
@@ -483,6 +489,7 @@ def test_about_payload():
                 "min_view_for_ranking": 2000,
                 "video_shrink_minutes": 60,
                 "shrink_minutes": None,
+                "home_growth_days": "auto",
             }
             assert about["heat_weights"]["view"] == 0.30
             assert about["followers"]["videos_total"] == 3
@@ -553,6 +560,65 @@ def test_stats_query():
             db.close()
 
 
+
+def test_growth_auto_falls_back_to_one_day():
+    """3 天快照撑不起 7 日增长；auto 应降到 1 日。"""
+    import json
+    import os
+    import time
+
+    with tempfile.TemporaryDirectory() as tmp:
+        db, q = setup(tmp)
+        try:
+            auto = q.list_videos(sort="growth", growth_days="auto", page_size=6)
+            assert auto["available"] is True
+            assert auto["growth_days"] == 1
+            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  # 相邻日
+
+            forced7 = q.list_videos(sort="growth", growth_days=7)
+            assert forced7["available"] is False
+            assert "至少 8 天快照" in forced7["reason"]
+
+            # 配置钉死 7 天：auto 不再降级
+            path = Path(tmp) / "config.json"
+            data = json.loads(path.read_text(encoding="utf-8"))
+            data.setdefault("analytics", {})["home_growth_days"] = 7
+            path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
+            target = max(path.stat().st_mtime_ns, time.time_ns()) + 1_000_000
+            os.utime(path, ns=(target, target))
+            DASHBOARD_CACHE.clear()
+            pinned = q.list_videos(sort="growth", growth_days="auto")
+            assert pinned["growth_days"] == 7
+            assert pinned["available"] is False
+        finally:
+            db.close()
+
+
+def test_snapshot_recent_next_run_follows_completion():
+    import json
+
+    with tempfile.TemporaryDirectory() as tmp:
+        db, q = setup(tmp)
+        try:
+            repo = Repository(db)
+            finished = FIXED_NOW - 10
+            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()
+            assert system["next_runs"]["snapshot_recent"] == finished + 3600
+        finally:
+            db.close()
+
+
 def main_test():
     test_overview()
     test_video_listing_filters_and_sorting()
@@ -568,6 +634,8 @@ 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_growth_auto_falls_back_to_one_day()
+    test_snapshot_recent_next_run_follows_completion()
     print("OK: Web 查询层总览/榜单/详情/UP/时长效率/粉丝/候选/系统/访问统计测试全部通过")
 
 
@@ -605,7 +673,9 @@ def test_system_status_uses_completed_date_and_running_flags():
             system = q.system()
             assert "snapshot" in system["running"]
             assert system["running"]["snapshot"]["running"] is True
-            assert set(system["next_runs"]) >= {"fans", "snapshot", "discover"}
+            assert set(system["next_runs"]) >= {
+                "fans", "snapshot", "discover", "snapshot_recent",
+            }
 
             # 失败记录保留，完成日不变
             repo.kv_set(

tests/web_fixtures.py

@@ -50,6 +50,8 @@ def write_config(
             "discover_minutes": 60,
             "snapshot_at": "00:01",
             "candidate_ttl_days": 7,
+            "recent_snapshot_minutes": 60,
+            "recent_video_days": 2,
         },
     }, ensure_ascii=False, indent=2), encoding="utf-8")
     return path