398 lines
14 KiB
Python
398 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""Simple RomM staging import worker.
|
|
|
|
Watches a single landing folder, /staging/incoming, for ROM files, infers the
|
|
RomM platform slug when it can do so safely, moves files through
|
|
processing/verified, copies them into /romm/library/roms/<platform_slug>/, and
|
|
then triggers RomM's library scan API.
|
|
|
|
The worker intentionally quarantines files when the platform cannot be inferred
|
|
with reasonable confidence. Add mappings with IMPORT_PLATFORM_MAP_JSON if needed.
|
|
Example:
|
|
IMPORT_PLATFORM_MAP_JSON={".iso":"ngc",".rvz":"ngc"}
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
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", "")
|
|
ROMM_API_TOKEN = os.environ.get("ROMM_API_TOKEN", "")
|
|
ROMM_SCAN_TASK = os.environ.get("ROMM_SCAN_TASK", "socket_quick_scan")
|
|
ROMM_SCAN_APIS = [api.strip() for api in os.environ.get("ROMM_SCAN_APIS", "igdb").split(",") if api.strip()]
|
|
|
|
IGNORE_SUFFIXES = {".part", ".partial", ".tmp", ".crdownload"}
|
|
|
|
# Safe extension mappings that are normally unambiguous for RomM platform slugs.
|
|
EXTENSION_PLATFORM_MAP: dict[str, str] = {
|
|
".nes": "nes",
|
|
".fds": "fds",
|
|
".sfc": "snes",
|
|
".smc": "snes",
|
|
".gb": "gb",
|
|
".gbc": "gbc",
|
|
".gba": "gba",
|
|
".n64": "n64",
|
|
".z64": "n64",
|
|
".v64": "n64",
|
|
".ndS".lower(): "nds",
|
|
".3ds": "3ds",
|
|
".cia": "3ds",
|
|
".gcm": "ngc",
|
|
".rvz": "ngc",
|
|
".wbfs": "wii",
|
|
".wad": "wii",
|
|
}
|
|
|
|
# Extensions that require header detection or an explicit override because they
|
|
# can belong to many systems.
|
|
AMBIGUOUS_EXTENSIONS = {".iso", ".bin", ".cue", ".chd", ".zip", ".7z", ".rar"}
|
|
|
|
try:
|
|
user_map = json.loads(os.environ.get("IMPORT_PLATFORM_MAP_JSON", "{}") or "{}")
|
|
if isinstance(user_map, dict):
|
|
for ext, platform in user_map.items():
|
|
if isinstance(ext, str) and isinstance(platform, str) and ext and platform:
|
|
normalized_ext = ext.lower() if ext.startswith(".") else f".{ext.lower()}"
|
|
EXTENSION_PLATFORM_MAP[normalized_ext] = platform.strip()
|
|
except json.JSONDecodeError as exc:
|
|
print(f"WARNING: failed to parse IMPORT_PLATFORM_MAP_JSON: {exc}", flush=True)
|
|
|
|
|
|
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 []
|
|
# Single landing folder mode: process files dropped directly in incoming.
|
|
# Ignore files already inside subdirectories so processing/verified layouts do
|
|
# not accidentally get re-imported if someone creates folders in incoming.
|
|
return sorted([p for p in INCOMING.iterdir() if is_ready_file(p)])
|
|
|
|
|
|
def detect_gamecube_or_wii(path: Path) -> str | None:
|
|
"""Return ngc/wii for disc images with Nintendo magic, otherwise None."""
|
|
try:
|
|
with path.open("rb") as fh:
|
|
fh.seek(0x18)
|
|
raw = fh.read(8)
|
|
if len(raw) < 8:
|
|
return None
|
|
# GameCube magic at 0x1C: 0xC2339F3D. Wii magic at 0x18: 0x5D1C9EA3.
|
|
wii_magic = struct.unpack(">I", raw[0:4])[0]
|
|
gc_magic = struct.unpack(">I", raw[4:8])[0]
|
|
if gc_magic == 0xC2339F3D:
|
|
return "ngc"
|
|
if wii_magic == 0x5D1C9EA3:
|
|
return "wii"
|
|
except OSError:
|
|
return None
|
|
return None
|
|
|
|
|
|
def platform_for(path: Path) -> tuple[str | None, str]:
|
|
suffix = path.suffix.lower()
|
|
|
|
detected = None
|
|
if suffix in {".iso", ".gcm", ".rvz", ".wbfs"}:
|
|
detected = detect_gamecube_or_wii(path)
|
|
if detected:
|
|
return detected, f"Nintendo disc image header detected as {detected}"
|
|
|
|
if suffix in EXTENSION_PLATFORM_MAP:
|
|
platform = EXTENSION_PLATFORM_MAP[suffix]
|
|
if suffix in AMBIGUOUS_EXTENSIONS:
|
|
return platform, f"explicit/import env extension override {suffix}->{platform}"
|
|
return platform, f"extension mapping {suffix}->{platform}"
|
|
|
|
if suffix in AMBIGUOUS_EXTENSIONS:
|
|
return None, f"ambiguous extension {suffix}; add IMPORT_PLATFORM_MAP_JSON override or use a less ambiguous file format"
|
|
|
|
return None, f"unknown extension {suffix or '<none>'}; add IMPORT_PLATFORM_MAP_JSON override if this is expected"
|
|
|
|
|
|
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)
|
|
safe_name = "".join(c if c.isalnum() or c in "-. _" else "_" for c in name)
|
|
path = REPORTS / safe_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:
|
|
"""Return a RomM bearer token for task calls.
|
|
|
|
Prefer ROMM_API_TOKEN because RomM task execution requires the tasks.run
|
|
scope. Username/password logins can authenticate successfully while still
|
|
lacking that scope, which causes HTTP 403 on /api/tasks/run/scan_library.
|
|
"""
|
|
if ROMM_API_TOKEN:
|
|
return ROMM_API_TOKEN
|
|
|
|
if not (ROMM_SERVER_URL and ROMM_USERNAME and ROMM_PASSWORD):
|
|
raise RuntimeError(
|
|
"ROMM_SERVER_URL plus either ROMM_API_TOKEN or ROMM_USERNAME/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_via_task_api(task_name: str) -> bool:
|
|
token = romm_token()
|
|
req = request.Request(
|
|
f"{ROMM_SERVER_URL}/api/tasks/run/{task_name}",
|
|
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 task {task_name}: HTTP {resp.status} {body[:300]}")
|
|
return True
|
|
except error.HTTPError as exc:
|
|
body = exc.read().decode("utf-8", errors="replace")
|
|
if exc.code == 403:
|
|
raise RuntimeError(
|
|
"RomM scan trigger failed: HTTP 403 Forbidden. The RomM credential/token can log in, "
|
|
"but it does not have the tasks.run scope required to run tasks. "
|
|
"Create a RomM API token with Run tasks/tasks.run permission and set ROMM_API_TOKEN."
|
|
) from exc
|
|
log(f"RomM task API trigger {task_name} did not run: HTTP {exc.code} {body[:300]}")
|
|
return False
|
|
|
|
|
|
def trigger_romm_scan_via_socket() -> None:
|
|
"""Trigger the same Socket.IO quick scan used by RomM's web UI.
|
|
|
|
Current RomM versions expose /api/tasks/run/scan_library, but that scheduled
|
|
task is not manual-runnable unless RomM's scheduled rescan is enabled. The web
|
|
UI manual scan uses the /ws/socket.io "scan" event instead.
|
|
"""
|
|
if not ROMM_SERVER_URL:
|
|
raise RuntimeError("ROMM_SERVER_URL is required for socket quick scan")
|
|
|
|
options = {
|
|
"platforms": [],
|
|
"type": "quick",
|
|
"roms_ids": [],
|
|
"apis": ROMM_SCAN_APIS,
|
|
"launchbox_remote_enabled": True,
|
|
}
|
|
node_script = r"""
|
|
const { io } = require('socket.io-client');
|
|
const url = process.env.ROMM_SERVER_URL;
|
|
const options = JSON.parse(process.env.ROMM_SOCKET_SCAN_OPTIONS || '{}');
|
|
const socket = io(url, { path: '/ws/socket.io', transports: ['websocket'], timeout: 10000 });
|
|
socket.on('connect', () => {
|
|
console.log('socket_connected');
|
|
socket.emit('scan', options);
|
|
setTimeout(() => { console.log('socket_scan_emitted'); socket.close(); process.exit(0); }, 2000);
|
|
});
|
|
socket.on('connect_error', (err) => {
|
|
console.error('socket_connect_error: ' + err.message);
|
|
process.exit(2);
|
|
});
|
|
"""
|
|
with tempfile.NamedTemporaryFile("w", suffix=".js", delete=False) as fh:
|
|
fh.write(node_script)
|
|
script_path = fh.name
|
|
try:
|
|
env = os.environ.copy()
|
|
env["ROMM_SOCKET_SCAN_OPTIONS"] = json.dumps(options)
|
|
try:
|
|
npm_root = subprocess.check_output(["npm", "root", "-g"], text=True, timeout=10).strip()
|
|
env["NODE_PATH"] = npm_root
|
|
except Exception:
|
|
pass
|
|
result = subprocess.run(
|
|
["node", script_path],
|
|
env=env,
|
|
text=True,
|
|
capture_output=True,
|
|
timeout=30,
|
|
check=False,
|
|
)
|
|
output = (result.stdout + result.stderr).strip().replace("\n", " | ")
|
|
if result.returncode != 0:
|
|
raise RuntimeError(f"RomM socket quick scan failed rc={result.returncode}: {output}")
|
|
log(f"triggered RomM socket quick scan: {output}")
|
|
finally:
|
|
try:
|
|
Path(script_path).unlink(missing_ok=True)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def trigger_romm_scan() -> None:
|
|
if not SCAN_AFTER_IMPORT:
|
|
return
|
|
|
|
if ROMM_SCAN_TASK and ROMM_SCAN_TASK != "socket_quick_scan":
|
|
if trigger_romm_scan_via_task_api(ROMM_SCAN_TASK):
|
|
return
|
|
|
|
# Preferred/default for this RomM install: mirrors the UI's manual quick scan.
|
|
trigger_romm_scan_via_socket()
|
|
|
|
|
|
def import_file(path: Path) -> bool:
|
|
platform, platform_reason = platform_for(path)
|
|
if not platform:
|
|
quarantine(path, f"Could not infer RomM platform from single-folder incoming mode: {platform_reason}")
|
|
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} ({platform_reason})")
|
|
|
|
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,
|
|
"platform_reason": platform_reason,
|
|
"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; single-folder 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())
|