"""安全、幂等地将旧版 JSON/日志快照补录到统计数据库。""" from __future__ import annotations import argparse import asyncio import hashlib import json import re from collections import Counter from collections.abc import Mapping from datetime import UTC, datetime, timedelta, timezone from pathlib import Path from typing import Any try: from .stats_store import DEFAULT_DATABASE_PATH, StatsStore, business_date except ImportError: # 允许直接执行文件 from stats_store import DEFAULT_DATABASE_PATH, StatsStore, business_date BEIJING_TZ = timezone(timedelta(hours=8), name="Asia/Shanghai") _SOURCE_FILES = ( "users.json", "song_requests.json", "admin_audit.log", "queue_state.json", "music_state.json", "tts_state.json", ) _USER_FIELDS = { "uname", "points", "last_signin_date", "created_at", "blocked_all", "blocked_queue", "blocked_song_request", "note", } _SONG_FIELDS = { "id", "name", "artist", "duration_ms", "duration_sec", "keyword", "uid", "uname", "requested_at", "source", "remove_after_play", "started_at", "finished_at", "status", } _AUDIT_FIELDS = {"at", "action", "target", "client_ip", "session_id", "detail"} TABLE_COLUMNS = { "users": { "platform", "platform_user_id", "display_name", "avatar_url", "user_level", "is_admin", "first_seen_at_utc", "last_seen_at_utc", "snapshot_json", }, "song_requests": { "request_id", "platform", "platform_user_id", "requested_at_utc", "business_date", "song_id", "song_name", "artist", "source", "status", "points_cost", "queue_position", "payload_json", }, "playback_sessions": { "playback_id", "request_id", "song_id", "song_name", "started_at_utc", "ended_at_utc", "business_date", "status", "duration_ms", "played_ms", "stop_reason", "payload_json", }, "admin_audit_events": { "event_id", "occurred_at_utc", "business_date", "actor", "action", "target_type", "target_id", "success", "remote_address_hash", "payload_json", }, "point_transactions": { "transaction_id", "platform", "platform_user_id", "occurred_at_utc", "business_date", "amount", "balance_after", "reason", "reference_type", "reference_id", "payload_json", }, "events": { "event_id", "event_type", "category", "occurred_at_utc", "business_date", "payload_json", }, } _POINT_DETAIL_RE = re.compile(r"^delta=(-?\d+)\s+now=(-?\d+)$") _COUNT_DETAIL_RE = re.compile(r"^count=(\d+)$") _STATUS_MAP = { "played": "completed", "complete": "completed", "finished": "completed", "success": "completed", "skipped": "skipped", "skip": "skipped", "interrupted": "interrupted", "cancelled": "cancelled", "canceled": "cancelled", "play_error": "failed", "error": "failed", "failed": "failed", "queued": "queued", "pending": "queued", "playing": "playing", "active": "playing", } def _hash(*parts: Any) -> str: encoded = json.dumps(parts, ensure_ascii=False, separators=(",", ":"), default=str).encode("utf-8", "replace") return hashlib.sha256(encoded).hexdigest() def _text(value: Any, limit: int) -> str | None: if value is None: return None result = str(value).strip() return result[:limit] if result else None def _int(value: Any, *, minimum: int | None = None) -> int | None: try: result = int(value) except (TypeError, ValueError, OverflowError): return None if minimum is not None and result < minimum: return None return result def _bool(value: Any) -> bool: return value is True or value == 1 def _utc_iso( value: Any, *, naive_is_beijing: bool = True, fallback: datetime | None = None, timespec: str = "milliseconds", ) -> str: parsed: datetime if isinstance(value, (int, float)) and not isinstance(value, bool): parsed = datetime.fromtimestamp(float(value), UTC) elif isinstance(value, str) and value.strip(): raw = value.strip() try: parsed = datetime.fromisoformat(raw.replace("Z", "+00:00")) except ValueError: parsed = fallback or datetime.now(UTC) else: parsed = fallback or datetime.now(UTC) if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=BEIJING_TZ if naive_is_beijing else UTC) return parsed.astimezone(UTC).isoformat(timespec=timespec).replace("+00:00", "Z") def _file_time(path: Path) -> str: return _utc_iso(datetime.fromtimestamp(path.stat().st_mtime, UTC), naive_is_beijing=False) def _read_bytes(path: Path) -> bytes: return path.read_bytes() def _read_json(raw: bytes) -> Any: return json.loads(raw.decode("utf-8-sig")) def _fingerprint(raw: bytes) -> str: return hashlib.sha256(raw).hexdigest() def _normalize_status(value: Any, default: str) -> str: key = str(value or "").strip().casefold() return _STATUS_MAP.get(key, default) def _json_payload(value: Mapping[str, Any]) -> dict[str, Any]: return {key: item for key, item in value.items() if item is not None} def _user_records(path: Path, raw: bytes) -> list[tuple[str, dict[str, Any]]]: data = _read_json(raw) if not isinstance(data, Mapping): return [] snapshot_at = _file_time(path) records: list[tuple[str, dict[str, Any]]] = [] for uid, original in data.items(): if not isinstance(original, Mapping): continue item = {key: original[key] for key in _USER_FIELDS if key in original} platform_user_id = _text(uid, 64) if not platform_user_id: continue created_at = _utc_iso(item.get("created_at"), fallback=datetime.fromtimestamp(path.stat().st_mtime, UTC)) snapshot = _json_payload({ "points": _int(item.get("points")), "last_signin_date": _text(item.get("last_signin_date"), 10), "blocked_all": _bool(item.get("blocked_all")), "blocked_queue": _bool(item.get("blocked_queue")), "blocked_song_request": _bool(item.get("blocked_song_request")), "note_length": len(str(item.get("note") or "")), "legacy_snapshot": True, }) records.append(("users", { "platform": "bilibili", "platform_user_id": platform_user_id, "display_name": _text(item.get("uname"), 128), "avatar_url": None, "user_level": None, "is_admin": False, "first_seen_at_utc": created_at, "last_seen_at_utc": snapshot_at, "snapshot_json": snapshot, })) return records def _song_records(path: Path, raw: bytes) -> list[tuple[str, dict[str, Any]]]: data = _read_json(raw) if not isinstance(data, Mapping): return [] snapshot_at = _file_time(path) fallback = datetime.fromtimestamp(path.stat().st_mtime, UTC) entries: list[dict[str, Any]] = [] sections = (("queue", data.get("queue")), ("active", [data.get("active")]), ("history", data.get("history"))) for section, values in sections: if not isinstance(values, list): continue section_entries: dict[tuple[str | None, str, str | None, str], list[dict[str, Any]]] = {} for queue_position, original in enumerate(values, 1): if not isinstance(original, Mapping): continue item = {key: original[key] for key in _SONG_FIELDS if key in original} requested = _utc_iso(item.get("requested_at"), fallback=fallback, timespec="microseconds") source = _text(item.get("source"), 64) or "legacy" business_key = ( _text(item.get("uid"), 64), requested, _text(item.get("id"), 128), source, ) entry = { "section": section, "item": item, "requested": requested, "started": _utc_iso(item["started_at"], timespec="microseconds") if item.get("started_at") is not None else None, "ended": _utc_iso(item["finished_at"], timespec="microseconds") if item.get("finished_at") is not None else None, "source": source, "business_key": business_key, "queue_position": queue_position if section == "queue" else None, } section_entries.setdefault(business_key, []).append(entry) for business_key, duplicates in section_entries.items(): duplicates.sort(key=lambda entry: ( entry["started"] or "", entry["ended"] or "", str(entry["item"].get("status") or ""), str(entry["item"].get("name") or ""), str(entry["item"].get("artist") or ""), str(entry["item"].get("duration_ms") or ""), )) for duplicate_ordinal, entry in enumerate(duplicates): entry["request_id"] = _hash("legacy-song-request", *business_key, duplicate_ordinal) entries.append(entry) section_priority = {"queue": 0, "active": 1, "history": 2} requests: dict[str, dict[str, Any]] = {} playbacks: dict[str, dict[str, Any]] = {} for entry in sorted(entries, key=lambda value: section_priority[value["section"]]): section = entry["section"] item = entry["item"] request_id = entry["request_id"] confidence = "snapshot" if section in {"queue", "active"} else "history" default_status = "queued" if section == "queue" else "playing" if section == "active" else "unknown" status = _normalize_status(item.get("status"), default_status) requests[request_id] = { "request_id": request_id, "platform": "bilibili", "platform_user_id": _text(item.get("uid"), 64), "requested_at_utc": entry["requested"], "business_date": business_date(entry["requested"]), "song_id": _text(item.get("id"), 128), "song_name": _text(item.get("name"), 256), "artist": _text(item.get("artist"), 256), "source": entry["source"], "status": status, "points_cost": 0, "queue_position": entry["queue_position"], "payload_json": _json_payload({ "legacy_section": section, "confidence": confidence, "latest_80_only": section == "history", "keyword_length": len(str(item.get("keyword") or "")), "remove_after_play": _bool(item.get("remove_after_play")), "snapshot_at_utc": snapshot_at if confidence == "snapshot" else None, }), } started = entry["started"] if not started: continue ended = entry["ended"] played_ms = None if ended: start_dt = datetime.fromisoformat(started.replace("Z", "+00:00")) end_dt = datetime.fromisoformat(ended.replace("Z", "+00:00")) played_ms = max(0, round((end_dt - start_dt).total_seconds() * 1000)) playback_id = _hash("legacy-playback", request_id, started) playbacks[playback_id] = { "playback_id": playback_id, "request_id": request_id, "song_id": _text(item.get("id"), 128), "song_name": _text(item.get("name"), 256), "started_at_utc": started, "ended_at_utc": ended, "business_date": business_date(started), "status": status, "duration_ms": _int(item.get("duration_ms"), minimum=0), "played_ms": played_ms, "stop_reason": "snapshot" if confidence == "snapshot" else status, "payload_json": { "legacy_section": section, "confidence": confidence, "latest_80_only": section == "history", }, } return [ *(("song_requests", record) for record in requests.values()), *(("playback_sessions", record) for record in playbacks.values()), ] def _safe_detail(detail: Any) -> dict[str, Any]: text = str(detail or "") result: dict[str, Any] = { "detail_length": len(text), "detail_sha256": _hash("admin-detail", text), } count_match = _COUNT_DETAIL_RE.fullmatch(text) if count_match: result.update({"detail_kind": "count", "count": int(count_match.group(1))}) return result try: parsed = json.loads(text) except (TypeError, ValueError): parsed = None if isinstance(parsed, Mapping): allowed_flags = {key: _bool(parsed[key]) for key in ("blocked_all", "blocked_queue", "blocked_song_request") if key in parsed} if allowed_flags: result.update({"detail_kind": "flags", "flags": allowed_flags}) return result result["detail_kind"] = "opaque" return result def _audit_records(raw: bytes) -> list[tuple[str, dict[str, Any]]]: records: list[tuple[str, dict[str, Any]]] = [] duplicate_counts: Counter[str] = Counter() for raw_line in raw.decode("utf-8-sig", "replace").splitlines(): line = raw_line.strip() if not line: continue try: original = json.loads(line) except json.JSONDecodeError: continue if not isinstance(original, Mapping): continue line_hash = _hash("legacy-admin-audit-line", line) duplicate_ordinal = duplicate_counts[line_hash] duplicate_counts[line_hash] += 1 item = {key: original[key] for key in _AUDIT_FIELDS if key in original} occurred = _utc_iso(item.get("at"), naive_is_beijing=True) action = _text(item.get("action"), 96) or "unknown" target = _text(item.get("target"), 256) event_id = _hash("legacy-admin-audit", line_hash, duplicate_ordinal) payload = _safe_detail(item.get("detail")) session = _text(item.get("session_id"), 512) if session: payload["session_hash"] = _hash("admin-session", session) records.append(("admin_audit_events", { "event_id": event_id, "occurred_at_utc": occurred, "business_date": business_date(occurred), "actor": "admin", "action": action, "target_type": "legacy_target", "target_id": target, "success": True, "remote_address_hash": _hash("admin-ip", item.get("client_ip")) if item.get("client_ip") else None, "payload_json": payload, })) point_match = _POINT_DETAIL_RE.fullmatch(str(item.get("detail") or "")) if action == "user_add_points" else None if point_match and target: records.append(("point_transactions", { "transaction_id": _hash("legacy-admin-points", event_id), "platform": "bilibili", "platform_user_id": target, "occurred_at_utc": occurred, "business_date": business_date(occurred), "amount": int(point_match.group(1)), "balance_after": int(point_match.group(2)), "reason": "admin_adjustment", "reference_type": "admin_audit", "reference_id": event_id, "payload_json": {"legacy_import": True}, })) return records def _snapshot_event(path: Path, raw: bytes) -> list[tuple[str, dict[str, Any]]]: data = _read_json(raw) if not isinstance(data, Mapping): return [] monitor_updated_at = (data.get("monitor") or {}).get("updated_at") if isinstance(data.get("monitor"), Mapping) else None occurred = _utc_iso( data.get("updated_at") or monitor_updated_at, fallback=datetime.fromtimestamp(path.stat().st_mtime, UTC), ) if path.name == "queue_state.json": queue = data.get("queue") if isinstance(data.get("queue"), list) else [] payload = { "queue_size": len(queue), "has_active_user": bool(data.get("current_admin_uid")), "has_group": bool(data.get("current_group")), "default_running": _bool(data.get("default_running")), "login_status": _text(data.get("login_status"), 64), "has_user_finished_once": _bool(data.get("has_user_finished_once")), "confidence": "snapshot", } event_type = "legacy.queue_snapshot" category = "queue" elif path.name == "music_state.json": current = data.get("current") if isinstance(data.get("current"), Mapping) else {} monitor = data.get("monitor") if isinstance(data.get("monitor"), Mapping) else {} payload = { "playing": _bool(data.get("playing")), "current_title": _text(current.get("title"), 256), "current_artist": _text(current.get("artist"), 256), "duration": _int(current.get("duration"), minimum=0), "progress": _int(current.get("progress"), minimum=0), "playlist_size": len(data.get("playlist")) if isinstance(data.get("playlist"), list) else 0, "request_size": len(data.get("requests")) if isinstance(data.get("requests"), list) else 0, "monitor_online": _bool(monitor.get("online")), "platform": _text(monitor.get("platform"), 32), "confidence": "snapshot", } event_type = "legacy.music_snapshot" category = "music" else: recent = data.get("recent_events") if isinstance(data.get("recent_events"), list) else [] payload = { "enabled": _bool(data.get("enabled")), "provider": _text(data.get("provider"), 64), "model_loaded": _bool(data.get("model_loaded")), "last_duration_ms": _int(data.get("last_duration_ms"), minimum=0), "has_last_error": bool(data.get("last_error")), "total_synthesized": _int(data.get("total_synthesized"), minimum=0), "total_errors": _int(data.get("total_errors"), minimum=0), "recent_event_count": len(recent), "last_text_length": len(str(data.get("last_text") or "")), "confidence": "snapshot", } event_type = "legacy.tts_snapshot" category = "tts" payload = _json_payload(payload) return [("events", { "event_id": _hash("legacy-snapshot", path.name, _fingerprint(raw)), "event_type": event_type, "category": category, "occurred_at_utc": occurred, "business_date": business_date(occurred), "payload_json": payload, })] def _build_records(path: Path, raw: bytes) -> list[tuple[str, dict[str, Any]]]: if path.name == "users.json": return _user_records(path, raw) if path.name == "song_requests.json": return _song_records(path, raw) if path.name == "admin_audit.log": return _audit_records(raw) return _snapshot_event(path, raw) def validate_records(records: list[tuple[str, dict[str, Any]]]) -> None: """拒绝未知表、未知字段及非映射记录，避免导入边界被意外扩大。""" for record_number, record in enumerate(records, 1): if not isinstance(record, tuple) or len(record) != 2: raise ValueError(f"第 {record_number} 条导入记录格式无效") table, fields = record allowed_columns = TABLE_COLUMNS.get(table) if allowed_columns is None: raise ValueError(f"第 {record_number} 条记录使用未知表: {table}") if not isinstance(fields, Mapping): raise ValueError(f"第 {record_number} 条记录字段不是映射") unknown_columns = set(fields) - allowed_columns if unknown_columns: raise ValueError(f"{table} 包含未知字段: {sorted(unknown_columns)}") if not fields: raise ValueError(f"{table} 导入记录不能为空") async def backfill_legacy_statistics(store: StatsStore, data_dir: str | Path, dry_run: bool = False) -> dict[str, Any]: """扫描旧数据并通过 ``StatsStore.import_once`` 原子、幂等地补录。""" root = Path(data_dir) result: dict[str, Any] = {"dry_run": bool(dry_run), "sources": {}, "record_counts": {}} totals: Counter[str] = Counter() for filename in _SOURCE_FILES: path = root / filename if not path.is_file(): result["sources"][filename] = {"status": "missing", "records": 0} continue raw = _read_bytes(path) records = _build_records(path, raw) validate_records(records) counts = Counter(table for table, _ in records) totals.update(counts) imported = True if not dry_run: imported = await store.import_once( f"legacy-history:{filename}", "legacy_history", records, fingerprint=_fingerprint(raw), checkpoint_key="sha256", checkpoint_value=_fingerprint(raw), metadata={"filename": filename, "record_count": len(records), "schema": 1}, ) result["sources"][filename] = { "status": "dry_run" if dry_run else "completed" if imported else "failed", "records": len(records), "tables": dict(sorted(counts.items())), } result["record_counts"] = dict(sorted(totals.items())) result["total_records"] = sum(totals.values()) return result async def _main_async(args: argparse.Namespace) -> int: data_dir = Path(args.data_dir) if args.dry_run: result = await backfill_legacy_statistics(StatsStore(args.database), data_dir, dry_run=True) else: store = StatsStore(args.database) if not await store.start(): print(json.dumps({"status": "failed", "reason": "database_start_failed"}, ensure_ascii=False)) return 1 try: result = await backfill_legacy_statistics(store, data_dir) await store.flush() finally: await store.close() print(json.dumps(result, ensure_ascii=False, sort_keys=True)) return 0 if all(source["status"] != "failed" for source in result["sources"].values()) else 1 def main() -> int: parser = argparse.ArgumentParser(description="安全、幂等地补录旧版直播统计") parser.add_argument("--data-dir", default=str(Path(__file__).resolve().parents[1] / "data")) parser.add_argument("--database", default=str(DEFAULT_DATABASE_PATH)) parser.add_argument("--dry-run", action="store_true") return asyncio.run(_main_async(parser.parse_args())) if __name__ == "__main__": raise SystemExit(main())