Files
homelabstack/stacks/romhacks/orchestrator/analyze_catalog.py
T

131 lines
4.6 KiB
Python

#!/usr/bin/env python3
"""Analysis pass for the Obsidian catalog publisher (read-only).
For every routed library/<slug>/ game, derive its catalog fields:
- name, platform (UPPERCASE), kind, base, version, status, generation,
library_path, source link, #guides, #art.
Generation comes from which forum the thread lived in (exports/<channelId>/);
base/version/status are parsed from the staged description.
Emits one JSON object per line (so the caller can dedupe against the existing
vault notes) plus a summary. Writes nothing outside stdout.
"""
import json
import os
import pathlib
import re
LIBRARY = pathlib.Path(os.environ.get("LIBRARY_DIR", "/library"))
METADATA = pathlib.Path(os.environ.get("METADATA_DIR", "/metadata"))
EXPORTS = pathlib.Path(os.environ.get("EXPORTS_DIR", "/exports"))
CHANNEL_GEN = {
"1474174360926290105": "Gen I", "1474174532196761802": "Gen II",
"1474174608587489416": "Gen III", "1474174672210755714": "Gen IV",
"1474174736438132756": "Gen V", "1474174789680632084": "Gen VI",
"1474174859671240898": "Gen VII", "1474174950138183880": "Gen VIII",
"1474175004248903935": "Gen IX", "1474175244209098986": "Joiplay/RPGXP",
"1475605095805747351": "Unique",
}
PLAT_UP = {"gba": "GBA", "gbc": "GBC", "gb": "GB", "nds": "NDS",
"3ds": "3DS", "switch": "Switch", "pc": "PC", "patch": "Patch",
"unknown": "—"}
def norm(name):
n = name.lower()
n = re.sub(r"pok[eé]mon", "", n)
n = re.sub(r"[^a-z0-9]+", "", n)
return n
def gen_map():
m = {}
for chan_dir in (EXPORTS.iterdir() if EXPORTS.exists() else []):
if not chan_dir.is_dir():
continue
gen = CHANNEL_GEN.get(chan_dir.name)
if not gen:
continue
for jf in chan_dir.glob("*.json"):
try:
d = json.loads(jf.read_text(encoding="utf-8"))
nm = d.get("channel", {}).get("name")
if nm:
m[norm(nm)] = gen
except Exception:
pass
return m
def parse_desc(desc):
base = ver = status = "—"
m = re.search(r"based on\s*:?\s*([A-Za-z0-9 .&'+-]{2,40})", desc, re.I)
if m:
base = m.group(1).strip().rstrip(".").strip()
m = re.search(r"version\s*:?\s*([^\n]{1,30})", desc, re.I)
if m:
ver = m.group(1).strip()
low = desc.lower()
if "completed" in low or re.search(r"\bcomplete\b", low):
status = "Complete"
elif "beta" in low:
status = "Beta"
elif "ongoing" in low or "in development" in low:
status = "Ongoing"
return base, ver, status
def main():
gens = gen_map()
rows = []
for folder in sorted(p for p in LIBRARY.iterdir() if p.is_dir()):
hj = folder / "handoff.json"
if not hj.exists():
continue
h = json.loads(hj.read_text(encoding="utf-8"))
name = h.get("name", folder.name)
slug = h.get("slug", folder.name)
plat = PLAT_UP.get(h.get("platform", "unknown"), "—")
desc = ""
mp = METADATA / slug / "metadata.json"
if mp.exists():
try:
desc = json.loads(mp.read_text(encoding="utf-8")).get("description", "")
except Exception:
pass
base, ver, status = parse_desc(desc)
gen = gens.get(norm(name), "—")
art = h.get("art", [])
guides = h.get("guides", [])
links = h.get("links", [])
artifact = h.get("artifact")
kind = h.get("kind")
lib_path = (f"/storage1/Emulation/roms/{h.get('platform')}/{artifact}"
if kind == "console" and artifact else
f"/storage1/labdata/romhacks/library/{slug}/")
rows.append({
"name": name, "slug": slug, "norm": norm(name), "platform": plat,
"kind": kind, "base": base, "version": ver, "status": status,
"generation": gen, "library_path": lib_path,
"source": links[0] if links else None, "n_links": len(links),
"n_guides": len(guides), "n_art": len(art), "artifact": artifact,
})
for r in rows:
print(json.dumps(r, ensure_ascii=False))
# summary to stderr-ish (still stdout, prefixed)
from collections import Counter
pc = Counter(r["platform"] for r in rows)
kc = Counter(r["kind"] for r in rows)
print("@@@SUMMARY", json.dumps({
"total": len(rows), "by_platform": dict(pc), "by_kind": dict(kc),
"with_base": sum(1 for r in rows if r["base"] != "—"),
"with_gen": sum(1 for r in rows if r["generation"] != "—"),
}, ensure_ascii=False))
if __name__ == "__main__":
main()