320 lines
9.0 KiB
Python
320 lines
9.0 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Dict, List, Optional, Tuple
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
from main import fetch_all_roms, session, base
|
|
|
|
DEFAULT_HOST_ROMS_PATH = "/home/jawz/Games/roms"
|
|
|
|
|
|
def translate_docker_path(
|
|
docker_path: str, host_path: str = DEFAULT_HOST_ROMS_PATH
|
|
) -> str:
|
|
"""Translate Docker container path to host path."""
|
|
if not docker_path:
|
|
return docker_path
|
|
|
|
if docker_path.startswith(host_path):
|
|
return docker_path
|
|
|
|
docker_prefixes = ["/roms", "/data/roms", "/mnt/roms"]
|
|
|
|
matching_prefix = next(
|
|
(prefix for prefix in docker_prefixes if docker_path.startswith(prefix)), None
|
|
)
|
|
|
|
if matching_prefix:
|
|
relative_path = docker_path[len(matching_prefix) :].lstrip("/")
|
|
return os.path.join(host_path, relative_path)
|
|
|
|
path_parts = Path(docker_path).parts
|
|
if len(path_parts) > 1:
|
|
return str(Path(host_path) / Path(*path_parts[1:]))
|
|
|
|
return docker_path
|
|
|
|
|
|
def is_matched(rom: Dict) -> bool:
|
|
"""Check if ROM is matched (has metadata)."""
|
|
metadata_keys = ["igdb_id", "moby_id", "ss_id", "launchbox_id"]
|
|
return any(rom.get(key) for key in metadata_keys)
|
|
|
|
|
|
def should_skip_rom(rom: Dict) -> bool:
|
|
"""Check if ROM should be skipped (e.g., notgame, ZZZ prefix, etc.)."""
|
|
name = rom.get("name", "")
|
|
if not name:
|
|
return True
|
|
|
|
name_lower = name.lower()
|
|
skip_patterns = ["(notgame)", "zzz", "zzz:"]
|
|
|
|
return any(pattern in name_lower for pattern in skip_patterns)
|
|
|
|
|
|
def already_has_proper_name(rom: Dict) -> bool:
|
|
"""Check if ROM already has the proper name format."""
|
|
if not (fs_name := rom.get("fs_name")):
|
|
return False
|
|
|
|
if not (rom_name := rom.get("name")):
|
|
return False
|
|
|
|
if not (extension := get_extension(rom)):
|
|
return False
|
|
|
|
sanitized_name = sanitize_filename(rom_name)
|
|
region_code = get_region_code(rom)
|
|
|
|
if region_code:
|
|
expected_name = f"{sanitized_name} ({region_code}).{extension}"
|
|
else:
|
|
expected_name = f"{sanitized_name}.{extension}"
|
|
|
|
current_name = os.path.basename(fs_name)
|
|
return current_name.lower() == expected_name.lower()
|
|
|
|
|
|
def is_valid_for_rename(rom: Dict) -> bool:
|
|
"""Check if ROM is valid for renaming (matched, not skipped, and needs renaming)."""
|
|
return (
|
|
is_matched(rom)
|
|
and not should_skip_rom(rom)
|
|
and not already_has_proper_name(rom)
|
|
)
|
|
|
|
|
|
def get_region_code(rom: Dict) -> Optional[str]:
|
|
"""Extract region code from ROM regions.
|
|
Returns single letter code like 'U', 'E', 'J' etc.
|
|
"""
|
|
regions = rom.get("regions", [])
|
|
if not regions:
|
|
return None
|
|
|
|
region_map = {
|
|
"usa": "U",
|
|
"north america": "U",
|
|
"ntsc-u": "U",
|
|
"eur": "E",
|
|
"europe": "E",
|
|
"pal": "E",
|
|
"jpn": "J",
|
|
"japan": "J",
|
|
"ntsc-j": "J",
|
|
"asia": "A",
|
|
"australia": "A",
|
|
"world": "W",
|
|
}
|
|
|
|
def find_matching_code(region: str) -> Optional[str]:
|
|
region_lower = region.lower()
|
|
return next(
|
|
(code for key, code in region_map.items() if key in region_lower), None
|
|
)
|
|
|
|
matching_code = next(
|
|
(code for region in regions if (code := find_matching_code(region))), None
|
|
)
|
|
|
|
if matching_code:
|
|
return matching_code
|
|
|
|
if regions and len(regions[0]) >= 1:
|
|
return regions[0].upper()[0]
|
|
|
|
return None
|
|
|
|
|
|
def sanitize_filename(name: str) -> str:
|
|
"""Sanitize filename to remove invalid characters."""
|
|
invalid_chars = '<>:"/\\|?*'
|
|
return name.translate(str.maketrans("", "", invalid_chars)).strip(" .")
|
|
|
|
|
|
def get_extension(rom: Dict) -> Optional[str]:
|
|
"""Extract file extension from ROM."""
|
|
if extension := rom.get("fs_extension"):
|
|
return extension
|
|
|
|
if fs_name := rom.get("fs_name"):
|
|
ext = os.path.splitext(fs_name)[1].lstrip(".")
|
|
return ext if ext else None
|
|
|
|
return None
|
|
|
|
|
|
def generate_new_filename(rom: Dict) -> Optional[str]:
|
|
"""Generate new filename in format: {rom_name} ({region}).{ext} or {rom_name}.{ext} if no region"""
|
|
if not (rom_name := rom.get("name")):
|
|
return None
|
|
|
|
if not (extension := get_extension(rom)):
|
|
return None
|
|
|
|
sanitized_name = sanitize_filename(rom_name)
|
|
region_code = get_region_code(rom)
|
|
|
|
if region_code:
|
|
return f"{sanitized_name} ({region_code}).{extension}"
|
|
|
|
return f"{sanitized_name}.{extension}"
|
|
|
|
|
|
def rename_rom_file(
|
|
rom: Dict, dry_run: bool = True, host_path: str = DEFAULT_HOST_ROMS_PATH
|
|
) -> Tuple[bool, str, str]:
|
|
"""Rename a ROM file.
|
|
Returns: (success, old_path, new_path_or_error)
|
|
"""
|
|
if not (full_path := rom.get("full_path")):
|
|
return False, "", "No full_path available"
|
|
|
|
translated_path = translate_docker_path(full_path, host_path)
|
|
|
|
if not os.path.exists(translated_path):
|
|
return False, translated_path, f"File does not exist: {translated_path}"
|
|
|
|
if not (new_filename := generate_new_filename(rom)):
|
|
return False, translated_path, "Could not generate new filename"
|
|
|
|
old_path = Path(translated_path)
|
|
new_path = old_path.parent / new_filename
|
|
|
|
if new_path.exists() and new_path != old_path:
|
|
return False, str(old_path), f"Target file already exists: {new_path}"
|
|
|
|
if dry_run:
|
|
return True, str(old_path), str(new_path)
|
|
|
|
try:
|
|
os.rename(str(old_path), str(new_path))
|
|
return True, str(old_path), str(new_path)
|
|
except Exception as e:
|
|
return False, str(old_path), f"Error: {e}"
|
|
|
|
|
|
def process_rom_rename(
|
|
rom: Dict, dry_run: bool, host_path: str = DEFAULT_HOST_ROMS_PATH
|
|
) -> Tuple[Dict, bool, str, str]:
|
|
"""Process a single ROM rename."""
|
|
success, old_path, new_path = rename_rom_file(
|
|
rom, dry_run=dry_run, host_path=host_path
|
|
)
|
|
return (rom, success, old_path, new_path)
|
|
|
|
|
|
def format_rename_result(rom: Dict, old_path: str, new_path: str) -> str:
|
|
"""Format a rename result for display."""
|
|
old_name = os.path.basename(old_path)
|
|
new_name = os.path.basename(new_path)
|
|
return (
|
|
f" {rom.get('name', 'Unknown')} (ID: {rom['id']})\n"
|
|
f" {old_name}\n"
|
|
f" -> {new_name}\n"
|
|
)
|
|
|
|
|
|
def format_error_result(rom: Dict, old_path: str, error: str) -> str:
|
|
"""Format an error result for display."""
|
|
return (
|
|
f" {rom.get('name', 'Unknown')} (ID: {rom['id']})\n"
|
|
f" {old_path}\n"
|
|
f" Error: {error}\n"
|
|
)
|
|
|
|
|
|
def print_results(results: List[Tuple[Dict, bool, str, str]]) -> None:
|
|
"""Print rename results."""
|
|
successful = [r for r in results if r[1]]
|
|
failed = [r for r in results if not r[1]]
|
|
|
|
print(f"Successful: {len(successful)}")
|
|
print(f"Failed: {len(failed)}\n")
|
|
|
|
if successful:
|
|
print("Successfully processed:")
|
|
print(
|
|
"".join(
|
|
format_rename_result(rom, old_path, new_path)
|
|
for rom, _, old_path, new_path in successful
|
|
)
|
|
)
|
|
|
|
if failed:
|
|
print("Failed:")
|
|
print(
|
|
"".join(
|
|
format_error_result(rom, old_path, error)
|
|
for rom, _, old_path, error in failed
|
|
)
|
|
)
|
|
|
|
|
|
def rename_matched_roms(
|
|
dry_run: bool = True, host_path: str = DEFAULT_HOST_ROMS_PATH
|
|
) -> None:
|
|
"""Rename all matched ROMs."""
|
|
print("Fetching all ROMs...")
|
|
roms = fetch_all_roms(session, base)
|
|
print(f"Fetched {len(roms)} ROMs\n")
|
|
|
|
matched_roms = list(filter(is_matched, roms))
|
|
valid_roms = list(filter(is_valid_for_rename, roms))
|
|
already_named = [
|
|
rom
|
|
for rom in matched_roms
|
|
if not should_skip_rom(rom) and already_has_proper_name(rom)
|
|
]
|
|
|
|
print(f"Found {len(matched_roms)} matched ROMs")
|
|
print(f"Found {len(valid_roms)} valid ROMs for renaming")
|
|
if already_named:
|
|
print(f"Skipped {len(already_named)} ROMs that already have proper names\n")
|
|
else:
|
|
print()
|
|
|
|
if not valid_roms:
|
|
print("No valid ROMs to rename.")
|
|
return
|
|
|
|
mode = "DRY RUN" if dry_run else "RENAMING"
|
|
print(f"{'='*60}")
|
|
print(f"{mode} MODE")
|
|
print(f"Host path: {host_path}")
|
|
print(f"{'='*60}\n")
|
|
|
|
results = [process_rom_rename(rom, dry_run, host_path) for rom in valid_roms]
|
|
|
|
print_results(results)
|
|
|
|
if dry_run and any(r[1] for r in results):
|
|
print(f"{'='*60}")
|
|
print("This was a DRY RUN. No files were actually renamed.")
|
|
print("Run with --execute to perform the renames.")
|
|
print(f"{'='*60}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(description="Rename matched ROMs")
|
|
parser.add_argument(
|
|
"--execute",
|
|
action="store_true",
|
|
help="Actually perform the renames (default is dry run)",
|
|
)
|
|
parser.add_argument(
|
|
"--host-path",
|
|
type=str,
|
|
default=os.getenv("ROMS_HOST_PATH", DEFAULT_HOST_ROMS_PATH),
|
|
help=f"Host path for ROMs (default: {DEFAULT_HOST_ROMS_PATH})",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
rename_matched_roms(dry_run=not args.execute, host_path=args.host_path)
|