feat(snapshot): refresh API video metadata; keep removed visible

入库 / 小时近新片 / 全量快照成功时以 view_detail 覆盖 title、duration、
pages、cover、bvid、up_*、pubdate 等接口字段，并照旧写七项快照。
原站失效仅标 removed，不删行、不空写元数据；列表与详情继续展示，
快照队列仍跳过 removed。

tests/fixtures.py

@@ -256,11 +256,41 @@ def view_for(
     reply: int = 16,
     danmaku: int = 7,
     bvid: str | None = None,
+    title: str | None = None,
+    duration: int | None = None,
+    pages: int | None = None,
+    cover: str | None = None,
+    up_mid: int | None = None,
+    up_name: str | None = None,
+    up_face: str | None = None,
+    pubdate: int | None = None,
 ) -> dict[str, Any]:
     item = copy.deepcopy(VIEW_PAYLOAD)
     data = item["data"]
     data["aid"] = aid
     data["bvid"] = bvid or f"BV{aid}"
+    if title is not None:
+        data["title"] = title
+    if duration is not None:
+        data["duration"] = duration
+    if cover is not None:
+        data["pic"] = cover
+    if pubdate is not None:
+        data["pubdate"] = pubdate
+        data["ctime"] = pubdate
+    if pages is not None:
+        data["videos"] = pages
+        data["pages"] = [
+            {"cid": 1000 + i, "page": i + 1, "from": "vupload", "part": f"P{i + 1}"}
+            for i in range(max(1, int(pages)))
+        ]
+    owner = data.setdefault("owner", {})
+    if up_mid is not None:
+        owner["mid"] = up_mid
+    if up_name is not None:
+        owner["name"] = up_name
+    if up_face is not None:
+        owner["face"] = up_face
     data["stat"].update({
         "aid": aid,
         "view": view,

tests/test_repository.py

@@ -408,6 +408,72 @@ def test_up_follower_snapshot_and_backfill():
             db.close()
 
 
+def test_refresh_video_metadata_overwrites_api_fields_keeps_nulls():
+    with tempfile.TemporaryDirectory() as tmp:
+        db, repo = setup(tmp)
+        try:
+            repo.upsert_from_fav(resource(9, pages=1), F1.media_id, was_added=True)
+            ok = repo.refresh_video_metadata(
+                aid=9,
+                bvid="BVnew9",
+                title="刷新后",
+                up_mid=200,
+                up_name="新名",
+                pubdate=1_710_000_000,
+                duration=999,
+                pages=5,
+                cover="http://example/new.png",
+                up_face="http://example/face.png",
+                now=1_800_000_100,
+            )
+            assert ok is True
+            video = repo.get_video(9)
+            assert video["title"] == "刷新后"
+            assert video["duration"] == 999
+            assert video["pages"] == 5
+            assert video["bvid"] == "BVnew9"
+            assert video["up_mid"] == 200
+            assert video["cover"] == "http://example/new.png"
+            assert video["status"] == "active"
+            # 空值不冲掉
+            ok = repo.refresh_video_metadata(
+                aid=9,
+                bvid=None,
+                title="",
+                up_mid=None,
+                up_name=None,
+                pubdate=0,
+                duration=None,
+                pages=None,
+                cover=None,
+                up_face=None,
+                now=1_800_000_200,
+            )
+            assert ok is True
+            video = repo.get_video(9)
+            assert video["title"] == "刷新后"
+            assert video["duration"] == 999
+            assert video["pages"] == 5
+            assert video["bvid"] == "BVnew9"
+            # 不存在的 aid
+            assert repo.refresh_video_metadata(
+                aid=404,
+                bvid=None, title="x", up_mid=None, up_name=None, pubdate=1,
+            ) is False
+            # removed 后 refresh 也不改 status
+            repo.set_video_status(9, "removed")
+            repo.refresh_video_metadata(
+                aid=9, bvid="BVkeep", title="仍在", up_mid=200, up_name="新名",
+                pubdate=1_710_000_000, duration=1000, pages=6,
+            )
+            video = repo.get_video(9)
+            assert video["status"] == "removed"
+            assert video["title"] == "仍在"
+            assert video["pages"] == 6
+        finally:
+            db.close()
+
+
 def main():
     test_upsert_and_folder_resource()
     test_invalid_attr_and_missing()
@@ -422,8 +488,10 @@ def main():
     test_pending_snapshot_aids()
     test_upsert_snapshot_preserves_view_open()
     test_up_follower_snapshot_and_backfill()
+    test_refresh_video_metadata_overwrites_api_fields_keeps_nulls()
     print("OK: 收藏夹/视频状态机与黑名单测试全部通过")
 
 
+
 if __name__ == "__main__":
     main()

tests/test_snapshot.py

@@ -216,14 +216,86 @@ def test_unavailable_marks_removed():
         try:
             add_video(repo, 1)
             add_video(repo, 2)
+            before = repo.get_video(1)
             result = service.run(date=DATE)
             assert result.removed == 2
             assert result.failed == 0
             assert result.completed is True
-            assert repo.get_video(1)["status"] == "removed"
-            assert repo.get_video(2)["status"] == "removed"
+            video1 = repo.get_video(1)
+            video2 = repo.get_video(2)
+            assert video1["status"] == "removed"
+            assert video2["status"] == "removed"
+            # 失效只改 status，不得抹元数据 / 不得写空快照
+            assert video1["title"] == before["title"]
+            assert video1["duration"] == before["duration"]
+            assert video1["bvid"] == before["bvid"]
             assert repo.get_snapshot(DATE, 1) is None
             assert repo.get_snapshot(DATE, 2) is None
+            # removed 不再进入后续快照队列
+            assert 1 not in repo.video_aids_for_snapshot()
+            assert 2 not in repo.video_aids_for_snapshot()
+        finally:
+            db.close()
+
+
+def test_snapshot_refreshes_video_metadata():
+    """快照成功路径以 view_detail 覆盖 pages/duration/title 等元数据。"""
+    with tempfile.TemporaryDirectory() as tmp:
+        transport = SnapshotTransport({
+            1: view_for(
+                1, view=111, title="新标题P3", duration=3600, pages=3,
+                cover="http://example/new.jpg", up_name="新UP",
+            ),
+        })
+        db, repo, service = make_service(tmp, transport, track_fans=())
+        try:
+            add_video(repo, 1)
+            old = repo.get_video(1)
+            assert old["title"] == "视频1"
+            assert old["duration"] == 600
+            assert old.get("pages") in (None, 0) or old["pages"] != 3
+            result = service.run_videos(date=DATE)
+            assert result.success == 1
+            video = repo.get_video(1)
+            assert video["title"] == "新标题P3"
+            assert video["duration"] == 3600
+            assert video["pages"] == 3
+            assert video["cover"] == "http://example/new.jpg"
+            assert video["up_name"] == "新UP"
+            assert video["status"] == "active"
+            snap = repo.get_snapshot(DATE, 1)
+            assert snap["view"] == 111 and snap["ok"] == 1
+        finally:
+            db.close()
+
+
+def test_recent_snapshot_refreshes_metadata():
+    with tempfile.TemporaryDirectory() as tmp:
+        now = 1_800_000_000
+        transport = SnapshotTransport({
+            7: view_for(7, view=77, title="近新片加P", duration=1200, pages=2,
+                        pubdate=now - 3600),
+        })
+        db, repo, service = make_service(tmp, transport, track_fans=())
+        try:
+            repo.upsert_from_fav(
+                FavResource(
+                    aid=7, bvid="BV7", title="旧标题", up_mid=100,
+                    up_name="UP", pubdate=now - 3600, duration=300,
+                    cover=None, attr=0, fav_time=None, pages=1,
+                ),
+                FOLDER.media_id,
+                was_added=True,
+            )
+            result = service.run_recent_videos(date=DATE, recent_days=2)
+            assert result.success == 1
+            video = repo.get_video(7)
+            assert video["title"] == "近新片加P"
+            assert video["duration"] == 1200
+            assert video["pages"] == 2
+            assert repo.get_snapshot(DATE, 7)["view"] == 77
+            # recent 不推进完成日
+            assert repo.kv_get("snapshot:last_completed") in (None, "")
         finally:
             db.close()
 
@@ -440,11 +512,12 @@ def test_run_recent_videos_refreshes_without_completion_markers():
     """小时任务只刷新近窗新片，不写完成日 / 备份。"""
     clock = 1_800_000_000
     with tempfile.TemporaryDirectory() as tmp:
+        # view_for 须带与库内一致的 pubdate，否则快照回写会把近窗片挤出窗口
         transport = SnapshotTransport({
-            1: view_for(1, view=111),
-            2: view_for(2, view=222),
-            3: view_for(3, view=333),
-            4: view_for(4, view=444),
+            1: view_for(1, view=111, pubdate=clock - 3600),
+            2: view_for(2, view=222, pubdate=clock - 86400),
+            3: view_for(3, view=333, pubdate=clock - 10 * 86400),
+            4: view_for(4, view=444, pubdate=clock - 10 * 86400),
         })
         db, repo, service = make_service(tmp, transport, track_fans=())
         try:
@@ -776,6 +849,8 @@ def main():
     test_snapshot_success_resume_and_force()
     test_resume_skips_ok_and_retries_failed()
     test_unavailable_marks_removed()
+    test_snapshot_refreshes_video_metadata()
+    test_recent_snapshot_refreshes_metadata()
     test_other_error_writes_failed_placeholder()
     test_fan_failure_placeholder()
     test_auth_invalid_is_warned_but_snapshot_continues()

tests/test_web_queries.py

@@ -68,13 +68,16 @@ def test_video_listing_filters_and_sorting():
 
             heat = q.list_videos(sort="heat")
             aids = [item["aid"] for item in heat["items"]]
-            assert 104 not in aids, "removed 视频不应进入榜单"
+            # 热度榜仍按门槛过滤；104 removed 无成功快照，不进热度
+            assert 104 not in aids
             # 综合热度榜应用最小播放门槛 2000：101（1200）与 103（120）被排除
             assert aids == [102], aids
             assert heat["items"][0]["heat_percentile"] == 100.0
 
             views = q.list_videos(sort="view")
-            assert [item["aid"] for item in views["items"]] == [102, 101, 103]
+            # removed(104) 站内继续展示；无指标排在末尾
+            assert [item["aid"] for item in views["items"]] == [102, 101, 103, 104]
+            assert views["items"][-1]["status"] == "removed"
 
             interactions = q.list_videos(sort="interaction")
             assert interactions["items"][0]["aid"] == 101
@@ -91,14 +94,19 @@ def test_video_listing_filters_and_sorting():
             assert [item["aid"] for item in searched["items"]] == [102]
 
             folder = q.list_videos(sort="view", folder_id=3700023351)
-            assert folder["total"] == 3
+            assert folder["total"] == 4  # 含 removed 104
             folder = q.list_videos(sort="view", folder_id=999)
             assert folder["total"] == 0
 
             page = q.list_videos(sort="view", page=2, page_size=1)
-            assert page["total"] == 3
-            assert page["pages"] == 3
+            assert page["total"] == 4
+            assert page["pages"] == 4
             assert [item["aid"] for item in page["items"]] == [101]
+
+            detail = q.video_detail(104)
+            assert detail is not None
+            assert detail["status"] == "removed"
+            assert detail["aid"] == 104
         finally:
             db.close()
 
@@ -457,7 +465,7 @@ def test_year_filter_and_published_sort():
         db, q = setup(tmp)
         try:
             by_year = q.list_videos(sort="view", year=2026)
-            assert [item["aid"] for item in by_year["items"]] == [102, 101, 103]
+            assert [item["aid"] for item in by_year["items"]] == [102, 101, 103, 104]
             assert by_year["year"] == 2026
 
             none = q.list_videos(sort="view", year=2025)
@@ -465,10 +473,10 @@ def test_year_filter_and_published_sort():
             assert none["total"] == 0
 
             published = q.list_videos(sort="published")
-            assert [item["aid"] for item in published["items"]] == [103, 101, 102]
+            assert [item["aid"] for item in published["items"]] == [103, 101, 102, 104]
 
             added = q.list_videos(sort="added")
-            assert added["total"] == 3
+            assert added["total"] == 4  # 含 removed
 
             ups = q.list_ups(sort="efficiency", year=2026)
             assert ups["year"] == 2026 and ups["total"] == 2