Files
re-xxm/tools/seed_preview_db.py
T

721 lines
25 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- 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_task_records(
conn: sqlite3.Connection, latest_day: date, video_count: int
) -> None:
"""写入可读的最近任务状态,让预览系统页有内容可看。"""
base = _timestamp(latest_day, hour=0, minute=1)
records = {
"snapshot": {
"name": "snapshot", "ok": True,
"started_at": int(base),
"finished_at": int(base + 1124),
"error": None,
"summary": {
"stat_date": latest_day.isoformat(), "total": video_count,
"ok": video_count, "failed": 0, "removed": 0, "completed": True,
"backup_path": f"data/backup/app-{latest_day.strftime('%Y%m%d')}.db",
},
},
"folder_sync": {
"name": "folder_sync", "ok": True,
"started_at": int(base + 20 * 3600),
"finished_at": int(base + 20 * 3600 + 8),
"error": None,
"summary": {"folders": 2, "added": 65, "removed": 0, "ok": True},
},
"discovery": {
"name": "discovery", "ok": True,
"started_at": int(base + 21 * 3600),
"finished_at": int(base + 21 * 3600 + 24),
"error": None,
"summary": {
"searches": 4, "items": 168, "candidates": 4,
"notified": 3, "errors": [],
},
},
"candidate_cleanup": {
"name": "candidate_cleanup", "ok": True,
"started_at": int(base + 22 * 3600),
"finished_at": int(base + 22 * 3600 + 1),
"error": None,
"summary": {"candidates_removed": 0, "fav_actions_expired": 0},
},
}
for name, record in records.items():
conn.execute(
"INSERT OR REPLACE INTO kv(k, v) VALUES (?, ?)",
(f"task:last:{name}", json.dumps(record, ensure_ascii=False)),
)
conn.execute(
"INSERT OR REPLACE INTO kv(k, v) VALUES ('snapshot:last_attempt', ?)",
(latest_day.isoformat(),),
)
conn.execute(
"INSERT OR REPLACE INTO kv(k, v) VALUES ('snapshot:last_completed', ?)",
(latest_day.isoformat(),),
)
conn.execute(
"INSERT OR REPLACE INTO kv(k, v) VALUES (?, '1')",
(f"snapshot:backup:{latest_day.isoformat()}",),
)
conn.execute(
"INSERT OR REPLACE INTO kv(k, v) VALUES (?, '')",
(f"snapshot:backup_error:{latest_day.isoformat()}",),
)
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
# values[0] 对应 latest-1,values[-1] 对应 latest-days。
values: list[int] = []
for _ in range(days):
decrement = max(1, int(round(current * rng.uniform(0.00015, 0.0006))))
current = max(0, current - decrement)
values.append(current)
for offset in range(days, 0, -1):
stat_date = (latest_day - timedelta(days=offset)).isoformat()
conn.execute(
"""
INSERT OR REPLACE INTO up_stat(
stat_date, mid, follower, following, captured_at, ok
) VALUES (?, ?, ?, ?, ?, 1)
""",
(
stat_date,
int(mid),
values[offset - 1],
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
expected_rows = 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
expected_rows += synthetic_days
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))
_seed_task_records(conn, latest_day, len(selected))
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": expected_rows - 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 PermissionError as exc:
print(json.dumps({
"ok": False,
"error": f"目标数据库被占用,请先停止预览服务: {exc}",
}, ensure_ascii=False))
return 2
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())