277 lines
9.6 KiB
Python
277 lines
9.6 KiB
Python
# -*- 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,
|
||
allow_lossy_replace: bool = False,
|
||
dry_run: bool = False,
|
||
) -> dict[str, Any]:
|
||
"""把解析后的序列写入 ``up_stat``。"""
|
||
points = list(series.get("points", []))
|
||
repo = Repository(db)
|
||
deleted = 0
|
||
last_csv = str(series.get("last_date") or "")
|
||
if replace and last_csv:
|
||
newer = db.read(lambda conn: int(conn.execute(
|
||
"SELECT COUNT(*) FROM up_stat WHERE mid = ? AND stat_date > ?",
|
||
(int(mid), last_csv),
|
||
).fetchone()[0]))
|
||
if newer and not allow_lossy_replace:
|
||
raise ValueError(
|
||
f"数据库中存在 {newer} 条晚于 CSV 末尾 {last_csv} 的粉丝记录；"
|
||
"确认要用 CSV 覆盖并删除这些记录时请加 --allow-lossy-replace"
|
||
)
|
||
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(
|
||
"--allow-lossy-replace", action="store_true",
|
||
help="配合 --replace：允许删除数据库中晚于 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,
|
||
allow_lossy_replace=args.allow_lossy_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()) |