feat: add RomM database path mismatch resolver script
This commit is contained in:
ginnoir
@@ -0,0 +1,176 @@
|
||||
import os
|
||||
import csv
|
||||
import re
|
||||
import sys
|
||||
from config.config_manager import ConfigManager
|
||||
from sqlalchemy import create_engine, text
|
||||
|
||||
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 main():
|
||||
dry_run = "--apply" not in sys.argv
|
||||
if dry_run:
|
||||
print("=== DRY RUN MODE: No database updates will be applied ===")
|
||||
print("To apply changes, run the script with --apply flag.")
|
||||
else:
|
||||
print("=== EXECUTION MODE: Database updates WILL be applied ===")
|
||||
|
||||
# 1. Parse CSV files and build mapping
|
||||
csv_dir = "/tmp/reports"
|
||||
mapping = {} # key: (platform, old_filename), value: new_filename
|
||||
|
||||
if os.path.exists(csv_dir):
|
||||
try:
|
||||
csv_files = [f for f in os.listdir(csv_dir) if f.startswith("_dry_") and f.endswith(".csv")]
|
||||
for csv_file in csv_files:
|
||||
platform = csv_file.replace("_dry_", "").replace(".csv", "")
|
||||
csv_path = os.path.join(csv_dir, csv_file)
|
||||
print(f"Parsing CSV {csv_file} (platform: {platform})...")
|
||||
try:
|
||||
with open(csv_path, "r", encoding="utf-8", errors="ignore") as f:
|
||||
reader = csv.DictReader(f)
|
||||
for row in reader:
|
||||
status = row.get("Status")
|
||||
if status != "FOUND":
|
||||
continue
|
||||
game_name = row.get("Game Name")
|
||||
rom_files_str = row.get("ROM Files")
|
||||
if not game_name or not rom_files_str:
|
||||
continue
|
||||
rom_files = [r.strip() for r in rom_files_str.split(",") if r.strip()]
|
||||
for rom_file in rom_files:
|
||||
old_filename = os.path.basename(rom_file)
|
||||
_, ext = os.path.splitext(old_filename)
|
||||
new_filename = game_name.replace("/", "-").replace("\\", "-") + ext
|
||||
mapping[(platform, old_filename)] = new_filename
|
||||
except Exception as e:
|
||||
print(f"Error parsing CSV {csv_file}: {e}")
|
||||
except Exception as e:
|
||||
print(f"Error listing CSV files: {e}")
|
||||
else:
|
||||
print("CSV directory /tmp/reports does not exist in container.")
|
||||
|
||||
print(f"Loaded {len(mapping)} mappings from CSVs.")
|
||||
|
||||
# 2. Connect to database
|
||||
engine = create_engine(ConfigManager.get_db_engine())
|
||||
|
||||
with engine.begin() as conn:
|
||||
result = conn.execute(text("""
|
||||
SELECT rf.id, rf.rom_id, rf.file_name, rf.file_path, r.fs_name, r.fs_path, rf.missing_from_fs, r.missing_from_fs
|
||||
FROM rom_files rf
|
||||
JOIN roms r ON rf.rom_id = r.id
|
||||
"""))
|
||||
|
||||
mismatches_resolved = 0
|
||||
mismatches_unresolved = 0
|
||||
missing_marked = 0
|
||||
found_cleared = 0
|
||||
|
||||
for row in result:
|
||||
rf_id, rom_id, file_name, file_path, fs_name, fs_path, rf_missing, r_missing = row
|
||||
|
||||
# Check if file exists on disk
|
||||
full_path = os.path.join("/romm/library", file_path, file_name)
|
||||
if os.path.exists(full_path):
|
||||
# File exists on disk at the stored name
|
||||
if rf_missing or r_missing:
|
||||
# Clear missing flag
|
||||
found_cleared += 1
|
||||
print(f"Found existing file on disk: Clear missing flag for rf_id={rf_id} rom_id={rom_id} ('{file_name}')")
|
||||
if not dry_run:
|
||||
conn.execute(text("UPDATE rom_files SET missing_from_fs = 0 WHERE id = :id"), {"id": rf_id})
|
||||
conn.execute(text("UPDATE roms SET missing_from_fs = 0 WHERE id = :id"), {"id": rom_id})
|
||||
continue
|
||||
|
||||
# File does NOT exist at stored path. Try to resolve.
|
||||
platform = file_path.split("/")[-1] if "/" in file_path else file_path
|
||||
new_name = None
|
||||
|
||||
# Match 1: CSV mapping
|
||||
csv_mapped_name = mapping.get((platform, file_name))
|
||||
if csv_mapped_name:
|
||||
csv_mapped_path = os.path.join("/romm/library", file_path, csv_mapped_name)
|
||||
if os.path.exists(csv_mapped_path):
|
||||
new_name = csv_mapped_name
|
||||
else:
|
||||
# Match 1b: CSV mapping with extension-insensitive search
|
||||
new_base, _ = os.path.splitext(csv_mapped_name)
|
||||
dir_path = os.path.join("/romm/library", file_path)
|
||||
if os.path.exists(dir_path):
|
||||
try:
|
||||
files_in_dir = os.listdir(dir_path)
|
||||
for f in files_in_dir:
|
||||
f_base, _ = os.path.splitext(f)
|
||||
if f_base.lower() == new_base.lower() and not os.path.isdir(os.path.join(dir_path, f)):
|
||||
new_name = f
|
||||
break
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Match 2: Directory scan normalization
|
||||
if not new_name:
|
||||
dir_path = os.path.join("/romm/library", file_path)
|
||||
if os.path.exists(dir_path):
|
||||
try:
|
||||
files_in_dir = os.listdir(dir_path)
|
||||
norm_old = normalize(file_name)
|
||||
|
||||
# Match 2a: Exact normalized match
|
||||
for f in files_in_dir:
|
||||
if os.path.isdir(os.path.join(dir_path, f)):
|
||||
continue
|
||||
if normalize(f) == norm_old:
|
||||
new_name = f
|
||||
break
|
||||
|
||||
# Match 2b: Substring normalized match
|
||||
if not new_name:
|
||||
for f in files_in_dir:
|
||||
if os.path.isdir(os.path.join(dir_path, f)):
|
||||
continue
|
||||
norm_f = normalize(f)
|
||||
if (norm_old and norm_f) and (norm_old in norm_f or norm_f in norm_old):
|
||||
new_name = f
|
||||
break
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if new_name:
|
||||
mismatches_resolved += 1
|
||||
print(f"RESOLVED: rf_id={rf_id} rom_id={rom_id} path='{file_path}': '{file_name}' -> '{new_name}'")
|
||||
if not dry_run:
|
||||
conn.execute(text("""
|
||||
UPDATE rom_files
|
||||
SET file_name = :new_name, missing_from_fs = 0, updated_at = NOW()
|
||||
WHERE id = :id
|
||||
"""), {"new_name": new_name, "id": rf_id})
|
||||
conn.execute(text("""
|
||||
UPDATE roms
|
||||
SET fs_name = :new_name, missing_from_fs = 0, updated_at = NOW()
|
||||
WHERE id = :id
|
||||
"""), {"new_name": new_name, "id": rom_id})
|
||||
else:
|
||||
mismatches_unresolved += 1
|
||||
if not rf_missing or not r_missing:
|
||||
missing_marked += 1
|
||||
print(f"UNRESOLVED: Mark as missing: rf_id={rf_id} rom_id={rom_id} ('{file_name}')")
|
||||
if not dry_run:
|
||||
conn.execute(text("UPDATE rom_files SET missing_from_fs = 1 WHERE id = :id"), {"id": rf_id})
|
||||
conn.execute(text("UPDATE roms SET missing_from_fs = 1 WHERE id = :id"), {"id": rom_id})
|
||||
|
||||
print("\n=== SUMMARY ===")
|
||||
print(f"Total resolved mismatches (updated names): {mismatches_resolved}")
|
||||
print(f"Total unresolved mismatches: {mismatches_unresolved}")
|
||||
print(f"Marked as missing (unresolved): {missing_marked}")
|
||||
print(f"Cleared missing flag (found on disk): {found_cleared}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user