diff --git a/tests/test_preview_seed.py b/tests/test_preview_seed.py new file mode 100644 index 0000000..8ef5468 --- /dev/null +++ b/tests/test_preview_seed.py @@ -0,0 +1,290 @@ +# -*- 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() diff --git a/tools/__init__.py b/tools/__init__.py new file mode 100644 index 0000000..0057617 --- /dev/null +++ b/tools/__init__.py @@ -0,0 +1,2 @@ +# -*- coding: utf-8 -*- +"""开发辅助脚本。""" diff --git a/tools/seed_preview_db.py b/tools/seed_preview_db.py new file mode 100644 index 0000000..0347eba --- /dev/null +++ b/tools/seed_preview_db.py @@ -0,0 +1,642 @@ +# -*- coding: utf-8 -*- +"""生成独立预览数据库。 + +真实库只读：使用 SQLite backup API 复制到 ``data/preview/app.db``，再向副本 +回填可重复的合成快照。合成行写入 ``src='preview'``，并在 ``kv`` 留下 +``preview:seeded=1`` 标记，供只读 Web 预览模式校验。 + +用法:: + + uv run python tools/seed_preview_db.py --days 7 --seed 20261031 + uv run python app.py serve --preview --config data/preview/config.json +""" + +from __future__ import annotations + +import argparse +import json +import os +import random +import sqlite3 +import sys +from datetime import date, datetime, time, timedelta +from pathlib import Path +from typing import Any, Mapping, Sequence + +ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT)) + +from reaction_board.analytics import METRICS # noqa: E402 +from reaction_board.timeutil import SHANGHAI_TZ, shanghai_datetime # noqa: E402 + +DEFAULT_SOURCE = ROOT / "data" / "app.db" +DEFAULT_TARGET = ROOT / "data" / "preview" / "app.db" +DEFAULT_CONFIG_SOURCE = ROOT / "config.json" + +BASE_DAILY_RATES: dict[str, float] = { + "view": 0.0040, + "like": 0.0050, + "coin": 0.0060, + "favorite": 0.0040, + "share": 0.0080, + "reply": 0.0050, + "danmaku": 0.0045, +} +MISSING_RATE = 0.02 +FAILED_ROWS = 3 +FAKE_CANDIDATE_AIDS = ( + 990000000000000001, + 990000000000000002, + 990000000000000003, + 990000000000000004, +) +FAKE_BLACKLIST_MID = 990000000000000099 + + +def _day_start(day: date) -> float: + return datetime.combine(day, time(0, 1), tzinfo=SHANGHAI_TZ).timestamp() + + +def _timestamp(day: date, *, hour: int = 0, minute: int = 1, second: int = 0) -> float: + return datetime.combine(day, time(hour, minute, second), tzinfo=SHANGHAI_TZ).timestamp() + + +def parse_day(value: str) -> date: + try: + return date.fromisoformat(value) + except ValueError as exc: + raise ValueError(f"日期必须是 YYYY-MM-DD: {value!r}") from exc + + +def backfill_metrics( + latest: Mapping[str, Any], + days: int, + rng: random.Random, + *, + multiplier: float = 1.0, +) -> list[dict[str, int]]: + """从最新值向前回填 ``days`` 天，返回按时间升序的指标列表。 + + 累计指标单调不降；每次向前的减量至少为 1（值大于 0 时），避免大量视频 + 出现完全相同的水平线。互动类指标不会超过当天播放量。 + """ + if days < 0: + raise ValueError("days 不能为负") + current = {metric: max(0, int(latest.get(metric) or 0)) for metric in METRICS} + series: list[dict[str, int]] = [] + for _ in range(days): + previous: dict[str, int] = {} + for metric in METRICS: + value = current[metric] + if value <= 0: + previous[metric] = 0 + continue + rate = BASE_DAILY_RATES.get(metric, 0.004) * max(multiplier, 0.0) + rate *= rng.uniform(0.4, 1.6) + decrement = int(round(value * rate)) + if value > 0: + decrement = max(decrement, 1) + previous[metric] = max(0, value - decrement) + view = previous["view"] + for metric in METRICS: + if metric != "view": + previous[metric] = min(previous[metric], view) + series.append(previous) + current = previous + series.reverse() + return series + + +def spread_first_seen( + aids: Sequence[int], + start_day: date, + latest_day: date, + rng: random.Random, + *, + spread_ratio: float = 0.30, +) -> dict[int, float]: + """把一部分视频的 ``first_seen`` 打散到回填区间内。 + + 其余视频保持“首批导入”，即回填起始日。返回 aid -> 时间戳。 + """ + total_days = max((latest_day - start_day).days, 0) + result: dict[int, float] = {} + for aid in aids: + if rng.random() >= spread_ratio: + result[int(aid)] = _timestamp(start_day, hour=0, minute=30) + continue + offset = rng.randint(0, total_days) + day = start_day + timedelta(days=offset) + result[int(aid)] = _timestamp( + day, hour=rng.randint(8, 22), minute=rng.randint(0, 59) + ) + return result + + +def build_preview_config( + source_config: Mapping[str, Any] | None, + *, + target: Path, + config_out: Path, + folders: Sequence[Mapping[str, Any]], + track_fans: Sequence[int], + blacklist: Sequence[int], +) -> dict[str, Any]: + """生成不含凭据的预览配置。""" + source = dict(source_config or {}) + result: dict[str, Any] = { + "bilibili": { + "read": {"mid": 0, "csrf": "", "cookie": ""}, + "write": {"mid": 0, "csrf": "", "cookie": ""}, + }, + "telegram": {"enabled": False, "bot_token": "", "chat_id": 0}, + "web": { + "host": "127.0.0.1", + "port": 8081, + "auth_token": "", + "public_base_url": "", + }, + "storage": { + "database_path": os.path.relpath(target, config_out.parent).replace("\\", "/"), + "backup_dir": "backup", + "backup_keep_days": 7, + }, + "folders": [dict(item) for item in folders], + "track_fans": [int(mid) for mid in track_fans], + "up_blacklist_seed": [int(mid) for mid in blacklist], + } + for key in ("schedule", "notify", "discover", "classify", "heat_weights"): + value = source.get(key) + if isinstance(value, Mapping): + result[key] = json.loads(json.dumps(value, ensure_ascii=False)) + return result + + +def _copy_database(source: Path, target: Path) -> None: + source_uri = source.resolve().as_posix() + src = sqlite3.connect(f"file:{source_uri}?mode=ro", uri=True) + dst = sqlite3.connect(target) + try: + src.backup(dst) + finally: + dst.close() + src.close() + + +def _read_source_config(path: Path | None) -> dict[str, Any] | None: + if path is None or not path.exists(): + return None + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return None + return data if isinstance(data, dict) else None + + +def _latest_snapshots(conn: sqlite3.Connection) -> dict[int, dict[str, Any]]: + rows = conn.execute( + """ + SELECT s.* + FROM snapshot s + JOIN ( + SELECT aid, MAX(stat_date) AS stat_date + FROM snapshot WHERE ok = 1 GROUP BY aid + ) latest ON latest.aid = s.aid AND latest.stat_date = s.stat_date + """ + ).fetchall() + result: dict[int, dict[str, Any]] = {} + for row in rows: + data = dict(row) + result[int(data["aid"])] = data + return result + + +def _video_rows(conn: sqlite3.Connection) -> list[dict[str, Any]]: + rows = conn.execute( + "SELECT aid, pubdate, first_seen FROM video " + "WHERE status IN ('active', 'unfaved') ORDER BY aid" + ).fetchall() + return [dict(row) for row in rows] + + +def _insert_snapshot( + conn: sqlite3.Connection, + stat_date: str, + aid: int, + metrics: Mapping[str, int] | None, + *, + captured_at: float, + ok: bool, + err: str | None, +) -> None: + values = metrics or {} + conn.execute( + """ + INSERT OR REPLACE INTO snapshot( + stat_date, aid, "view", "like", favorite, coin, share, reply, + danmaku, captured_at, ok, err, src + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'preview') + """, + ( + stat_date, + int(aid), + values.get("view"), + values.get("like"), + values.get("favorite"), + values.get("coin"), + values.get("share"), + values.get("reply"), + values.get("danmaku"), + int(captured_at), + 1 if ok else 0, + err, + ), + ) + + +def _seed_blacklist(conn: sqlite3.Connection, now: float) -> None: + conn.execute( + """ + INSERT OR REPLACE INTO up_blacklist(mid, name, reason, created_at) + VALUES (?, ?, 'preview', ?) + """, + (FAKE_BLACKLIST_MID, "[预览] 切片号", int(now)), + ) + + +def _seed_candidates(conn: sqlite3.Connection, latest_day: date) -> None: + base = _timestamp(latest_day, hour=14, minute=0) + rows = [ + (6, "高分候选：咻咻满 reaction 全程震惊", "预览UP A", 0, 1), + (4, "中等候选：第一次听咻咻满", "预览UP B", 1, 1), + (2, "低分候选：咻咻满 充电专场", "预览UP C", 2, 0), + (1, "失败样本 [预览-失败]：咻咻满", "预览UP D", 3, 0), + ] + for index, (score, title, up_name, day_offset, notified) in enumerate(rows): + aid = FAKE_CANDIDATE_AIDS[index] + first_seen = base - day_offset * 86400 - index * 900 + conn.execute( + """ + INSERT OR REPLACE INTO candidate( + aid, bvid, title, up_mid, up_name, pubdate, cover, score, channel, + first_seen, notified_at, decision, decided_at, + duration, play, like, favorite, danmaku, review + ) VALUES (?, ?, ?, ?, ?, ?, NULL, ?, 'preview', ?, ?, NULL, NULL, + ?, ?, ?, ?, ?, ?) + """, + ( + aid, + f"BVpreview{index + 1:04d}", + title, + 990000000000000100 + index, + up_name, + int(first_seen - 3600), + float(score), + int(first_seen), + int(first_seen + 600) if notified else None, + 5400 + index * 1200, + 12000 + index * 3000, + 300 + index * 50, + 120 + index * 20, + 80 + index * 10, + 40 + index * 5, + ), + ) + + +def _seed_up_stats( + conn: sqlite3.Connection, + latest_day: date, + days: int, + rng: random.Random, +) -> int: + latest: dict[int, dict[str, Any]] = {} + for row in conn.execute( + """ + SELECT u.* + FROM up_stat u + JOIN (SELECT mid, MAX(stat_date) AS stat_date FROM up_stat WHERE ok=1 GROUP BY mid) m + ON m.mid = u.mid AND m.stat_date = u.stat_date + """ + ).fetchall(): + data = dict(row) + latest[int(data["mid"])] = data + + inserted = 0 + for mid, data in latest.items(): + follower = int(data.get("follower") or 0) + following = data.get("following") + current = follower + for offset in range(days, 0, -1): + stat_date = (latest_day - timedelta(days=offset)).isoformat() + decrement = max(1, int(round(current * rng.uniform(0.00015, 0.0006)))) + current = max(0, current - decrement) + conn.execute( + """ + INSERT OR REPLACE INTO up_stat( + stat_date, mid, follower, following, captured_at, ok + ) VALUES (?, ?, ?, ?, ?, 1) + """, + ( + stat_date, + int(mid), + current, + following, + int(_day_start(latest_day - timedelta(days=offset)) + 60), + ), + ) + inserted += 1 + return inserted + + +def seed_preview( + source: str | Path = DEFAULT_SOURCE, + target: str | Path = DEFAULT_TARGET, + *, + days: int = 7, + seed: int = 20261031, + profile: str = "live", + force: bool = False, + config_source: str | Path | None = DEFAULT_CONFIG_SOURCE, + config_out: str | Path | None = None, + with_anomalies: bool = True, +) -> dict[str, Any]: + """复制并回填预览库；返回摘要。真实库不会被写入。""" + if days < 1: + raise ValueError("--days 必须至少为 1") + if profile not in {"live", "first-day"}: + raise ValueError("--profile 必须是 live 或 first-day") + + source_path = Path(source).resolve() + target_path = Path(target).resolve() + if not source_path.exists(): + raise FileNotFoundError(f"源数据库不存在: {source_path}") + if source_path == target_path: + raise ValueError("目标数据库不能与源数据库相同") + if target_path.exists() and not force: + raise FileExistsError(f"目标已存在，使用 --force 覆盖: {target_path}") + + target_path.parent.mkdir(parents=True, exist_ok=True) + for suffix in ("", "-wal", "-shm"): + candidate = Path(str(target_path) + suffix) + if candidate.exists(): + candidate.unlink() + + _copy_database(source_path, target_path) + + conn = sqlite3.connect(target_path, timeout=10.0) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA foreign_keys = ON") + try: + latest_row = conn.execute( + "SELECT MAX(stat_date) AS d FROM snapshot WHERE ok = 1" + ).fetchone() + if latest_row is None or not latest_row["d"]: + raise RuntimeError("源数据库没有可用快照，无法确定回填日期") + latest_day = parse_day(str(latest_row["d"])) + start_day = latest_day - timedelta(days=days) + rng = random.Random(seed) + anomaly_rng = random.Random(seed ^ 0x5EED) + + videos = _video_rows(conn) + latest = _latest_snapshots(conn) + selected = [video for video in videos if int(video["aid"]) in latest] + + if profile == "live": + first_seen_map = spread_first_seen( + [int(video["aid"]) for video in selected], + start_day, + latest_day, + rng, + ) + else: + first_seen_map = { + int(video["aid"]): float(video["first_seen"] or 0) + for video in selected + } + + rows: list[tuple[str, int, dict[str, int], float]] = [] + inserted_first_seen = 0 + for video in selected: + aid = int(video["aid"]) + first_seen_ts = first_seen_map[aid] + if profile == "live": + conn.execute( + "UPDATE video SET first_seen = ? WHERE aid = ?", + (int(first_seen_ts), aid), + ) + inserted_first_seen += 1 + snapshot_start = max(start_day, shanghai_datetime(first_seen_ts).date()) + else: + snapshot_start = latest_day + + synthetic_days = max(0, (latest_day - snapshot_start).days) + if synthetic_days <= 0: + continue + age_days = max( + 0.0, + (latest_day - shanghai_datetime(int(video["pubdate"] or 0)).date()).days, + ) + recency = 1.0 + min(age_days, 365.0) / 365.0 * 0.8 + multiplier = rng.choices( + [0.25, 0.8, 1.5, 4.0, 9.0], + weights=[3.0, 5.0, 2.0, 1.0, 0.25], + k=1, + )[0] + multiplier *= recency + series = backfill_metrics( + latest[aid], synthetic_days, rng, multiplier=multiplier + ) + for index, metrics in enumerate(series): + stat_day = snapshot_start + timedelta(days=index) + if stat_day >= latest_day: + continue + if with_anomalies and anomaly_rng.random() < MISSING_RATE: + continue + captured = _timestamp(stat_day, hour=0, minute=1) + anomaly_rng.randint(0, 240) + rows.append( + (stat_day.isoformat(), aid, metrics, captured) + ) + + failed_indices: set[int] = set() + failed = 0 + if with_anomalies and len(rows) > FAILED_ROWS: + for index in anomaly_rng.sample(range(len(rows)), FAILED_ROWS): + failed_indices.add(index) + failed += 1 + + for index, (stat_date, aid, metrics, captured) in enumerate(rows): + if index in failed_indices: + _insert_snapshot( + conn, + stat_date, + aid, + None, + captured_at=captured, + ok=False, + err="preview: simulated failure", + ) + else: + _insert_snapshot( + conn, + stat_date, + aid, + metrics, + captured_at=captured, + ok=True, + err=None, + ) + + if profile == "live": + up_rows = _seed_up_stats(conn, latest_day, days, anomaly_rng) + _seed_candidates(conn, latest_day) + _seed_blacklist(conn, _timestamp(latest_day, hour=14)) + else: + up_rows = 0 + conn.execute("DELETE FROM candidate WHERE channel = 'preview'") + conn.execute("DELETE FROM up_blacklist WHERE reason = 'preview'") + + seeded_at = _timestamp(latest_day, hour=15) + for key, value in { + "preview:seeded": "1", + "preview:source": str(source_path), + "preview:target": str(target_path), + "preview:latest_date": latest_day.isoformat(), + "preview:date_from": start_day.isoformat(), + "preview:days": str(days), + "preview:seed": str(seed), + "preview:profile": profile, + "preview:anomalies": "1" if with_anomalies else "0", + "preview:generated_at": str(int(seeded_at)), + }.items(): + conn.execute( + "INSERT OR REPLACE INTO kv(k, v) VALUES (?, ?)", (key, value) + ) + conn.commit() + finally: + conn.close() + + source_cfg = _read_source_config(Path(config_source) if config_source else None) + + config_target = ( + Path(config_out).resolve() + if config_out is not None + else target_path.parent / "config.json" + ) + config_target.parent.mkdir(parents=True, exist_ok=True) + + folder_map: dict[int, dict[str, Any]] = {} + if source_cfg and isinstance(source_cfg.get("folders"), list): + for item in source_cfg["folders"]: + if isinstance(item, Mapping) and "media_id" in item and "year" in item: + folder_map[int(item["media_id"])] = { + "year": int(item["year"]), + "media_id": int(item["media_id"]), + "title": str(item.get("title") or ""), + } + if not folder_map: + conn = sqlite3.connect(target_path) + conn.row_factory = sqlite3.Row + try: + for row in conn.execute("SELECT media_id, year, title FROM folder ORDER BY year"): + folder_map[int(row["media_id"])] = { + "year": int(row["year"]), + "media_id": int(row["media_id"]), + "title": str(row["title"] or ""), + } + finally: + conn.close() + + if source_cfg and isinstance(source_cfg.get("track_fans"), list): + track_fans = [int(mid) for mid in source_cfg["track_fans"]] + else: + conn = sqlite3.connect(target_path) + try: + track_fans = [ + int(row[0]) for row in conn.execute("SELECT DISTINCT mid FROM up_stat ORDER BY mid") + ] + finally: + conn.close() + + if source_cfg and isinstance(source_cfg.get("up_blacklist_seed"), list): + blacklist = [int(mid) for mid in source_cfg["up_blacklist_seed"]] + else: + conn = sqlite3.connect(target_path) + try: + blacklist = [int(row[0]) for row in conn.execute("SELECT mid FROM up_blacklist ORDER BY mid")] + finally: + conn.close() + + config = build_preview_config( + source_cfg, + target=target_path, + config_out=config_target, + folders=list(folder_map.values()), + track_fans=track_fans, + blacklist=blacklist, + ) + config_target.write_text( + json.dumps(config, ensure_ascii=False, indent=2) + "\n", encoding="utf-8" + ) + + return { + "ok": True, + "source": str(source_path), + "target": str(target_path), + "config": str(config_target), + "profile": profile, + "seed": seed, + "days": days, + "date_from": start_day.isoformat(), + "date_to": latest_day.isoformat(), + "videos": len(selected), + "first_seen_updated": inserted_first_seen, + "snapshots_inserted": len(rows) - failed, + "snapshots_failed": failed, + "snapshots_missing": (len(selected) * days) - len(rows), + "up_stat_rows": up_rows, + "preview_marker": "preview:seeded=1", + } + + +def main(argv: Sequence[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="生成只读预览数据库（真实库不会被修改）") + parser.add_argument("--source", default=str(DEFAULT_SOURCE), help="源数据库") + parser.add_argument("--target", default=str(DEFAULT_TARGET), help="目标预览数据库") + parser.add_argument("--days", type=int, default=7, help="向前回填天数（默认 7）") + parser.add_argument("--seed", type=int, default=20261031, help="随机种子") + parser.add_argument( + "--profile", choices=("live", "first-day"), default="live", + help="live：打散入库日并回填；first-day：保持首次导入状态", + ) + parser.add_argument("--force", action="store_true", help="覆盖已存在的目标库") + parser.add_argument( + "--config-source", default=str(DEFAULT_CONFIG_SOURCE), + help="读取 folders/track_fans/heat_weights 的源配置（不会复制凭据）", + ) + parser.add_argument("--config-out", default=None, help="生成的预览配置路径") + parser.add_argument( + "--no-anomalies", action="store_true", help="不生成缺失日和失败快照" + ) + args = parser.parse_args(list(argv) if argv is not None else None) + try: + summary = seed_preview( + args.source, + args.target, + days=args.days, + seed=args.seed, + profile=args.profile, + force=args.force, + config_source=args.config_source, + config_out=args.config_out, + with_anomalies=not args.no_anomalies, + ) + except (ValueError, FileNotFoundError, FileExistsError, RuntimeError, sqlite3.Error) as exc: + print(json.dumps({"ok": False, "error": str(exc)}, ensure_ascii=False)) + return 2 + print(json.dumps(summary, ensure_ascii=False, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())