import os import re import sys import shutil import datetime from sqlalchemy import create_engine, text # We run inside the romm container, so we can import ConfigManager from config.config_manager import ConfigManager def normalize(name): # Strip extension base, _ = os.path.splitext(name) # Strip everything inside parentheses/brackets base = re.sub(r"\([^)]*\)", "", base) base = re.sub(r"\[[^\]]*\]", "", base) # Remove non-alphanumeric characters and lowercase return re.sub(r"[^a-zA-Z0-9]", "", base).lower() def get_emulator_slug(platform): mapping = { "gba": "mgba", "nds": "melonds", "n64": "retroarch-mupen64plus_next", "psx": "retroarch-pcsx_rearmed", "switch": "yuzu" } return mapping.get(platform, "retroarch") def main(): dry_run = "--apply" not in sys.argv if dry_run: print("=== DRY RUN MODE: No changes will be applied ===") print("To apply changes, run with --apply flag.\n") else: print("=== EXECUTION MODE: Appending files and updating database ===\n") # Paths inside container staging_dir = "/romm/library/tmp/saves_staging" assets_dir = "/romm/assets" user_id = 1 user_hex = "557365723a31" # User:1 in hex if not os.path.exists(staging_dir): print(f"Staging directory {staging_dir} does not exist!") sys.exit(1) # Manual overrides dictionary: (platform, normalized_save_name) -> rom_id manual_overrides = { ("switch", "armsglobaltestpunch"): 9599, ("nds", "pokemondefinitiveheartgoldversion"): 3582, ("nds", "heartgoldqol"): 3586, } # 2. Connect to database engine = create_engine(ConfigManager.get_db_engine()) # Fetch all ROMs in database to build a matching index # Map (platform_id, normalized_name) -> list of (rom_id, rom_name, file_name, file_path) rom_index = {} # Map rom_id -> (rom_id, rom_name, file_name, file_path) rom_by_id = {} # Map title_id (16 hex chars starting with 0100) -> list of (rom_id, rom_name, file_name, file_path) title_id_index = {} platform_slug_to_id = { "gba": 72, "nds": 116, "n64": 111, "psx": 148, "switch": 179 } with engine.begin() as conn: print("Fetching ROMs from database...") result = conn.execute(text(""" SELECT r.id, r.name, r.platform_id, rf.file_name, rf.file_path FROM roms r JOIN rom_files rf ON r.id = rf.rom_id """)) for row in result: rom_id, rom_name, platform_id, file_name, file_path = row rom_tuple = (rom_id, rom_name, file_name, file_path) rom_by_id[rom_id] = rom_tuple # Index by normalized name norm_name = normalize(file_name) key = (platform_id, norm_name) if key not in rom_index: rom_index[key] = [] rom_index[key].append(rom_tuple) # Index Switch by title ID if present in filename if platform_id == 179: match = re.search(r"0100[0-9a-fA-F]{12}", file_name) if match: title_id = match.group(0).lower() if title_id not in title_id_index: title_id_index[title_id] = [] title_id_index[title_id].append(rom_tuple) print(f"Indexed {len(rom_index)} ROM name patterns and {len(title_id_index)} Switch title IDs.") # 3. Process staged files platforms = [d for d in os.listdir(staging_dir) if os.path.isdir(os.path.join(staging_dir, d))] copied_count = 0 db_count = 0 for platform_slug in platforms: platform_id = platform_slug_to_id.get(platform_slug) if not platform_id: print(f"Skipping unknown platform: {platform_slug}") continue platform_path = os.path.join(staging_dir, platform_slug) save_files = [f for f in os.listdir(platform_path) if os.path.isfile(os.path.join(platform_path, f))] print(f"\nProcessing {len(save_files)} saves for platform '{platform_slug}'...") for save_file in save_files: save_path = os.path.join(platform_path, save_file) save_base, save_ext = os.path.splitext(save_file) save_ext_clean = save_ext.replace(".", "").lower() # Match logic matched_rom = None norm_save_name = normalize(save_file) # Check manual overrides first override_rom_id = manual_overrides.get((platform_slug, norm_save_name)) if override_rom_id: matched_rom = rom_by_id.get(override_rom_id) if matched_rom: print(f" [*] MANUAL OVERRIDE MATCH: '{save_file}' -> ROM ID: {override_rom_id}") # Switch Title ID match if not matched_rom and platform_slug == "switch" and re.match(r"^0100[0-9a-fA-F]{12}$", save_base): title_id = save_base.lower() matches = title_id_index.get(title_id, []) if len(matches) == 1: matched_rom = matches[0] elif len(matches) > 1: print(f" [!] Ambiguous title ID {title_id} ({len(matches)} matches). Skipping.") continue # Fallback to normalized name match if not matched_rom: key = (platform_id, norm_save_name) matches = rom_index.get(key, []) if len(matches) == 1: matched_rom = matches[0] elif len(matches) > 1: # Try exact match on file name without extension exact_matches = [m for m in matches if os.path.splitext(m[2])[0].lower() == save_base.lower()] if len(exact_matches) == 1: matched_rom = exact_matches[0] else: print(f" [!] Ambiguous normalized name {norm_save_name} ({len(matches)} matches). Skipping.") continue if not matched_rom: print(f" [-] UNMATCHED: '{save_file}' (normalized: {norm_save_name})") continue rom_id, rom_name, rom_file_name, rom_file_path = matched_rom rom_base, _ = os.path.splitext(rom_file_name) # Format file name on disk mtime = os.path.getmtime(save_path) dt = datetime.datetime.fromtimestamp(mtime) timestamp_str = dt.strftime("%Y-%m-%d_%H-%M-%S") new_save_file_name = f"{rom_base} [{timestamp_str}]{save_ext}" new_save_file_no_ext = f"{rom_base} [{timestamp_str}]" # Emulator slug emulator = get_emulator_slug(platform_slug) # Path relative to assets folder rel_save_path = f"users/{user_hex}/saves/{platform_slug}/{rom_id}/{emulator}" dest_dir = os.path.join(assets_dir, rel_save_path) dest_file_path = os.path.join(dest_dir, new_save_file_name) file_size = os.path.getsize(save_path) print(f" [+] MATCHED: '{save_file}' -> ROM '{rom_name}' (ID: {rom_id})") print(f" Target Name: '{new_save_file_name}'") print(f" Target Path: '{rel_save_path}'") if not dry_run: # Create destination directory os.makedirs(dest_dir, exist_ok=True) # Copy file shutil.copy2(save_path, dest_file_path) copied_count += 1 # Database insert with engine.begin() as conn: # Check if save is already registered exist_check = conn.execute(text(""" SELECT id FROM saves WHERE rom_id = :rom_id AND user_id = :user_id AND file_name = :file_name """), {"rom_id": rom_id, "user_id": user_id, "file_name": new_save_file_name}).fetchone() if not exist_check: conn.execute(text(""" INSERT INTO saves ( emulator, created_at, updated_at, file_name, file_name_no_tags, file_name_no_ext, file_extension, file_path, file_size_bytes, rom_id, user_id, missing_from_fs, slot ) VALUES ( :emulator, :created_at, :updated_at, :file_name, :file_name_no_tags, :file_name_no_ext, :file_extension, :file_path, :file_size_bytes, :rom_id, :user_id, 0, 'default' ) """), { "emulator": emulator, "created_at": dt, "updated_at": dt, "file_name": new_save_file_name, "file_name_no_tags": rom_name, "file_name_no_ext": new_save_file_no_ext, "file_extension": save_ext_clean, "file_path": rel_save_path, "file_size_bytes": file_size, "rom_id": rom_id, "user_id": user_id }) db_count += 1 print(" [DB] Inserted database record.") else: print(" [DB] Save record already exists. Skipped DB insert.") print("\n=== SUMMARY ===") if dry_run: print("Dry run completed. Run with --apply flag to execute.") else: print(f"Successfully copied {copied_count} files to assets.") print(f"Successfully registered {db_count} saves in database.") if __name__ == "__main__": main()