From c4c9c0b0382f3c6172a15f1cf68085758ab1a7be Mon Sep 17 00:00:00 2001 From: kaireewu Date: Sun, 1 Nov 2026 02:23:09 +0800 Subject: [PATCH] feat(tools): replace xxm fan import with bilibili csv --- tests/test_fans_csv.py | 132 ++++++++++++++++ tests/test_fans_history.py | 174 --------------------- tools/import_fans_csv.py | 260 ++++++++++++++++++++++++++++++ tools/import_fans_history.py | 295 ----------------------------------- 4 files changed, 392 insertions(+), 469 deletions(-) create mode 100644 tests/test_fans_csv.py delete mode 100644 tests/test_fans_history.py create mode 100644 tools/import_fans_csv.py delete mode 100644 tools/import_fans_history.py diff --git a/tests/test_fans_csv.py b/tests/test_fans_csv.py new file mode 100644 index 0000000..8dac3ae --- /dev/null +++ b/tests/test_fans_csv.py @@ -0,0 +1,132 @@ +# -*- coding: utf-8 -*- +"""粉丝历史 CSV 回填测试（离线）: uv run python tests/test_fans_csv.py""" + +import os +import sys +import tempfile +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_csv import ( # noqa: E402 + import_fans_series, + parse_fans_csv, +) + +CSV_TEXT = """\ufeff时间,粉丝,7日变化,1日变化 +2026-01-03 10:00,1003,0,1 +2026-01-01 09:00,1001,0,0 +2026-01-02 08:30,1002,0,1 +2026-01-02 23:30,1005,0,4 +2026-01-04,1007,0,1 +""" + + +def test_parse_csv(): + series = parse_fans_csv(CSV_TEXT) + assert series["first_date"] == "2026-01-01" + assert series["last_date"] == "2026-01-04" + assert [point["date"] for point in series["points"]] == [ + "2026-01-01", "2026-01-02", "2026-01-03", "2026-01-04", + ] + # 同一天两行，保留时间较晚的一条 + assert [point["follower"] for point in series["points"]] == [1001, 1005, 1003, 1007] + assert series["unique_days"] == 4 + assert series["total_rows"] == 5 + assert series["gaps"] == 0 + assert any("多行" in warning for warning in series["warnings"]) + + for bad in ("", "时间,粉丝\n2026-01-01\n", "date,follower\nx,1\n", "时间,粉丝\n"): + try: + parse_fans_csv(bad) + except ValueError: + pass + else: + raise AssertionError(f"非法 CSV 应报错: {bad!r}") + + +def test_parse_rejects_bad_values(): + for bad in ( + "时间,粉丝\n2026-01-01,nope\n", + "时间,粉丝\nnot-a-date,100\n", + "时间,粉丝\n2026-01-01,-1\n", + ): + try: + parse_fans_csv(bad) + except ValueError: + pass + else: + raise AssertionError(f"非法行应报错: {bad!r}") + + +def test_import_default_force_and_replace(): + series = parse_fans_csv(CSV_TEXT) + with tempfile.TemporaryDirectory() as tmp: + db = Database(Path(tmp) / "app.db") + db.initialize() + try: + repo = Repository(db) + repo.upsert_up_stat("2026-01-01", 37754047, follower=1, following=9, ok=True, captured_at=1) + repo.upsert_up_stat("2026-01-02", 37754047, follower=None, following=None, ok=False, captured_at=1) + repo.upsert_up_stat("2026-01-03", 99999, follower=1, following=9, ok=True, captured_at=1) + + dry = import_fans_series(db, series, dry_run=True) + assert dry["inserted"] == 2 + assert dry["updated"] == 1 + assert dry["skipped"] == 1 + assert repo.get_up_stat("2026-01-02", 37754047)["ok"] == 0 + + summary = import_fans_series(db, series) + assert summary["inserted"] == 2 + assert summary["updated"] == 1 + assert summary["skipped"] == 1 + assert repo.get_up_stat("2026-01-01", 37754047)["follower"] == 1 + assert repo.get_up_stat("2026-01-02", 37754047)["follower"] == 1005 + assert repo.get_up_stat("2026-01-02", 37754047)["ok"] == 1 + assert repo.get_up_stat("2026-01-04", 37754047)["follower"] == 1007 + # 另一个 mid 不受影响 + assert repo.get_up_stat("2026-01-03", 99999)["follower"] == 1 + + forced = import_fans_series(db, series, force=True) + assert forced["updated"] == 4 + assert repo.get_up_stat("2026-01-01", 37754047)["follower"] == 1001 + + replaced = import_fans_series(db, series, replace=True) + assert replaced["deleted"] == 4 + assert replaced["inserted"] == 4 + assert replaced["updated"] == 0 + assert repo.get_up_stat("2026-01-01", 37754047)["follower"] == 1001 + assert repo.get_up_stat("2026-01-03", 99999)["follower"] == 1 + finally: + db.close() + + +def test_replace_dry_run_counts_only(): + series = parse_fans_csv(CSV_TEXT) + with tempfile.TemporaryDirectory() as tmp: + db = Database(Path(tmp) / "app.db") + db.initialize() + try: + repo = Repository(db) + repo.upsert_up_stat("2026-01-01", 37754047, follower=1, following=None, ok=True, captured_at=1) + result = import_fans_series(db, series, replace=True, dry_run=True) + assert result["deleted"] == 1 + assert result["inserted"] == 4 + assert repo.get_up_stat("2026-01-01", 37754047)["follower"] == 1 + finally: + db.close() + + +def main_test(): + test_parse_csv() + test_parse_rejects_bad_values() + test_import_default_force_and_replace() + test_replace_dry_run_counts_only() + print("OK: 粉丝 CSV 解析/导入/替换/幂等测试全部通过") + + +if __name__ == "__main__": + main_test() \ No newline at end of file diff --git a/tests/test_fans_history.py b/tests/test_fans_history.py deleted file mode 100644 index 028aa36..0000000 --- a/tests/test_fans_history.py +++ /dev/null @@ -1,174 +0,0 @@ -# -*- 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_csv.py b/tools/import_fans_csv.py new file mode 100644 index 0000000..28727d4 --- /dev/null +++ b/tools/import_fans_csv.py @@ -0,0 +1,260 @@ +# -*- coding: utf-8 -*- +"""从 B 站粉丝历史 CSV 回填 ``up_stat``。 + +CSV 形如:: + + 时间,粉丝,7日变化,1日变化 + 2026-10-31 22:42,2767269,139,-33 + 2026-10-29 23:53,2767334,496,15 + +脚本只写 ``up_stat``，不会修改视频快照、收藏夹或其它表。默认跳过已有的 +``ok=1`` 记录；``--force`` 覆盖同日期记录；``--replace`` 先清空该 mid 的 +全部粉丝历史再导入，用于一次性用完整 CSV 替换旧数据。 + +用法:: + + uv run python tools/import_fans_csv.py --csv E:\\Downloads\\history.csv --dry-run + uv run python tools/import_fans_csv.py --csv E:\\Downloads\\history.csv --replace +""" + +from __future__ import annotations + +import argparse +import csv +import io +import json +import sys +from datetime import date, datetime, time +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_MID = 37754047 +TIME_KEYS = ("时间", "日期", "date", "time", "datetime") +FOLLOWER_KEYS = ("粉丝", "粉丝数", "follower", "followers") + + +def _parse_day(value: str) -> date: + text = str(value).strip().replace("/", "-") + for fmt in ("%Y-%m-%d %H:%M", "%Y-%m-%d %H:%M:%S", "%Y-%m-%d"): + try: + return datetime.strptime(text, fmt).date() + except ValueError: + continue + try: + return date.fromisoformat(text) + except ValueError as exc: + raise ValueError(f"时间格式不正确: {value!r}") from exc + + +def _pick(row: Mapping[str, str], keys: Sequence[str], label: str) -> str: + for key in keys: + if key in row and row[key] not in (None, ""): + return str(row[key]) + raise ValueError(f"CSV 缺少{label}列（支持列名: {', '.join(keys)}）") + + +def parse_fans_csv(text: str) -> dict[str, Any]: + """解析粉丝 CSV，返回按日期升序、每日最多一条的序列。 + + 同一天出现多行时保留时间最晚的一条，并在 ``warnings`` 中记录。 + """ + if not isinstance(text, str) or not text.strip(): + raise ValueError("CSV 内容为空") + text = text.lstrip("\ufeff") + reader = csv.DictReader(io.StringIO(text)) + if reader.fieldnames is None: + raise ValueError("CSV 缺少表头") + fieldnames = [str(name).strip() for name in reader.fieldnames] + reader.fieldnames = fieldnames + + by_date: dict[date, tuple[datetime | None, int]] = {} + warnings: list[str] = [] + total_rows = 0 + for line_number, row in enumerate(reader, start=2): + if not any(value not in (None, "") for value in row.values()): + continue + total_rows += 1 + try: + raw_time = _pick(row, TIME_KEYS, "时间") + raw_follower = _pick(row, FOLLOWER_KEYS, "粉丝") + except ValueError: + raise + try: + follower = int(str(raw_follower).replace(",", "").strip()) + except ValueError as exc: + raise ValueError( + f"第 {line_number} 行粉丝不是整数: {raw_follower!r}" + ) from exc + if follower < 0: + raise ValueError(f"第 {line_number} 行粉丝不能为负: {follower}") + day = _parse_day(raw_time) + try: + moment = datetime.strptime(str(raw_time).strip(), "%Y-%m-%d %H:%M") + except ValueError: + moment = None + existing = by_date.get(day) + if existing is not None: + previous_moment, previous_value = existing + keep_new = moment is not None and ( + previous_moment is None or moment > previous_moment + ) + warnings.append( + f"{day.isoformat()} 出现多行，保留" + f"{'较晚' if keep_new else '原'}记录" + ) + if keep_new: + by_date[day] = (moment, follower) + continue + by_date[day] = (moment, follower) + + if not by_date: + raise ValueError("CSV 没有可导入的数据行") + points = [ + {"date": day.isoformat(), "follower": by_date[day][1]} + for day in sorted(by_date) + ] + return { + "points": points, + "total_rows": total_rows, + "unique_days": len(points), + "first_date": points[0]["date"], + "last_date": points[-1]["date"], + "warnings": warnings, + "gaps": _count_gaps([date.fromisoformat(p["date"]) for p in points]), + } + + +def _count_gaps(days: Sequence[date]) -> int: + return sum( + 1 for previous, current in zip(days, days[1:]) + if (current - previous).days > 1 + ) + + +def _day_timestamp(day: date) -> int: + return int(datetime.combine(day, time(0, 1), tzinfo=SHANGHAI_TZ).timestamp()) + + +def import_fans_series( + db: Database, + series: Mapping[str, Any], + *, + mid: int = DEFAULT_MID, + force: bool = False, + replace: bool = False, + dry_run: bool = False, +) -> dict[str, Any]: + """把解析后的序列写入 ``up_stat``。""" + points = list(series.get("points", [])) + repo = Repository(db) + deleted = 0 + if replace: + if dry_run: + deleted = db.read(lambda conn: int(conn.execute( + "SELECT COUNT(*) FROM up_stat WHERE mid = ?", (int(mid),) + ).fetchone()[0])) + else: + deleted = db.write(lambda conn: max(conn.execute( + "DELETE FROM up_stat WHERE mid = ?", (int(mid),) + ).rowcount, 0)) + + inserted = 0 + updated = 0 + skipped = 0 + for point in points: + day = str(point["date"]) + follower = int(point["follower"]) + existing = None if (dry_run and replace) else repo.get_up_stat(day, mid) + if existing is not None and int(existing.get("ok") or 0) == 1 and not (force or replace): + 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), + "replace": bool(replace), + "mid": int(mid), + "deleted": deleted, + "inserted": inserted, + "updated": updated, + "skipped": skipped, + "total": len(points), + "first_date": series.get("first_date"), + "last_date": series.get("last_date"), + "gaps": series.get("gaps", 0), + "warnings": list(series.get("warnings", [])), + } + + +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="从 B 站粉丝历史 CSV 回填 up_stat") + parser.add_argument("--csv", required=True, help="粉丝历史 CSV 路径") + parser.add_argument("--config", default=str(ROOT / "config.json"), help="配置文件") + parser.add_argument("--db", default=None, help="直接指定数据库（覆盖配置）") + parser.add_argument("--mid", type=int, default=DEFAULT_MID, help="写入的 B 站 mid") + parser.add_argument("--force", action="store_true", help="覆盖同日期已有 ok=1 记录") + parser.add_argument( + "--replace", action="store_true", + help="先清空该 mid 的全部粉丝历史，再用 CSV 重建", + ) + parser.add_argument("--dry-run", action="store_true", help="只计算，不写数据库") + args = parser.parse_args(list(argv) if argv is not None else None) + + try: + text = Path(args.csv).read_text(encoding="utf-8-sig") + series = parse_fans_csv(text) + db_path = _database_path(args.config, args.db) + db = Database(db_path) + db.initialize() + try: + summary = import_fans_series( + db, + series, + mid=args.mid, + force=args.force, + replace=args.replace, + dry_run=args.dry_run, + ) + finally: + db.close() + summary.update({"csv": args.csv, "database": str(db_path)}) + summary["total_rows"] = series["total_rows"] + summary["unique_days"] = series["unique_days"] + 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()) \ No newline at end of file diff --git a/tools/import_fans_history.py b/tools/import_fans_history.py deleted file mode 100644 index 901ed28..0000000 --- a/tools/import_fans_history.py +++ /dev/null @@ -1,295 +0,0 @@ -# -*- 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())