diff --git a/scripts/artifacts.py b/scripts/artifacts.py index 0bc28eff..8251944b 100644 --- a/scripts/artifacts.py +++ b/scripts/artifacts.py @@ -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") diff --git a/scripts/generate_pack.py b/scripts/generate_pack.py index bace9ef0..f526536e 100644 --- a/scripts/generate_pack.py +++ b/scripts/generate_pack.py @@ -12,6 +12,7 @@ and 3-tier storage (embedded/external/user_provided). from __future__ import annotations +from artifacts import _build_timestamp import argparse import contextlib import hashlib @@ -53,6 +54,7 @@ from common import ( sanitize_pack_path, yaml_load, ) +import packresolve import region as region_mod import slot as slot_mod from deterministic_zip import _FIXED_DATE_TIME, rebuild_zip_deterministic @@ -74,11 +76,6 @@ DEFAULT_DB_FILE = "database.json" DEFAULT_OUTPUT_DIR = "dist" DEFAULT_BIOS_DIR = "bios" -# CLI-wide network policy. Public helpers can override it explicitly, while -# every CLI mode inherits --offline (including large-file release fallback). -_OFFLINE = False - -_HEX_RE = re.compile(r"\b([0-9a-fA-F]{8,40})\b") def _write_generated_member(zf: zipfile.ZipFile, arcname: str, text: str) -> None: @@ -95,184 +92,12 @@ def _write_generated_member(zf: zipfile.ZipFile, arcname: str, text: str) -> Non zf.writestr(info, text) -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") -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 _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 _pack_member_groups( @@ -366,94 +191,9 @@ def _target_tag(target_name: str) -> str: ) -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_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" def download_external(file_entry: dict, dest_path: str) -> bool: @@ -504,860 +244,13 @@ def download_external(file_entry: dict, dest_path: str) -> bool: return True -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'}" - - -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" def generate_pack( @@ -2159,31 +1052,6 @@ def _normalize_zip_for_pack( # Emulator/system mode pack generation -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 - def generate_emulator_pack( profile_names: list[str], @@ -3301,7 +2169,6 @@ def _run_platform_packs( def main(): - global _OFFLINE parser = argparse.ArgumentParser(description="Generate platform BIOS ZIP packs") parser.add_argument("--platform", "-p", help="Platform name (e.g., retroarch)") @@ -3403,7 +2270,7 @@ def main(): help="Extract and verify pack integrity (path + hash)", ) args = parser.parse_args() - _OFFLINE = bool(args.offline) + packresolve.set_offline(bool(args.offline)) # Parsed before the quick-exit modes: --verify-packs returns early and # still needs the region priority list to narrow its expectation. @@ -3972,193 +2839,6 @@ def generate_manifest( # Post-generation pack verification + manifest + SHA256SUMS -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 inject_manifest(zip_path: str, manifest: dict) -> None: """Inject manifest.json into an existing ZIP pack.""" @@ -4198,441 +2878,10 @@ def inject_manifest(zip_path: str, manifest: dict) -> None: raise -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, - ) def _narrows_contents(pack_name: str) -> bool: @@ -4739,6 +2988,50 @@ def verify_and_finalize_packs( generate_sha256sums(output_dir) return all_ok +from packpaths import ( # noqa: E402,F401 + _path_parents, + _has_path_conflict, + _register_path, + _flat, + _resolve_destination, +) +from packextras import ( # noqa: E402,F401 + _detect_extras_prefix, + _detect_slug_structure, + _map_emulator_to_slug, + _emulator_systems_index, + _collect_emulator_extras, + _extra_system_ids, + platform_region_groups, + _emulator_region_group, +) + + +from packreadme import ( # noqa: E402,F401 + _build_readme, + _build_agnostic_rename_readme, +) +from packverify import ( # noqa: E402,F401 + _check_member_hash, + _hash_matches, + _repo_satisfies_declaration, + _intentional_hash_exclusion, + verify_pack_against_platform, + verify_pack, + generate_sha256sums, +) + + +# _OFFLINE is deliberately not re-exported: importing a value copies it, so a +# facade name would keep saying False after set_offline() changed the flag. +from packresolve import ( # noqa: E402,F401 + _detect_hash_type, + parse_hash_input, + parse_hash_file, + lookup_hashes, + _find_candidate_satisfying_both, + resolve_file, +) if __name__ == "__main__": main() diff --git a/scripts/packextras.py b/scripts/packextras.py new file mode 100644 index 00000000..8cb67773 --- /dev/null +++ b/scripts/packextras.py @@ -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'}" diff --git a/scripts/packpaths.py b/scripts/packpaths.py new file mode 100644 index 00000000..63910345 --- /dev/null +++ b/scripts/packpaths.py @@ -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 diff --git a/scripts/packreadme.py b/scripts/packreadme.py new file mode 100644 index 00000000..f5ad8c08 --- /dev/null +++ b/scripts/packreadme.py @@ -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" diff --git a/scripts/packresolve.py b/scripts/packresolve.py new file mode 100644 index 00000000..9c0fcaef --- /dev/null +++ b/scripts/packresolve.py @@ -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" diff --git a/scripts/packverify.py b/scripts/packverify.py new file mode 100644 index 00000000..768d1748 --- /dev/null +++ b/scripts/packverify.py @@ -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, + )