# -*- 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())