570 lines
22 KiB
Python
570 lines
22 KiB
Python
"""安全、幂等地将旧版 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())
|