refactor: split the pack builder into its layers

generate_pack.py held six responsibilities in 4744 lines. Five move out
in dependency order, so nothing above reaches back down: destinations,
core extras, resolution with its storage tiers, the notes shipped inside
a pack, and the pack verifier. generate_pack.py keeps the build and the
command line, and re-exports the rest.

Two things the move surfaced. The offline switch was a module global the
command line assigned, which a re-export would have copied and frozen at
False; it is set through a call now and not re-exported. And a facade
placed after the entry point binds too late: importing the module worked,
running it did not, so the manifest run died on a name the tests never
exercised because tests import.

Verified against the previous revision on frozen inputs: every generated
artefact is identical, save the catalogue that embeds hashes of files
carrying a build timestamp.
This commit is contained in:
Abdessamad Derraz committed 2026-08-12 12:20:17 +02:00
1 parent b5fd643ccf
commit c31ce0b693
7 files changed
+1872 -1754

No files matched your search

+14
View File
@@ -86,3 +86,17 @@ def artifact_lock(directory: str, exclusive: bool = True):
yield
finally:
fcntl.flock(handle, fcntl.LOCK_UN)
def _build_timestamp(db: dict | None = None) -> str:
"""Timestamp for generated artifacts.
Reads the database snapshot the artifact was built from, so rebuilding
the same data twice yields the same value. Falls back to the clock only
when no database is at hand.
"""
stamp = (db or {}).get("generated_at")
if isinstance(stamp, str) and stamp:
return stamp
from datetime import datetime, timezone
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+47 -1754
View File
File diff suppressed because it is too large. Load diff
+566
View File
@@ -0,0 +1,566 @@
"""What the cores need that the platform does not declare.
A pack is the platform's own list plus the files its cores load. The
second half is derived here, and the region pass reads the grouping
this module builds so the report and the pack withdraw the same files."""
from __future__ import annotations
from common import expand_platform_declared_names
from common import load_emulator_profiles
from common import resolution_is_hash_exact
from common import resolve_local_file
from common import sanitize_pack_path
def _emulator_systems_index(emu_profiles: dict | None) -> dict[str, list[str]]:
"""Map both the profile key and its display name to the profile's systems.
find_undeclared_files reports the display name ("Beetle PSX (Mednafen
PSX)"), while the profile dictionary is keyed by slug. A key-only lookup
therefore missed almost every core, dropping its files into one shared
bucket and losing the per-system grouping the narrowing passes rely on.
"""
index: dict[str, list[str]] = {}
for key, profile in (emu_profiles or {}).items():
systems = list(profile.get("systems", []))
index[key] = systems
display = profile.get("emulator", "")
if display:
index.setdefault(display, systems)
return index
def _detect_extras_prefix(config: dict, base_dest: str) -> str:
"""Detect the effective BIOS prefix for core extras.
When base_destination is empty (RetroDECK), infer the prefix from
the dominant root of YAML-declared destinations. Returns the prefix
to prepend to every core-extra destination (may be empty).
"""
if base_dest:
return base_dest
dests: list[str] = []
for sys_data in config.get("systems", {}).values():
for f in sys_data.get("files", []):
d = f.get("destination", "")
if d and "/" in d:
dests.append(d)
if not dests:
return ""
from collections import Counter
roots = Counter(d.split("/", 1)[0] for d in dests)
most_common, count = roots.most_common(1)[0]
if count / len(dests) > 0.9:
return most_common
return ""
def _detect_slug_structure(config: dict) -> tuple[bool, dict[str, str]]:
"""Detect whether a platform uses per-system slug destinations.
Returns ``(is_slug_based, system_to_slug)`` where ``system_to_slug``
maps system IDs to their destination slug prefix. Slug-based means
each system's files live under a per-system subfolder (e.g. RomM's
``bios/{platform_slug}/{file}``), with varying slugs across systems.
Only returns True when nearly ALL destinations have a subfolder and
nearly ALL systems map to a consistent slug, distinguishing true
slug-based layouts (RomM) from platforms that happen to have some
subfoldered files (RetroArch ``dc/``, ``neocd/``).
"""
total_files = 0
files_with_slash = 0
sys_to_slug: dict[str, str] = {}
total_systems_with_files = 0
for sys_id, sys_data in config.get("systems", {}).items():
files = sys_data.get("files", [])
if not files:
continue
total_systems_with_files += 1
slugs: set[str] = set()
for f in files:
d = f.get("destination", "")
if d:
total_files += 1
if "/" in d:
files_with_slash += 1
slugs.add(d.split("/", 1)[0])
if len(slugs) == 1:
sys_to_slug[sys_id] = slugs.pop()
if not sys_to_slug or total_files == 0:
return False, {}
# All conditions must hold for slug-based detection:
# 1. Nearly all files have a subfolder
# 2. Multiple distinct slugs (not a constant prefix)
# 3. Nearly all systems with files map to a slug
# 4. Files are exactly slug/filename (depth 2), not deeper
unique_slugs = set(sys_to_slug.values())
all_have_slash = files_with_slash / total_files > 0.95
varying_slugs = len(unique_slugs) > 1
high_coverage = len(sys_to_slug) / total_systems_with_files > 0.9
# Count files deeper than slug/filename (e.g., amiga/bios/kick.rom)
deep_files = 0
for sys_data in config.get("systems", {}).values():
for f in sys_data.get("files", []):
d = f.get("destination", "")
if d and d.count("/") > 1:
deep_files += 1
shallow = deep_files / total_files < 0.05 if total_files else True
return (all_have_slash and varying_slugs and high_coverage and shallow), sys_to_slug
def _map_emulator_to_slug(
profile: dict,
platform_systems: set[str],
norm_map: dict[str, str],
sys_to_slug: dict[str, str],
) -> str:
"""Map an emulator to a destination slug for slug-based platforms."""
from common import _norm_system_id
emu_systems = set(profile.get("systems", []))
# Direct match
direct = emu_systems & platform_systems
if direct:
target = sorted(direct)[0]
return sys_to_slug.get(target, "")
# Normalized match
for es in sorted(emu_systems):
norm = _norm_system_id(es)
if norm in norm_map:
target = norm_map[norm]
return sys_to_slug.get(target, "")
return ""
def _collect_emulator_extras(
config: dict,
emulators_dir: str,
db: dict,
seen: set,
base_dest: str,
emu_profiles: dict | None = None,
target_cores: set[str] | None = None,
include_all: bool = False,
) -> list[dict]:
"""Collect core requirement files from emulator profiles not in the platform pack.
Uses the same system-overlap matching as verify.py cross-reference:
- Matches emulators by shared system IDs with the platform
- Filters mode: standalone, type: launcher, type: alias
- Respects data_directories coverage
- Only returns files that exist in the repo (packable)
When the same file is needed at multiple destinations by different cores
(e.g. cdimono1.zip at root for cdi2015 and at same_cdi/bios/ for same_cdi),
all destinations are included so every core finds its files.
Works for ANY platform (RetroArch, Batocera, Recalbox, etc.)
"""
from common import _norm_system_id, resolve_platform_cores
from verify import find_undeclared_files
profiles = (
emu_profiles
if emu_profiles is not None
else load_emulator_profiles(emulators_dir)
)
# Detect destination conventions for core extras
extras_prefix = _detect_extras_prefix(config, base_dest)
is_slug_based, sys_to_slug = _detect_slug_structure(config)
platform_systems = set(config.get("systems", {}).keys())
norm_map: dict[str, str] = {}
if is_slug_based:
for sid in platform_systems:
norm_map[_norm_system_id(sid)] = sid
# Use strict YAML names (no DB alias enrichment) so that files known
# under an alias still get packed at the emulator's expected path.
yaml_names: set[str] = set()
for system in config.get("systems", {}).values():
for fe in system.get("files", []):
name = fe.get("name", "")
if name:
yaml_names.add(name)
undeclared = find_undeclared_files(
config, emulators_dir, db, emu_profiles, target_cores=target_cores,
include_all=include_all, declared_names=yaml_names,
)
extras = []
seen_dests: set[str] = set(seen)
for u in undeclared:
if not u["in_repo"]:
continue
# For archive entries, use the archive name for resolution
archive = u.get("archive")
name = archive if archive else u["name"]
raw_dest = archive if archive else (u.get("path") or u["name"])
# Directory path: append filename (e.g. "cafeLibs/" + "snd_user.rpl")
dest = f"{raw_dest}{u['name']}" if raw_dest.endswith("/") else raw_dest
# Slug-based platforms: prefix dest with system slug
if is_slug_based:
emu_name = u.get("profile") or u.get("emulator", "")
profile = profiles.get(emu_name, {})
# Try finding profile by display name if key lookup failed
if not profile:
for pn, pp in profiles.items():
if pp.get("emulator") == emu_name:
profile = pp
break
slug = _map_emulator_to_slug(
profile,
platform_systems,
norm_map,
sys_to_slug,
)
if not slug:
continue # can't place without slug
dest = f"{slug}/{dest}"
full_dest = f"{extras_prefix}/{dest}" if extras_prefix else dest
if full_dest in seen_dests:
continue
seen_dests.add(full_dest)
extra = {
"name": name,
"destination": dest,
"required": u.get("required", False),
"hle_fallback": u.get("hle_fallback", False),
"source_emulator": u.get("emulator", ""),
"source_profile": u.get("profile", ""),
# Identity of the report entry this extra came from. The name
# alone does not identify it: Dolphin declares three IPL.bin that
# differ only by path, and collapsing them onto one key withdraws
# the wrong regional variant from the report.
"source_name": u.get("name", ""),
"source_path": u.get("path") or "",
"source_system": u.get("system"),
"source_systems": u.get("systems", []),
"region": u.get("region"),
"variant_group": u.get("variant_group"),
}
# Keep every reproducible identity constraint carried by the source
# profile. Some profiles expose CRC32/SHA-256 only through their
# validation block; dropping those fields here would let a same-named
# but different payload enter a generated pack.
expected = u.get("expected") or {}
for identity in ("sha1", "md5", "sha256", "crc32", "size"):
declared = u.get(identity)
if declared in (None, "", []):
declared = expected.get(identity)
if declared not in (None, "", []):
extra[identity] = declared
extras.append(extra)
# Second pass: find alternative destinations for files already in the pack.
# A file declared by the platform or emitted above may also be needed at a
# different path by another core (e.g. neocd/ vs root, same_cdi/bios/ vs root).
# Only adds a copy when the file is ALREADY covered at a different path -
# never introduces a file that wasn't selected by the first pass.
#
# Skip for slug-based platforms (RomM): alternative paths don't map to
# the required {platform_slug}/{file} structure.
if is_slug_based:
return extras
relevant = resolve_platform_cores(config, profiles, target_cores=target_cores)
standalone_set = {str(c) for c in config.get("standalone_cores", [])}
by_name = db.get("indexes", {}).get("by_name", {})
by_path_suffix = db.get("indexes", {}).get("by_path_suffix", {})
# Build set of filenames already covered (platform baseline + first pass extras)
# Enriched with canonical names from DB via MD5 (handles platform renaming)
covered_names = expand_platform_declared_names(config, db)
for e in extras:
covered_names.add(e["name"])
for emu_name, profile in sorted(profiles.items()):
if profile.get("type") in ("launcher", "alias"):
continue
if emu_name not in relevant:
continue
is_standalone = emu_name in standalone_set or bool(
standalone_set & {str(c) for c in profile.get("cores", [])}
)
for f in profile.get("files", []):
fname = f.get("name", "")
if not fname:
continue
# Only duplicate files already covered at another destination
if fname not in covered_names:
continue
file_mode = f.get("mode")
if file_mode == "standalone" and not is_standalone:
continue
if file_mode == "libretro" and is_standalone:
continue
# Skip files loaded from non-system directories (save_dir, content_dir)
load_from = f.get("load_from", "")
if load_from and load_from != "system_dir":
continue
if is_standalone:
raw = f.get("standalone_path") or f.get("path") or fname
else:
raw = f.get("path") or fname
dest = f"{raw}{fname}" if raw.endswith("/") else raw
if dest == fname:
continue # no alternative destination
full_dest = f"{extras_prefix}/{dest}" if extras_prefix else dest
if full_dest in seen_dests:
continue
# Check file exists in repo or data dirs
if not (
by_name.get(fname)
or by_name.get(dest.rsplit("/", 1)[-1])
or by_path_suffix.get(dest)
):
continue
seen_dests.add(full_dest)
extras.append(
{
"name": fname,
"destination": dest,
"required": f.get("required", False),
"hle_fallback": f.get("hle_fallback", False),
"source_emulator": profile.get("emulator", emu_name),
"source_profile": emu_name,
"source_system": f.get("system"),
"source_systems": list(profile.get("systems", [])),
"region": f.get("region"),
"variant_group": f.get("variant_group"),
}
)
# Archive prefix pass: cores that store BIOS archives in a subdirectory
# (e.g. system/fbneo/neogeo.zip). When the archive is already covered at
# the root, add a copy at the prefixed path so the core's .info firmware
# check finds it.
for emu_name, profile in sorted(profiles.items()):
if profile.get("type") in ("launcher", "alias"):
continue
if emu_name not in relevant:
continue
prefix = profile.get("archive_prefix", "")
if not prefix:
continue
profile_archives: dict[str, dict] = {}
for f in profile.get("files", []):
archive = f.get("archive", "")
if archive:
profile_archives.setdefault(archive, f)
for archive_name, archive_entry in sorted(profile_archives.items()):
if archive_name not in covered_names:
continue
dest = f"{prefix}/{archive_name}"
full_dest = f"{extras_prefix}/{dest}" if extras_prefix else dest
if full_dest in seen_dests:
continue
if not by_name.get(archive_name):
continue
seen_dests.add(full_dest)
extras.append(
{
"name": archive_name,
"destination": dest,
"required": True,
"hle_fallback": False,
"source_emulator": profile.get("emulator", emu_name),
"source_profile": emu_name,
"source_system": archive_entry.get("system"),
"source_systems": list(profile.get("systems", [])),
"region": archive_entry.get("region"),
"variant_group": archive_entry.get("variant_group"),
}
)
# Third pass: agnostic scan — for filename-agnostic cores, include all
# DB files matching the system path prefix and size criteria.
files_db = db.get("files", {})
for emu_name, profile in sorted(profiles.items()):
if profile.get("type") in ("launcher", "alias"):
continue
if emu_name not in relevant:
continue
is_profile_agnostic = profile.get("bios_mode") == "agnostic"
if not is_profile_agnostic:
if not any(f.get("agnostic") for f in profile.get("files", [])):
continue
# Where each of this profile's files resolves. A scan anchored on one
# filename walks into whatever tree that name happens to hit: rom1.bin
# is a PS2 ROM and a Roland SC-55 ROM, GameIndex.yaml belongs to four
# PS2 profiles and to an Android package. One match is not evidence
# of a directory; two files of the same profile agreeing is.
resolved_dirs: dict[int, str] = {}
named_only: set[int] = set()
agnostic_votes: dict[str, int] = {}
for candidate in profile.get("files", []):
if not isinstance(candidate, dict):
continue
local, status = resolve_local_file(
candidate, db, dest_hint=candidate.get("path", "")
)
if not local or "/" not in local:
continue
directory = local.rsplit("/", 1)[0]
if directory.endswith("/.variants"):
directory = directory[: -len("/.variants")]
resolved_dirs[id(candidate)] = directory + "/"
if not (resolution_is_hash_exact(status) or status == "path_exact"):
named_only.add(id(candidate))
agnostic_votes[directory + "/"] = (
agnostic_votes.get(directory + "/", 0) + 1
)
for f in profile.get("files", []):
if not is_profile_agnostic and not f.get("agnostic"):
continue
fname = f.get("name", "")
if not fname:
continue
# An agnostic BIOS mode says the BIOS filename is free, not that
# every file the emulator loads is. A free filename still has a
# known shape, and that shape is what identifies a candidate: an
# entry declaring no size takes the whole directory, which is how
# the flag icons of an Android package walked into a PS2 pack.
if f.get("category", "bios") != "bios":
continue
if not (f.get("size") or f.get("min_size") or f.get("max_size")):
continue
path_prefix = resolved_dirs.get(id(f), "")
if not path_prefix:
continue
# A name matching one file identifies it. A name matching several
# identifies nothing on its own, so the profile's other files have
# to agree on the directory before the scan walks it.
ambiguous = id(f) in named_only and len(by_name.get(fname, [])) > 1
if ambiguous and agnostic_votes.get(path_prefix, 0) < 2:
continue
# Size criteria from the file entry
min_size = f.get("min_size", 0)
max_size = f.get("max_size", float("inf"))
exact_size = f.get("size")
if exact_size and not min_size:
min_size = exact_size
max_size = exact_size
# Scan DB for all files under this prefix matching size
for sha1, entry in files_db.items():
path = entry.get("path", "")
if not path.startswith(path_prefix):
continue
size = entry.get("size", 0)
if not (min_size <= size <= max_size):
continue
scan_name = entry.get("name", "")
if not scan_name:
continue
dest = scan_name
full_dest = f"{extras_prefix}/{dest}" if extras_prefix else dest
if full_dest in seen_dests:
continue
seen_dests.add(full_dest)
extras.append(
{
"name": scan_name,
"destination": dest,
# The scan already holds the entry it selected, so it
# names it by content. Emitting the filename alone
# sent the packing step back to a by-name lookup, and
# three PS2 resources came back from another
# emulator's copy of the same filename.
"sha1": sha1,
"required": False,
"hle_fallback": False,
"source_emulator": profile.get("emulator", emu_name),
"source_profile": emu_name,
"source_system": f.get("system"),
"source_systems": list(profile.get("systems", [])),
"region": f.get("region"),
"variant_group": f.get("variant_group"),
"agnostic_scan": True,
}
)
return extras
def _extra_system_ids(extra: dict) -> list[str]:
"""Return the narrowest system ownership preserved on a core extra."""
explicit = extra.get("source_system")
if explicit:
return [str(explicit)]
return [str(value) for value in extra.get("source_systems", []) if value]
def platform_region_groups(
config: dict,
systems: dict,
emulators_dir: str,
db: dict | None,
base_dest: str,
emu_profiles: dict | None,
*,
target_cores: set[str] | None = None,
include_extras: bool = True,
include_all: bool = False,
) -> tuple[dict[str, list[tuple[str, str]]], dict[tuple[str, str, str], str]]:
"""Group a platform's pack candidates the way region filtering reads them.
Returns the groups and, for every core extra, the destination it was
grouped under, keyed by (emulator, name). verify.py needs that mapping to
withdraw from its report exactly what the builder withdraws from the pack:
grouping the declared files here and the core extras there would let the
two answer differently on the same request.
"""
groups: dict[str, list[tuple[str, str]]] = {}
for sys_id, system in systems.items():
members = groups.setdefault(sys_id, [])
for file_entry in system.get("files", []):
dest = sanitize_pack_path(
file_entry.get("destination", file_entry.get("name", ""))
)
if dest:
members.append((dest, file_entry.get("name", "")))
extra_dests: dict[tuple[str, str, str], str] = {}
if not include_extras or db is None:
return groups, extra_dests
for extra in _collect_emulator_extras(
config,
emulators_dir,
db,
set(),
base_dest,
emu_profiles,
target_cores=target_cores,
include_all=include_all,
):
dest = sanitize_pack_path(extra.get("destination", extra.get("name", "")))
if not dest:
continue
name = extra.get("name", "")
extra_dests[
(
extra.get("source_emulator", ""),
extra.get("source_name", ""),
extra.get("source_path", ""),
)
] = dest
variant = extra.get("variant_group")
for sys_id in _extra_system_ids(extra) or ["_extras"]:
group_id = f"{sys_id}:variant:{variant}" if variant else sys_id
groups.setdefault(group_id, []).append((dest, name))
return groups, extra_dests
def _emulator_region_group(emu_name: str, profile: dict, file_entry: dict) -> str:
"""Stable group ID for regional alternatives within an emulator profile."""
variant = file_entry.get("variant_group")
if variant:
return f"{emu_name}:variant:{variant}"
system = file_entry.get("system")
profile_systems = list(profile.get("systems", []))
if not system and len(profile_systems) == 1:
system = profile_systems[0]
return f"{emu_name}:system:{system or '_profile'}"
+64
View File
@@ -0,0 +1,64 @@
"""Where a file lands inside a pack.
Two files may not claim the same destination, and a destination may
not sit inside another file's path: a pack that cannot be extracted
on Windows is not a pack."""
from __future__ import annotations
from common import sanitize_pack_path
def _path_parents(dest: str) -> list[str]:
"""Return all parent directory segments of a path."""
parts = dest.split("/")
return ["/".join(parts[:i]) for i in range(1, len(parts))]
def _has_path_conflict(dest: str, seen_files: set[str], seen_parents: set[str]) -> bool:
"""Check if dest conflicts with existing paths (file vs directory).
Returns True if adding dest would create an impossible extraction:
- A parent of dest is already a file (e.g., adding X/Y when X is a file)
- dest itself is already used as a directory (e.g., adding X when X/Y exists)
"""
for parent in _path_parents(dest):
if parent in seen_files:
return True
if dest in seen_parents:
return True
return False
def _register_path(dest: str, seen_files: set[str], seen_parents: set[str]) -> None:
"""Track a file path and its parent directories."""
seen_files.add(dest)
for parent in _path_parents(dest):
seen_parents.add(parent)
def _flat(arcname: str, prefix: str, flatten: bool) -> str:
"""Strip base_destination prefix from ZIP arcname when flattening."""
if flatten and prefix and arcname.startswith(prefix + "/"):
return arcname[len(prefix) + 1:]
return arcname
def _resolve_destination(
file_entry: dict, pack_structure: dict | None, standalone: bool
) -> str:
"""Resolve the ZIP destination path for a file entry."""
# 1. standalone_path override
if standalone and file_entry.get("standalone_path"):
rel = file_entry["standalone_path"]
# 2. path field
elif file_entry.get("path"):
rel = file_entry["path"]
# 3. name fallback
else:
rel = file_entry.get("name", "")
rel = sanitize_pack_path(rel)
# Prepend pack_structure prefix
if pack_structure:
mode_key = "standalone" if standalone else "libretro"
prefix = pack_structure.get(mode_key, "")
if prefix:
rel = f"{prefix}/{rel}"
return rel
+316
View File
@@ -0,0 +1,316 @@
"""The notes shipped inside a pack.
A pack has to explain itself offline: what it holds, where each file
goes, and which ones the user has to rename."""
from __future__ import annotations
def _build_readme(
platform_name: str,
platform_display: str,
base_dest: str,
total_files: int,
num_systems: int,
source: str = "full",
contributors: list[dict] | None = None,
regions: list[str] | None = None,
fallback_systems: list[str] | None = None,
one_per_slot: bool = False,
undecidable_slots: list[str] | None = None,
narrowings: list[tuple[str, str]] | None = None,
system_filter: list[str] | None = None,
) -> str:
"""Build a personalized step-by-step README for each platform pack."""
narrowings = narrowings or []
sep = "=" * 50
header = (
f"{sep}\n"
f" RETROBIOS - {platform_display} BIOS Pack\n"
f" {total_files} files for {num_systems} systems\n"
f"{sep}\n\n"
)
guides: dict[str, str] = {
"retroarch": (
"INSTALLATION GUIDE\n\n"
" Option A: Automatic (recommended)\n"
" ---------------------------------\n"
" Run this in a terminal:\n\n"
" curl -fsSL https://raw.githubusercontent.com/Abdess/retrobios/main/install.sh | sh\n\n"
" The script auto-detects your RetroArch install and copies\n"
" files to the correct location.\n\n"
" Option B: Manual (PC)\n"
" ---------------------\n"
" 1. Find your RetroArch system directory:\n"
" - RetroArch > Settings > Directory > System/BIOS\n"
" - Default: retroarch/system/\n"
" 2. Extract all files from this archive directly into your system directory\n"
" 3. Overwrite if asked\n\n"
" Option C: Manual (handheld / SD card)\n"
" -------------------------------------\n"
" Anbernic, Retroid, Miyoo, Trimui, etc.:\n"
" 1. Connect your SD card to your PC\n"
" 2. Find the BIOS folder (usually BIOS/ or system/)\n"
" 3. Extract all files from this archive directly into that folder\n"
" 4. Eject SD card and reboot your device\n\n"
" Common paths by device:\n"
" Anbernic (ArkOS/JELOS): BIOS/\n"
" Retroid (RetroArch): RetroArch/system/\n"
" Miyoo Mini (Onion OS): BIOS/\n"
" Steam Deck (RetroArch): ~/.config/retroarch/system/\n\n"
),
"batocera": (
"INSTALLATION GUIDE\n\n"
" Option A: Automatic (recommended)\n"
" ---------------------------------\n"
" Open a terminal (F1 from Batocera menu) and run:\n\n"
" curl -fsSL https://raw.githubusercontent.com/Abdess/retrobios/main/install.sh | sh\n\n"
" Option B: Manual (network share)\n"
" --------------------------------\n"
" 1. On your PC, open the Batocera network share:\n"
" - Windows: \\\\BATOCERA\\share\\bios\\\n"
" - Mac/Linux: smb://batocera/share/bios/\n"
" 2. Extract all files from this archive directly into the share\n"
" 3. Overwrite if asked\n\n"
" Option C: Manual (SD card)\n"
" --------------------------\n"
" 1. Put the SD card in your PC\n"
" 2. Navigate to /userdata/bios/ on the SHARE partition\n"
" 3. Extract all files from this archive directly into that folder\n\n"
" NOTE: Dreamcast flash memory is named dc_nvmem.bin\n"
" (if your setup asks for dc_flash.bin, same file).\n\n"
),
"recalbox": (
"INSTALLATION GUIDE\n\n"
" Option A: Automatic\n"
" -------------------\n"
" curl -fsSL https://raw.githubusercontent.com/Abdess/retrobios/main/install.sh | sh\n\n"
" Option B: Manual (network share)\n"
" --------------------------------\n"
" 1. On your PC, open the Recalbox network share:\n"
" - Windows: \\\\RECALBOX\\share\\bios\\\n"
" - Mac/Linux: smb://recalbox/share/bios/\n"
" 2. Extract all files from this archive directly into the share\n\n"
" Option C: Manual (SD card)\n"
" --------------------------\n"
" 1. Put the SD card in your PC\n"
" 2. Navigate to /recalbox/share/bios/\n"
" 3. Extract all files from this archive directly into that folder\n\n"
),
"emudeck": (
"INSTALLATION GUIDE (Steam Deck / Linux)\n\n"
" Option A: Automatic (recommended)\n"
" ---------------------------------\n"
" Open Konsole (or any terminal) and run:\n\n"
" curl -fsSL https://raw.githubusercontent.com/Abdess/retrobios/main/install.sh | sh\n\n"
" The script places BIOS files AND sets up standalone\n"
" emulator keys automatically.\n\n"
" Option B: Manual\n"
" ----------------\n"
" 1. Open Dolphin file manager\n"
" 2. Navigate to ~/Emulation/bios/\n"
" 3. Extract all files from this archive directly into ~/Emulation/bios/\n\n"
" STANDALONE EMULATORS (extra step)\n"
" Switch and 3DS emulators need keys in specific folders:\n"
" prod.keys -> ~/.local/share/yuzu/keys/\n"
" prod.keys -> ~/.local/share/eden/keys/\n"
" prod.keys -> ~/.config/Ryujinx/system/\n"
" aes_keys.txt -> ~/Emulation/bios/citra/keys/\n"
" The automatic installer handles this for you.\n\n"
),
"retrodeck": (
"INSTALLATION GUIDE (Steam Deck / Linux)\n\n"
" Option A: Automatic (recommended)\n"
" ---------------------------------\n"
" Open Konsole (or any terminal) and run:\n\n"
" curl -fsSL https://raw.githubusercontent.com/Abdess/retrobios/main/install.sh | sh\n\n"
" Option B: Manual\n"
" ----------------\n"
" 1. Open Dolphin file manager\n"
" 2. Show hidden files (Ctrl+H)\n"
" 3. Navigate to ~/retrodeck/bios/\n"
" 4. Extract all files from this archive directly into ~/retrodeck/bios/\n\n"
" NOTE: RetroDECK uses its own BIOS checker. After\n"
" copying, open RetroDECK > Tools > BIOS Checker to\n"
" verify everything is detected.\n\n"
),
"retrobat": (
"INSTALLATION GUIDE (Windows)\n\n"
" Option A: Automatic (recommended)\n"
" ---------------------------------\n"
" Download and run install.bat from:\n"
" https://github.com/Abdess/retrobios/releases\n\n"
" Option B: Manual\n"
" ----------------\n"
" 1. Open your RetroBat installation folder\n"
" 2. Navigate to the bios\\ subfolder\n"
" (default: C:\\RetroBat\\bios\\)\n"
" 3. Extract all files from this archive directly into your bios\\ folder\n"
" 4. Overwrite if asked\n\n"
),
"bizhawk": (
"INSTALLATION GUIDE\n\n"
" 1. Open your BizHawk installation folder\n"
" 2. Navigate to the Firmware subfolder:\n"
" - Windows: BizHawk\\Firmware\\\n"
" - Linux: ~/.config/BizHawk/Firmware/\n"
" 3. Extract all files from this archive directly into your Firmware folder\n"
" 4. In BizHawk: Config > Paths > Firmware should\n"
" point to this folder\n\n"
),
"romm": (
"INSTALLATION GUIDE (RomM server)\n\n"
" 1. Locate your RomM library folder\n"
" 2. Navigate to the bios/ subdirectory\n"
" 3. Extract all files from this archive directly into that folder\n"
" 4. Restart the RomM service to detect new files\n\n"
),
"retropie": (
"INSTALLATION GUIDE (Raspberry Pi)\n\n"
" Option A: Via network share\n"
" --------------------------\n"
" 1. On your PC, open: \\\\RETROPIE\\bios\\\n"
" 2. Extract all files from this archive directly into that folder\n\n"
" Option B: Via SSH\n"
" -----------------\n"
" 1. SSH into your Pi: ssh pi@retropie\n"
" 2. Copy files to ~/RetroPie/BIOS/\n\n"
" Option C: Via SD card\n"
" ---------------------\n"
" 1. Put the SD card in your PC\n"
" 2. Navigate to /home/pi/RetroPie/BIOS/\n"
" 3. Extract all files from this archive directly into that folder\n\n"
),
}
# Lakka uses same guide as RetroArch
guides["lakka"] = guides["retroarch"]
guide = guides.get(
platform_name,
(
f"INSTALLATION\n\n"
f" 1. Extract all files from this archive directly into your BIOS directory\n"
f" 2. Overwrite if asked\n\n"
),
)
if regions:
region_help = (
" - Wrong region? This pack was filtered. Only the\n"
" best-matching BIOS was kept per system. Use the\n"
" unfiltered pack to play imports.\n"
)
else:
region_help = (
" - Wrong region? Some systems have regional BIOS\n"
" variants (USA/EUR/JAP). All are included.\n"
)
footer = (
"TROUBLESHOOTING\n\n"
" - Core says BIOS missing? Check the exact filename\n"
" and make sure it's in the right subfolder.\n"
f"{region_help}"
" - Need help? https://github.com/Abdess/retrobios/issues\n\n"
f"{sep}\n"
f" https://github.com/Abdess/retrobios\n"
f"{sep}\n"
)
source_info = ""
if source == "platform":
source_info = (
"PACK TYPE: Platform Only\n\n"
f" This pack contains only files declared by {platform_display}.\n"
" Core extras from emulator profiles are not included.\n"
" Use the Full pack for maximum coverage.\n\n"
)
elif source == "truth":
source_info = (
"PACK TYPE: Ground Truth\n\n"
" This pack contains files that emulators actually load,\n"
" based on source code analysis of emulator profiles.\n"
" Independent of platform scraper accuracy.\n\n"
)
region_info = ""
if regions:
pretty = ", ".join(
" ".join(w.title() for w in slug.split("-")) for slug in regions
)
region_info = (
"PACK TYPE: Region Filtered\n\n"
f" Region priority: {pretty}\n\n"
" Only the best-matching BIOS was kept for each system.\n"
)
if fallback_systems:
listed = "\n".join(f" {s}" for s in fallback_systems)
region_info += (
"\n These systems have no BIOS in those regions, so all\n"
" of theirs were kept:\n"
f"{listed}\n"
)
region_info += (
"\n This shrinks the pack. It does not change how cores pick\n"
" a BIOS: most already select per region from fixed filename\n"
" lists driven by the game's region. Loading imports from\n"
" another region may need the unfiltered pack.\n\n"
)
slot_info = ""
if one_per_slot:
slot_info = (
" Where a core declares which BIOS it prefers, only that one was\n"
" kept for each system and region. Systems whose cores declare no\n"
" order keep all of theirs.\n"
)
if undecidable_slots:
slot_info += (
f" {len(undecidable_slots)} slot(s) had no declared order.\n"
)
slot_info += "\n"
narrowed = ""
labels = [label for _tag, label in narrowings]
if system_filter:
labels.append(f"systems {', '.join(system_filter)}")
if labels:
listed = "".join(f" {label}\n" for label in labels)
narrowed = (
"PACK TYPE: Narrowed\n\n"
" This pack holds fewer files than the full one:\n"
f"{listed}"
"\n The unfiltered pack is the one to use when in doubt.\n\n"
)
credits = ""
if contributors:
credits = "\nCONTRIBUTORS\n\n"
for cb in contributors:
username = cb.get("username", "")
credits += f" @{username}\n"
credits += "\n"
return (
header + narrowed + source_info + region_info + slot_info
+ guide + credits + footer
)
def _build_agnostic_rename_readme(
destination: str,
original: str,
alternatives: list[str],
) -> str:
"""Build a README explaining an agnostic file rename."""
lines = [
"This file was renamed for compatibility:",
f" {destination} <- {original}",
"",
]
if alternatives:
lines.append("All variants included in this pack:")
for alt in sorted(alternatives):
lines.append(f" {alt}")
lines.append("")
lines.append(f"To use a different variant, rename it to: {destination}")
return "\n".join(lines) + "\n"
+220
View File
@@ -0,0 +1,220 @@
"""Finding the bytes a pack entry names.
Resolution proper lives in common; this layer adds what only a build
needs: storage tiers, release assets for files too large to commit,
and the offline switch that forbids reaching for the network."""
from __future__ import annotations
import re
_HEX_RE = re.compile(r"\b([0-9a-fA-F]{8,40})\b")
from validation import check_file_validation
from largefiles import fetch_large_file
import os
from hashing import parse_md5_list
from common import resolve_local_file
# CLI-wide network policy. Public helpers can override it explicitly, while
# every CLI mode inherits --offline (including large-file release fallback).
_OFFLINE = False
def set_offline(value: bool) -> None:
"""Forbid every network reach for the rest of the run."""
global _OFFLINE
_OFFLINE = value
def _detect_hash_type(h: str) -> str:
n = len(h)
if n == 40:
return "sha1"
if n == 32:
return "md5"
if n == 8:
return "crc32"
return "md5"
def parse_hash_input(raw: str) -> list[tuple[str, str]]:
"""Parse comma-separated hash string into (type, hash) tuples."""
results: list[tuple[str, str]] = []
for part in raw.split(","):
part = part.strip().lower()
if not part:
continue
m = _HEX_RE.search(part)
if m:
h = m.group(1)
results.append((_detect_hash_type(h), h))
return results
def parse_hash_file(path: str) -> list[tuple[str, str]]:
"""Parse hash file (one per line, comments with #, mixed formats)."""
results: list[tuple[str, str]] = []
with open(path) as f:
for line in f:
line = line.strip()
if not line or line.startswith("#"):
continue
m = _HEX_RE.search(line.lower())
if m:
h = m.group(1)
results.append((_detect_hash_type(h), h))
return results
def lookup_hashes(
hashes: list[tuple[str, str]],
db: dict,
bios_dir: str,
emulators_dir: str,
platforms_dir: str,
) -> None:
"""Print diagnostic info for each hash."""
files_db = db.get("files", {})
by_md5 = db.get("indexes", {}).get("by_md5", {})
by_crc32 = db.get("indexes", {}).get("by_crc32", {})
for hash_type, hash_val in hashes:
sha1 = None
if hash_type == "sha1" and hash_val in files_db:
sha1 = hash_val
elif hash_type == "md5":
sha1 = by_md5.get(hash_val)
elif hash_type == "crc32":
sha1 = by_crc32.get(hash_val)
if not sha1 or sha1 not in files_db:
print(f"\n{hash_type.upper()}: {hash_val}")
print(" NOT FOUND in database")
continue
entry = files_db[sha1]
name = entry.get("name", "?")
md5 = entry.get("md5", "?")
paths = entry.get("paths") or []
aliases = entry.get("aliases") or []
print(f"\n{hash_type.upper()}: {hash_val}")
print(f" SHA1: {sha1}")
print(f" MD5: {md5}")
print(f" Name: {name}")
if paths:
print(f" Path: {paths[0]}")
if aliases:
print(f" Aliases: {aliases}")
# Check if file exists in repo (by path or by resolve_local_file)
in_repo = False
if paths:
primary = os.path.join(bios_dir, paths[0])
if os.path.exists(primary):
in_repo = True
if not in_repo:
try:
fe_check = {"name": name, "sha1": sha1, "md5": md5}
local, status = resolve_file(fe_check, db, bios_dir, {})
if local and status != "not_found":
in_repo = True
except (KeyError, OSError):
pass
print(f" In repo: {'YES' if in_repo else 'NO'}")
def _find_candidate_satisfying_both(
file_entry: dict,
db: dict,
local_path: str,
validation_index: dict,
bios_dir: str,
) -> str | None:
"""Search for a repo file that satisfies both platform MD5 and emulator validation.
When the current file passes platform verification but fails emulator checks,
search all candidates with the same name for one that passes both.
Returns a better path, or None if no upgrade found.
"""
fname = file_entry.get("name", "")
if not fname:
return None
entry = validation_index.get(fname)
if not entry:
return None
md5_expected = file_entry.get("md5", "")
md5_set = (
{m.strip().lower() for m in md5_expected.split(",") if m.strip()}
if md5_expected
else set()
)
by_name = db.get("indexes", {}).get("by_name", {})
files_db = db.get("files", {})
for sha1 in by_name.get(fname, []):
candidate = files_db.get(sha1, {})
path = candidate.get("path", "")
if (
not path
or not os.path.exists(path)
or os.path.realpath(path) == os.path.realpath(local_path)
):
continue
# Must still satisfy platform MD5
if md5_set and candidate.get("md5", "").lower() not in md5_set:
continue
# Check emulator validation
reason = check_file_validation(path, fname, validation_index, bios_dir)
if reason is None:
return path
return None
def resolve_file(
file_entry: dict,
db: dict,
bios_dir: str,
zip_contents: dict | None = None,
dest_hint: str = "",
data_dir_registry: dict | None = None,
*,
offline: bool | None = None,
) -> tuple[str | None, str]:
"""Resolve a BIOS file with storage tiers and release asset fallback.
Wraps common.resolve_local_file() with pack-specific logic for
storage tiers (external/user_provided), large file release assets,
and MAME clone mapping (deduped ZIPs).
"""
storage = file_entry.get("storage", "embedded")
if storage == "user_provided":
return None, "user_provided"
if storage == "external":
return None, "external"
path, status = resolve_local_file(
file_entry,
db,
zip_contents,
dest_hint=dest_hint,
data_dir_registry=data_dir_registry,
)
if path and status != "hash_mismatch":
return path, status
# Large files from GitHub release assets -tried when local file is
# missing OR has a hash mismatch (wrong variant on disk)
name = file_entry.get("name", "")
sha1 = file_entry.get("sha1")
first_sha1 = (sha1[0] if sha1 else "") if isinstance(sha1, list) else (sha1 or "")
md5_list = parse_md5_list(file_entry.get("md5"))
first_md5 = md5_list[0] if md5_list else ""
cached = fetch_large_file(
name,
expected_sha1=first_sha1,
expected_md5=first_md5,
offline=_OFFLINE if offline is None else offline,
)
if cached:
return cached, "release_asset"
# Fall back to hash_mismatch local file if release asset unavailable
if path:
return path, status
return None, "not_found"
+645
View File
@@ -0,0 +1,645 @@
"""Checking a built pack the way the frontend would.
The builder and the verifier have to reach the same verdict on the
same file, so both read the native mode from one place: a pack and a
coverage report that disagree describe different collections."""
from __future__ import annotations
from auto_fetch import DEFAULT_BIOS_DIR
from artifacts import _build_timestamp
from packextras import _collect_emulator_extras
from packextras import _detect_extras_prefix
from packpaths import _has_path_conflict
from packpaths import _register_path
from ziptools import build_zip_contents_index
from ziptools import check_inside_zip
from nativemode import digest_algorithm
from nativemode import hash_mismatch_excludes_file
import hashlib
from common import load_emulator_profiles
from common import load_platform_config
import os
from packextras import platform_region_groups
from nativemode import reads_file_contents
import region as region_mod
from packresolve import resolve_file
from common import sanitize_pack_path
import zipfile
def verify_pack(
zip_path: str, db: dict, data_registry: dict | None = None
) -> tuple[bool, dict]:
"""Verify a generated pack ZIP by re-hashing every file inside.
Checks against database.json, data directory caches, and verifies
rebuilt ZIP content by comparing inner CRC32s against source.
Returns (all_ok, manifest_dict).
"""
files_db = db.get("files", {}) # SHA1 -> file_info
by_md5 = db.get("indexes", {}).get("by_md5", {}) # MD5 -> SHA1
by_name = db.get("indexes", {}).get("by_name", {}) # name -> [SHA1]
# Data directory file index
_data_index: dict[str, list[str]] = {}
_data_path_index: dict[str, str] = {}
if data_registry:
for _dk, _de in data_registry.items():
cache = _de.get("local_cache", "")
if not cache or not os.path.isdir(cache):
continue
for _r, _d, _fns in os.walk(cache):
for _fn in _fns:
_fp = os.path.join(_r, _fn)
_rel = os.path.relpath(_fp, cache)
_data_path_index[_rel] = _fp
_data_index.setdefault(_fn, []).append(_fp)
manifest = {
"schema_version": 1,
"version": 1,
"generator": "retrobios generate_pack.py",
"generated": _build_timestamp(db),
"files": [],
}
errors = []
with zipfile.ZipFile(zip_path, "r") as zf:
for info in zf.infolist():
if info.is_dir():
continue
name = info.filename
if name.startswith("INSTRUCTIONS_") or name in (
"manifest.json",
"README.txt",
):
continue
with zf.open(info) as f:
sha1_h = hashlib.sha1()
md5_h = hashlib.md5()
size = 0
for chunk in iter(lambda: f.read(65536), b""):
sha1_h.update(chunk)
md5_h.update(chunk)
size += len(chunk)
sha1 = sha1_h.hexdigest()
md5 = md5_h.hexdigest()
# Look up in database: files_db keyed by SHA1
db_entry = files_db.get(sha1)
status = "verified"
file_name = ""
if db_entry:
file_name = db_entry.get("name", "")
else:
# Try MD5 -> SHA1 lookup
ref_sha1 = by_md5.get(md5)
if ref_sha1:
db_entry = files_db.get(ref_sha1)
if db_entry:
file_name = db_entry.get("name", "")
status = "verified_md5"
else:
status = "untracked"
else:
status = "untracked"
# Rebuilt ZIP: verify inner ROM CRC32s match source
if status == "untracked" and name.endswith(".zip"):
_bn = os.path.basename(name)
for _src_sha1 in by_name.get(_bn, []):
if _src_sha1 not in files_db:
continue
_src_path = files_db[_src_sha1]["path"]
if not os.path.exists(_src_path):
continue
try:
import io as _io
with zipfile.ZipFile(_src_path) as _sz:
_sc = {
i.filename: i.CRC
for i in _sz.infolist()
if not i.is_dir()
}
with zipfile.ZipFile(_io.BytesIO(zf.read(name))) as _pz:
_pc = {
i.filename: i.CRC
for i in _pz.infolist()
if not i.is_dir()
}
if _sc == _pc:
status = "verified_rebuild"
file_name = _bn
break
except (zipfile.BadZipFile, OSError):
continue
# Data directory: check against cached files
if status == "untracked" and _data_index:
_bn = os.path.basename(name)
_pr = name
for _known_prefix in ("system/", "bios/", "BIOS/", "Firmware/"):
if name.startswith(_known_prefix):
_pr = name[len(_known_prefix):]
break
_cands = []
if _pr in _data_path_index:
_cands.append(_data_path_index[_pr])
for _dp in _data_index.get(_bn, []):
if _dp not in _cands:
_cands.append(_dp)
for _dp in _cands:
if not os.path.exists(_dp):
continue
if os.path.getsize(_dp) == size:
status = "verified_data"
file_name = _bn
break
if name.endswith(".zip") and _dp.endswith(".zip"):
try:
import io as _io2
with zipfile.ZipFile(_io2.BytesIO(zf.read(name))) as _pz2:
_pc2 = {
i.filename: i.CRC
for i in _pz2.infolist()
if not i.is_dir()
}
with zipfile.ZipFile(_dp) as _dz:
_dc = {
i.filename: i.CRC
for i in _dz.infolist()
if not i.is_dir()
}
if _pc2 == _dc:
status = "verified_data"
file_name = _bn
break
except (zipfile.BadZipFile, OSError):
continue
manifest["files"].append(
{
"path": name,
"sha1": sha1,
"md5": md5,
"size": size,
"status": status,
"name": file_name,
}
)
# Corruption check: SHA1 in DB but doesn't match what we computed
# This should never happen (we looked up by SHA1), but catches
# edge cases where by_md5 resolved to a different SHA1
if db_entry and status == "verified_md5":
expected_sha1 = db_entry.get("sha1", "")
if expected_sha1 and expected_sha1.lower() != sha1.lower():
errors.append(
f"{name}: SHA1 mismatch (expected {expected_sha1}, got {sha1})"
)
verified = sum(1 for f in manifest["files"] if f["status"].startswith("verified"))
untracked = sum(1 for f in manifest["files"] if f["status"] == "untracked")
total = len(manifest["files"])
manifest["summary"] = {
"total_files": total,
"verified": verified,
"untracked": untracked,
"errors": len(errors),
}
manifest["errors"] = errors
all_ok = len(errors) == 0
return all_ok, manifest
def generate_sha256sums(output_dir: str) -> str | None:
"""Generate SHA256SUMS.txt for all ZIP files in output_dir."""
sums_path = os.path.join(output_dir, "SHA256SUMS.txt")
entries = []
for name in sorted(os.listdir(output_dir)):
if not name.endswith(".zip"):
continue
path = os.path.join(output_dir, name)
sha256 = hashlib.sha256()
with open(path, "rb") as f:
for chunk in iter(lambda: f.read(65536), b""):
sha256.update(chunk)
entries.append(f"{sha256.hexdigest()} {name}")
if not entries:
return None
with open(sums_path, "w") as f:
f.write("\n".join(entries) + "\n")
print(f"\n{sums_path}: {len(entries)} pack checksums")
return sums_path
def _hash_matches(declared: str, actual: str) -> bool:
"""Compare a declared hash value against an actual hex digest.
Handles comma-separated multi-hash lists and uppercase (Recalbox)
and truncated MD5s (Batocera 29-char): a declared value shorter
than the digest matches by prefix.
"""
actual = actual.lower()
for cand in declared.split(","):
cand = cand.strip().lower()
if not cand:
continue
if len(cand) < len(actual):
if actual.startswith(cand):
return True
elif actual == cand:
return True
return False
def _check_member_hash(
zf: zipfile.ZipFile, member: str, file_entry: dict, mode: str
) -> str | None:
"""Verify a pack member against its platform-declared hash.
Reproduces the platform's native check inside the pack: md5/sha1 of
the member bytes, or checkInsideZip when zipped_file is set
(Batocera hashes a ROM inside the ZIP, matched case-insensitively).
Returns an error string, or None when the member passes.
"""
declared = str(file_entry.get(mode) or "").strip()
if not declared:
return None
zipped_file = file_entry.get("zipped_file")
if zipped_file:
import io
try:
with zipfile.ZipFile(io.BytesIO(zf.read(member))) as inner:
want = zipped_file.casefold()
target = next(
(n for n in inner.namelist() if n.casefold() == want), None
)
if target is None:
return f"{member}: {zipped_file} not found inside ZIP"
actual = hashlib.md5(inner.read(target)).hexdigest()
except zipfile.BadZipFile:
return f"{member}: not a valid ZIP"
if not _hash_matches(declared, actual):
return (
f"{member}: {zipped_file} inside-zip md5 {actual} "
f"!= declared {declared}"
)
return None
h = hashlib.md5() if mode == "md5" else hashlib.sha1()
with zf.open(member) as f:
for chunk in iter(lambda: f.read(65536), b""):
h.update(chunk)
actual = h.hexdigest()
if _hash_matches(declared, actual):
return None
# Recalbox Md5Composite: MD5 over sorted inner contents of a ZIP,
# independent of compression and metadata (Zip::Md5Composite()).
if mode == "md5" and member.endswith(".zip"):
import io
try:
with zipfile.ZipFile(io.BytesIO(zf.read(member))) as inner:
names = sorted(n for n in inner.namelist() if not n.endswith("/"))
ch = hashlib.md5()
for n in names:
ch.update(inner.read(n))
if _hash_matches(declared, ch.hexdigest()):
return None
except (zipfile.BadZipFile, OSError):
pass
return f"{member}: {mode} {actual} != declared {declared}"
def _repo_satisfies_declaration(
entries: list[dict], db: dict, mode: str
) -> bool:
"""Check whether any repo file matches one of the declared hashes.
Used to separate pack divergence (a matching file exists but was not
packed) from data coverage gaps (no repo file matches the upstream
declaration): only the former is a pack generation error.
"""
from common import md5_composite
files_db = db.get("files", {})
by_md5 = db.get("indexes", {}).get("by_md5", {})
by_name = db.get("indexes", {}).get("by_name", {})
for fe in entries:
declared = str(fe.get(mode) or "").strip()
if not declared:
continue
hashes = [h.strip().lower() for h in declared.split(",") if h.strip()]
for h in hashes:
if mode == "sha1":
entry = files_db.get(h)
else:
entry = files_db.get(by_md5.get(h, ""))
if entry and os.path.exists(entry.get("path", "")):
return True
if mode == "md5":
hash_set = set(hashes)
for cand_sha in by_name.get(fe.get("name", ""), []):
entry = files_db.get(cand_sha)
if not entry:
continue
path = entry.get("path", "")
if not path.endswith(".zip") or not os.path.exists(path):
continue
try:
if md5_composite(path).lower() in hash_set:
return True
except (zipfile.BadZipFile, OSError):
continue
return False
def _intentional_hash_exclusion(
entries: list[dict],
db: dict,
bios_dir: str = DEFAULT_BIOS_DIR,
zip_contents: dict | None = None,
data_dir_registry: dict | None = None,
verification_mode: str = "md5",
) -> bool:
"""Return whether the builder must omit every declaration as unsafe.
A hash platform never substitutes a same-named payload for an explicitly
hash-identified one, so a missing pack member is accounted for when every
declaration for that destination resolves to a local hash mismatch. An
existence platform reads no bytes and the builder ships the file anyway,
so nothing is ever excluded on its behalf. Genuine absence,
external-download failure and a packable alternative stay conformance
errors in both modes.
"""
if not entries or not hash_mismatch_excludes_file(verification_mode):
return False
archive_index = zip_contents if zip_contents is not None else {}
for entry in entries:
local_path, status = resolve_file(
entry,
db,
bios_dir,
archive_index,
data_dir_registry=data_dir_registry,
offline=True,
)
if status != "hash_mismatch":
return False
# A container can mismatch the outer declaration while still carrying
# the exact inner ROM requested by Batocera-style zipped_file entries.
zipped_file = entry.get("zipped_file")
if zipped_file and local_path:
declared = str(entry.get("md5") or "")
candidates = [value.strip() for value in declared.split(",") if value.strip()]
if not candidates:
candidates = [""]
if any(
check_inside_zip(local_path, zipped_file, candidate) == "ok"
for candidate in candidates
):
return False
return True
def verify_pack_against_platform(
zip_path: str,
platform_name: str,
platforms_dir: str,
db: dict | None = None,
emulators_dir: str = "emulators",
emu_profiles: dict | None = None,
regions: list[str] | None = None,
data_registry: dict | None = None,
) -> tuple[bool, int, int, list[str], int, int, int, int, int]:
"""Verify a pack ZIP against its platform config and core requirements.
A region priority list narrows the expectation to what the builder would
have packed, using the same selection function.
Checks:
1. Every baseline file declared by the platform exists in the ZIP
at the correct destination path
2. Every in-repo core extra file (from emulator profiles) is present
3. No duplicate entries
4. No path anomalies (double slash, absolute, traversal)
5. No unexpected zero-byte BIOS files
Returns ``(all_ok, checked, present, errors, baseline_checked,
baseline_present, core_checked, core_present, baseline_excluded)``.
"""
from collections import Counter
config = load_platform_config(platform_name, platforms_dir)
base_dest = config.get("base_destination", "")
errors: list[str] = []
checked = 0
present = 0
if emu_profiles is None:
emu_profiles = load_emulator_profiles(emulators_dir)
region_drops: set[str] = set()
if regions:
region_index = region_mod.build_region_index(emu_profiles)
region_groups, _extra_dests = platform_region_groups(
config,
config.get("systems", {}),
emulators_dir,
db,
base_dest,
emu_profiles,
)
region_drops = region_mod.resolve_region_drops(
region_groups, region_index, regions
)
with zipfile.ZipFile(zip_path, "r") as zf:
zip_set = set(zf.namelist())
zip_lower = {n.lower(): n for n in zip_set}
# Auto-detect flat vs nested ZIP
is_flat = bool(base_dest) and not any(
n.startswith(base_dest + "/")
for n in zip_set
if n not in ("README.txt", "manifest.json") and not n.endswith("/")
)
# Structural checks
dupes = sum(1 for c in Counter(zf.namelist()).values() if c > 1)
if dupes:
errors.append(f"{dupes} duplicate entries")
for n in zip_set:
if "//" in n:
errors.append(f"double slash: {n}")
if n.startswith("/"):
errors.append(f"absolute path: {n}")
if ".." in n:
errors.append(f"path traversal: {n}")
# Zero-byte check (exclude Dolphin GraphicMods markers)
for info in zf.infolist():
if info.file_size == 0 and not info.is_dir():
if "GraphicMods" not in info.filename and info.filename not in (
"manifest.json",
"README.txt",
):
errors.append(f"zero-byte: {info.filename}")
# 1. Baseline file presence + native hash check
verification_mode = config.get("verification_mode", "existence")
baseline_checked = 0
baseline_present = 0
baseline_excluded = 0
decl_by_member: dict[str, list[dict]] = {}
# Mirror the builder's path-conflict logic: a declaration whose path
# collides file-vs-directory with a packed member was skipped by the
# builder (upstream declares e.g. both SGB1.sfc and SGB1.sfc/program.rom)
zip_parents: set[str] = set()
for n in zip_set:
parts = n.split("/")
for i in range(1, len(parts)):
zip_parents.add("/".join(parts[:i]))
baseline_groups: dict[str, list[dict]] = {}
for _sys_id, system in config.get("systems", {}).items():
for fe in system.get("files", []):
dest = sanitize_pack_path(fe.get("destination", fe.get("name", "")))
if not dest:
continue
if region_drops and dest in region_drops:
continue
expected = f"{base_dest}/{dest}" if base_dest and not is_flat else dest
baseline_groups.setdefault(expected, []).append(fe)
baseline_checked = len(baseline_groups)
exclusion_index = build_zip_contents_index(db) if db is not None else {}
for expected, declarations in baseline_groups.items():
if expected in zip_set:
member = expected
elif expected.lower() in zip_lower:
member = zip_lower[expected.lower()]
elif _has_path_conflict(expected, zip_set, zip_parents):
# Skipped by the builder for the same reason: not an error
baseline_present += 1
continue
elif db is not None and _intentional_hash_exclusion(
declarations,
db,
zip_contents=exclusion_index,
data_dir_registry=data_registry,
verification_mode=verification_mode,
):
baseline_excluded += 1
continue
else:
errors.append(f"baseline missing: {expected}")
continue
baseline_present += 1
decl_by_member.setdefault(member, []).extend(declarations)
# Reproduce the platform's native hash check on pack bytes.
# A destination declared by several entries passes when the packed
# member satisfies any of them. A failure only counts as a pack
# error when the repo holds a file matching a declaration: without
# one, the pack ships its best effort and the gap is a data issue
# reported by verify.py, not a generation bug.
digest = digest_algorithm(verification_mode)
if reads_file_contents(verification_mode) and digest:
for member, decl_entries in decl_by_member.items():
checkable = [
fe
for fe in decl_entries
if str(fe.get(digest) or "").strip()
]
if not checkable:
continue
member_errors = []
satisfied = False
for fe in checkable:
err = _check_member_hash(zf, member, fe, digest)
if err is None:
satisfied = True
break
member_errors.append(err)
if satisfied:
continue
if db is None or _repo_satisfies_declaration(
checkable, db, verification_mode
):
errors.append(member_errors[0])
# 2. Core extras presence (files from emulator profiles, in repo)
# Mirror the pack builder's skip logic: only count files that
# can actually be resolved and don't have path conflicts.
core_checked = 0
core_present = 0
if db is not None:
core_files = _collect_emulator_extras(
config,
emulators_dir,
db,
set(),
base_dest,
emu_profiles,
)
seen_conformance: set[str] = set(zip_set)
seen_parents: set[str] = set()
for n in zip_set:
parts = n.split("/")
for i in range(1, len(parts)):
seen_parents.add("/".join(parts[:i]))
extras_pfx = _detect_extras_prefix(config, base_dest)
for fe in core_files:
raw_dest = fe.get("destination", fe.get("name", ""))
dest = sanitize_pack_path(raw_dest)
if not dest:
continue
if region_drops and sanitize_pack_path(dest) in region_drops:
continue
if extras_pfx and not (is_flat and extras_pfx == base_dest):
if not dest.startswith(f"{extras_pfx}/"):
full = f"{extras_pfx}/{dest}"
else:
full = dest
else:
full = dest
# Skip path conflicts (same logic as pack builder)
if _has_path_conflict(full, seen_conformance, seen_parents):
continue
# Skip unresolvable files (game_data dirs, etc.)
local_path, status = resolve_file(
fe,
db,
"bios",
{},
dest_hint=raw_dest,
data_dir_registry=data_registry,
offline=True,
)
if status in ("not_found", "external", "user_provided"):
continue
core_checked += 1
if full in zip_set or full.lower() in zip_lower:
core_present += 1
seen_conformance.add(full)
_register_path(full, seen_conformance, seen_parents)
else:
errors.append(f"core missing: {full}")
checked = baseline_checked + core_checked
present = baseline_present + core_present
return (
len(errors) == 0,
checked,
present,
errors,
baseline_checked,
baseline_present,
core_checked,
core_present,
baseline_excluded,
)