feat(tools): import 90-day fan history from public data source

tests/test_fans_history.py

@@ -0,0 +1,174 @@
+# -*- coding: utf-8 -*-
+"""粉丝历史回填测试（离线）: uv run python tests/test_fans_history.py"""
+
+import json
+import os
+import sys
+import tempfile
+from datetime import date, timedelta
+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.database import Database  # noqa: E402
+from reaction_board.repository import Repository  # noqa: E402
+from tools.import_fans_history import (  # noqa: E402
+    build_follower_series,
+    import_follower_series,
+    parse_data_js,
+)
+
+
+def make_payload(*, days=10, gap_day=None, mismatch=False):
+    start = date(2026, 1, 1)
+    dates = [start + timedelta(days=index) for index in range(days)]
+    anchor = dates[-1]
+    # 从 900 起每天 +10，最后一天为 900 + (days-1)*10
+    values = {day: 900 + index * 10 for index, day in enumerate(dates)}
+    corr = []
+    for index, day in enumerate(dates):
+        delta = 0 if index == 0 else values[day] - values[dates[index - 1]]
+        corr.append([day.isoformat(), 0, delta])
+    if gap_day is not None:
+        corr = [row for row in corr if row[0] != gap_day]
+    week = [
+        [day.strftime("%Y/%m/%d"), values[day], 0]
+        for day in dates[-3:]
+    ]
+    if mismatch:
+        week[-1][1] += 7
+    payload = {
+        "meta": {"updated": anchor.isoformat()},
+        "followers": {
+            "咻咻满": {
+                "total": values[anchor],
+                "week": week,
+            }
+        },
+        "corr": corr,
+    }
+    return payload, values
+
+
+def test_parse_data_js():
+    payload, _ = make_payload()
+    text = "/* demo */\nwindow.XXM = " + json.dumps(payload, ensure_ascii=False) + ";\n"
+    parsed = parse_data_js(text)
+    assert parsed["meta"]["updated"] == "2026-01-10"
+
+    for bad in ("", "<html></html>", "window.XXM = {}"):
+        try:
+            parse_data_js(bad)
+        except ValueError:
+            pass
+        else:
+            raise AssertionError(f"非法输入应报错: {bad!r}")
+
+
+def test_build_follower_series():
+    payload, values = make_payload(days=10)
+    series = build_follower_series(payload, days=10)
+    assert series["anchor_follower"] == values[date(2026, 1, 10)]
+    assert series["first_date"] == "2026-01-01"
+    assert series["last_date"] == "2026-01-10"
+    assert series["points"][0] == {"date": "2026-01-01", "follower": 900}
+    assert series["points"][-1] == {"date": "2026-01-10", "follower": 990}
+    assert series["warnings"] == []
+    assert series["stopped_at_gap"] is False
+
+    limited = build_follower_series(payload, days=3)
+    assert [point["date"] for point in limited["points"]] == [
+        "2026-01-08", "2026-01-09", "2026-01-10",
+    ]
+
+    try:
+        build_follower_series(payload, days=0)
+    except ValueError:
+        pass
+    else:
+        raise AssertionError("days=0 应报错")
+
+
+def test_gap_and_mismatch():
+    payload, _ = make_payload(days=10, gap_day="2026-01-05")
+    series = build_follower_series(payload, days=10)
+    assert series["stopped_at_gap"] is True
+    assert series["first_date"] == "2026-01-06"
+    assert series["last_date"] == "2026-01-10"
+    assert any("不连续" in warning for warning in series["warnings"])
+
+    payload, _ = make_payload(days=10, mismatch=True)
+    try:
+        build_follower_series(payload, days=10)
+    except ValueError as exc:
+        assert "不一致" in str(exc)
+    else:
+        raise AssertionError("week 与 corr 不一致时应报错")
+
+
+def test_import_and_force():
+    payload, values = make_payload(days=10)
+    series = build_follower_series(payload, days=10)
+    with tempfile.TemporaryDirectory() as tmp:
+        db = Database(Path(tmp) / "app.db")
+        db.initialize()
+        try:
+            repo = Repository(db)
+            repo.upsert_up_stat(
+                "2026-01-08", 37754047, follower=1, following=5,
+                ok=True, captured_at=1,
+            )
+            repo.upsert_up_stat(
+                "2026-01-09", 37754047, follower=None, following=None,
+                ok=False, captured_at=1,
+            )
+            repo.upsert_up_stat(
+                "2026-01-10", 37754047, follower=2, following=None,
+                ok=True, captured_at=1,
+            )
+
+            dry = import_follower_series(
+                db, series, mid=37754047, dry_run=True
+            )
+            assert dry["inserted"] == 7
+            assert dry["updated"] == 1
+            assert dry["skipped"] == 2
+            assert repo.get_up_stat("2026-01-09", 37754047)["ok"] == 0
+
+            summary = import_follower_series(db, series, mid=37754047)
+            assert summary["inserted"] == 7
+            assert summary["updated"] == 1
+            assert summary["skipped"] == 2
+            assert repo.get_up_stat("2026-01-01", 37754047)["follower"] == 900
+            row = repo.get_up_stat("2026-01-09", 37754047)
+            assert row["follower"] == values[date(2026, 1, 9)]
+            assert row["ok"] == 1
+            assert row["following"] is None
+            preserved = repo.get_up_stat("2026-01-08", 37754047)
+            assert preserved["follower"] == 1
+            assert preserved["following"] == 5
+
+            again = import_follower_series(db, series, mid=37754047)
+            assert again["inserted"] == 0
+            assert again["skipped"] == 10
+
+            forced = import_follower_series(
+                db, series, mid=37754047, force=True
+            )
+            assert forced["updated"] == 10
+            assert repo.get_up_stat("2026-01-08", 37754047)["follower"] == values[date(2026, 1, 8)]
+        finally:
+            db.close()
+
+
+def main_test():
+    test_parse_data_js()
+    test_build_follower_series()
+    test_gap_and_mismatch()
+    test_import_and_force()
+    print("OK: 粉丝历史解析/回填/缺口/幂等测试全部通过")
+
+
+if __name__ == "__main__":
+    main_test()

tools/import_fans_history.py

@@ -0,0 +1,295 @@
+# -*- coding: utf-8 -*-
+"""从小满虫之家公开数据回填粉丝历史。
+
+数据源是站点 ``assets/js/data.js`` 中的 ``window.XXM`` 快照：``followers``
+提供锚点，``corr`` 提供每日净增。脚本只写入 ``up_stat``，不会修改视频快照、
+收藏夹或其它表。
+
+用法::
+
+    uv run python tools/import_fans_history.py --days 90 --dry-run
+    uv run python tools/import_fans_history.py --days 90
+
+如果站点暂时不可达，可先保存 data.js 再离线导入::
+
+    uv run python tools/import_fans_history.py --from-file data.js --days 90
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+import urllib.request
+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.database import Database  # noqa: E402
+from reaction_board.repository import Repository  # noqa: E402
+from reaction_board.timeutil import SHANGHAI_TZ  # noqa: E402
+
+DEFAULT_URL = "https://www.xxm0319.com/assets/js/data.js"
+DEFAULT_ACCOUNT = "咻咻满"
+DEFAULT_MID = 37754047
+USER_AGENT = (
+    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
+    "(KHTML, like Gecko) Chrome/155.0.0.0 Safari/537.36"
+)
+
+
+def parse_data_js(text: str) -> dict[str, Any]:
+    """从 ``window.XXM = {...};`` 中提取 JSON 对象。"""
+    if not isinstance(text, str) or not text.strip():
+        raise ValueError("data.js 内容为空")
+    start = text.find("window.XXM")
+    if start < 0:
+        raise ValueError("data.js 中找不到 window.XXM")
+    equals = text.find("=", start)
+    if equals < 0:
+        raise ValueError("data.js 中 window.XXM 缺少赋值")
+    end = text.rfind(";")
+    if end <= equals:
+        raise ValueError("data.js 中 window.XXM 缺少结尾分号")
+    raw = text[equals + 1:end].strip()
+    try:
+        payload = json.loads(raw)
+    except json.JSONDecodeError as exc:
+        raise ValueError(f"window.XXM 不是合法 JSON: {exc}") from exc
+    if not isinstance(payload, dict):
+        raise ValueError("window.XXM 顶层必须是对象")
+    return payload
+
+
+def _parse_date(value: str) -> date:
+    text = str(value).strip().replace("/", "-")
+    try:
+        return date.fromisoformat(text)
+    except ValueError as exc:
+        raise ValueError(f"日期格式不正确: {value!r}") from exc
+
+
+def _day_timestamp(day: date) -> int:
+    return int(datetime.combine(day, time(0, 1), tzinfo=SHANGHAI_TZ).timestamp())
+
+
+def build_follower_series(
+    payload: Mapping[str, Any],
+    *,
+    account: str = DEFAULT_ACCOUNT,
+    days: int = 90,
+) -> dict[str, Any]:
+    """根据锚点和每日净增构建连续粉丝序列。
+
+    返回的 ``points`` 按日期升序，包含 ``date`` 和 ``follower``。若数据出现
+    日期缺口，只回填锚点之前连续可计算的部分，不插值。
+    """
+    if days < 1:
+        raise ValueError("days 必须至少为 1")
+    followers = payload.get("followers")
+    if not isinstance(followers, Mapping):
+        raise ValueError("data.js 缺少 followers")
+    account_data = followers.get(account)
+    if not isinstance(account_data, Mapping):
+        raise ValueError(f"data.js 缺少账号: {account}")
+    week = account_data.get("week")
+    if not isinstance(week, list) or not week:
+        raise ValueError(f"{account} 缺少 week 锚点")
+    last = week[-1]
+    if not isinstance(last, list) or len(last) < 2:
+        raise ValueError("week 锚点结构不正确")
+    anchor_day = _parse_date(last[0])
+    anchor_value = int(last[1])
+    total = account_data.get("total")
+    warnings: list[str] = []
+    if isinstance(total, int) and total != anchor_value:
+        warnings.append(f"week 锚点 {anchor_value} 与 total {total} 不一致，使用 week 锚点")
+
+    corr = payload.get("corr")
+    if not isinstance(corr, list) or not corr:
+        raise ValueError("data.js 缺少 corr 每日净增")
+    deltas: dict[date, int] = {}
+    for item in corr:
+        if not isinstance(item, list) or len(item) < 3:
+            raise ValueError("corr 条目必须包含日期、占位值和净增")
+        day = _parse_date(item[0])
+        try:
+            delta = int(item[2])
+        except (TypeError, ValueError) as exc:
+            raise ValueError(f"corr 净增不是整数: {item!r}") from exc
+        deltas[day] = delta
+
+    values: dict[date, int] = {anchor_day: anchor_value}
+    ordered = sorted(day for day in deltas if day <= anchor_day)
+    if anchor_day not in deltas:
+        ordered.append(anchor_day)
+        ordered = sorted(set(ordered))
+    stopped = False
+    for current, next_day in zip(reversed(ordered[:-1]), reversed(ordered[1:])):
+        if (next_day - current).days != 1:
+            warnings.append(
+                f"粉丝数据在 {current.isoformat()} 与 {next_day.isoformat()} 之间不连续"
+            )
+            stopped = True
+            break
+        delta = deltas.get(next_day)
+        if delta is None:
+            warnings.append(f"缺少 {next_day.isoformat()} 的净增")
+            stopped = True
+            break
+        value = values[next_day] - delta
+        if value < 0:
+            raise ValueError(
+                f"计算结果为负: {current.isoformat()} follower={value}"
+            )
+        values[current] = value
+
+    # 用 week 中的多个点校验每日净增口径，防止站点字段语义变化。
+    for item in week:
+        if not isinstance(item, list) or len(item) < 2:
+            continue
+        day = _parse_date(item[0])
+        if day in values and int(item[1]) != values[day]:
+            raise ValueError(
+                f"week 与 corr 计算结果不一致: {day.isoformat()} "
+                f"{item[1]} != {values[day]}"
+            )
+
+    points = [
+        {"date": day.isoformat(), "follower": values[day]}
+        for day in sorted(values)[-days:]
+    ]
+    if not points:
+        raise ValueError("没有可用的粉丝历史数据")
+    return {
+        "account": account,
+        "source_updated": str((payload.get("meta") or {}).get("updated") or anchor_day),
+        "anchor_date": anchor_day.isoformat(),
+        "anchor_follower": anchor_value,
+        "first_date": points[0]["date"],
+        "last_date": points[-1]["date"],
+        "points": points,
+        "warnings": warnings,
+        "stopped_at_gap": stopped,
+    }
+
+
+def import_follower_series(
+    db: Database,
+    series: Mapping[str, Any],
+    *,
+    mid: int = DEFAULT_MID,
+    force: bool = False,
+    dry_run: bool = False,
+) -> dict[str, Any]:
+    """把序列写入 ``up_stat``；默认跳过已有成功记录。"""
+    repo = Repository(db)
+    inserted = 0
+    updated = 0
+    skipped = 0
+    for point in series.get("points", []):
+        day = str(point["date"])
+        follower = int(point["follower"])
+        existing = repo.get_up_stat(day, mid)
+        if existing is not None and int(existing.get("ok") or 0) == 1 and not force:
+            skipped += 1
+            continue
+        if existing is None:
+            inserted += 1
+        else:
+            updated += 1
+        if dry_run:
+            continue
+        following = None if existing is None else existing.get("following")
+        repo.upsert_up_stat(
+            day,
+            mid,
+            follower=follower,
+            following=following,
+            ok=True,
+            captured_at=_day_timestamp(date.fromisoformat(day)),
+        )
+    return {
+        "ok": True,
+        "dry_run": bool(dry_run),
+        "mid": int(mid),
+        "inserted": inserted,
+        "updated": updated,
+        "skipped": skipped,
+        "total": len(series.get("points", [])),
+        "first_date": series.get("first_date"),
+        "last_date": series.get("last_date"),
+        "source_updated": series.get("source_updated"),
+        "warnings": list(series.get("warnings", [])),
+    }
+
+
+def fetch_data_js(url: str, *, timeout: float = 30.0) -> str:
+    request = urllib.request.Request(
+        url,
+        headers={
+            "User-Agent": USER_AGENT,
+            "Accept": "application/javascript,text/javascript,*/*",
+            "Referer": "https://www.xxm0319.com/data.html",
+        },
+    )
+    with urllib.request.urlopen(request, timeout=timeout) as response:
+        return response.read().decode("utf-8")
+
+
+def _database_path(config_path: str, override: str | None) -> Path:
+    if override:
+        return Path(override)
+    from reaction_board.config import ConfigStore
+
+    return ConfigStore(config_path, require_auth=False).get().storage.database_path
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="从小满虫之家回填粉丝历史到 up_stat")
+    parser.add_argument("--config", default=str(ROOT / "config.json"), help="配置文件")
+    parser.add_argument("--db", default=None, help="直接指定数据库（覆盖配置）")
+    parser.add_argument("--url", default=DEFAULT_URL, help="data.js 地址")
+    parser.add_argument("--from-file", default=None, help="离线 data.js 文件")
+    parser.add_argument("--account", default=DEFAULT_ACCOUNT, help="数据源账号名")
+    parser.add_argument("--mid", type=int, default=DEFAULT_MID, help="写入的 B 站 mid")
+    parser.add_argument("--days", type=int, default=90, help="保留最近天数")
+    parser.add_argument("--force", action="store_true", help="覆盖已有 ok=1 记录")
+    parser.add_argument("--dry-run", action="store_true", help="只计算，不写数据库")
+    args = parser.parse_args(list(argv) if argv is not None else None)
+
+    try:
+        if args.from_file:
+            text = Path(args.from_file).read_text(encoding="utf-8")
+        else:
+            text = fetch_data_js(args.url)
+        payload = parse_data_js(text)
+        series = build_follower_series(
+            payload, account=args.account, days=args.days
+        )
+        db_path = _database_path(args.config, args.db)
+        db = Database(db_path)
+        db.initialize()
+        try:
+            summary = import_follower_series(
+                db, series, mid=args.mid, force=args.force, dry_run=args.dry_run
+            )
+        finally:
+            db.close()
+        summary.update({
+            "database": str(db_path),
+            "source": args.from_file or args.url,
+        })
+        print(json.dumps(summary, ensure_ascii=False, indent=2))
+        return 0
+    except (ValueError, OSError, RuntimeError) as exc:
+        print(json.dumps({"ok": False, "error": str(exc)}, ensure_ascii=False))
+        return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tools/seed_preview_db.py

@@ -389,6 +389,12 @@ def _seed_up_stats(
 
     inserted = 0
     for mid, data in latest.items():
+        existing = conn.execute(
+            "SELECT COUNT(*) FROM up_stat WHERE mid = ? AND ok = 1", (int(mid),)
+        ).fetchone()[0]
+        if int(existing) >= 2:
+            # 已有真实历史时保留原样，不合成粉丝数据。
+            continue
         follower = int(data.get("follower") or 0)
         following = data.get("following")
         current = follower