291 lines
10 KiB
Python
291 lines
10 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""预览库生成器测试（离线）: uv run python tests/test_preview_seed.py"""
|
|
|
|
import json
|
|
import os
|
|
import random
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
from datetime import date
|
|
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.analytics import METRICS # noqa: E402
|
|
from reaction_board.bilibili.models import FavResource # noqa: E402
|
|
from reaction_board.config import FolderConfig # noqa: E402
|
|
from reaction_board.database import Database # noqa: E402
|
|
from reaction_board.repository import Repository # noqa: E402
|
|
from tools.seed_preview_db import ( # noqa: E402
|
|
FAKE_BLACKLIST_MID,
|
|
FAKE_CANDIDATE_AIDS,
|
|
backfill_metrics,
|
|
seed_preview,
|
|
spread_first_seen,
|
|
)
|
|
|
|
LATEST = "2026-10-31"
|
|
START = "2026-10-24"
|
|
|
|
|
|
def resource(aid: int) -> FavResource:
|
|
return FavResource(
|
|
aid=aid,
|
|
bvid=f"BV{aid}",
|
|
title=f"视频 {aid}",
|
|
up_mid=1000 + (aid % 5),
|
|
up_name=f"UP {aid % 5}",
|
|
pubdate=1_790_000_000 - aid * 3600,
|
|
duration=7200,
|
|
cover=f"//i0.hdslb.com/{aid}.jpg",
|
|
attr=0,
|
|
fav_time=None,
|
|
)
|
|
|
|
|
|
def build_source(tmp: str, *, count: int = 40) -> Path:
|
|
path = Path(tmp) / "app.db"
|
|
db = Database(path)
|
|
db.initialize()
|
|
try:
|
|
repo = Repository(db)
|
|
folder = FolderConfig(year=2026, media_id=3700023351, title="2026")
|
|
repo.ensure_folder(folder)
|
|
repo.seed_blacklist([12345])
|
|
for aid in range(1, count + 1):
|
|
repo.upsert_from_fav(resource(aid), folder.media_id, was_added=True)
|
|
repo.upsert_snapshot(
|
|
LATEST,
|
|
aid,
|
|
metrics={
|
|
"view": 1000 + aid * 500,
|
|
"like": 50 + aid * 10,
|
|
"coin": 20 + aid * 5,
|
|
"favorite": 30 + aid * 4,
|
|
"share": 2 + aid,
|
|
"reply": 10 + aid * 2,
|
|
"danmaku": 15 + aid * 3,
|
|
},
|
|
ok=True,
|
|
err=None,
|
|
captured_at=1_793_445_517,
|
|
)
|
|
repo.upsert_up_stat(
|
|
LATEST, 37754047, follower=2_767_270, following=95, ok=True,
|
|
captured_at=1_793_445_517,
|
|
)
|
|
finally:
|
|
db.close()
|
|
return path
|
|
|
|
|
|
def snapshot_rows(path: Path) -> list[dict]:
|
|
conn = sqlite3.connect(path)
|
|
conn.row_factory = sqlite3.Row
|
|
try:
|
|
return [dict(row) for row in conn.execute(
|
|
"SELECT * FROM snapshot ORDER BY aid, stat_date"
|
|
)]
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def test_backfill_metrics():
|
|
latest = {
|
|
"view": 10_000, "like": 500, "coin": 200, "favorite": 300,
|
|
"share": 20, "reply": 100, "danmaku": 150,
|
|
}
|
|
first = backfill_metrics(latest, 5, random.Random(7), multiplier=1.0)
|
|
second = backfill_metrics(latest, 5, random.Random(7), multiplier=1.0)
|
|
assert first == second, "同一随机种子必须生成相同序列"
|
|
assert len(first) == 5
|
|
previous = first[0]
|
|
for row in first[1:]:
|
|
for metric in METRICS:
|
|
assert row[metric] >= previous[metric], (metric, previous, row)
|
|
previous = row
|
|
for row in first:
|
|
for metric in METRICS:
|
|
if metric != "view":
|
|
assert row[metric] <= row["view"]
|
|
assert all(value >= 0 for row in first for value in row.values())
|
|
|
|
zero = backfill_metrics({}, 3, random.Random(1))
|
|
assert all(value == 0 for row in zero for value in row.values())
|
|
|
|
try:
|
|
backfill_metrics(latest, -1, random.Random(1))
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
raise AssertionError("负数天数应抛 ValueError")
|
|
|
|
|
|
def test_spread_first_seen():
|
|
aids = list(range(1, 101))
|
|
rng = random.Random(42)
|
|
result = spread_first_seen(aids, date.fromisoformat(START), date.fromisoformat(LATEST), rng)
|
|
assert set(result) == set(aids)
|
|
assert len(set(result.values())) > 1
|
|
for ts in result.values():
|
|
day = date.fromtimestamp(ts).isoformat()
|
|
assert START <= day <= LATEST
|
|
|
|
|
|
def test_refuses_same_path_and_existing_target():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
source = build_source(tmp)
|
|
try:
|
|
seed_preview(source, source, days=7)
|
|
except ValueError as exc:
|
|
assert "不能与源数据库相同" in str(exc)
|
|
else:
|
|
raise AssertionError("同路径应拒绝")
|
|
|
|
target = Path(tmp) / "preview.db"
|
|
seed_preview(source, target, days=7, config_source=None)
|
|
try:
|
|
seed_preview(source, target, days=7, config_source=None)
|
|
except FileExistsError:
|
|
pass
|
|
else:
|
|
raise AssertionError("已存在目标且未 --force 应拒绝")
|
|
|
|
|
|
def test_seed_preview_end_to_end():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
source = build_source(tmp)
|
|
target = Path(tmp) / "preview" / "app.db"
|
|
config_out = Path(tmp) / "preview" / "config.json"
|
|
summary = seed_preview(
|
|
source, target, days=7, seed=20261031, config_source=None,
|
|
config_out=config_out,
|
|
)
|
|
assert summary["ok"] is True
|
|
assert summary["snapshots_failed"] == 3
|
|
assert summary["date_from"] == START
|
|
assert summary["date_to"] == LATEST
|
|
assert summary["videos"] == 40
|
|
|
|
# 源库快照数量不变
|
|
assert len(snapshot_rows(source)) == 40
|
|
rows = snapshot_rows(target)
|
|
assert len(rows) > 40
|
|
dates = sorted({row["stat_date"] for row in rows})
|
|
assert dates == [
|
|
START, "2026-10-25", "2026-10-26", "2026-10-27",
|
|
"2026-10-28", "2026-10-29", "2026-10-30", LATEST,
|
|
]
|
|
synthetic = [row for row in rows if row["stat_date"] != LATEST]
|
|
assert synthetic
|
|
assert all(row["src"] == "preview" for row in synthetic)
|
|
assert sum(1 for row in synthetic if row["ok"] == 0) == 3
|
|
assert any(row["ok"] == 0 for row in synthetic)
|
|
|
|
conn = sqlite3.connect(target)
|
|
conn.row_factory = sqlite3.Row
|
|
try:
|
|
assert conn.execute(
|
|
"SELECT v FROM kv WHERE k='preview:seeded'"
|
|
).fetchone()[0] == "1"
|
|
first_seen = {
|
|
int(row["aid"]): int(row["first_seen"])
|
|
for row in conn.execute("SELECT aid, first_seen FROM video")
|
|
}
|
|
for row in synthetic:
|
|
day = date.fromtimestamp(first_seen[int(row["aid"])]).isoformat()
|
|
assert row["stat_date"] >= day, (row["aid"], row["stat_date"], day)
|
|
assert len(conn.execute(
|
|
"SELECT * FROM candidate WHERE channel='preview'"
|
|
).fetchall()) == len(FAKE_CANDIDATE_AIDS)
|
|
assert conn.execute(
|
|
"SELECT 1 FROM up_blacklist WHERE mid=?", (FAKE_BLACKLIST_MID,)
|
|
).fetchone() is not None
|
|
finally:
|
|
conn.close()
|
|
|
|
# 每个视频的累计指标单调不降
|
|
by_aid: dict[int, list[dict]] = {}
|
|
for row in rows:
|
|
if row["ok"] == 1:
|
|
by_aid.setdefault(int(row["aid"]), []).append(row)
|
|
for series in by_aid.values():
|
|
series.sort(key=lambda item: item["stat_date"])
|
|
for metric in METRICS:
|
|
values = [item[metric] for item in series if item[metric] is not None]
|
|
assert values == sorted(values), (metric, values)
|
|
|
|
config = json.loads(config_out.read_text(encoding="utf-8"))
|
|
assert config["telegram"]["enabled"] is False
|
|
assert config["bilibili"]["read"]["cookie"] == ""
|
|
assert config["bilibili"]["write"]["csrf"] == ""
|
|
assert config["web"]["port"] == 8081
|
|
assert config["storage"]["database_path"].endswith("app.db")
|
|
|
|
# 同 seed 生成结果一致
|
|
target2 = Path(tmp) / "preview2" / "app.db"
|
|
seed_preview(source, target2, days=7, seed=20261031, config_source=None)
|
|
rows2 = snapshot_rows(target2)
|
|
assert [
|
|
(row["stat_date"], row["aid"], row["ok"], row["view"], row["like"])
|
|
for row in rows
|
|
] == [
|
|
(row["stat_date"], row["aid"], row["ok"], row["view"], row["like"])
|
|
for row in rows2
|
|
]
|
|
|
|
|
|
def test_first_day_profile_and_config_no_secrets():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
source = build_source(tmp)
|
|
config_source = Path(tmp) / "config.json"
|
|
config_source.write_text(json.dumps({
|
|
"bilibili": {
|
|
"read": {"mid": 1, "csrf": "read-csrf", "cookie": "SESSDATA=fake-read-secret"},
|
|
"write": {"mid": 2, "csrf": "write-csrf", "cookie": "SESSDATA=secret2"},
|
|
},
|
|
"telegram": {"enabled": True, "bot_token": "secret-token", "chat_id": 1},
|
|
"folders": [{"year": 2026, "media_id": 3700023351, "title": "2026"}],
|
|
"track_fans": [37754047],
|
|
"heat_weights": {"view": 1.0},
|
|
}), encoding="utf-8")
|
|
target = Path(tmp) / "preview" / "app.db"
|
|
config_out = Path(tmp) / "preview" / "config.json"
|
|
summary = seed_preview(
|
|
source, target, days=7, profile="first-day",
|
|
config_source=config_source, config_out=config_out,
|
|
)
|
|
assert summary["snapshots_inserted"] == 0
|
|
assert len(snapshot_rows(target)) == 40
|
|
config = json.loads(config_out.read_text(encoding="utf-8"))
|
|
assert "secret" not in json.dumps(config)
|
|
assert config["telegram"]["enabled"] is False
|
|
assert config["heat_weights"] == {"view": 1.0}
|
|
assert config["folders"][0]["media_id"] == 3700023351
|
|
|
|
conn = sqlite3.connect(target)
|
|
try:
|
|
assert conn.execute(
|
|
"SELECT COUNT(*) FROM candidate WHERE channel='preview'"
|
|
).fetchone()[0] == 0
|
|
assert conn.execute(
|
|
"SELECT COUNT(*) FROM up_blacklist WHERE reason='preview'"
|
|
).fetchone()[0] == 0
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def main_test():
|
|
test_backfill_metrics()
|
|
test_spread_first_seen()
|
|
test_refuses_same_path_and_existing_target()
|
|
test_seed_preview_end_to_end()
|
|
test_first_day_profile_and_config_no_secrets()
|
|
print("OK: 预览库生成/单调/确定性/安全边界测试全部通过")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main_test()
|