diff --git a/tests/test_fans_history.py b/tests/test_fans_history.py new file mode 100644 index 0000000..028aa36 --- /dev/null +++ b/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 ("", "", "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() diff --git a/tools/import_fans_history.py b/tools/import_fans_history.py new file mode 100644 index 0000000..901ed28 --- /dev/null +++ b/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()) diff --git a/tools/seed_preview_db.py b/tools/seed_preview_db.py index 02dc764..b88738e 100644 --- a/tools/seed_preview_db.py +++ b/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