mirror of
https://github.com/Abdess/retroarch_system.git
synced 2026-04-13 12:22:33 -05:00
bb1855d3f7
Add md5_composite() to verify.py replicating Recalbox Zip::Md5Composite (sorted filenames, sequential content hash). Independent of ZIP compression level, resolves all 9 MAME arcade untested entries. Add Recalbox-specific MAME ZIP variants from Recalbox 10 pack. Batocera 671/680 (9 untested MAME-specific), all others 100%.
129 lines
4.5 KiB
Python
129 lines
4.5 KiB
Python
"""Shared utilities for retrobios scripts.
|
|
|
|
Single source of truth for platform config loading, hash computation,
|
|
and file resolution - eliminates DRY violations across scripts.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import zipfile
|
|
import zlib
|
|
from pathlib import Path
|
|
|
|
try:
|
|
import yaml
|
|
except ImportError:
|
|
yaml = None
|
|
|
|
|
|
def compute_hashes(filepath: str | Path) -> dict[str, str]:
|
|
"""Compute SHA1, MD5, SHA256, CRC32 for a file."""
|
|
sha1 = hashlib.sha1()
|
|
md5 = hashlib.md5()
|
|
sha256 = hashlib.sha256()
|
|
crc = 0
|
|
with open(filepath, "rb") as f:
|
|
for chunk in iter(lambda: f.read(65536), b""):
|
|
sha1.update(chunk)
|
|
md5.update(chunk)
|
|
sha256.update(chunk)
|
|
crc = zlib.crc32(chunk, crc)
|
|
return {
|
|
"sha1": sha1.hexdigest(),
|
|
"md5": md5.hexdigest(),
|
|
"sha256": sha256.hexdigest(),
|
|
"crc32": format(crc & 0xFFFFFFFF, "08x"),
|
|
}
|
|
|
|
|
|
def load_database(db_path: str) -> dict:
|
|
"""Load database.json and return parsed dict."""
|
|
with open(db_path) as f:
|
|
return json.load(f)
|
|
|
|
|
|
def md5sum(filepath: str | Path) -> str:
|
|
"""Compute MD5 of a file - matches Batocera's md5sum()."""
|
|
h = hashlib.md5()
|
|
with open(filepath, "rb") as f:
|
|
for chunk in iter(lambda: f.read(65536), b""):
|
|
h.update(chunk)
|
|
return h.hexdigest()
|
|
|
|
|
|
def md5_composite(filepath: str | Path) -> str:
|
|
"""Compute composite MD5 of a ZIP - matches Recalbox's Zip::Md5Composite().
|
|
|
|
Sorts filenames alphabetically, reads each file's contents in order,
|
|
feeds everything into a single MD5 hasher. The result is independent
|
|
of ZIP compression level or metadata.
|
|
"""
|
|
with zipfile.ZipFile(filepath) as zf:
|
|
names = sorted(n for n in zf.namelist() if not n.endswith("/"))
|
|
h = hashlib.md5()
|
|
for name in names:
|
|
h.update(zf.read(name))
|
|
return h.hexdigest()
|
|
|
|
|
|
def load_platform_config(platform_name: str, platforms_dir: str = "platforms") -> dict:
|
|
"""Load a platform config with inheritance and shared group resolution.
|
|
|
|
This is the SINGLE implementation used by generate_pack, generate_readme,
|
|
verify, and auto_fetch. No other copy should exist.
|
|
"""
|
|
if yaml is None:
|
|
raise ImportError("PyYAML required: pip install pyyaml")
|
|
|
|
config_file = os.path.join(platforms_dir, f"{platform_name}.yml")
|
|
if not os.path.exists(config_file):
|
|
raise FileNotFoundError(f"Platform config not found: {config_file}")
|
|
|
|
with open(config_file) as f:
|
|
config = yaml.safe_load(f) or {}
|
|
|
|
# Resolve inheritance
|
|
if "inherits" in config:
|
|
parent = load_platform_config(config["inherits"], platforms_dir)
|
|
merged = {**parent}
|
|
merged.update({k: v for k, v in config.items() if k not in ("inherits", "overrides")})
|
|
if "overrides" in config and "systems" in config["overrides"]:
|
|
merged.setdefault("systems", {})
|
|
for sys_id, override in config["overrides"]["systems"].items():
|
|
if sys_id in merged["systems"]:
|
|
merged["systems"][sys_id] = {**merged["systems"][sys_id], **override}
|
|
else:
|
|
merged["systems"][sys_id] = override
|
|
config = merged
|
|
|
|
# Resolve shared group includes
|
|
shared_path = os.path.join(platforms_dir, "_shared.yml")
|
|
if os.path.exists(shared_path):
|
|
with open(shared_path) as f:
|
|
shared = yaml.safe_load(f) or {}
|
|
shared_groups = shared.get("shared_groups", {})
|
|
for system in config.get("systems", {}).values():
|
|
for group_name in system.get("includes", []):
|
|
if group_name in shared_groups:
|
|
existing_names = {f.get("name") for f in system.get("files", [])}
|
|
for gf in shared_groups[group_name]:
|
|
if gf.get("name") not in existing_names:
|
|
system.setdefault("files", []).append(gf)
|
|
existing_names.add(gf.get("name"))
|
|
|
|
return config
|
|
|
|
|
|
def safe_extract_zip(zip_path: str, dest_dir: str) -> None:
|
|
"""Extract a ZIP file safely, preventing zip-slip path traversal."""
|
|
dest = os.path.realpath(dest_dir)
|
|
with zipfile.ZipFile(zip_path, "r") as zf:
|
|
for member in zf.infolist():
|
|
member_path = os.path.realpath(os.path.join(dest, member.filename))
|
|
if not member_path.startswith(dest + os.sep) and member_path != dest:
|
|
raise ValueError(f"Zip slip detected: {member.filename}")
|
|
zf.extract(member, dest)
|