NotSoToothless
123 lines
4.2 KiB
Python
123 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Migrate flat STORAGE/REPLAYS/<session_id>/ dirs → STORAGE/REPLAYS/SRE/<session_id>/
|
|
|
|
Run in dry-run first (default), then with --execute to actually move files.
|
|
Pass --move to delete each source dir after a verified copy.
|
|
|
|
Usage:
|
|
python scripts/migrate_replays_to_sre.py
|
|
python scripts/migrate_replays_to_sre.py --execute
|
|
python scripts/migrate_replays_to_sre.py --execute --move
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import os
|
|
import re
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
_storage_env = os.environ.get("STORAGE_VOL_PATH", "").strip()
|
|
if not _storage_env:
|
|
raise RuntimeError("STORAGE_VOL_PATH must be set")
|
|
|
|
STORAGE_DIR = Path(_storage_env)
|
|
DEFAULT_SOURCE = STORAGE_DIR / "REPLAYS"
|
|
DEFAULT_DEST = STORAGE_DIR / "REPLAYS" / "SRE"
|
|
|
|
HEX_RE = re.compile(r"^[0-9a-fA-F]+$")
|
|
|
|
|
|
def _iter_files(root: Path) -> list[Path]:
|
|
return [p for p in root.rglob("*") if p.is_file()]
|
|
|
|
|
|
def _copy_dir(src: Path, dst: Path, dry_run: bool) -> tuple[int, int]:
|
|
copied = skipped = 0
|
|
for src_file in _iter_files(src):
|
|
dst_file = dst / src_file.relative_to(src)
|
|
if dst_file.exists() and dst_file.stat().st_size == src_file.stat().st_size:
|
|
skipped += 1
|
|
continue
|
|
copied += 1
|
|
if not dry_run:
|
|
dst_file.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(src_file, dst_file)
|
|
return copied, skipped
|
|
|
|
|
|
def _fully_copied(src: Path, dst: Path) -> bool:
|
|
for src_file in _iter_files(src):
|
|
dst_file = dst / src_file.relative_to(src)
|
|
if not dst_file.exists() or dst_file.stat().st_size != src_file.stat().st_size:
|
|
return False
|
|
return True
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--source", type=Path, default=DEFAULT_SOURCE,
|
|
help=f"Flat REPLAYS root (default: {DEFAULT_SOURCE})")
|
|
parser.add_argument("--dest", type=Path, default=DEFAULT_DEST,
|
|
help=f"REPLAYS/SRE target (default: {DEFAULT_DEST})")
|
|
parser.add_argument("--execute", action="store_true",
|
|
help="Actually copy files (default is dry-run).")
|
|
parser.add_argument("--move", action="store_true",
|
|
help="Remove each source dir after a verified copy (requires --execute).")
|
|
args = parser.parse_args()
|
|
|
|
source: Path = args.source.expanduser().resolve()
|
|
dest: Path = args.dest.expanduser().resolve()
|
|
dry_run = not args.execute
|
|
|
|
if args.move and dry_run:
|
|
parser.error("--move requires --execute")
|
|
if not source.exists():
|
|
raise SystemExit(f"Source does not exist: {source}")
|
|
if source == dest:
|
|
raise SystemExit("Source and destination are the same directory")
|
|
|
|
print(f"Source : {source}")
|
|
print(f"Dest : {dest}")
|
|
print(f"Mode : {'DRY-RUN' if dry_run else 'EXECUTE'}" + (" + MOVE" if args.move else ""))
|
|
print()
|
|
|
|
dirs_found = dirs_skipped = files_copied = files_skipped = 0
|
|
|
|
for entry in sorted(source.iterdir()):
|
|
if not entry.is_dir():
|
|
continue
|
|
# Skip SRE/TSS subdirs if they already exist (idempotent)
|
|
if entry.name in {"SRE", "TSS"}:
|
|
continue
|
|
# Only migrate hex session-ID dirs
|
|
if not HEX_RE.fullmatch(entry.name):
|
|
print(f" skip {entry.name} (not a hex session dir)")
|
|
continue
|
|
|
|
dirs_found += 1
|
|
target = dest / entry.name.lower()
|
|
c, s = _copy_dir(entry, target, dry_run)
|
|
files_copied += c
|
|
files_skipped += s
|
|
print(f" {'[dry] ' if dry_run else ''}{'copy' if c else 'skip'} "
|
|
f"{entry.name} → SRE/{entry.name.lower()} "
|
|
f"(copy {c}, already-there {s})")
|
|
|
|
if args.move and not dry_run and _fully_copied(entry, target):
|
|
shutil.rmtree(entry)
|
|
print(f" removed source dir: {entry}")
|
|
|
|
print()
|
|
print(f"Dirs processed : {dirs_found}")
|
|
print(f"Files copied : {files_copied}")
|
|
print(f"Files skipped : {files_skipped}")
|
|
if dry_run:
|
|
print("\nDry-run only — re-run with --execute to copy.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|