NotSoToothless
128 lines
3.6 KiB
Python
128 lines
3.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
migrate_replay_ids.py
|
|
|
|
One-shot migration: renames decimal replay directories to hex and updates
|
|
the _id field inside each replay_data.json.gz to match.
|
|
|
|
Run on the server:
|
|
python3 migrate_replay_ids.py
|
|
python3 migrate_replay_ids.py --dry-run # preview without touching anything
|
|
|
|
Reads STORAGE_VOL_PATH from environment or TSSBOT/.env.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import gzip
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
_HERE = Path(__file__).resolve().parent
|
|
|
|
# Try loading .env if python-dotenv is available
|
|
try:
|
|
from dotenv import load_dotenv
|
|
load_dotenv(dotenv_path=_HERE / ".env")
|
|
except ImportError:
|
|
pass
|
|
|
|
STORAGE_VOL_PATH = os.environ.get("STORAGE_VOL_PATH", "").strip()
|
|
if not STORAGE_VOL_PATH:
|
|
print("ERROR: STORAGE_VOL_PATH not set", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
REPLAYS_DIR = Path(STORAGE_VOL_PATH) / "REPLAYS" / "TSS"
|
|
|
|
|
|
def is_decimal_id(name: str) -> bool:
|
|
"""True if the directory name looks like a plain decimal integer (not hex)."""
|
|
try:
|
|
int(name)
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def to_hex(decimal_str: str) -> str:
|
|
return hex(int(decimal_str))[2:].lower()
|
|
|
|
|
|
def migrate_one(src_dir: Path, *, dry_run: bool) -> tuple[str, str] | None:
|
|
"""
|
|
Rename src_dir from decimal to hex, update _id inside replay_data.json.gz.
|
|
Returns (old_name, new_name) on success, None if already hex or skipped.
|
|
"""
|
|
name = src_dir.name
|
|
if not is_decimal_id(name):
|
|
return None
|
|
|
|
hex_name = to_hex(name)
|
|
dst_dir = src_dir.parent / hex_name
|
|
|
|
if dst_dir.exists():
|
|
print(f" SKIP {name} → {hex_name} (destination already exists)")
|
|
return None
|
|
|
|
gz_path = src_dir / "replay_data.json.gz"
|
|
|
|
if gz_path.exists():
|
|
try:
|
|
with gzip.open(gz_path, "rb") as fh:
|
|
data = json.loads(fh.read().decode("utf-8"))
|
|
except Exception as exc:
|
|
print(f" ERROR reading {gz_path}: {exc}")
|
|
return None
|
|
|
|
data["_id"] = hex_name
|
|
|
|
if not dry_run:
|
|
with gzip.open(gz_path, "wb") as fh:
|
|
fh.write(json.dumps(data, ensure_ascii=False).encode("utf-8"))
|
|
|
|
if not dry_run:
|
|
src_dir.rename(dst_dir)
|
|
|
|
return name, hex_name
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
parser.add_argument("--dry-run", action="store_true", help="Print what would happen without changing anything")
|
|
args = parser.parse_args()
|
|
|
|
if not REPLAYS_DIR.is_dir():
|
|
print(f"ERROR: {REPLAYS_DIR} does not exist", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
entries = sorted(p for p in REPLAYS_DIR.iterdir() if p.is_dir())
|
|
candidates = [p for p in entries if is_decimal_id(p.name)]
|
|
|
|
if not candidates:
|
|
print("No decimal-named replay directories found. Nothing to do.")
|
|
return
|
|
|
|
mode = "DRY RUN — " if args.dry_run else ""
|
|
print(f"{mode}Found {len(candidates)} decimal director(ies) to migrate in {REPLAYS_DIR}\n")
|
|
|
|
ok = skipped = errors = 0
|
|
for src_dir in candidates:
|
|
result = migrate_one(src_dir, dry_run=args.dry_run)
|
|
if result is None:
|
|
skipped += 1
|
|
else:
|
|
old, new = result
|
|
tag = "[dry-run] " if args.dry_run else ""
|
|
print(f" {tag}{old} → {new}")
|
|
ok += 1
|
|
|
|
print(f"\nDone. migrated={ok} skipped={skipped} errors={errors}")
|
|
if args.dry_run:
|
|
print("(dry run — nothing was changed)")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|