281 lines
11 KiB
Python
281 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Import a drop-folder of mixed Pokémon hack files into the valhalla library.
|
|
|
|
Handles a folder containing any mix of:
|
|
- completed ROMs (.gba/.gbc/.gb/.nds) -> validated, clean-named, copied to roms/<plat>/Hacks/
|
|
- patches (.ips/.bps/.ups/.xdelta) -> applied to an owned base, placed in Hacks/, archived
|
|
- documentation (.pdf/.txt/.png/.md) -> archived under PATCHES/_docs/
|
|
- archives (.zip/.rar/.7z) -> extracted and recursed (same rules), source kept
|
|
|
|
Naming matches the rest of the library: "Pokemon - <Hack> (Hack).<ext>"
|
|
(version/junk parentheticals are stripped from the source filename).
|
|
|
|
Base ROM for a patch is auto-detected: BPS/UPS embed source CRC32 so the correct
|
|
base is the one that applies cleanly. IPS has no checksum, so IPS patches must be
|
|
named in IPS_BASE_MAP or they are reported unresolved (never blindly applied).
|
|
|
|
Run on valhalla: python3 romhack-import.py <incoming_dir> [--place]
|
|
"""
|
|
import sys, os, re, zlib, shutil, subprocess, zipfile, unicodedata, tempfile
|
|
|
|
ROMS = "/storage1/Emulation/roms"
|
|
PATCHES = "/storage1/igir/romhacks/patches"
|
|
APPLY = "/storage1/igir/romhacks/tools/apply.py"
|
|
|
|
BASES = {
|
|
"crystal_rev1": f"{ROMS}/gbc/Pokemon - Crystal Version (USA, Europe) (Rev 1).gbc",
|
|
"gold": f"{ROMS}/gbc/Pokemon - Gold Version (USA, Europe) (SGB Enhanced) (GB Compatible).gbc",
|
|
"silver": f"{ROMS}/gbc/Pokemon - Silver Version (USA, Europe) (SGB Enhanced) (GB Compatible).gbc",
|
|
"ruby": f"{ROMS}/gba/Pokemon - Ruby Version (USA, Europe).gba",
|
|
"sapphire": f"{ROMS}/gba/Pokemon - Sapphire Version (USA, Europe).gba",
|
|
"emerald": f"{ROMS}/gba/Pokemon - Emerald Version (USA, Europe).gba",
|
|
"firered": f"{ROMS}/gba/Pokemon - FireRed Version (USA, Europe).gba",
|
|
"leafgreen": f"{ROMS}/gba/Pokemon - LeafGreen Version (USA, Europe).gba",
|
|
"red": f"{ROMS}/gb/Pokemon - Red Version (USA, Europe) (SGB Enhanced).gb",
|
|
}
|
|
# candidate bases to try per target platform (order = preference)
|
|
CANDIDATES = {
|
|
"gba": ["emerald", "firered", "leafgreen", "ruby", "sapphire"],
|
|
"gbc": ["crystal_rev1", "gold", "silver"],
|
|
"gb": ["red"],
|
|
}
|
|
# IPS patches (no checksum) need an explicit base by name substring
|
|
IPS_BASE_MAP = [
|
|
("seaglass", "emerald", "gba"),
|
|
("crystal kaizo", "crystal_rev1", "gbc"),
|
|
("kaizo", "crystal_rev1", "gbc"), # fallback for crystal kaizo rar
|
|
]
|
|
|
|
ROM_EXT = {".gba": "gba", ".gbc": "gbc", ".gb": "gb", ".nds": "nds"}
|
|
PATCH_EXT = {".ips", ".bps", ".ups", ".xdelta"}
|
|
DOC_EXT = {".pdf", ".txt", ".png", ".md", ".jpg", ".jpeg", ".html",
|
|
".xlsx", ".docx", ".bmp", ".avi", ".gif"}
|
|
ARCHIVE_EXT = {".zip", ".rar", ".7z"}
|
|
|
|
# Clean display names for files whose filenames are messy (patch/zip artifacts).
|
|
# (substr-of-original-filename, lowercased) -> canonical hack name. First match wins.
|
|
NAME_OVERRIDES = [
|
|
("emeraldseaglass", "Emerald Seaglass"),
|
|
("emerald seaglass", "Emerald Seaglass"),
|
|
("lazarus", "Lazarus"),
|
|
("saiph 2", "Saiph 2"), ("saiph2", "Saiph 2"),
|
|
("saiph", "Saiph"),
|
|
("flora-sky", "Flora Sky"), ("flora sky", "Flora Sky"),
|
|
("sots", "Sovereign of the Skies"),
|
|
("sors", "Sors"),
|
|
("aesthetic red", "Aesthetic Red"),
|
|
("crystal kaizo", "Crystal Kaizo"),
|
|
]
|
|
# Files we will NOT auto-place (unidentified / need user to name). Reported instead.
|
|
SKIP_NAME_SUBSTR = ["beta 15 + expansion"]
|
|
|
|
PLACE = False
|
|
report = []
|
|
ARCH_SUBDIR = "" # set while recursing an archive, groups its docs under _docs/<subdir>/
|
|
|
|
|
|
def strip_trailing_groups(name):
|
|
"""Remove trailing (...) / [...] groups that look like version/variant junk."""
|
|
verkw = re.compile(r"(\d|version|hotfix|bug.?fix|final|anniversary|classic\+|"
|
|
r"beta|\brev\b|english|main|patch|suloku|pss|complete|\bhack\b)", re.I)
|
|
while name and name[-1] in ")]":
|
|
close = name[-1]; open_ch = "(" if close == ")" else "["
|
|
depth = 0; start = -1
|
|
for i in range(len(name) - 1, -1, -1):
|
|
if name[i] == close: depth += 1
|
|
elif name[i] == open_ch:
|
|
depth -= 1
|
|
if depth == 0: start = i; break
|
|
if start < 0: break
|
|
inner = name[start:]
|
|
if verkw.search(inner):
|
|
name = name[:start].rstrip()
|
|
else:
|
|
break
|
|
return name
|
|
|
|
|
|
def clean_name(fn):
|
|
base = os.path.basename(fn).lower()
|
|
for sub, canon in NAME_OVERRIDES:
|
|
if sub in base:
|
|
return canon
|
|
name = os.path.splitext(os.path.basename(fn))[0]
|
|
name = unicodedata.normalize("NFKD", name).encode("ascii", "ignore").decode()
|
|
name = name.replace("_", " ")
|
|
name = re.sub(r"^\s*pokemon\s*-\s*", "", name, flags=re.I)
|
|
name = re.sub(r"^\s*pokemon\s+", "", name, flags=re.I)
|
|
name = strip_trailing_groups(name)
|
|
# strip trailing version/junk tokens left outside parens
|
|
name = re.sub(r"\s*(full release|full version|completed|\bby .+|v\d[\d._]*|"
|
|
r"version|\d{4}|\d{1,2}-\d{1,2}-\d{2,4})\s*$", "", name, flags=re.I).strip(" -_")
|
|
return name or os.path.splitext(os.path.basename(fn))[0]
|
|
|
|
|
|
def is_skip_name(fn):
|
|
base = os.path.basename(fn).lower()
|
|
return any(s in base for s in SKIP_NAME_SUBSTR)
|
|
|
|
|
|
def crc32(path):
|
|
return zlib.crc32(open(path, "rb").read()) & 0xffffffff
|
|
|
|
|
|
def validate(path, plat):
|
|
with open(path, "rb") as f:
|
|
data = f.read()
|
|
n = len(data)
|
|
if plat == "gba":
|
|
if data[0xB2] != 0x96:
|
|
return "not a GBA ROM (byte 0xB2 != 0x96)", None
|
|
if n not in (4*1024*1024, 8*1024*1024, 16*1024*1024, 32*1024*1024):
|
|
for t in (8, 16, 32):
|
|
if n <= t*1024*1024:
|
|
return None, t*1024*1024 # needs pad to t MB
|
|
elif plat in ("gbc", "gb"):
|
|
if data[0x104:0x10A] != bytes.fromhex("ceed6666cc0d"):
|
|
return "GB/GBC Nintendo logo missing", None
|
|
elif plat == "nds":
|
|
if data[0x15C:0x15E] != b"\x56\xCF":
|
|
return "NDS logo CRC16 != 0xCF56", None
|
|
return None, None
|
|
|
|
|
|
def place_rom(src, plat, hackname, origin):
|
|
if is_skip_name(origin):
|
|
report.append(("ROM", "NEEDS-ID", origin, "unidentified hack — tell me the name", ""))
|
|
return
|
|
out_name = f"Pokemon - {hackname} (Hack).{plat}"
|
|
err, padto = validate(src, plat)
|
|
if err:
|
|
report.append(("ROM", "FAIL", origin, err, "")); return
|
|
data = open(src, "rb").read()
|
|
if padto and len(data) < padto:
|
|
data += b"\xff" * (padto - len(data))
|
|
if PLACE:
|
|
d = os.path.join(ROMS, plat, "Hacks"); os.makedirs(d, exist_ok=True)
|
|
with open(os.path.join(d, out_name), "wb") as f:
|
|
f.write(data)
|
|
report.append(("ROM", "OK", origin, f"{len(data):,}b crc={zlib.crc32(data)&0xffffffff:08x}",
|
|
f"{plat}/Hacks/{out_name}"))
|
|
|
|
|
|
def try_apply(patch, base_key, out):
|
|
base = BASES[base_key]
|
|
if not os.path.exists(base):
|
|
return False, "base file missing"
|
|
r = subprocess.run([sys.executable, APPLY, patch, base, out],
|
|
capture_output=True, text=True)
|
|
return (r.returncode == 0), (r.stdout or r.stderr).strip()
|
|
|
|
|
|
def place_patch(patch, origin):
|
|
ext = os.path.splitext(patch)[1].lower()
|
|
hackname = clean_name(patch)
|
|
# archive the patch file itself
|
|
if PLACE:
|
|
ad = os.path.join(PATCHES, "_imported"); os.makedirs(ad, exist_ok=True)
|
|
shutil.copy2(patch, os.path.join(ad, os.path.basename(patch)))
|
|
tmp_out = patch + ".out"
|
|
if ext == ".ips":
|
|
hit = next(((bk, pl) for sub, bk, pl in IPS_BASE_MAP if sub in patch.lower()), None)
|
|
if not hit:
|
|
report.append(("PATCH", "UNRESOLVED", origin, "IPS needs known base (not in map)", ""))
|
|
return
|
|
bk, plat = hit
|
|
ok, msg = try_apply(patch, bk, tmp_out)
|
|
if not ok:
|
|
report.append(("PATCH", "FAIL", origin, f"ips/{bk}: {msg[:50]}", "")); return
|
|
place_rom(tmp_out, plat, hackname, origin); return
|
|
if ext == ".xdelta":
|
|
report.append(("PATCH", "SKIP", origin, "xdelta: handle via xdelta tool", "")); return
|
|
# BPS / UPS: auto-detect base across all platforms (CRC-verified)
|
|
for plat, keys in CANDIDATES.items():
|
|
for bk in keys:
|
|
ok, msg = try_apply(patch, bk, tmp_out)
|
|
if ok:
|
|
place_rom(tmp_out, plat, hackname, origin + f" [base={bk}]")
|
|
return
|
|
report.append(("PATCH", "FAIL", origin, "no owned base matched (BPS/UPS CRC)", ""))
|
|
|
|
|
|
def archive_doc(path, origin):
|
|
sub = os.path.join("_docs", ARCH_SUBDIR) if ARCH_SUBDIR else "_docs"
|
|
rel = os.path.join(sub, os.path.basename(path))
|
|
if PLACE:
|
|
d = os.path.join(PATCHES, sub); os.makedirs(d, exist_ok=True)
|
|
shutil.copy2(path, os.path.join(PATCHES, rel))
|
|
report.append(("DOC", "OK", origin, f"{os.path.getsize(path):,}b", rel))
|
|
|
|
|
|
def process_file(path, origin=None):
|
|
origin = origin or os.path.basename(path)
|
|
ext = os.path.splitext(path)[1].lower()
|
|
if ext in ROM_EXT:
|
|
place_rom(path, ROM_EXT[ext], clean_name(path), origin)
|
|
elif ext in PATCH_EXT:
|
|
place_patch(path, origin)
|
|
elif ext in DOC_EXT:
|
|
archive_doc(path, origin)
|
|
elif ext in ARCHIVE_EXT:
|
|
process_archive(path, origin)
|
|
else:
|
|
report.append(("SKIP", "?", origin, f"unknown ext {ext}", ""))
|
|
|
|
|
|
def process_archive(path, origin):
|
|
# keep the source archive
|
|
if PLACE:
|
|
ad = os.path.join(PATCHES, "_archives"); os.makedirs(ad, exist_ok=True)
|
|
shutil.copy2(path, os.path.join(ad, os.path.basename(path)))
|
|
global ARCH_SUBDIR
|
|
tmp = tempfile.mkdtemp(prefix="imp_")
|
|
ext = os.path.splitext(path)[1].lower()
|
|
prev_subdir = ARCH_SUBDIR
|
|
# group this archive's docs under a sanitized folder name
|
|
ARCH_SUBDIR = re.sub(r"[^\w.\- ]", "_", os.path.splitext(os.path.basename(path))[0])[:40]
|
|
try:
|
|
if ext == ".zip":
|
|
with zipfile.ZipFile(path) as z:
|
|
z.extractall(tmp)
|
|
else: # rar/7z via unar in container
|
|
subprocess.run(["docker", "run", "--rm", "-v", f"{os.path.dirname(path)}:/in:ro",
|
|
"-v", f"{tmp}:/out", "node:lts", "bash", "-c",
|
|
"apt-get update -qq>/dev/null 2>&1; apt-get install -y -qq unar>/dev/null 2>&1; "
|
|
f"unar -force-overwrite -o /out '/in/{os.path.basename(path)}' >/dev/null"],
|
|
capture_output=True, text=True)
|
|
for root, _, files in os.walk(tmp):
|
|
for fn in files:
|
|
fp = os.path.join(root, fn)
|
|
process_file(fp, origin=f"{os.path.basename(path)} :: {fn}")
|
|
finally:
|
|
ARCH_SUBDIR = prev_subdir
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
|
|
def main():
|
|
global PLACE
|
|
args = [a for a in sys.argv[1:] if not a.startswith("--")]
|
|
PLACE = "--place" in sys.argv
|
|
incoming = args[0]
|
|
for fn in sorted(os.listdir(incoming)):
|
|
fp = os.path.join(incoming, fn)
|
|
if os.path.isfile(fp):
|
|
process_file(fp)
|
|
# report
|
|
order = {"OK": 0, "UNRESOLVED": 1, "SKIP": 2, "FAIL": 3}
|
|
print(f"\n=== IMPORT REPORT ({'PLACED' if PLACE else 'DRY RUN'}) ===")
|
|
for typ, st, origin, detail, dest in sorted(report, key=lambda r: (r[0], order.get(r[1], 9))):
|
|
line = f"[{st:10}] {typ:5} {origin[:46]:46} {detail}"
|
|
print(line)
|
|
if dest:
|
|
print(f"{'':19}-> {dest}")
|
|
counts = {}
|
|
for r in report:
|
|
counts[(r[0], r[1])] = counts.get((r[0], r[1]), 0) + 1
|
|
print("\nsummary:", dict(sorted(counts.items())))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|