feat(romhacks): automated hand-off — route downloads to RomM + library, render notes
handoff.py classifies completed downloads (console rom vs RPG-Maker fangame vs patch, via bsdtar archive peek incl. RAR5), copies console roms into Emulation/roms/<platform> for RomM, moves the full folder (rom + guides) into the library archive, and renders note.md + handoff.json per hack for the Obsidian publish step. Wired into the cycle; single /data mount keeps moves as fast renames. Validated on 4 real games.
This commit is contained in:
ginnoir
@@ -0,0 +1,247 @@
|
||||
#!/usr/bin/env python3
|
||||
"""romhacks hand-off — route completed downloads into the library + RomM, and
|
||||
render an Obsidian note per hack.
|
||||
|
||||
For each *completed* `incoming/<slug>/` folder (no in-progress JDownloader files):
|
||||
1. Classify it — console rom (by file/archive extension) vs RPG-Maker style
|
||||
fan-game (Game.exe / rgss / rxdata) vs patch-only.
|
||||
2. Route — console roms get the rom artifact copied into
|
||||
`/storage1/Emulation/<platform>/` so RomM scans it; fan-games stay out of
|
||||
RomM. The WHOLE folder (rom + guides + spreadsheets) is moved to
|
||||
`/storage1/labdata/romhacks/library/<slug>/` as the permanent archive
|
||||
(folder contents are a feature, not noise — keep them all).
|
||||
3. Render `note.md` + `handoff.json` into the library folder for the Obsidian
|
||||
publish step (done separately via the Obsidian MCP — livesync's chunked
|
||||
CouchDB format isn't safe to write from here).
|
||||
|
||||
Idempotent via a handled-slug state file. HANDOFF_DRY=true (default) classifies
|
||||
and renders to a scratch dir but moves/copies nothing.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import pathlib
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
INCOMING = pathlib.Path(os.environ.get("INCOMING_DIR", "/incoming"))
|
||||
LIBRARY = pathlib.Path(os.environ.get("LIBRARY_DIR", "/library"))
|
||||
EMU = pathlib.Path(os.environ.get("EMULATION_DIR", "/emulation"))
|
||||
METADATA = pathlib.Path(os.environ.get("METADATA_DIR", "/metadata"))
|
||||
HANDLED = pathlib.Path(os.environ.get("HANDOFF_STATE", "/state/handled.json"))
|
||||
DRY = os.environ.get("HANDOFF_DRY", "true").lower() != "false"
|
||||
|
||||
ROM_EXT_PLATFORM = {
|
||||
".gba": "gba", ".gb": "gb", ".gbc": "gbc", ".nds": "nds",
|
||||
".3ds": "3ds", ".cia": "3ds", ".nsp": "switch", ".xci": "switch",
|
||||
}
|
||||
ARCHIVE_EXTS = {".zip", ".rar", ".7z"}
|
||||
PATCH_EXTS = {".ips", ".bps", ".ups", ".xdelta"}
|
||||
INPROGRESS = (".part", ".jdownload", ".jdownload.cfg", ".tmp", ".crdownload")
|
||||
FANGAME_HINTS = ("game.exe", "game.ini", ".rxdata", ".rvdata", ".rvdata2", ".rgssad", ".rgss2a", ".rgss3a")
|
||||
ART_EXTS = (".png", ".jpg", ".jpeg", ".webp", ".gif")
|
||||
# Document / extra files worth keeping (guides, spreadsheets, changelogs, and
|
||||
# supplementary or alternate-version archives). Deliberately excludes raw game
|
||||
# assets (png/audio/scripts) so an extracted tree isn't counted as "guides".
|
||||
GUIDE_EXTS = {".html", ".htm", ".pdf", ".xlsx", ".xls", ".ods", ".csv", ".doc",
|
||||
".docx", ".txt", ".md", ".rtf", ".zip", ".rar", ".7z"}
|
||||
|
||||
|
||||
def slug(s):
|
||||
return re.sub(r"[^A-Za-z0-9._-]+", "_", s or "").strip("_")[:120] or "unknown"
|
||||
|
||||
|
||||
def load_handled():
|
||||
try:
|
||||
return set(json.loads(HANDLED.read_text())["handled"])
|
||||
except Exception:
|
||||
return set()
|
||||
|
||||
|
||||
def save_handled(handled):
|
||||
HANDLED.parent.mkdir(parents=True, exist_ok=True)
|
||||
HANDLED.write_text(json.dumps({"handled": sorted(handled)}))
|
||||
|
||||
|
||||
def is_complete(folder):
|
||||
for p in folder.rglob("*"):
|
||||
if p.is_file() and p.name.lower().endswith(INPROGRESS):
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def archive_listing(path):
|
||||
"""Return lowercased entry names inside an archive. bsdtar (libarchive) lists
|
||||
zip/rar/7z uniformly, including RAR5 which the Alpine 7zip build can't read."""
|
||||
try:
|
||||
out = subprocess.run(["bsdtar", "-tf", str(path)], capture_output=True,
|
||||
text=True, timeout=180)
|
||||
return out.stdout.lower()
|
||||
except Exception:
|
||||
return ""
|
||||
|
||||
|
||||
def classify(folder):
|
||||
"""Return (kind, platform, primary_artifact). kind in console|fangame|patch|unknown."""
|
||||
files = [p for p in folder.rglob("*") if p.is_file()]
|
||||
# 1. Loose rom file wins outright.
|
||||
for p in files:
|
||||
plat = ROM_EXT_PLATFORM.get(p.suffix.lower())
|
||||
if plat:
|
||||
return "console", plat, p
|
||||
# 2. Fan-game markers (loose).
|
||||
for p in files:
|
||||
name = p.name.lower()
|
||||
if name in ("game.exe", "game.ini") or p.suffix.lower() in FANGAME_HINTS:
|
||||
archive = max((a for a in files if a.suffix.lower() in ARCHIVE_EXTS),
|
||||
key=lambda a: a.stat().st_size, default=None)
|
||||
return "fangame", "pc", archive
|
||||
# 3. Peek inside archives (largest first).
|
||||
archives = sorted((p for p in files if p.suffix.lower() in ARCHIVE_EXTS),
|
||||
key=lambda a: a.stat().st_size, reverse=True)
|
||||
for a in archives:
|
||||
listing = archive_listing(a)
|
||||
if any(h in listing for h in ("game.exe", "rgssad", ".rxdata", ".rvdata")):
|
||||
return "fangame", "pc", a
|
||||
for ext, plat in ROM_EXT_PLATFORM.items():
|
||||
if ext in listing:
|
||||
return "console", plat, a
|
||||
# 4. Patch-only.
|
||||
if any(p.suffix.lower() in PATCH_EXTS for p in files):
|
||||
patch = max((p for p in files if p.suffix.lower() in PATCH_EXTS),
|
||||
key=lambda p: p.stat().st_size)
|
||||
return "patch", "patch", patch
|
||||
# 5. Fall back to the biggest archive as the artifact, platform unknown.
|
||||
return "unknown", "unknown", (archives[0] if archives else None)
|
||||
|
||||
|
||||
def collect_guides(folder, primary):
|
||||
"""Files that aren't the primary artifact or art — the guides/spreadsheets."""
|
||||
guides = []
|
||||
for p in folder.rglob("*"):
|
||||
if not p.is_file() or p == primary:
|
||||
continue
|
||||
if p.suffix.lower() in GUIDE_EXTS:
|
||||
guides.append(p)
|
||||
return guides
|
||||
|
||||
|
||||
def render_note(meta, kind, platform, primary, guides, art, folder):
|
||||
name = meta.get("name", folder.name)
|
||||
lines = [
|
||||
"---",
|
||||
f"title: {name}",
|
||||
f"platform: {platform}",
|
||||
f"kind: {kind}",
|
||||
"tags: [pokemon-romhack]",
|
||||
"---",
|
||||
"",
|
||||
f"# {name}",
|
||||
"",
|
||||
]
|
||||
for a in art:
|
||||
lines.append(f"![[{a.name}]]")
|
||||
if art:
|
||||
lines.append("")
|
||||
desc = (meta.get("description") or "").strip()
|
||||
if desc:
|
||||
lines += ["## Description", "", desc, ""]
|
||||
links = meta.get("links") or []
|
||||
if links:
|
||||
lines += ["## Download links", ""]
|
||||
lines += [f"- {l}" for l in links]
|
||||
lines.append("")
|
||||
lines += ["## Files", "",
|
||||
f"- **Artifact:** `{primary.name if primary else 'n/a'}` ({kind}/{platform})"]
|
||||
if guides:
|
||||
lines.append(f"- **Guides / extras ({len(guides)}):**")
|
||||
for g in sorted(guides):
|
||||
lines.append(f" - `{g.relative_to(folder)}`")
|
||||
lines.append("")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def route(folder, slug_name, kind, platform, primary):
|
||||
dest_lib = LIBRARY / slug_name
|
||||
# console roms become RomM-scannable; fan-games/patches stay in the library only.
|
||||
if kind == "console" and primary is not None:
|
||||
plat_dir = EMU / platform
|
||||
if not DRY:
|
||||
plat_dir.mkdir(parents=True, exist_ok=True)
|
||||
shutil.copy2(primary, plat_dir / primary.name)
|
||||
if not DRY:
|
||||
LIBRARY.mkdir(parents=True, exist_ok=True)
|
||||
if dest_lib.exists():
|
||||
shutil.rmtree(dest_lib)
|
||||
shutil.move(str(folder), str(dest_lib))
|
||||
return dest_lib
|
||||
|
||||
|
||||
def process(folder, handled):
|
||||
slug_name = folder.name
|
||||
if slug_name in handled:
|
||||
return None
|
||||
if not is_complete(folder):
|
||||
print(f"[handoff] skip (in progress): {slug_name}")
|
||||
return None
|
||||
meta = {}
|
||||
mp = METADATA / slug_name / "metadata.json"
|
||||
if mp.exists():
|
||||
try:
|
||||
meta = json.loads(mp.read_text())
|
||||
except Exception:
|
||||
pass
|
||||
art = sorted(p for p in (METADATA / slug_name).glob("*") if p.suffix.lower() in ART_EXTS) \
|
||||
if (METADATA / slug_name).exists() else []
|
||||
|
||||
kind, platform, primary = classify(folder)
|
||||
guides = collect_guides(folder, primary)
|
||||
note = render_note(meta, kind, platform, primary, guides, art, folder)
|
||||
|
||||
print(f"[handoff] {slug_name}: kind={kind} platform={platform} "
|
||||
f"artifact={primary.name if primary else 'NONE'} guides={len(guides)} art={len(art)}")
|
||||
|
||||
if DRY:
|
||||
scratch = pathlib.Path("/tmp/handoff-preview") / slug_name
|
||||
scratch.mkdir(parents=True, exist_ok=True)
|
||||
(scratch / "note.md").write_text(note, encoding="utf-8")
|
||||
(scratch / "handoff.json").write_text(json.dumps(
|
||||
{"name": meta.get("name", slug_name), "slug": slug_name, "kind": kind,
|
||||
"platform": platform, "artifact": primary.name if primary else None,
|
||||
"guides": [str(g.relative_to(folder)) for g in guides],
|
||||
"art": [a.name for a in art], "links": meta.get("links", [])}, indent=2), encoding="utf-8")
|
||||
return None
|
||||
|
||||
# Snapshot guide paths relative to the folder BEFORE moving it; the same
|
||||
# relative layout holds under dest_lib after the move.
|
||||
guide_rels = [str(g.relative_to(folder)) for g in guides]
|
||||
dest_lib = route(folder, slug_name, kind, platform, primary)
|
||||
for a in art:
|
||||
shutil.copy2(a, dest_lib / a.name)
|
||||
(dest_lib / "note.md").write_text(note, encoding="utf-8")
|
||||
(dest_lib / "handoff.json").write_text(json.dumps(
|
||||
{"name": meta.get("name", slug_name), "slug": slug_name, "kind": kind,
|
||||
"platform": platform, "artifact": primary.name if primary else None,
|
||||
"guides": guide_rels, "art": [a.name for a in art],
|
||||
"links": meta.get("links", [])}, indent=2), encoding="utf-8")
|
||||
handled.add(slug_name)
|
||||
return slug_name
|
||||
|
||||
|
||||
def main():
|
||||
if not INCOMING.exists():
|
||||
print(f"[handoff] no incoming dir {INCOMING}")
|
||||
return
|
||||
handled = load_handled()
|
||||
done = 0
|
||||
for folder in sorted(p for p in INCOMING.iterdir() if p.is_dir()):
|
||||
if process(folder, handled):
|
||||
done += 1
|
||||
if not DRY:
|
||||
save_handled(handled)
|
||||
print(f"[handoff] done — {done} game(s) routed; DRY={DRY}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user