+ ${cover(detail, "hero-cover")}
+ ${removedCoverMark(detail)}
+
${escapeHtml(detail.title)}
diff --git a/tests/fixtures.py b/tests/fixtures.py
index 65fcd52..cc4f68d 100644
--- a/tests/fixtures.py
+++ b/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,
diff --git a/tests/test_removed_cover_mark.py b/tests/test_removed_cover_mark.py
new file mode 100644
index 0000000..cf9d3ca
--- /dev/null
+++ b/tests/test_removed_cover_mark.py
@@ -0,0 +1,34 @@
+# -*- coding: utf-8 -*-
+"""失效封面徽标静态落地检查: uv run python tests/test_removed_cover_mark.py"""
+
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+STATIC = ROOT / "reaction_board" / "web" / "static"
+
+
+def test_removed_cover_mark_markup():
+ components = (STATIC / "components.js").read_text(encoding="utf-8")
+ assert "export function removedCoverMark" in components
+ assert 'data-removed-tip' in components
+ assert "该视频已失效" in components
+ assert "${removedCoverMark(item)}" in components
+
+ app = (STATIC / "app.js").read_text(encoding="utf-8")
+ assert 'closest("[data-removed-tip]")' in app
+ assert 'toast("该视频已失效")' in app
+
+ css = (STATIC / "style.css").read_text(encoding="utf-8")
+ assert ".cover-removed-mark" in css
+
+ video = (STATIC / "views" / "video.js").read_text(encoding="utf-8")
+ assert "removedCoverMark(detail)" in video
+
+
+def main():
+ test_removed_cover_mark_markup()
+ print("OK: 失效封面徽标静态落地检查通过")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/tests/test_repository.py b/tests/test_repository.py
index 8c37ca2..4293e89 100644
--- a/tests/test_repository.py
+++ b/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()
diff --git a/tests/test_snapshot.py b/tests/test_snapshot.py
index 79da107..3d7204a 100644
--- a/tests/test_snapshot.py
+++ b/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()
diff --git a/tests/test_web_queries.py b/tests/test_web_queries.py
index 56a0e9b..88f0d46 100644
--- a/tests/test_web_queries.py
+++ b/tests/test_web_queries.py
@@ -38,11 +38,11 @@ def test_overview():
assert overview["data_date"] == DATES[-1]
assert overview["snapshot_days"] == 3
assert overview["is_first_day"] is False
- assert overview["initial_import_count"] == 2
+ assert overview["initial_import_count"] == 3 # 含 removed 104
assert overview["counts"]["active"] == 2
assert overview["counts"]["unfaved"] == 1
assert overview["counts"]["removed"] == 1
- assert overview["counts"]["active_ups"] == 2
+ assert overview["counts"]["active_ups"] == 3 # 含 UP C（仅 removed）
assert overview["counts"]["candidates_pending"] == 1
assert overview["totals"]["view"] == 1200 + 5200 + 120
assert overview["interactions"] > 0
@@ -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 视频不应进入榜单"
+ # 热度门槛榜口径含 removed；104 无成功快照仍不进热度
+ 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,40 @@ 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()
+
+
+
+def test_heat_ranking_includes_removed_with_metrics():
+ """removed 进入热度门槛榜口径；有成功快照且过门槛时可见。"""
+ with tempfile.TemporaryDirectory() as tmp:
+ db, q = setup(tmp)
+ try:
+ db.write(lambda conn: conn.execute(
+ "UPDATE snapshot SET ok = 1, view = 5000, like = 10, coin = 1, "
+ "favorite = 1, share = 0, reply = 0, danmaku = 0, err = NULL "
+ "WHERE aid = 104 AND stat_date = ?",
+ (DATES[1],),
+ ))
+ DASHBOARD_CACHE.clear()
+ heat = q.list_videos(sort="heat")
+ aids = [item["aid"] for item in heat["items"]]
+ assert 104 in aids
+ assert next(item for item in heat["items"] if item["aid"] == 104)["status"] == "removed"
finally:
db.close()
@@ -183,13 +212,14 @@ def test_up_queries():
try:
ups = q.list_ups()
assert ups["sort"] == "avg_views"
- assert ups["total"] == 2
+ assert ups["total"] == 3 # 含仅有 removed 视频的 UP C
assert ups["items"][0]["up_mid"] == 1
assert ups["items"][0]["video_count"] == 2
assert ups["items"][0]["total_views"] == 6400
assert ups["items"][0]["avg_views"] == 3200.0
assert ups["items"][0]["up_face"] == "https://i0.hdslb.com/bfs/face/a.jpg"
assert ups["items"][0]["follower_latest"] == 12345
+ assert {item["up_mid"] for item in ups["items"]} == {1, 2, 3}
one = q.list_ups(min_videos=2)
assert one["total"] == 1
@@ -220,8 +250,8 @@ def test_up_queries():
# 大小写不敏感
lower = q.list_ups(q="up b")
assert lower["total"] == 1 and lower["items"][0]["up_mid"] == 2
- # 命中多个 UP
- assert q.list_ups(q="UP")["total"] == 2
+ # 命中多个 UP（含 UP C）
+ assert q.list_ups(q="UP")["total"] == 3
finally:
db.close()
@@ -457,7 +487,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,13 +495,13 @@ 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
+ assert ups["year"] == 2026 and ups["total"] == 3
ups_none = q.list_ups(sort="views", year=2025)
assert ups_none["total"] == 0
finally:
@@ -498,9 +528,9 @@ def test_about_payload():
"growth_rate_prior": 1000,
}
assert about["heat_weights"]["view"] == 0.30
- assert about["followers"]["videos_total"] == 3
+ assert about["followers"]["videos_total"] == 4 # 含 removed
assert about["followers"]["videos_covered"] == 3
- assert about["followers"]["videos_missing"] == 0
+ assert about["followers"]["videos_missing"] == 1
assert about["folders"][0]["media_id"] == 3700023351
assert about["preview"] is False
finally: