From 9227b260bd77a84b3515f9d2f099b3f76a58c444 Mon Sep 17 00:00:00 2001 From: wheelz Date: Wed, 24 Jun 2026 18:54:26 +0000 Subject: [PATCH] feat: add RomM staging import worker --- romm-automation-import-worker.py | 230 ++++++++++++++++++++++++++++ romm-automation-truenas-compose.yml | 11 +- romm-automation-truenas.env.example | 7 + 3 files changed, 247 insertions(+), 1 deletion(-) create mode 100644 romm-automation-import-worker.py diff --git a/romm-automation-import-worker.py b/romm-automation-import-worker.py new file mode 100644 index 0000000..dc8465c --- /dev/null +++ b/romm-automation-import-worker.py @@ -0,0 +1,230 @@ +#!/usr/bin/env python3 +"""Simple RomM staging import worker. + +Watches /staging/incoming// for files, moves them through +processing/verified, copies them into /romm/library/roms//, 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// 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()) diff --git a/romm-automation-truenas-compose.yml b/romm-automation-truenas-compose.yml index 35c703b..d45b94e 100644 --- a/romm-automation-truenas-compose.yml +++ b/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 diff --git a/romm-automation-truenas.env.example b/romm-automation-truenas.env.example index 0abbc0f..8f133e0 100644 --- a/romm-automation-truenas.env.example +++ b/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//, 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