Deploy to valhalla / deploy (push) Has been cancelled
- proxy: read-only bind /storage1/labdata/romhacks/library -> /srv/romhacks - Caddyfile: romhacks-files.ginnoir.com (internal_only) file_server + browse - render_catalog_notes.py + handoff.py: embed box art and link guides/ spreadsheets via https://romhacks-files.ginnoir.com/<slug>/<file> (handoff art/guides; guide filter drops buried game-data txt, caps at 30) - regenerate bookmarks
303 lines
12 KiB
Python
303 lines
12 KiB
Python
#!/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
|
|
from urllib.parse import quote
|
|
|
|
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")
|
|
# Internal-only Caddy file_server over the library (romhacks-files.ginnoir.com →
|
|
# /srv/romhacks). Lets the Obsidian notes embed art / link guides off /storage1.
|
|
FILE_BASE = os.environ.get("FILE_BASE", "https://romhacks-files.ginnoir.com")
|
|
# Document/spreadsheet extensions that make a genuine guide (a subset of
|
|
# GUIDE_EXTS — bare game-data .txt under PBS/Graphics/Plugins is not a guide).
|
|
DOC_EXTS = {".html", ".htm", ".pdf", ".xlsx", ".xls", ".ods", ".csv",
|
|
".doc", ".docx", ".rtf"}
|
|
GAME_DATA_DIRS = {"pbs", "graphics", "plugins", "audio", "data", "fonts",
|
|
"scripts", "optional", "se", "bgm", "tilesets", "sound",
|
|
"movies", "backup"}
|
|
MAX_GUIDES = 30
|
|
|
|
|
|
def file_url(slug_name, rel):
|
|
"""URL into the served library, encoding each path segment."""
|
|
parts = [quote(slug_name)] + [quote(seg) for seg in str(rel).split("/")]
|
|
return FILE_BASE + "/" + "/".join(parts)
|
|
|
|
|
|
def pick_guide_rels(guide_rels, artifact_name):
|
|
"""Filter raw guide rel-paths to linkable docs/spreadsheets/extras."""
|
|
art_l = (artifact_name or "").lower()
|
|
scored = []
|
|
for rel in guide_rels:
|
|
low = rel.lower()
|
|
parts = low.split("/")
|
|
bn = parts[-1]
|
|
if bn in ("note.md", "handoff.json"):
|
|
continue
|
|
ext = os.path.splitext(bn)[1]
|
|
in_gamedata = any(seg in GAME_DATA_DIRS for seg in parts[:-1])
|
|
if ext in DOC_EXTS:
|
|
if in_gamedata:
|
|
continue
|
|
scored.append((0, rel))
|
|
elif ext in (".txt", ".md"):
|
|
if in_gamedata or len(parts) > 2:
|
|
continue
|
|
scored.append((1, rel))
|
|
elif ext in ARCHIVE_EXTS:
|
|
if bn == art_l:
|
|
continue
|
|
scored.append((2, rel))
|
|
scored.sort(key=lambda t: (t[0], t[1].lower()))
|
|
return [rel for _, rel in scored]
|
|
# 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, slug_name):
|
|
name = meta.get("name", folder.name)
|
|
lines = [
|
|
"---",
|
|
f"title: {name}",
|
|
f"platform: {platform}",
|
|
f"kind: {kind}",
|
|
"tags: [pokemon-romhack]",
|
|
"---",
|
|
"",
|
|
f"# {name}",
|
|
"",
|
|
]
|
|
# Box art — art is copied into the library folder root, so <slug>/<name>
|
|
# resolves on the file-server.
|
|
for a in art:
|
|
lines.append(f"})")
|
|
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("")
|
|
guide_rels = [str(g.relative_to(folder)) for g in guides]
|
|
picked = pick_guide_rels(guide_rels, primary.name if primary else None)
|
|
if picked:
|
|
lines += ["## Guides & extras", ""]
|
|
for rel in picked[:MAX_GUIDES]:
|
|
lines.append(f"- [{rel}]({file_url(slug_name, rel)})")
|
|
if len(picked) > MAX_GUIDES:
|
|
lines.append(f"- …and {len(picked) - MAX_GUIDES} more "
|
|
f"(browse the [library folder]({file_url(slug_name, '')}))")
|
|
lines.append("")
|
|
lines += ["## Files", "",
|
|
f"- **Artifact:** `{primary.name if primary else 'n/a'}` ({kind}/{platform})",
|
|
f"- **Library:** [browse files]({file_url(slug_name, '')})"]
|
|
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, slug_name)
|
|
|
|
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()
|