kairee
6a21b5d905
Backfill same-day snapshots for newly added or restored videos, preserve successful fan snapshots on resume, persist incomplete state before forced re-runs, and split collection completion from backup completion so failed backups can retry without API calls. Reject missing data payloads instead of treating them as empty folders, guard against mass-unfav on suspicious empty remote folders, rebuild Bilibili clients when credentials change, recompute primary folder ownership from active relations with per-folder fav_time, preserve original titles for dropped invalid videos, poll config every 60 seconds, and fix favsort verify interval and mixed resource types.
448 lines
16 KiB
Python
448 lines
16 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""每日快照与粉丝采集测试（transport 边界 Mock，不联网）:
|
|
uv run python tests/test_snapshot.py
|
|
"""
|
|
|
|
import io
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import urllib.parse
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
from fixtures import ( # noqa: E402
|
|
NAV_PAYLOAD,
|
|
RELATION_PAYLOAD,
|
|
VIEW_UNAVAILABLE,
|
|
view_for,
|
|
)
|
|
from reaction_board.bilibili.client import ( # noqa: E402
|
|
BilibiliClient,
|
|
HTTPResponse,
|
|
Pacer,
|
|
)
|
|
from reaction_board.bilibili.models import FavResource # noqa: E402
|
|
from reaction_board.config import ConfigStore, FolderConfig # noqa: E402
|
|
from reaction_board.database import Database # noqa: E402
|
|
from reaction_board.logging_json import JsonLogger # noqa: E402
|
|
from reaction_board.repository import Repository # noqa: E402
|
|
from reaction_board.services.snapshot import SnapshotService # noqa: E402
|
|
|
|
DATE = "2026-10-31"
|
|
FOLDER = FolderConfig(year=2026, media_id=3700023351, title="2026")
|
|
|
|
|
|
def http_json(payload):
|
|
return HTTPResponse(
|
|
200, {"content-type": "application/json"},
|
|
json.dumps(payload, ensure_ascii=False).encode("utf-8"),
|
|
)
|
|
|
|
|
|
class SnapshotTransport:
|
|
def __init__(self, views, *, nav=None, fan=None):
|
|
self.views = views
|
|
self.nav = NAV_PAYLOAD if nav is None else nav
|
|
self.fan = RELATION_PAYLOAD if fan is None else fan
|
|
self.calls = []
|
|
|
|
def request(self, method, url, headers, data=None, timeout=30):
|
|
self.calls.append(url)
|
|
if "/x/web-interface/nav" in url:
|
|
return http_json(self.nav)
|
|
if "/x/web-interface/view" in url:
|
|
aid = int(urllib.parse.parse_qs(
|
|
urllib.parse.urlparse(url).query
|
|
)["aid"][0])
|
|
payload = self.views[aid]
|
|
if isinstance(payload, BaseException):
|
|
raise payload
|
|
if isinstance(payload, HTTPResponse):
|
|
return payload
|
|
return http_json(payload)
|
|
if "/x/relation/stat" in url:
|
|
return http_json(self.fan)
|
|
raise AssertionError(f"未预期的 URL: {url}")
|
|
|
|
def count(self, part):
|
|
return sum(1 for url in self.calls if part in url)
|
|
|
|
|
|
def make_service(tmp, transport, *, track_fans=(37754047,), logger=None,
|
|
retries=3, cookie="SESSDATA=secret; buvid3=device"):
|
|
path = Path(tmp) / "config.json"
|
|
path.write_text(json.dumps({
|
|
"bilibili": {
|
|
"read": {"mid": 28802351, "csrf": "csrf", "cookie": cookie},
|
|
"write": {"mid": 2, "csrf": "write-csrf", "cookie": "write-cookie"},
|
|
},
|
|
"storage": {
|
|
"database_path": str(Path(tmp) / "app.db"),
|
|
"backup_dir": str(Path(tmp) / "backup"),
|
|
"backup_keep_days": 14,
|
|
},
|
|
"folders": [],
|
|
"track_fans": list(track_fans),
|
|
}, ensure_ascii=False), encoding="utf-8")
|
|
store = ConfigStore(path, env={}, require_auth=False)
|
|
db = Database(Path(tmp) / "app.db", backup_dir=Path(tmp) / "backup")
|
|
db.initialize()
|
|
repo = Repository(db, clock=lambda: 1_800_000_000)
|
|
repo.ensure_folder(FOLDER)
|
|
client = BilibiliClient(
|
|
cookie=cookie,
|
|
csrf="csrf",
|
|
mid=28802351,
|
|
transport=transport,
|
|
pacer=Pacer(0, 0),
|
|
retries=1,
|
|
)
|
|
service = SnapshotService(
|
|
store, repo, client,
|
|
clock=lambda: 1_800_000_000,
|
|
sleeper=lambda seconds: None,
|
|
logger=logger,
|
|
retries=retries,
|
|
backoff=0,
|
|
)
|
|
return db, repo, service
|
|
|
|
|
|
def add_video(repo, aid, *, status="active"):
|
|
repo.upsert_from_fav(
|
|
FavResource(
|
|
aid=aid, bvid=f"BV{aid}", title=f"视频{aid}", up_mid=100,
|
|
up_name="UP", pubdate=1_700_000_000, duration=600,
|
|
cover=None, attr=0, fav_time=None,
|
|
),
|
|
FOLDER.media_id,
|
|
was_added=True,
|
|
)
|
|
if status != "active":
|
|
repo.set_video_status(aid, status)
|
|
|
|
|
|
def test_snapshot_success_resume_and_force():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
transport = SnapshotTransport({
|
|
1: view_for(1, view=100, like=10, coin=2, favorite=3,
|
|
share=1, reply=4, danmaku=5),
|
|
2: view_for(2, view=200, like=20, coin=4, favorite=6,
|
|
share=2, reply=8, danmaku=10),
|
|
})
|
|
db, repo, service = make_service(tmp, transport)
|
|
try:
|
|
add_video(repo, 1)
|
|
add_video(repo, 2)
|
|
result = service.run(date=DATE)
|
|
assert result.completed is True
|
|
assert result.total == 2
|
|
assert result.success == 2
|
|
assert result.failed == 0
|
|
assert result.fans_ok == 1
|
|
assert result.fans_failed == 0
|
|
assert result.auth_ok is True
|
|
assert Path(result.backup_path or "").exists()
|
|
|
|
row = repo.get_snapshot(DATE, 1)
|
|
assert (row["view"], row["like"], row["favorite"], row["coin"],
|
|
row["share"], row["reply"], row["danmaku"]) == (100, 10, 3, 2, 1, 4, 5)
|
|
assert row["ok"] == 1 and row["src"] == "view"
|
|
fan = repo.get_up_stat(DATE, 37754047)
|
|
assert fan["follower"] == 2767268 and fan["ok"] == 1
|
|
assert repo.kv_get(service.completed_key(DATE)) == "1"
|
|
assert repo.kv_get("snapshot:last_completed") == DATE
|
|
assert repo.snapshot_counts(DATE) == {"total": 2, "ok": 2, "failed": 0}
|
|
|
|
# 已完成 -> 不再请求
|
|
view_before = transport.count("/web-interface/view")
|
|
again = service.run_if_needed(date=DATE)
|
|
assert again.already_completed is True
|
|
assert transport.count("/web-interface/view") == view_before
|
|
|
|
# force 重新采集
|
|
forced = service.run(date=DATE, force=True)
|
|
assert forced.success == 2
|
|
assert transport.count("/web-interface/view") == view_before + 2
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_resume_skips_ok_and_retries_failed():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
transport = SnapshotTransport({
|
|
1: view_for(1, view=111),
|
|
2: view_for(2, view=222),
|
|
})
|
|
db, repo, service = make_service(tmp, transport)
|
|
try:
|
|
add_video(repo, 1)
|
|
add_video(repo, 2)
|
|
repo.upsert_snapshot(
|
|
DATE, 1, metrics={"view": 1, "like": 1, "favorite": 1,
|
|
"coin": 1, "share": 1, "reply": 1, "danmaku": 1},
|
|
ok=True, err=None, captured_at=1,
|
|
)
|
|
repo.upsert_snapshot(
|
|
DATE, 2, metrics=None, ok=False, err="previous error",
|
|
captured_at=1,
|
|
)
|
|
result = service.run(date=DATE)
|
|
assert result.skipped == 1
|
|
assert result.success == 1
|
|
assert result.completed is True
|
|
assert transport.count("/web-interface/view") == 1
|
|
assert repo.get_snapshot(DATE, 1)["view"] == 1 # 未被覆盖
|
|
assert repo.get_snapshot(DATE, 2)["view"] == 222
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_unavailable_marks_removed():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
transport = SnapshotTransport({
|
|
1: VIEW_UNAVAILABLE,
|
|
2: HTTPResponse(404, {}, b"not found"),
|
|
})
|
|
db, repo, service = make_service(tmp, transport, track_fans=())
|
|
try:
|
|
add_video(repo, 1)
|
|
add_video(repo, 2)
|
|
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"
|
|
assert repo.get_snapshot(DATE, 1) is None
|
|
assert repo.get_snapshot(DATE, 2) is None
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_other_error_writes_failed_placeholder():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
transport = SnapshotTransport({
|
|
1: {"code": -500, "message": "服务异常", "data": None},
|
|
})
|
|
db, repo, service = make_service(tmp, transport, track_fans=(), retries=3)
|
|
try:
|
|
add_video(repo, 1)
|
|
result = service.run(date=DATE)
|
|
assert result.failed == 1
|
|
assert result.success == 0
|
|
assert result.completed is False
|
|
row = repo.get_snapshot(DATE, 1)
|
|
assert row["ok"] == 0
|
|
assert row["view"] is None and row["like"] is None
|
|
assert "code=-500" in row["err"]
|
|
assert repo.kv_get(service.completed_key(DATE)) == "0"
|
|
# 每次业务失败会走匿名+完整 Cookie 两条请求，服务层重试 3 次
|
|
assert transport.count("/web-interface/view") == 6
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_fan_failure_placeholder():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
transport = SnapshotTransport({}, fan={"code": -500, "message": "x", "data": None})
|
|
db, repo, service = make_service(tmp, transport, retries=2)
|
|
try:
|
|
result = service.run(date=DATE)
|
|
assert result.total == 0
|
|
assert result.fans_failed == 1
|
|
assert result.completed is False
|
|
row = repo.get_up_stat(DATE, 37754047)
|
|
assert row["ok"] == 0
|
|
assert row["follower"] is None and row["following"] is None
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_auth_invalid_is_warned_but_snapshot_continues():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
stream = io.StringIO()
|
|
logger = JsonLogger(stream)
|
|
transport = SnapshotTransport(
|
|
{}, nav={"code": 0, "message": "0", "ttl": 1,
|
|
"data": {"isLogin": False, "mid": 0, "uname": ""}},
|
|
fan=RELATION_PAYLOAD,
|
|
)
|
|
db, repo, service = make_service(tmp, transport, logger=logger)
|
|
try:
|
|
result = service.run(date=DATE)
|
|
assert result.auth_ok is False
|
|
assert result.completed is True
|
|
assert repo.kv_get("auth:ok") == "0"
|
|
events = [json.loads(line) for line in stream.getvalue().splitlines()]
|
|
assert any(e["event"] == "auth.invalid" for e in events)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_no_cookie_skips_nav_request():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
transport = SnapshotTransport({}, fan=RELATION_PAYLOAD)
|
|
db, repo, service = make_service(tmp, transport, cookie="")
|
|
try:
|
|
result = service.run(date=DATE)
|
|
assert result.auth_ok is False
|
|
assert result.auth_message == "未配置 cookie"
|
|
assert transport.count("/web-interface/nav") == 0
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_force_failure_clears_last_completed():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
transport = SnapshotTransport({1: view_for(1, view=100)})
|
|
db, repo, service = make_service(tmp, transport, track_fans=(), retries=1)
|
|
try:
|
|
add_video(repo, 1)
|
|
first = service.run(date=DATE)
|
|
assert first.completed is True
|
|
assert repo.kv_get("snapshot:last_completed") == DATE
|
|
|
|
transport.views[1] = {"code": -500, "message": "x", "data": None}
|
|
second = service.run(date=DATE, force=True)
|
|
assert second.failed == 1
|
|
assert second.completed is False
|
|
assert repo.kv_get("snapshot:last_completed") == ""
|
|
assert repo.get_snapshot(DATE, 1)["ok"] == 0
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_new_video_after_completion_backfills_same_day():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
transport = SnapshotTransport({1: view_for(1, view=100)})
|
|
db, repo, service = make_service(tmp, transport, track_fans=())
|
|
try:
|
|
add_video(repo, 1)
|
|
first = service.run(date=DATE)
|
|
assert first.completed is True
|
|
assert transport.count("/web-interface/view") == 1
|
|
|
|
# 当天 11:00 新入库：完成标记仍是 1，但必须补采新视频
|
|
add_video(repo, 2)
|
|
transport.views[2] = view_for(2, view=222)
|
|
second = service.run_if_needed(date=DATE)
|
|
assert second.already_completed is False
|
|
assert second.success == 1
|
|
assert second.skipped == 1
|
|
assert transport.count("/web-interface/view") == 2
|
|
assert repo.get_snapshot(DATE, 2)["view"] == 222
|
|
assert repo.get_snapshot(DATE, 1)["view"] == 100
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_fan_resume_preserves_successful_value():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
transport = SnapshotTransport(
|
|
{1: {"code": -500, "message": "x", "data": None}},
|
|
)
|
|
db, repo, service = make_service(tmp, transport, retries=1)
|
|
try:
|
|
add_video(repo, 1)
|
|
first = service.run(date=DATE)
|
|
assert first.failed == 1
|
|
assert first.fans_ok == 1
|
|
assert repo.get_up_stat(DATE, 37754047)["follower"] == 2767268
|
|
|
|
transport.views[1] = view_for(1, view=111)
|
|
transport.fan = {"code": -500, "message": "x", "data": None}
|
|
second = service.run(date=DATE)
|
|
assert second.fans_skipped == 1
|
|
assert second.fans_failed == 0
|
|
fan = repo.get_up_stat(DATE, 37754047)
|
|
assert fan["follower"] == 2767268
|
|
assert fan["ok"] == 1
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_force_interrupt_leaves_incomplete_marker():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
transport = SnapshotTransport({
|
|
1: view_for(1),
|
|
2: RuntimeError("simulated interrupt"),
|
|
})
|
|
db, repo, service = make_service(tmp, transport, track_fans=())
|
|
try:
|
|
add_video(repo, 1)
|
|
add_video(repo, 2)
|
|
repo.kv_set(service.completed_key(DATE), "1")
|
|
repo.kv_set("snapshot:last_completed", DATE)
|
|
try:
|
|
service.run(date=DATE, force=True)
|
|
except RuntimeError:
|
|
pass
|
|
else:
|
|
raise AssertionError("应模拟中断")
|
|
assert repo.kv_get(service.completed_key(DATE)) == "0"
|
|
assert repo.kv_get("snapshot:last_completed") == ""
|
|
|
|
transport.views[2] = view_for(2)
|
|
resumed = service.run_if_needed(date=DATE)
|
|
assert resumed.completed is True
|
|
assert resumed.success == 1
|
|
assert resumed.skipped == 1
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_backup_failure_retries_without_api():
|
|
import reaction_board.services.snapshot as snapshot_module
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
transport = SnapshotTransport({1: view_for(1)})
|
|
db, repo, service = make_service(tmp, transport, track_fans=())
|
|
original = snapshot_module.backup_daily
|
|
|
|
def fail_backup(*args, **kwargs):
|
|
raise OSError("disk full")
|
|
|
|
snapshot_module.backup_daily = fail_backup
|
|
try:
|
|
first = service.run(date=DATE)
|
|
finally:
|
|
snapshot_module.backup_daily = original
|
|
assert first.completed is True
|
|
assert first.backup_error
|
|
assert repo.kv_get(service.backup_key(DATE)) == "0"
|
|
view_calls = transport.count("/web-interface/view")
|
|
|
|
retried = service.run_if_needed(date=DATE)
|
|
assert retried.already_completed is True
|
|
assert retried.backup_error is None
|
|
assert Path(retried.backup_path or "").exists()
|
|
assert repo.kv_get(service.backup_key(DATE)) == "1"
|
|
assert transport.count("/web-interface/view") == view_calls
|
|
db.close()
|
|
|
|
|
|
def main():
|
|
test_snapshot_success_resume_and_force()
|
|
test_resume_skips_ok_and_retries_failed()
|
|
test_unavailable_marks_removed()
|
|
test_other_error_writes_failed_placeholder()
|
|
test_fan_failure_placeholder()
|
|
test_auth_invalid_is_warned_but_snapshot_continues()
|
|
test_no_cookie_skips_nav_request()
|
|
test_force_failure_clears_last_completed()
|
|
test_new_video_after_completion_backfills_same_day()
|
|
test_fan_resume_preserves_successful_value()
|
|
test_force_interrupt_leaves_incomplete_marker()
|
|
test_backup_failure_retries_without_api()
|
|
print("OK: 每日快照、断点续跑、失败占位与粉丝采集测试全部通过")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|