feat: add RomM staging import worker

romm-automation-import-worker.py

@@ -0,0 +1,230 @@
+#!/usr/bin/env python3
+"""Simple RomM staging import worker.
+
+Watches /staging/incoming/<platform_slug>/ for files, moves them through
+processing/verified, copies them into /romm/library/roms/<platform_slug>/, and
+then triggers RomM's library scan API.
+
+This intentionally requires a platform subfolder under incoming. Example:
+  /staging/incoming/ngc/Paper Mario.iso
+  /staging/incoming/nes/Homebrew.nes
+
+Files dropped directly into /staging/incoming are quarantined so the worker does
+not guess the wrong RomM platform.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import sys
+import time
+import traceback
+from datetime import datetime, timezone
+from pathlib import Path
+from urllib import error, parse, request
+
+STAGING = Path(os.environ.get("STAGING_PATH", "/staging"))
+INCOMING = STAGING / "incoming"
+PROCESSING = STAGING / "processing"
+VERIFIED = STAGING / "verified"
+QUARANTINE = STAGING / "quarantine"
+REPORTS = Path(os.environ.get("REPORTS_PATH", "/reports"))
+ROMM_LIBRARY = Path(os.environ.get("ROMM_LIBRARY_CONTAINER_PATH", "/romm/library"))
+ROMM_ROMS = ROMM_LIBRARY / "roms"
+POLL_SECONDS = int(os.environ.get("IMPORT_POLL_SECONDS", "60"))
+SCAN_AFTER_IMPORT = os.environ.get("ROMM_SCAN_AFTER_IMPORT", "true").lower() in {"1", "true", "yes", "on"}
+QUIET_IDLE = os.environ.get("IMPORT_QUIET_IDLE", "true").lower() in {"1", "true", "yes", "on"}
+
+ROMM_SERVER_URL = os.environ.get("ROMM_SERVER_URL", "").rstrip("/")
+ROMM_USERNAME = os.environ.get("ROMM_USERNAME", "")
+ROMM_PASSWORD = os.environ.get("ROMM_PASSWORD", "")
+
+IGNORE_SUFFIXES = {".part", ".partial", ".tmp", ".crdownload"}
+
+
+def log(message: str) -> None:
+    ts = datetime.now(timezone.utc).isoformat(timespec="seconds")
+    print(f"[{ts}] {message}", flush=True)
+
+
+def ensure_dirs() -> None:
+    for path in (INCOMING, PROCESSING, VERIFIED, QUARANTINE, REPORTS, ROMM_ROMS):
+        path.mkdir(parents=True, exist_ok=True)
+
+
+def is_ready_file(path: Path) -> bool:
+    if not path.is_file():
+        return False
+    if path.name.startswith("."):
+        return False
+    if path.suffix.lower() in IGNORE_SUFFIXES:
+        return False
+    try:
+        size1 = path.stat().st_size
+        time.sleep(1)
+        size2 = path.stat().st_size
+        return size1 == size2 and size2 > 0
+    except FileNotFoundError:
+        return False
+
+
+def iter_incoming_files() -> list[Path]:
+    if not INCOMING.exists():
+        return []
+    return sorted([p for p in INCOMING.rglob("*") if is_ready_file(p)])
+
+
+def platform_for(path: Path) -> str | None:
+    rel = path.relative_to(INCOMING)
+    if len(rel.parts) < 2:
+        return None
+    platform = rel.parts[0].strip()
+    if platform in {"", ".", ".."} or "/" in platform:
+        return None
+    return platform
+
+
+def unique_dest(dest: Path) -> Path:
+    if not dest.exists():
+        return dest
+    stem = dest.stem
+    suffix = dest.suffix
+    parent = dest.parent
+    for i in range(1, 1000):
+        candidate = parent / f"{stem} ({i}){suffix}"
+        if not candidate.exists():
+            return candidate
+    raise RuntimeError(f"could not find unique destination for {dest}")
+
+
+def write_report(name: str, data: dict) -> None:
+    REPORTS.mkdir(parents=True, exist_ok=True)
+    path = REPORTS / name
+    path.write_text(json.dumps(data, indent=2, sort_keys=True), encoding="utf-8")
+
+
+def quarantine(path: Path, reason: str) -> None:
+    dest = unique_dest(QUARANTINE / path.name)
+    dest.parent.mkdir(parents=True, exist_ok=True)
+    shutil.move(str(path), str(dest))
+    report = {
+        "time_utc": datetime.now(timezone.utc).isoformat(),
+        "source": str(path),
+        "destination": str(dest),
+        "reason": reason,
+    }
+    write_report(f"quarantine-{int(time.time())}-{dest.name}.json", report)
+    log(f"quarantined {path} -> {dest}: {reason}")
+
+
+def romm_token() -> str:
+    if not (ROMM_SERVER_URL and ROMM_USERNAME and ROMM_PASSWORD):
+        raise RuntimeError("ROMM_SERVER_URL, ROMM_USERNAME, and ROMM_PASSWORD are required for scan trigger")
+    data = parse.urlencode({"username": ROMM_USERNAME, "password": ROMM_PASSWORD}).encode()
+    req = request.Request(
+        f"{ROMM_SERVER_URL}/api/token",
+        data=data,
+        method="POST",
+        headers={"Content-Type": "application/x-www-form-urlencoded"},
+    )
+    with request.urlopen(req, timeout=20) as resp:
+        body = json.loads(resp.read().decode("utf-8"))
+    token = body.get("access_token") or body.get("token")
+    if not token:
+        raise RuntimeError(f"RomM token response did not include access_token/token: {body}")
+    return token
+
+
+def trigger_romm_scan() -> None:
+    if not SCAN_AFTER_IMPORT:
+        return
+    token = romm_token()
+    req = request.Request(
+        f"{ROMM_SERVER_URL}/api/tasks/run/scan_library",
+        data=b"{}",
+        method="POST",
+        headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
+    )
+    try:
+        with request.urlopen(req, timeout=30) as resp:
+            body = resp.read().decode("utf-8", errors="replace")
+            log(f"triggered RomM scan_library: HTTP {resp.status} {body[:300]}")
+    except error.HTTPError as exc:
+        body = exc.read().decode("utf-8", errors="replace")
+        raise RuntimeError(f"RomM scan trigger failed: HTTP {exc.code} {body}") from exc
+
+
+def import_file(path: Path) -> bool:
+    platform = platform_for(path)
+    if not platform:
+        quarantine(path, "File was dropped directly into incoming. Put ROMs under incoming/<romm_platform_slug>/ so the worker does not guess the platform.")
+        return False
+
+    processing_dir = PROCESSING / platform
+    verified_dir = VERIFIED / platform
+    romm_dir = ROMM_ROMS / platform
+    for d in (processing_dir, verified_dir, romm_dir):
+        d.mkdir(parents=True, exist_ok=True)
+
+    processing_path = unique_dest(processing_dir / path.name)
+    shutil.move(str(path), str(processing_path))
+    log(f"processing {processing_path} as platform={platform}")
+
+    verified_path = unique_dest(verified_dir / processing_path.name)
+    shutil.copy2(str(processing_path), str(verified_path))
+
+    romm_path = romm_dir / processing_path.name
+    if romm_path.exists():
+        log(f"RomM destination already exists, skipping library copy: {romm_path}")
+        imported = False
+    else:
+        shutil.copy2(str(processing_path), str(romm_path))
+        imported = True
+        log(f"copied into RomM library: {romm_path}")
+
+    processing_path.unlink(missing_ok=True)
+
+    report = {
+        "time_utc": datetime.now(timezone.utc).isoformat(),
+        "platform": platform,
+        "incoming": str(path),
+        "verified": str(verified_path),
+        "romm_path": str(romm_path),
+        "imported": imported,
+        "validation": "staging-flow-only; add DAT-backed Igir validation before trusting unknown files",
+    }
+    write_report(f"import-{int(time.time())}-{romm_path.name}.json", report)
+
+    if imported:
+        trigger_romm_scan()
+    return imported
+
+
+def main() -> int:
+    ensure_dirs()
+    log(f"RomM import worker started; incoming={INCOMING}; roms={ROMM_ROMS}; poll={POLL_SECONDS}s; scan_after_import={SCAN_AFTER_IMPORT}")
+    while True:
+        try:
+            files = iter_incoming_files()
+            if not files and not QUIET_IDLE:
+                log("no incoming files")
+            for path in files:
+                try:
+                    import_file(path)
+                except Exception as exc:  # keep worker alive, quarantine when possible
+                    log(f"ERROR processing {path}: {exc}")
+                    traceback.print_exc()
+                    if path.exists() and path.is_file():
+                        try:
+                            quarantine(path, f"worker error: {exc}")
+                        except Exception:
+                            traceback.print_exc()
+        except Exception:
+            traceback.print_exc()
+        time.sleep(POLL_SECONDS)
+
+
+if __name__ == "__main__":
+    sys.exit(main())

romm-automation-truenas-compose.yml

@@ -173,8 +173,17 @@ services:
     restart: unless-stopped
     profiles: ["validation"]
     working_dir: /work
-    command: ["sh", "-c", "apk add --no-cache python3 make g++ && npm install -g igir@latest && sleep infinity"]
+    command: ["sh", "-c", "apk add --no-cache python3 make g++ && npm install -g igir@latest && python3 /work/romm-automation-import-worker.py"]
+    environment:
+      TZ: ${TZ:-America/Chicago}
+      ROMM_SERVER_URL: ${ROMM_SERVER_URL:?set ROMM_SERVER_URL in .env}
+      ROMM_USERNAME: ${ROMM_USERNAME:?set ROMM_USERNAME in .env}
+      ROMM_PASSWORD: ${ROMM_PASSWORD:?set ROMM_PASSWORD in .env}
+      ROMM_SCAN_AFTER_IMPORT: ${ROMM_SCAN_AFTER_IMPORT:-true}
+      IMPORT_POLL_SECONDS: ${IMPORT_POLL_SECONDS:-60}
+      IMPORT_QUIET_IDLE: ${IMPORT_QUIET_IDLE:-true}
     volumes:
+      - ./romm-automation-import-worker.py:/work/romm-automation-import-worker.py:ro
       - ${TRUENAS_DOCKER_ROOT:-/mnt/HomeStorage02/Docker}/RomAutomation/staging:/staging
       - ${TRUENAS_DOCKER_ROOT:-/mnt/HomeStorage02/Docker}/RomAutomation/dat:/dat:ro
       - ${TRUENAS_DOCKER_ROOT:-/mnt/HomeStorage02/Docker}/RomAutomation/reports:/reports

romm-automation-truenas.env.example

@@ -23,6 +23,13 @@ ROMARR_PORT=9797
 # Valid values: romarr,discord,validation or comma-separated combinations.
 COMPOSE_PROFILES=validation
 
+# Import worker: watches staging/incoming/<romm_platform_slug>/, imports into RomM,
+# then triggers RomM's scan_library task. Keep immediate scan enabled while testing;
+# later we can debounce/batch scans if needed.
+ROMM_SCAN_AFTER_IMPORT=true
+IMPORT_POLL_SECONDS=60
+IMPORT_QUIET_IDLE=true
+
 # GG Requestz support services
 GGREQUESTZ_POSTGRES_DB=ggrequestz
 GGREQUESTZ_POSTGRES_USER=ggrequestz