mirror of
https://github.com/Abdess/retroarch_system.git
synced 2026-10-11 05:53:23 -05:00
1725 lines
64 KiB
Python
1725 lines
64 KiB
Python
"""Shared utilities for retrobios scripts.
|
|
|
|
Single source of truth for platform config loading, hash computation,
|
|
and file resolution - eliminates DRY violations across scripts.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import functools
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
|
|
try:
|
|
import yaml
|
|
except ImportError:
|
|
yaml = None
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def load_database(db_path: str) -> dict:
|
|
"""Load database.json and return parsed dict."""
|
|
with open(db_path) as f:
|
|
return json.load(f)
|
|
|
|
|
|
|
|
_casefold_index_cache: dict[int, dict[str, list[str]]] = {}
|
|
|
|
|
|
def _casefold_name_index(by_name: dict) -> dict[str, list[str]]:
|
|
"""Build (and cache) a casefolded view of the by_name index."""
|
|
key = id(by_name)
|
|
cached = _casefold_index_cache.get(key)
|
|
if cached is not None:
|
|
return cached
|
|
folded: dict[str, list[str]] = {}
|
|
for name, sha1s in by_name.items():
|
|
folded.setdefault(name.casefold(), []).extend(sha1s)
|
|
_casefold_index_cache[key] = folded
|
|
return folded
|
|
|
|
|
|
def name_match_size_ok(file_entry: dict, candidate_size: int | None) -> bool:
|
|
"""Whether a candidate found by name may be the file the entry describes.
|
|
|
|
Generic filenames collide across systems: rom1.bin is a Sony PlayStation 2
|
|
firmware and a Roland SC-55mk2 program ROM. A name carries no evidence on
|
|
its own, so a size the emulator verifies settles it. A size declared
|
|
without ``validation: [size]`` is informative and rejects nothing.
|
|
"""
|
|
validation = file_entry.get("validation")
|
|
if isinstance(validation, dict):
|
|
validation = validation.get("core", [])
|
|
if "size" not in (validation or []) or candidate_size is None:
|
|
return True
|
|
return size_fits(file_entry, candidate_size)
|
|
|
|
|
|
def size_fits(file_entry: dict, candidate_size: int) -> bool:
|
|
"""Whether a size is one the entry declares.
|
|
|
|
``size`` is an exact size or the list of sizes the code accepts; it
|
|
outranks a ``min_size``/``max_size`` range. Nothing declared fits all.
|
|
"""
|
|
declared = file_entry.get("size")
|
|
if declared is not None:
|
|
allowed = declared if isinstance(declared, list) else [declared]
|
|
return candidate_size in allowed
|
|
low = file_entry.get("min_size")
|
|
high = file_entry.get("max_size")
|
|
if low is not None and candidate_size < low:
|
|
return False
|
|
return not (high is not None and candidate_size > high)
|
|
|
|
|
|
|
|
|
|
_shared_yml_cache: dict[str, dict] = {}
|
|
_platform_config_cache: dict[tuple[str, str], dict] = {}
|
|
|
|
|
|
def load_platform_config(platform_name: str, platforms_dir: str = "platforms") -> dict:
|
|
"""Load a platform config with inheritance and shared group resolution.
|
|
|
|
This is the SINGLE implementation used by generate_pack, generate_readme,
|
|
verify, and auto_fetch. No other copy should exist.
|
|
"""
|
|
cache_key = (platform_name, os.path.realpath(platforms_dir))
|
|
if cache_key in _platform_config_cache:
|
|
return _platform_config_cache[cache_key]
|
|
|
|
if yaml is None:
|
|
raise ImportError("PyYAML required: pip install pyyaml")
|
|
|
|
config_file = os.path.join(platforms_dir, f"{platform_name}.yml")
|
|
if not os.path.exists(config_file):
|
|
raise FileNotFoundError(f"Platform config not found: {config_file}")
|
|
|
|
with open(config_file) as f:
|
|
config = yaml_load(f) or {}
|
|
|
|
# Resolve inheritance
|
|
if "inherits" in config:
|
|
parent = load_platform_config(config["inherits"], platforms_dir)
|
|
merged = {**parent}
|
|
merged.update(
|
|
{k: v for k, v in config.items() if k not in ("inherits", "overrides")}
|
|
)
|
|
if "overrides" in config and "systems" in config["overrides"]:
|
|
merged.setdefault("systems", {})
|
|
for sys_id, override in config["overrides"]["systems"].items():
|
|
if sys_id in merged["systems"]:
|
|
merged["systems"][sys_id] = {
|
|
**merged["systems"][sys_id],
|
|
**override,
|
|
}
|
|
else:
|
|
merged["systems"][sys_id] = override
|
|
config = merged
|
|
|
|
# Resolve shared group includes
|
|
shared_path = os.path.join(platforms_dir, "_shared.yml")
|
|
if os.path.exists(shared_path):
|
|
shared_real = os.path.realpath(shared_path)
|
|
if shared_real not in _shared_yml_cache:
|
|
with open(shared_path) as f:
|
|
_shared_yml_cache[shared_real] = yaml_load(f) or {}
|
|
shared = _shared_yml_cache[shared_real]
|
|
shared_groups = shared.get("shared_groups", {})
|
|
for system in config.get("systems", {}).values():
|
|
for group_name in system.get("includes", []):
|
|
if group_name in shared_groups:
|
|
existing = {
|
|
(f.get("name"), f.get("destination", f.get("name")))
|
|
for f in system.get("files", [])
|
|
}
|
|
existing_lower = {
|
|
f.get("destination", f.get("name", "")).lower()
|
|
for f in system.get("files", [])
|
|
}
|
|
for gf in shared_groups[group_name]:
|
|
key = (gf.get("name"), gf.get("destination", gf.get("name")))
|
|
dest_lower = gf.get("destination", gf.get("name", "")).lower()
|
|
if key not in existing and dest_lower not in existing_lower:
|
|
system.setdefault("files", []).append(gf)
|
|
existing.add(key)
|
|
|
|
# Merge metadata from _registry.yml. The registry is our curated source;
|
|
# the scraped YAML may be incomplete (missing cores, metadata fields).
|
|
# Registry fields supplement (not replace) the scraped config.
|
|
registry_path = os.path.join(platforms_dir, "_registry.yml")
|
|
if os.path.exists(registry_path):
|
|
reg_real = os.path.realpath(registry_path)
|
|
if reg_real not in _shared_yml_cache:
|
|
with open(registry_path) as f:
|
|
_shared_yml_cache[reg_real] = yaml_load(f) or {}
|
|
reg = _shared_yml_cache[reg_real]
|
|
reg_entry = reg.get("platforms", {}).get(platform_name, {})
|
|
|
|
# Merge cores (union for lists, override for all_libretro)
|
|
reg_cores = reg_entry.get("cores")
|
|
if reg_cores is not None:
|
|
cfg_cores = config.get("cores")
|
|
if reg_cores == "all_libretro":
|
|
config["cores"] = "all_libretro"
|
|
elif isinstance(reg_cores, list) and isinstance(cfg_cores, list):
|
|
merged_set = {str(c) for c in cfg_cores} | {str(c) for c in reg_cores}
|
|
config["cores"] = sorted(merged_set)
|
|
elif isinstance(reg_cores, list) and cfg_cores is None:
|
|
config["cores"] = reg_cores
|
|
|
|
# Merge all registry fields absent from config (except cores,
|
|
# handled above with union logic). No hardcoded list: any field
|
|
# added to the registry is automatically available in the config.
|
|
for key, val in reg_entry.items():
|
|
if key != "cores" and key not in config:
|
|
config[key] = val
|
|
|
|
_platform_config_cache[cache_key] = config
|
|
return config
|
|
|
|
|
|
def load_data_dir_registry(platforms_dir: str = "platforms") -> dict:
|
|
"""Load the data directory registry from _data_dirs.yml."""
|
|
registry_path = os.path.join(platforms_dir, "_data_dirs.yml")
|
|
if not os.path.exists(registry_path):
|
|
return {}
|
|
with open(registry_path) as f:
|
|
data = yaml_load(f) or {}
|
|
return data.get("data_directories", {})
|
|
|
|
|
|
def load_platform_registry(platforms_dir: str = "platforms") -> dict:
|
|
"""Return the `platforms:` mapping from _registry.yml, empty if absent."""
|
|
registry_path = os.path.join(platforms_dir, "_registry.yml")
|
|
if not os.path.exists(registry_path):
|
|
return {}
|
|
with open(registry_path) as f:
|
|
return (yaml_load(f) or {}).get("platforms", {})
|
|
|
|
|
|
def list_registered_platforms(
|
|
platforms_dir: str = "platforms",
|
|
include_archived: bool = False,
|
|
) -> list[str]:
|
|
"""List platforms registered in _registry.yml.
|
|
|
|
Only registered platforms generate packs and appear in CI.
|
|
Unregistered YAMLs (e.g., emulatorjs.yml) are base configs for inheritance.
|
|
"""
|
|
registry_path = os.path.join(platforms_dir, "_registry.yml")
|
|
if not os.path.exists(registry_path):
|
|
return []
|
|
with open(registry_path) as f:
|
|
registry = yaml_load(f) or {}
|
|
platforms = []
|
|
for name, meta in sorted(registry.get("platforms", {}).items()):
|
|
status = meta.get("status", "active")
|
|
if status == "archived" and not include_archived:
|
|
continue
|
|
config_path = os.path.join(platforms_dir, meta.get("config", f"{name}.yml"))
|
|
if os.path.exists(config_path):
|
|
platforms.append(name)
|
|
return platforms
|
|
|
|
|
|
def load_target_config(
|
|
platform_name: str,
|
|
target: str,
|
|
platforms_dir: str = "platforms",
|
|
) -> set[str]:
|
|
"""Load target config and return the set of core names for the given target.
|
|
|
|
Resolves aliases from _overrides.yml, applies add_cores/remove_cores.
|
|
Raises ValueError if target is unknown (with list of available targets).
|
|
Raises FileNotFoundError if no target file exists for the platform.
|
|
"""
|
|
targets_dir = os.path.join(platforms_dir, "targets")
|
|
target_file = os.path.join(targets_dir, f"{platform_name}.yml")
|
|
if not os.path.exists(target_file):
|
|
raise FileNotFoundError(
|
|
f"No target config for platform '{platform_name}': {target_file}"
|
|
)
|
|
with open(target_file) as f:
|
|
data = yaml_load(f) or {}
|
|
|
|
targets = data.get("targets", {})
|
|
|
|
overrides_file = os.path.join(targets_dir, "_overrides.yml")
|
|
overrides = {}
|
|
if os.path.exists(overrides_file):
|
|
with open(overrides_file) as f:
|
|
all_overrides = yaml_load(f) or {}
|
|
overrides = all_overrides.get(platform_name, {}).get("targets", {})
|
|
|
|
alias_index: dict[str, str] = {}
|
|
for tname in targets:
|
|
alias_index[tname] = tname
|
|
for alias in overrides.get(tname, {}).get("aliases", []):
|
|
alias_index[alias] = tname
|
|
|
|
canonical = alias_index.get(target)
|
|
if canonical is None:
|
|
available = sorted(targets.keys())
|
|
aliases = []
|
|
for tname, ovr in overrides.items():
|
|
for a in ovr.get("aliases", []):
|
|
aliases.append(f"{a} -> {tname}")
|
|
msg = f"Unknown target '{target}' for platform '{platform_name}'.\n"
|
|
msg += f"Available targets: {', '.join(available)}"
|
|
if aliases:
|
|
msg += f"\nAliases: {', '.join(sorted(aliases))}"
|
|
raise ValueError(msg)
|
|
|
|
cores = set(str(c) for c in targets[canonical].get("cores", []))
|
|
return apply_target_overrides(cores, overrides, canonical)
|
|
|
|
|
|
def apply_target_overrides(
|
|
cores: set[str], overrides: dict, target: str
|
|
) -> set[str]:
|
|
"""Cores of one target once the platform's overrides are applied.
|
|
|
|
``_default`` applies to every target of the platform, then the target's
|
|
own entry: a name the scraper reads as a core on every board (a machine
|
|
variant, a launcher) is written once instead of once per board.
|
|
"""
|
|
result = set(cores)
|
|
for scope in ("_default", target):
|
|
entry = overrides.get(scope) or {}
|
|
result |= {str(c) for c in entry.get("add_cores") or []}
|
|
result -= {str(c) for c in entry.get("remove_cores") or []}
|
|
return result
|
|
|
|
|
|
def list_available_targets(
|
|
platform_name: str,
|
|
platforms_dir: str = "platforms",
|
|
) -> list[dict]:
|
|
"""List available targets for a platform with their aliases.
|
|
|
|
Returns list of dicts with keys: name, architecture, core_count, aliases.
|
|
Returns empty list if no target file exists.
|
|
"""
|
|
targets_dir = os.path.join(platforms_dir, "targets")
|
|
target_file = os.path.join(targets_dir, f"{platform_name}.yml")
|
|
if not os.path.exists(target_file):
|
|
return []
|
|
with open(target_file) as f:
|
|
data = yaml_load(f) or {}
|
|
|
|
overrides_file = os.path.join(targets_dir, "_overrides.yml")
|
|
overrides = {}
|
|
if os.path.exists(overrides_file):
|
|
with open(overrides_file) as f:
|
|
all_overrides = yaml_load(f) or {}
|
|
overrides = all_overrides.get(platform_name, {}).get("targets", {})
|
|
|
|
result = []
|
|
for tname, tdata in sorted(data.get("targets", {}).items()):
|
|
aliases = overrides.get(tname, {}).get("aliases", [])
|
|
result.append(
|
|
{
|
|
"name": tname,
|
|
"architecture": tdata.get("architecture", ""),
|
|
"core_count": len(tdata.get("cores", [])),
|
|
"aliases": aliases,
|
|
}
|
|
)
|
|
return result
|
|
|
|
|
|
HASH_EXACT_RESOLUTION_STATUSES = frozenset(
|
|
{
|
|
"sha1_exact",
|
|
"sha256_exact",
|
|
"crc32_exact",
|
|
"md5_exact",
|
|
"md5_composite_exact",
|
|
"zip_exact",
|
|
"data_dir_hash_exact",
|
|
}
|
|
)
|
|
|
|
|
|
def resolution_is_hash_exact(status: str) -> bool:
|
|
"""Whether *status* proves content identity with a declared hash."""
|
|
return status in HASH_EXACT_RESOLUTION_STATUSES
|
|
|
|
|
|
def declared_hash_verdict(
|
|
candidate: str,
|
|
*,
|
|
has_strong_hash: bool,
|
|
sha1_candidates: set,
|
|
sha256_candidates: set,
|
|
md5_list: list,
|
|
crc_raw: str,
|
|
zipped_file: str | None,
|
|
declared_size,
|
|
) -> str:
|
|
"""Judge a candidate found by name against everything the entry declares.
|
|
|
|
A file reached through a directory walk was matched on its filename,
|
|
which is the weakest evidence there is. Whatever the entry declares
|
|
about its content is checked here before the walk may return it.
|
|
"""
|
|
if not has_strong_hash:
|
|
return "data_dir"
|
|
algorithms: set[str] = set()
|
|
if sha1_candidates:
|
|
algorithms.add("sha1")
|
|
if sha256_candidates:
|
|
algorithms.add("sha256")
|
|
if md5_list and not zipped_file:
|
|
algorithms.add("md5")
|
|
if crc_raw:
|
|
algorithms.add("crc32")
|
|
actual = compute_hashes(candidate, frozenset(algorithms)) if algorithms else {}
|
|
if sha1_candidates and actual.get("sha1", "").lower() not in sha1_candidates:
|
|
return "hash_mismatch"
|
|
if sha256_candidates and actual.get("sha256", "").lower() not in sha256_candidates:
|
|
return "hash_mismatch"
|
|
if md5_list and not zipped_file and not any(
|
|
actual.get("md5", "").lower().startswith(expected) for expected in md5_list
|
|
):
|
|
return "hash_mismatch"
|
|
if crc_raw and actual.get("crc32", "").lower() != crc_raw:
|
|
return "hash_mismatch"
|
|
if crc_raw and declared_size is not None:
|
|
allowed_sizes = declared_size if isinstance(declared_size, list) else [declared_size]
|
|
if os.path.getsize(candidate) not in allowed_sizes:
|
|
return "hash_mismatch"
|
|
if zipped_file and md5_list and not any(
|
|
check_inside_zip(candidate, zipped_file, expected) == "ok"
|
|
for expected in md5_list
|
|
):
|
|
return "hash_mismatch"
|
|
return "data_dir_hash_exact"
|
|
|
|
|
|
def _resolve_in_data_dirs(names_to_try, data_dir_registry, verdict):
|
|
"""Look for one of these names in the cached data directories.
|
|
|
|
Returns the first candidate whose content satisfies the entry, and
|
|
separately the first that answered to the name but contradicted it: a
|
|
name match alone never decides, so the caller reports the contradiction
|
|
rather than shipping the file.
|
|
"""
|
|
mismatch: str | None = None
|
|
for _key, dd_entry in data_dir_registry.items():
|
|
cache_dir = dd_entry.get("local_cache", "")
|
|
if not cache_dir or not os.path.isdir(cache_dir):
|
|
continue
|
|
for try_name in names_to_try:
|
|
candidate = os.path.join(cache_dir, try_name)
|
|
if os.path.isfile(candidate):
|
|
status = verdict(candidate)
|
|
if status != "hash_mismatch":
|
|
return (candidate, status), mismatch
|
|
mismatch = mismatch or candidate
|
|
# The declared path may not be where the cache put it, so the whole
|
|
# tree is walked for the bare filename, case-insensitively.
|
|
basename_targets = {
|
|
(n.rsplit("/", 1)[-1] if "/" in n else n).casefold()
|
|
for n in names_to_try
|
|
}
|
|
for root, _dirs, fnames in os.walk(cache_dir):
|
|
for fname in fnames:
|
|
if fname.casefold() in basename_targets:
|
|
candidate = os.path.join(root, fname)
|
|
status = verdict(candidate)
|
|
if status != "hash_mismatch":
|
|
return (candidate, status), mismatch
|
|
mismatch = mismatch or candidate
|
|
return None, mismatch
|
|
|
|
|
|
def _resolve_agnostic(file_entry: dict, files_db: dict, has_strong_hash: bool):
|
|
"""Any file of the right shape under the declared prefix.
|
|
|
|
Some cores accept whatever filename sits in their BIOS directory, so the
|
|
shape -a size or a size range -is all that identifies a candidate. A
|
|
declared hash outranks that, and stops this step being reached.
|
|
"""
|
|
if not file_entry.get("agnostic") or has_strong_hash:
|
|
return None
|
|
prefix = file_entry.get("agnostic_path_prefix", "")
|
|
if not prefix:
|
|
return None
|
|
for _sha1, entry in files_db.items():
|
|
path = entry.get("path", "")
|
|
if not path.startswith(prefix):
|
|
continue
|
|
if size_fits(file_entry, entry.get("size", 0)) and os.path.exists(path):
|
|
return path, "agnostic_fallback"
|
|
return None
|
|
|
|
|
|
def _affinity_tokens(text: str) -> set[str]:
|
|
return {
|
|
token
|
|
for token in (re.sub(r"[^0-9a-z]", "", part.casefold()) for part in text.split("/"))
|
|
if token
|
|
}
|
|
|
|
|
|
def _by_affinity(paths: list[str], file_entry: dict, dest_hint: str) -> list[str]:
|
|
"""Same-named candidates, the ones stored with the file's owner first.
|
|
|
|
A name step without a tie-break took whichever path sorted first: the
|
|
TheXTech coin.ogg for Syobon Action, a Hurrican sound for ARMSX2. The
|
|
destination's directories and the profile that asks name where its own
|
|
copy lives; the order is otherwise kept.
|
|
"""
|
|
hint = dest_hint or file_entry.get("path") or file_entry.get("destination") or ""
|
|
wanted = _affinity_tokens(hint.rsplit("/", 1)[0] if "/" in hint else "")
|
|
for owner in (file_entry.get("source_profile"), file_entry.get("source_emulator")):
|
|
if owner:
|
|
wanted |= _affinity_tokens(str(owner).replace(" ", ""))
|
|
if not wanted or len(paths) < 2:
|
|
return paths
|
|
|
|
def score(path: str) -> int:
|
|
return len(wanted & _affinity_tokens(path.rsplit("/", 1)[0]))
|
|
|
|
return sorted(paths, key=score, reverse=True)
|
|
|
|
|
|
def resolve_local_file(
|
|
file_entry: dict,
|
|
db: dict,
|
|
zip_contents: dict | None = None,
|
|
dest_hint: str = "",
|
|
_depth: int = 0,
|
|
data_dir_registry: dict | None = None,
|
|
) -> tuple[str | None, str]:
|
|
"""Resolve a BIOS file to its local path using database.json.
|
|
|
|
Single source of truth for file resolution, used by both verify.py
|
|
and generate_pack.py. Does NOT handle storage tiers (external/user_provided)
|
|
or release assets - callers handle those.
|
|
|
|
dest_hint: optional destination path (e.g., "GC/USA/IPL.bin") used to
|
|
disambiguate when multiple files share the same name. Matched against
|
|
the by_path_suffix index built from the repo's directory structure.
|
|
|
|
Returns ``(local_path, status)``. Statuses describe the evidence used:
|
|
``sha1_exact``, ``sha256_exact``, ``crc32_exact``, ``md5_exact``,
|
|
``md5_composite_exact``, ``zip_exact``, ``path_exact``, ``name_exact``,
|
|
``hash_mismatch`` or ``not_found`` (plus documented fallback statuses).
|
|
|
|
A path or filename is never allowed to override a declared strong hash.
|
|
"""
|
|
sha1 = file_entry.get("sha1")
|
|
name = file_entry.get("name", "")
|
|
zipped_file = file_entry.get("zipped_file")
|
|
aliases = file_entry.get("aliases", [])
|
|
names_to_try = [name] + [a for a in aliases if a != name]
|
|
|
|
# When name contains a path separator (e.g. "res/tilemap.bin"), also
|
|
# try the basename since by_name indexes filenames without directories
|
|
if "/" in name:
|
|
name_base = name.rsplit("/", 1)[-1]
|
|
if name_base and name_base not in names_to_try:
|
|
names_to_try.append(name_base)
|
|
|
|
# When dest_hint contains a path, also try its basename as a name
|
|
# (handles emulator profiles where name: is descriptive and path: is
|
|
# the actual filename, e.g. name: "MDA font ROM", path: "mda.rom")
|
|
if dest_hint:
|
|
hint_base = dest_hint.rsplit("/", 1)[-1] if "/" in dest_hint else dest_hint
|
|
if hint_base and hint_base not in names_to_try:
|
|
names_to_try.append(hint_base)
|
|
|
|
md5_list = parse_md5_list(file_entry.get("md5"))
|
|
sha1_candidates = [
|
|
str(value).strip().lower()
|
|
for value in (sha1 if isinstance(sha1, list) else [sha1] if sha1 else [])
|
|
if str(value).strip()
|
|
]
|
|
sha256_raw = file_entry.get("sha256")
|
|
sha256_values = sha256_raw if isinstance(sha256_raw, list) else [sha256_raw]
|
|
sha256_candidates = [
|
|
candidate.strip().lower()
|
|
for value in sha256_values
|
|
if value
|
|
for candidate in str(value).split(",")
|
|
if len(candidate.strip()) == 64
|
|
]
|
|
crc_raw = str(file_entry.get("crc32", "") or "").strip().lower()
|
|
declared_size = file_entry.get("size")
|
|
has_strong_hash = bool(
|
|
sha1_candidates or sha256_candidates or md5_list or crc_raw
|
|
)
|
|
files_db = db.get("files", {})
|
|
by_md5 = db.get("indexes", {}).get("by_md5", {})
|
|
by_name = db.get("indexes", {}).get("by_name", {})
|
|
by_path_suffix = db.get("indexes", {}).get("by_path_suffix", {})
|
|
|
|
def _record_match_status(match_sha1: str) -> str | None:
|
|
"""Return hash evidence when a DB record satisfies every declaration."""
|
|
entry = files_db.get(match_sha1)
|
|
if not entry:
|
|
return None
|
|
statuses: list[str] = []
|
|
if sha1_candidates:
|
|
if match_sha1.lower() not in sha1_candidates:
|
|
return None
|
|
statuses.append("sha1_exact")
|
|
if sha256_candidates:
|
|
if str(entry.get("sha256", "")).lower() not in sha256_candidates:
|
|
return None
|
|
statuses.append("sha256_exact")
|
|
# With zipped_file the MD5 identifies the member, not the container.
|
|
if md5_list and not zipped_file:
|
|
actual_md5 = str(entry.get("md5", "")).lower()
|
|
if not any(actual_md5.startswith(expected) for expected in md5_list):
|
|
return None
|
|
statuses.append("md5_exact")
|
|
if crc_raw:
|
|
if str(entry.get("crc32", "")).lower() != crc_raw:
|
|
return None
|
|
if declared_size is not None:
|
|
allowed_sizes = (
|
|
declared_size if isinstance(declared_size, list) else [declared_size]
|
|
)
|
|
if entry.get("size") not in allowed_sizes:
|
|
return None
|
|
statuses.append("crc32_exact")
|
|
return statuses[0] if statuses else None
|
|
|
|
# 1. SHA1 exact match (accept list-valued sha1 from profiles)
|
|
for cand in sha1_candidates:
|
|
if cand in files_db:
|
|
path = files_db[cand]["path"]
|
|
status = _record_match_status(cand)
|
|
if status and os.path.exists(path):
|
|
return path, status
|
|
|
|
# 1b. SHA256 exact match (profiles hashed from sources that publish
|
|
# sha256, e.g. MesenCE). A full sha256 is a strong identifier.
|
|
if sha256_candidates:
|
|
by_sha256 = db.get("indexes", {}).get("by_sha256", {})
|
|
for cand in sha256_candidates:
|
|
match = by_sha256.get(cand)
|
|
if match and match in files_db:
|
|
path = files_db[match]["path"]
|
|
status = _record_match_status(match)
|
|
if status and os.path.exists(path):
|
|
return path, status
|
|
|
|
# 1c. CRC32 lookup, only when no stronger hash is declared. A declared
|
|
# size confirms it when available; a handful of emulators validate CRC32
|
|
# alone, so those entries retain the same (weaker) evidence as the core.
|
|
if (
|
|
crc_raw
|
|
and not zipped_file
|
|
and not md5_list
|
|
and not sha1_candidates
|
|
and not sha256_candidates
|
|
):
|
|
by_crc32 = db.get("indexes", {}).get("by_crc32", {})
|
|
match = by_crc32.get(crc_raw)
|
|
if match and match in files_db:
|
|
entry = files_db[match]
|
|
path = entry["path"]
|
|
status = _record_match_status(match)
|
|
if status and os.path.exists(path):
|
|
return path, status
|
|
|
|
# 2. MD5 direct lookup (skip for zipped_file: md5 is inner ROM, not container)
|
|
# Guard: only accept if the found file's name matches the requested name
|
|
# (or is a .variants/ derivative). Prevents cross-contamination when an
|
|
# unrelated file happens to share the same MD5 in the index.
|
|
_name_set = set(names_to_try)
|
|
|
|
def _md5_name_ok(candidate_path: str) -> bool:
|
|
bn = os.path.basename(candidate_path)
|
|
if bn in _name_set:
|
|
return True
|
|
# .variants/ pattern: filename like "neogeo.zip.fc398ab4"
|
|
return any(bn.startswith(n + ".") for n in _name_set)
|
|
|
|
if md5_list and not zipped_file:
|
|
for md5_candidate in md5_list:
|
|
sha1_match = by_md5.get(md5_candidate)
|
|
if sha1_match and sha1_match in files_db:
|
|
path = files_db[sha1_match]["path"]
|
|
# Full MD5 (32 chars) is a strong identifier: trust it
|
|
# without name guard. Truncated MD5 still needs name check
|
|
# to avoid cross-contamination.
|
|
status = _record_match_status(sha1_match)
|
|
if status and os.path.exists(path):
|
|
if len(md5_candidate) >= 32 or _md5_name_ok(path):
|
|
return path, status
|
|
if len(md5_candidate) < 32:
|
|
for db_md5, db_sha1 in by_md5.items():
|
|
if db_md5.startswith(md5_candidate) and db_sha1 in files_db:
|
|
path = files_db[db_sha1]["path"]
|
|
status = _record_match_status(db_sha1)
|
|
if status and os.path.exists(path) and _md5_name_ok(path):
|
|
return path, status
|
|
|
|
# 2b. Path suffix lookup is useful for same-named regional files, but it
|
|
# is identity evidence only when no content hash was declared. A stale
|
|
# or incorrect destination can therefore never mask a hash mismatch.
|
|
own_file_absent = False
|
|
if dest_hint and by_path_suffix:
|
|
# A destination is written from the emulator's point of view
|
|
# ("pcsx2/resources/GameIndex.yaml") and the index from the repo's
|
|
# layout, so the two meet on a tail rather than on the whole string.
|
|
# Longest tail first, and never down to the bare filename: that is the
|
|
# weaker step below, and five files answer to GameIndex.yaml.
|
|
# Two kinds of ambiguity appear once every tail is indexed. A
|
|
# shortened tail can be claimed by several emulators shipping the same
|
|
# asset tree, and it then proves nothing: skip it. The full tail can be
|
|
# claimed twice as well, one storing it at the root of its own
|
|
# directory and one nested deeper; there the file whose path carries
|
|
# the least on top of the destination is the one that IS the
|
|
# destination.
|
|
# The descent drops at most one leading segment. A destination and the
|
|
# tree differ by the wrapper directory at most; beyond that the tail
|
|
# stops describing this file and starts describing whichever emulator
|
|
# happens to store the same asset tree.
|
|
hint_parts = dest_hint.split("/")
|
|
for start in range(min(2, len(hint_parts) - 1)):
|
|
tail = "/".join(hint_parts[start:])
|
|
candidates = by_path_suffix.get(tail, [])
|
|
if start and len(candidates) > 1:
|
|
continue
|
|
# The index names the file this destination designates. Not on
|
|
# disk, it is a release asset the checkout does not hold, and no
|
|
# other file of that name is it.
|
|
if not start and candidates and not any(
|
|
os.path.exists(files_db.get(h, {}).get("path", ""))
|
|
for h in candidates
|
|
):
|
|
own_file_absent = True
|
|
if len(candidates) > 1:
|
|
depth = len(tail.split("/"))
|
|
candidates = sorted(
|
|
candidates,
|
|
key=lambda h: len(
|
|
files_db.get(h, {}).get("path", "").split("/")
|
|
)
|
|
- depth,
|
|
)
|
|
for match_sha1 in candidates:
|
|
if match_sha1 not in files_db:
|
|
continue
|
|
path = files_db[match_sha1]["path"]
|
|
if not os.path.exists(path):
|
|
continue
|
|
if not has_strong_hash:
|
|
return path, "path_exact"
|
|
status = _record_match_status(match_sha1)
|
|
if status:
|
|
return path, status
|
|
|
|
# 3. No MD5 = any file with that name or alias (existence check)
|
|
def _size_ok(match_sha1: str) -> bool:
|
|
return name_match_size_ok(
|
|
file_entry, files_db.get(match_sha1, {}).get("size")
|
|
)
|
|
|
|
# An entry the profile documents as unsourceable is one the collection
|
|
# cannot hold: a font that only exists inside a paid package, cheat codes
|
|
# written per title. A file answering to its name is therefore a homonym
|
|
# from another system, not a stale dump of the same file - kanji.rom names
|
|
# an FPseNG font, an openMSX one and a 3DO ROM. Content still decides: the
|
|
# hash and path steps above run first, so collecting the real bytes makes
|
|
# the entry resolve.
|
|
unsourceable = bool(file_entry.get("unsourceable")) or own_file_absent
|
|
|
|
if not has_strong_hash and not unsourceable:
|
|
candidates = []
|
|
for try_name in names_to_try:
|
|
for match_sha1 in by_name.get(try_name, []):
|
|
if match_sha1 in files_db and _size_ok(match_sha1):
|
|
path = files_db[match_sha1]["path"]
|
|
if os.path.exists(path) and path not in candidates:
|
|
candidates.append(path)
|
|
if not candidates:
|
|
# Case-insensitive fallback: upstream platforms disagree on
|
|
# casing (VEC_MineStorm.vec vs VEC_Minestorm.vec). Safe by
|
|
# invariant: dedup keeps a single casing per content, so a
|
|
# casefold match cannot pick a different file.
|
|
folded = _casefold_name_index(by_name)
|
|
for try_name in names_to_try:
|
|
for match_sha1 in folded.get(try_name.casefold(), []):
|
|
if match_sha1 in files_db and _size_ok(match_sha1):
|
|
path = files_db[match_sha1]["path"]
|
|
if os.path.exists(path) and path not in candidates:
|
|
candidates.append(path)
|
|
if candidates:
|
|
if zipped_file:
|
|
candidates = [p for p in candidates if ".zip" in os.path.basename(p)]
|
|
candidates = _by_affinity(candidates, file_entry, dest_hint)
|
|
primary = [p for p in candidates if "/.variants/" not in p]
|
|
if primary or candidates:
|
|
return (primary[0] if primary else candidates[0]), "name_exact"
|
|
|
|
# 4. Name + alias fallback with md5_composite + direct MD5 per candidate
|
|
md5_set = set(md5_list)
|
|
candidates = []
|
|
seen_paths = set()
|
|
for try_name in names_to_try:
|
|
for match_sha1 in by_name.get(try_name, []):
|
|
if match_sha1 in files_db:
|
|
entry = files_db[match_sha1]
|
|
path = entry["path"]
|
|
if os.path.exists(path) and path not in seen_paths:
|
|
seen_paths.add(path)
|
|
candidates.append(
|
|
(path, entry.get("md5", ""), entry.get("size"))
|
|
)
|
|
|
|
if zipped_file:
|
|
candidates = [c for c in candidates if ".zip" in os.path.basename(c[0])]
|
|
# A size the emulator itself verifies settles a name collision, so a
|
|
# candidate that fits is preferred over one that does not: "ROM" answers
|
|
# both an Apple IIgs image and a Macintosh one. It ranks rather than
|
|
# excludes, because a present file of the wrong size is reported as
|
|
# present and untested, and dropping it would report it as absent.
|
|
sized = [c for c in candidates if name_match_size_ok(file_entry, c[2])]
|
|
if sized:
|
|
candidates = sized
|
|
|
|
if candidates:
|
|
if md5_set and not (sha1_candidates or sha256_candidates or crc_raw):
|
|
for path, db_md5, _size in candidates:
|
|
if ".zip" in os.path.basename(path):
|
|
try:
|
|
composite = md5_composite(path).lower()
|
|
if composite in md5_set:
|
|
return path, "md5_composite_exact"
|
|
except (zipfile.BadZipFile, OSError):
|
|
pass
|
|
if db_md5.lower() in md5_set:
|
|
return path, "md5_exact"
|
|
# When zipped_file is set, only accept candidates that contain it
|
|
if zipped_file:
|
|
valid = []
|
|
for path, m, size in candidates:
|
|
try:
|
|
with zipfile.ZipFile(path) as zf:
|
|
inner_names = {n.casefold() for n in zf.namelist()}
|
|
if zipped_file.casefold() in inner_names:
|
|
valid.append((path, m, size))
|
|
except (zipfile.BadZipFile, OSError):
|
|
pass
|
|
if valid:
|
|
primary = [c[0] for c in valid if "/.variants/" not in c[0]]
|
|
return (primary[0] if primary else valid[0][0]), "hash_mismatch"
|
|
# No candidate contains the zipped_file -fall through to step 5
|
|
elif not unsourceable:
|
|
ordered = _by_affinity([c[0] for c in candidates], file_entry, dest_hint)
|
|
primary = [c for c in ordered if "/.variants/" not in c]
|
|
return (primary[0] if primary else ordered[0]), "hash_mismatch"
|
|
|
|
# 5. zipped_file content match via pre-built index (last resort:
|
|
# matches inner ROM MD5 across ALL ZIPs in the repo, so only use
|
|
# when name-based resolution failed entirely)
|
|
if (
|
|
zipped_file
|
|
and md5_list
|
|
and zip_contents
|
|
and not (sha1_candidates or sha256_candidates or crc_raw)
|
|
):
|
|
for md5_candidate in md5_list:
|
|
if md5_candidate in zip_contents:
|
|
zip_sha1 = zip_contents[md5_candidate]
|
|
if zip_sha1 in files_db:
|
|
path = files_db[zip_sha1]["path"]
|
|
if os.path.exists(path):
|
|
return path, "zip_exact"
|
|
|
|
# MAME clone fallback: if a file was deduped, resolve via canonical
|
|
if _depth < 3 and not has_strong_hash:
|
|
clone_map = get_mame_clone_map()
|
|
canonical = clone_map.get(name)
|
|
if canonical and canonical != name:
|
|
canonical_entry = {"name": canonical}
|
|
result = resolve_local_file(
|
|
canonical_entry,
|
|
db,
|
|
zip_contents,
|
|
dest_hint,
|
|
_depth=_depth + 1,
|
|
data_dir_registry=data_dir_registry,
|
|
)
|
|
if result[0]:
|
|
return result[0], "mame_clone"
|
|
|
|
# Data directory fallback: scan data/ caches for matching filename
|
|
|
|
data_dir_mismatch: str | None = None
|
|
# Without a hash the cache walk matches on filename alone, which is the
|
|
# step an unsourceable entry has to skip: hiscore.dat names one file per
|
|
# driver set, so FBNeo's copy would answer for MAME's.
|
|
if data_dir_registry and (has_strong_hash or not unsourceable):
|
|
verdict = functools.partial(
|
|
declared_hash_verdict,
|
|
has_strong_hash=has_strong_hash,
|
|
sha1_candidates=sha1_candidates,
|
|
sha256_candidates=sha256_candidates,
|
|
md5_list=md5_list,
|
|
crc_raw=crc_raw,
|
|
zipped_file=zipped_file,
|
|
declared_size=declared_size,
|
|
)
|
|
hit, data_dir_mismatch = _resolve_in_data_dirs(
|
|
names_to_try, data_dir_registry, verdict
|
|
)
|
|
if hit:
|
|
return hit
|
|
|
|
if data_dir_mismatch and not unsourceable:
|
|
return data_dir_mismatch, "hash_mismatch"
|
|
|
|
agnostic = _resolve_agnostic(file_entry, files_db, has_strong_hash)
|
|
if agnostic:
|
|
return agnostic
|
|
|
|
return None, "not_found"
|
|
|
|
|
|
_mame_clone_map_cache: dict[str, str] | None = None
|
|
|
|
|
|
def get_mame_clone_map() -> dict[str, str]:
|
|
"""Load and cache the MAME clone map (clone_name -> canonical_name)."""
|
|
global _mame_clone_map_cache
|
|
if _mame_clone_map_cache is not None:
|
|
return _mame_clone_map_cache
|
|
clone_path = os.path.join(
|
|
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
|
"_mame_clones.json",
|
|
)
|
|
if os.path.exists(clone_path):
|
|
with open(clone_path) as f:
|
|
data = json.load(f)
|
|
_mame_clone_map_cache = {}
|
|
for canonical, info in data.items():
|
|
for clone in info.get("clones", []):
|
|
_mame_clone_map_cache[clone] = canonical
|
|
else:
|
|
_mame_clone_map_cache = {}
|
|
return _mame_clone_map_cache
|
|
|
|
|
|
|
|
|
|
|
|
_emulator_profiles_cache: dict[tuple[str, bool], dict[str, dict]] = {}
|
|
|
|
|
|
class ProfileSelectionError(ValueError):
|
|
"""A named profile cannot answer for itself."""
|
|
|
|
|
|
def select_emulator_profiles(
|
|
profile_names: list[str],
|
|
all_profiles: dict,
|
|
standalone: bool = False,
|
|
) -> list[tuple[str, dict]]:
|
|
"""Resolve named profiles, refusing the ones that answer for others.
|
|
|
|
An alias is the same binary under another name and a launcher only
|
|
starts an emulator, so neither has requirements of its own: naming one
|
|
is a question about the wrong profile, and the error says which one to
|
|
ask instead. Raising rather than exiting lets the verifier stop and the
|
|
builder return empty-handed from one rule.
|
|
"""
|
|
selected: list[tuple[str, dict]] = []
|
|
for name in profile_names:
|
|
if name not in all_profiles:
|
|
available = sorted(
|
|
key
|
|
for key, value in all_profiles.items()
|
|
if value.get("type") not in ("alias", "test")
|
|
)
|
|
raise ProfileSelectionError(
|
|
f"emulator '{name}' not found\n"
|
|
f"Available: {', '.join(available[:10])}..."
|
|
)
|
|
profile = all_profiles[name]
|
|
kind = profile.get("type", "libretro")
|
|
if kind == "alias":
|
|
raise ProfileSelectionError(
|
|
f"{name} is an alias of {profile.get('alias_of', '?')} "
|
|
f"-use --emulator {profile.get('alias_of', '?')}"
|
|
)
|
|
if kind == "launcher":
|
|
raise ProfileSelectionError(
|
|
f"{name} is a launcher -use the emulator it launches"
|
|
)
|
|
if standalone and "standalone" not in kind:
|
|
raise ProfileSelectionError(
|
|
f"{name} ({kind}) does not support --standalone"
|
|
)
|
|
selected.append((name, profile))
|
|
return selected
|
|
|
|
|
|
def load_emulator_profiles(
|
|
emulators_dir: str,
|
|
skip_aliases: bool = True,
|
|
) -> dict[str, dict]:
|
|
"""Load all emulator YAML profiles from a directory (cached)."""
|
|
cache_key = (os.path.realpath(emulators_dir), skip_aliases)
|
|
if cache_key in _emulator_profiles_cache:
|
|
return _emulator_profiles_cache[cache_key]
|
|
try:
|
|
import yaml
|
|
except ImportError:
|
|
# Every consumer reads {} as "this repo documents no emulator", which
|
|
# is what a broken install looks like from the outside: zero coverage,
|
|
# no gap, nothing to fix.
|
|
print(
|
|
"warning: pyyaml is not installed, no emulator profile was loaded",
|
|
file=sys.stderr,
|
|
)
|
|
return {}
|
|
profiles = {}
|
|
emu_path = Path(emulators_dir)
|
|
if not emu_path.exists():
|
|
print(
|
|
f"warning: no emulator profile directory at {emulators_dir}",
|
|
file=sys.stderr,
|
|
)
|
|
return profiles
|
|
for f in sorted(emu_path.glob("*.yml")):
|
|
if f.name.endswith(".old.yml"):
|
|
continue
|
|
with open(f) as fh:
|
|
profile = yaml_load(fh) or {}
|
|
if "emulator" not in profile:
|
|
continue
|
|
if skip_aliases and profile.get("type") == "alias":
|
|
continue
|
|
_normalize_hash_fields(profile)
|
|
profiles[f.stem] = profile
|
|
_emulator_profiles_cache[cache_key] = profiles
|
|
return profiles
|
|
|
|
|
|
_HASH_FIELDS = ("crc32", "md5", "sha1", "sha256")
|
|
|
|
|
|
def _normalize_hash_fields(node: object) -> None:
|
|
"""Coerce hash values to str in place.
|
|
|
|
YAML types unquoted digit-only hashes (e.g. crc32: 70295038) as int,
|
|
which breaks string comparisons and formatting downstream.
|
|
"""
|
|
if isinstance(node, dict):
|
|
for key, val in node.items():
|
|
if key in _HASH_FIELDS:
|
|
if isinstance(val, int):
|
|
node[key] = str(val)
|
|
elif isinstance(val, list):
|
|
node[key] = [str(v) if isinstance(v, int) else v for v in val]
|
|
else:
|
|
_normalize_hash_fields(val)
|
|
elif isinstance(node, list):
|
|
for item in node:
|
|
_normalize_hash_fields(item)
|
|
|
|
|
|
def unique_emulator_profiles(profiles: dict[str, dict]) -> dict[str, dict]:
|
|
"""Profiles counting as distinct emulators (aliases and tests excluded)."""
|
|
return {
|
|
k: v for k, v in profiles.items() if v.get("type") not in ("alias", "test")
|
|
}
|
|
|
|
|
|
# Trees that hold game and engine data rather than the system files of a
|
|
# machine: the engine trees, and the publishers whose games an engine runs.
|
|
GAME_DATA_TOPS = (
|
|
"RPG Maker",
|
|
"ScummVM",
|
|
"Game Engines",
|
|
"sdlpal",
|
|
"Dinothawr",
|
|
"3D Realms",
|
|
"Apogee",
|
|
"Blizzard",
|
|
"Crack dot Com",
|
|
"Electronic Arts",
|
|
"Epic MegaGames",
|
|
"Id Software",
|
|
"Interplay",
|
|
"Monolith",
|
|
"Softdisk",
|
|
"Valve",
|
|
)
|
|
|
|
|
|
def composition_tier(path: str) -> str:
|
|
"""Which composition bucket a repository path belongs to."""
|
|
parts = path.split("/")
|
|
top = parts[1] if len(parts) > 1 else ""
|
|
if top == "Arcade":
|
|
return "arcade"
|
|
if top in GAME_DATA_TOPS:
|
|
return "game_data"
|
|
return "systems"
|
|
|
|
|
|
def compute_composition(db: dict) -> dict:
|
|
"""File and byte counts by tree area.
|
|
|
|
Three buckets, each re-derivable from paths alone: arcade ROM sets
|
|
(Arcade/), game and engine data (the trees of GAME_DATA_TOPS), and
|
|
console or computer system files (everything else).
|
|
"""
|
|
buckets = {
|
|
"systems": {"files": 0, "size_bytes": 0},
|
|
"arcade": {"files": 0, "size_bytes": 0},
|
|
"game_data": {"files": 0, "size_bytes": 0},
|
|
}
|
|
for entry in db.get("files", {}).values():
|
|
bucket = buckets[composition_tier(entry.get("path", ""))]
|
|
bucket["files"] += 1
|
|
bucket["size_bytes"] += entry.get("size", 0)
|
|
return buckets
|
|
|
|
|
|
def count_catalog_matched(db: dict) -> int:
|
|
"""System files byte-identical to a dump-preservation catalog entry.
|
|
|
|
Scoped to the systems bucket, the denominator every surface pairs it
|
|
with: No-Intro, Redump and TOSEC index console and computer dumps, so
|
|
arcade ROM sets and engine data sit on neither side of the ratio.
|
|
"""
|
|
return sum(
|
|
1
|
|
for entry in db.get("files", {}).values()
|
|
if entry.get("provenance")
|
|
and composition_tier(entry.get("path", "")) == "systems"
|
|
)
|
|
|
|
|
|
def group_identical_platforms(
|
|
platforms: list[str],
|
|
platforms_dir: str,
|
|
target_cores_cache: dict[str, set[str] | None] | None = None,
|
|
) -> list[tuple[list[str], str]]:
|
|
"""Group platforms that produce identical packs.
|
|
|
|
A pack is the platform's own files plus those of the emulators it runs,
|
|
so two platforms share one only when both agree. base_destination does
|
|
not separate them: destinations inside a pack are relative to it.
|
|
|
|
Returns [(group_of_platform_names, representative), ...].
|
|
The representative is the root platform (one that does not inherit).
|
|
"""
|
|
fingerprints: dict[str, list[str]] = {}
|
|
representatives: dict[str, str] = {}
|
|
inherits: dict[str, bool] = {}
|
|
|
|
for platform in platforms:
|
|
try:
|
|
raw_path = os.path.join(platforms_dir, f"{platform}.yml")
|
|
with open(raw_path) as f:
|
|
raw = yaml_load(f) or {}
|
|
inherits[platform] = "inherits" in raw
|
|
config = load_platform_config(platform, platforms_dir)
|
|
except FileNotFoundError:
|
|
fingerprints.setdefault(platform, []).append(platform)
|
|
representatives.setdefault(platform, platform)
|
|
inherits[platform] = False
|
|
continue
|
|
|
|
entries = []
|
|
for sys_id, system in sorted(config.get("systems", {}).items()):
|
|
for fe in system.get("files", []):
|
|
dest = fe.get("destination", fe.get("name", ""))
|
|
sha1 = fe.get("sha1", "")
|
|
md5 = fe.get("md5", "")
|
|
entries.append(f"{dest}|{sha1}|{md5}")
|
|
|
|
fp = hashlib.sha1("|".join(sorted(entries)).encode()).hexdigest()
|
|
cores = config.get("cores")
|
|
emulators = [
|
|
cores if isinstance(cores, str) else sorted(map(str, cores or [])),
|
|
sorted(map(str, config.get("standalone_cores") or [])),
|
|
]
|
|
fp = hashlib.sha1(f"{fp}|{emulators}".encode()).hexdigest()
|
|
if target_cores_cache:
|
|
tc = target_cores_cache.get(platform)
|
|
if tc is not None:
|
|
tc_str = "|".join(sorted(tc))
|
|
fp = hashlib.sha1(f"{fp}|{tc_str}".encode()).hexdigest()
|
|
fingerprints.setdefault(fp, []).append(platform)
|
|
# Prefer the root platform (no inherits) as representative
|
|
if fp not in representatives or (
|
|
not inherits[platform] and inherits.get(representatives[fp], False)
|
|
):
|
|
representatives[fp] = platform
|
|
|
|
result = []
|
|
for fp, group in fingerprints.items():
|
|
rep = representatives[fp]
|
|
ordered = [rep] + [p for p in group if p != rep]
|
|
result.append((ordered, rep))
|
|
return result
|
|
|
|
|
|
_directory_members_cache: dict[int, tuple[dict, dict[str, dict[str, str]]]] = {}
|
|
|
|
|
|
def directory_members(db: dict, directory: str) -> dict[str, str]:
|
|
"""Files the collection holds under a directory an emulator reads whole.
|
|
|
|
Returns {destination: sha1}, each destination starting with the directory
|
|
as the profile writes it. The collection and the destination meet on a
|
|
tail, so the members are the indexed tails that begin with it.
|
|
|
|
A generic directory name can sit in two unrelated trees (``voice/``,
|
|
``data/``). Members that do not share one parent, or a destination two
|
|
contents claim, would merge them into one emulator's directory: nothing
|
|
is returned and the entry stays a reported gap.
|
|
"""
|
|
tail = directory.strip("/")
|
|
if not tail:
|
|
return {}
|
|
by_path_suffix = db.get("indexes", {}).get("by_path_suffix", {})
|
|
cached = _directory_members_cache.get(id(by_path_suffix))
|
|
if cached is None or cached[0] is not by_path_suffix:
|
|
cached = (by_path_suffix, {})
|
|
_directory_members_cache[id(by_path_suffix)] = cached
|
|
if tail in cached[1]:
|
|
return cached[1][tail]
|
|
|
|
prefix = f"{tail}/"
|
|
infix = f"/{prefix}"
|
|
held: dict[str, list[str]] = {}
|
|
parent: dict[str, str] = {}
|
|
for key, sha1s in by_path_suffix.items():
|
|
if key.startswith(prefix):
|
|
held[key] = sha1s
|
|
continue
|
|
at = key.find(infix)
|
|
if at > 0:
|
|
member = key[at + 1:]
|
|
if len(key[:at]) > len(parent.get(member, "")):
|
|
parent[member] = key[:at]
|
|
held = {
|
|
dest: sha1s for dest, sha1s in held.items() if "/.variants/" not in dest
|
|
}
|
|
one_tree = len({parent.get(dest, "") for dest in held}) <= 1
|
|
one_content = all(len(sha1s) == 1 for sha1s in held.values())
|
|
members = (
|
|
{dest: held[dest][0] for dest in sorted(held)}
|
|
if one_tree and one_content
|
|
else {}
|
|
)
|
|
cached[1][tail] = members
|
|
return members
|
|
|
|
|
|
def expand_directory_entries(
|
|
files: list[dict], db: dict, standalone: bool = False
|
|
) -> list[dict]:
|
|
"""Replace each ``type: directory`` entry by the files held under it.
|
|
|
|
A profile declares the directory its code walks; what a pack can carry
|
|
is the files. Each member keeps the entry's fields and is pinned to the
|
|
content the collection holds at that destination. A directory nothing
|
|
answers is kept as declared, one gap, and a member the profile also
|
|
names as a file is left to that entry.
|
|
"""
|
|
def destination(entry: dict) -> str:
|
|
chosen = entry.get("standalone_path") if standalone else None
|
|
return chosen or entry.get("path") or entry.get("name", "")
|
|
|
|
if not any(entry.get("type") == "directory" for entry in files):
|
|
return files
|
|
named = {
|
|
destination(entry) for entry in files if entry.get("type") != "directory"
|
|
}
|
|
expanded: list[dict] = []
|
|
for entry in files:
|
|
if entry.get("type") != "directory":
|
|
expanded.append(entry)
|
|
continue
|
|
members = directory_members(db, destination(entry))
|
|
if not members:
|
|
expanded.append(entry)
|
|
continue
|
|
inherited = {
|
|
key: value
|
|
for key, value in entry.items()
|
|
if key not in ("type", "standalone_path")
|
|
}
|
|
expanded.extend(
|
|
{**inherited, "name": dest, "path": dest, "sha1": sha1}
|
|
for dest, sha1 in members.items()
|
|
if dest not in named
|
|
)
|
|
return expanded
|
|
|
|
|
|
def runs_standalone(
|
|
emu_name: str, profile: dict, standalone_cores: set[str]
|
|
) -> bool:
|
|
"""Whether a platform lays this emulator's files out for its standalone build.
|
|
|
|
A platform names the emulators it launches standalone in
|
|
``standalone_cores``, by profile key or by any name in ``cores:``. The
|
|
name alone is not enough: Recalbox and Batocera call their standalone
|
|
ScummVM ``scummvm``, which is also the key of the libretro core's
|
|
profile, and that profile documents no standalone build. Laid out
|
|
"standalone" it would lose every ``path:`` and drop its files at the
|
|
root, so a profile whose ``type`` has no standalone build keeps its
|
|
libretro layout whatever the platform calls it.
|
|
"""
|
|
if not standalone_cores:
|
|
return False
|
|
if "standalone" not in str(profile.get("type", "")):
|
|
return False
|
|
return emu_name in standalone_cores or bool(
|
|
standalone_cores & {str(c) for c in profile.get("cores", [])}
|
|
)
|
|
|
|
|
|
# What identifies a profile entry's content, carried unchanged wherever the
|
|
# entry travels (cross-reference report, core extras, second-pass copies).
|
|
# Each copy once kept its own subset: np2kai's bios.rom lost min_size and
|
|
# validation on the way and resolved to an IBM PCjr ROM of the same name.
|
|
PROFILE_IDENTITY_FIELDS = (
|
|
"sha1", "md5", "sha256", "crc32", "size", "min_size", "max_size",
|
|
"validation", "aliases", "unsourceable",
|
|
)
|
|
|
|
|
|
def upstream_profile_index(
|
|
profiles: dict[str, dict], include_aliases: bool = False
|
|
) -> dict[str, set[str]]:
|
|
"""Every name a core is known by upstream -> the profiles that claim it.
|
|
|
|
A name can be claimed twice: the buildbot's `pcsx2` is the LRPS2 core,
|
|
which lrps2.yml lists in `cores:`, and also the key of the standalone
|
|
pcsx2 profile. A last-wins dict kept one of the two depending on file
|
|
order, and a RetroArch target filter dropped LRPS2's files. Callers
|
|
choose: a platform list prefers the exact key, a target filter keeps
|
|
every claimant.
|
|
"""
|
|
index: dict[str, set[str]] = {}
|
|
for name, profile in profiles.items():
|
|
if not include_aliases and profile.get("type") == "alias":
|
|
continue
|
|
index.setdefault(name, set()).add(name)
|
|
for core in profile.get("cores") or []:
|
|
index.setdefault(str(core), set()).add(name)
|
|
return index
|
|
|
|
|
|
def preferred_profile(index: dict[str, set[str]], name: str) -> str | None:
|
|
"""The profile a name designates: its own key first, else the first claimant."""
|
|
claimants = index.get(name)
|
|
if not claimants:
|
|
return None
|
|
return name if name in claimants else sorted(claimants)[0]
|
|
|
|
|
|
def resolve_platform_cores(
|
|
config: dict,
|
|
profiles: dict[str, dict],
|
|
target_cores: set[str] | None = None,
|
|
) -> set[str]:
|
|
"""Resolve which emulator profiles are relevant for a platform.
|
|
|
|
Resolution strategies (by priority):
|
|
1. cores: "all_libretro" -- all profiles with libretro in type
|
|
2. cores: [list] -- profiles whose dict key matches a core name
|
|
3. cores: absent -- fallback to systems intersection
|
|
|
|
Alias profiles are always excluded (they point to another profile).
|
|
If target_cores is provided, result is intersected with it.
|
|
"""
|
|
cores_config = config.get("cores")
|
|
|
|
if cores_config == "all_libretro":
|
|
result = {
|
|
name
|
|
for name, p in profiles.items()
|
|
if "libretro" in p.get("type", "") and p.get("type") != "alias"
|
|
}
|
|
elif isinstance(cores_config, list):
|
|
core_set = {str(c) for c in cores_config}
|
|
index = upstream_profile_index(profiles)
|
|
result = {
|
|
profile
|
|
for c in core_set
|
|
if (profile := preferred_profile(index, c)) is not None
|
|
}
|
|
# Support "all_libretro" as a list element: combines all libretro
|
|
# profiles with explicitly listed standalone cores (e.g. RetroDECK
|
|
# ships RetroArch + standalone emulators)
|
|
if "all_libretro" in core_set or "retroarch" in core_set:
|
|
result |= {
|
|
name
|
|
for name, p in profiles.items()
|
|
if "libretro" in p.get("type", "") and p.get("type") != "alias"
|
|
}
|
|
else:
|
|
# Fallback: system ID intersection with normalization
|
|
norm_plat_systems = {_norm_system_id(s) for s in config.get("systems", {})}
|
|
result = {
|
|
name
|
|
for name, p in profiles.items()
|
|
if {_norm_system_id(s) for s in p.get("systems", [])} & norm_plat_systems
|
|
and p.get("type") != "alias"
|
|
}
|
|
|
|
if target_cores is not None:
|
|
# Build reverse index: upstream name -> profile key
|
|
# Upstream sources (buildbot, es_systems) may use different names
|
|
# than our profile keys (e.g., mednafen_psx vs beetle_psx).
|
|
# The profiles' cores: field lists these alternate names.
|
|
index = upstream_profile_index(profiles, include_aliases=True)
|
|
expanded = {
|
|
profile for c in target_cores for profile in index.get(c, {c})
|
|
}
|
|
result = result & expanded
|
|
return result
|
|
|
|
|
|
MANUFACTURER_PREFIXES = (
|
|
"acorn-",
|
|
"apple-",
|
|
"microsoft-",
|
|
"nintendo-",
|
|
"sony-",
|
|
"sega-",
|
|
"snk-",
|
|
"panasonic-",
|
|
"nec-",
|
|
"epoch-",
|
|
"mattel-",
|
|
"fairchild-",
|
|
"hartung-",
|
|
"tiger-",
|
|
"magnavox-",
|
|
"philips-",
|
|
"bandai-",
|
|
"casio-",
|
|
"coleco-",
|
|
"commodore-",
|
|
"sharp-",
|
|
"sinclair-",
|
|
"atari-",
|
|
"sammy-",
|
|
"gce-",
|
|
"interton-",
|
|
"texas-instruments-",
|
|
"videoton-",
|
|
"wenquxing-",
|
|
"aquaplus-",
|
|
)
|
|
|
|
|
|
def derive_manufacturer(system_id: str, system_data: dict) -> str:
|
|
"""Derive manufacturer name for a system.
|
|
|
|
Priority: explicit manufacturer field > system ID prefix > 'Other'.
|
|
"""
|
|
mfr = system_data.get("manufacturer", "")
|
|
if mfr and mfr not in ("Various", "Other"):
|
|
return mfr.split("|")[0].strip()
|
|
s = system_id.lower().replace("_", "-")
|
|
for prefix in MANUFACTURER_PREFIXES:
|
|
if s.startswith(prefix):
|
|
return prefix.rstrip("-").title()
|
|
return "Other"
|
|
|
|
|
|
# Abbreviations that normalization alone cannot resolve.
|
|
# Maps platform-specific short names to canonical profile system IDs.
|
|
SYSTEM_ALIASES: dict[str, str] = {
|
|
"gmaster": "hartung-game-master",
|
|
"n64dd": "nintendo-64dd",
|
|
"neogeo64": "hyper-neogeo64",
|
|
"psvita": "sony-playstation-vita",
|
|
# Platform IDs missing the manufacturer-prefix hyphen
|
|
"atari5200": "atari-5200",
|
|
"atari7800": "atari-7800",
|
|
"atarist": "atari-st",
|
|
"sega32x": "sega-32x",
|
|
"segastv": "sega-stv",
|
|
"ti994a": "ti99",
|
|
}
|
|
|
|
|
|
def _norm_system_id(sid: str) -> str:
|
|
"""Normalize system ID for cross-platform matching.
|
|
|
|
Resolves known aliases, then strips manufacturer prefixes and separators
|
|
so that platform-specific IDs (e.g., "xbox", "nintendo-wiiu") match
|
|
profile IDs (e.g., "microsoft-xbox", "nintendo-wii-u").
|
|
"""
|
|
s = sid.lower().replace("_", "-")
|
|
s = SYSTEM_ALIASES.get(s, s)
|
|
for prefix in MANUFACTURER_PREFIXES:
|
|
if s.startswith(prefix):
|
|
s = s[len(prefix) :]
|
|
break
|
|
return s.replace("-", "")
|
|
|
|
|
|
def filter_systems_by_target(
|
|
systems: dict[str, dict],
|
|
profiles: dict[str, dict],
|
|
target_cores: set[str] | None,
|
|
platform_cores: set[str] | None = None,
|
|
) -> dict[str, dict]:
|
|
"""Filter platform systems to only those reachable by target cores.
|
|
|
|
A system is reachable if at least one core that emulates it is available
|
|
on the target. Only considers cores relevant to the platform (from
|
|
platform_cores). Systems whose cores are all outside the platform's
|
|
scope are kept (no information to exclude them).
|
|
|
|
Returns the filtered systems dict (or all if no target).
|
|
"""
|
|
if target_cores is None:
|
|
return systems
|
|
|
|
# Build reverse index for target core name resolution
|
|
index = upstream_profile_index(profiles, include_aliases=True)
|
|
expanded_target = {
|
|
profile for c in target_cores for profile in index.get(c, {c})
|
|
}
|
|
|
|
_norm_sid = _norm_system_id
|
|
|
|
# Build normalized system -> cores from ALL profiles
|
|
norm_system_cores: dict[str, set[str]] = {}
|
|
for name, p in profiles.items():
|
|
if p.get("type") == "alias":
|
|
continue
|
|
for sid in p.get("systems", []):
|
|
norm_key = _norm_sid(sid)
|
|
norm_system_cores.setdefault(norm_key, set()).add(name)
|
|
|
|
# Platform-scoped mapping (for distinguishing "no info" from "known but off-target")
|
|
norm_plat_system_cores: dict[str, set[str]] = {}
|
|
if platform_cores is not None:
|
|
for name in platform_cores:
|
|
p = profiles.get(name, {})
|
|
for sid in p.get("systems", []):
|
|
norm_key = _norm_sid(sid)
|
|
norm_plat_system_cores.setdefault(norm_key, set()).add(name)
|
|
|
|
filtered = {}
|
|
for sys_id, sys_data in systems.items():
|
|
norm_key = _norm_sid(sys_id)
|
|
all_cores = norm_system_cores.get(norm_key, set())
|
|
plat_cores_here = norm_plat_system_cores.get(norm_key, set())
|
|
|
|
if not all_cores and not plat_cores_here:
|
|
# No profile maps to this system -keep it
|
|
filtered[sys_id] = sys_data
|
|
elif all_cores & expanded_target:
|
|
# At least one core is on the target
|
|
filtered[sys_id] = sys_data
|
|
elif not plat_cores_here:
|
|
# Platform resolution didn't find cores for this system -keep it
|
|
filtered[sys_id] = sys_data
|
|
# else: known cores exist but none are on the target -exclude
|
|
return filtered
|
|
|
|
|
|
def expand_platform_declared_names(config: dict, db: dict) -> set[str]:
|
|
"""Build set of file names declared by a platform config.
|
|
|
|
Enriches the set with canonical names and aliases from the database
|
|
by resolving each platform file's MD5 through by_md5. This handles
|
|
cases where a platform declares a file under a different name than
|
|
the emulator profile (e.g. Batocera ROM1 vs gsplus ROM).
|
|
"""
|
|
declared: set[str] = set()
|
|
by_md5 = db.get("indexes", {}).get("by_md5", {})
|
|
files_db = db.get("files", {})
|
|
for system in config.get("systems", {}).values():
|
|
for fe in system.get("files", []):
|
|
name = fe.get("name", "")
|
|
if name:
|
|
declared.add(name)
|
|
md5 = fe.get("md5", "")
|
|
if not md5:
|
|
continue
|
|
# Skip multi-hash and zipped_file entries (inner ROM MD5, not file MD5)
|
|
if "," in md5 or fe.get("zipped_file"):
|
|
continue
|
|
sha1 = by_md5.get(md5.lower())
|
|
if not sha1:
|
|
continue
|
|
entry = files_db.get(sha1, {})
|
|
db_name = entry.get("name", "")
|
|
if db_name:
|
|
declared.add(db_name)
|
|
for alias in entry.get("aliases", []):
|
|
declared.add(alias)
|
|
return declared
|
|
|
|
|
|
import re
|
|
|
|
|
|
|
|
|
|
# Validation and mode filtering -extracted to validation.py for SoC.
|
|
# Re-exported below for backward compatibility.
|
|
|
|
|
|
|
|
def sanitize_pack_path(raw: str) -> str:
|
|
"""Strip traversal components from a relative destination.
|
|
|
|
The builder and the coverage report key their region grouping on this
|
|
value, so they have to derive it the same way: a destination normalized on
|
|
one side only would be looked up under a key the other side never emits.
|
|
"""
|
|
raw = raw.replace("\\", "/")
|
|
return "/".join(p for p in raw.split("/") if p and p not in ("..", "."))
|
|
|
|
|
|
|
|
|
|
def list_emulator_profiles(emulators_dir: str, skip_aliases: bool = True) -> None:
|
|
"""Print available emulator profiles."""
|
|
profiles = load_emulator_profiles(emulators_dir, skip_aliases=False)
|
|
for name in sorted(profiles):
|
|
p = profiles[name]
|
|
if p.get("type") in ("alias", "test"):
|
|
continue
|
|
display = p.get("emulator", name)
|
|
ptype = p.get("type", "libretro")
|
|
systems = ", ".join(p.get("systems", [])[:3])
|
|
more = "..." if len(p.get("systems", [])) > 3 else ""
|
|
print(f" {name:30s} {display:40s} [{ptype}] {systems}{more}")
|
|
|
|
|
|
def list_system_ids(emulators_dir: str) -> None:
|
|
"""Print available system IDs with emulator count."""
|
|
profiles = load_emulator_profiles(emulators_dir)
|
|
system_emus: dict[str, list[str]] = {}
|
|
for name, p in profiles.items():
|
|
if p.get("type") in ("alias", "test", "launcher"):
|
|
continue
|
|
for sys_id in p.get("systems", []):
|
|
system_emus.setdefault(sys_id, []).append(name)
|
|
for sys_id in sorted(system_emus):
|
|
count = len(system_emus[sys_id])
|
|
print(f" {sys_id:35s} ({count} emulator{'s' if count > 1 else ''})")
|
|
|
|
|
|
def list_platform_system_ids(platform_name: str, platforms_dir: str) -> None:
|
|
"""Print system IDs from a platform's YAML config."""
|
|
config = load_platform_config(platform_name, platforms_dir)
|
|
systems = config.get("systems", {})
|
|
for sys_id in sorted(systems):
|
|
file_count = len(systems[sys_id].get("files", []))
|
|
mfr = systems[sys_id].get("manufacturer", "")
|
|
mfr_display = f" [{mfr.split('|')[0]}]" if mfr else ""
|
|
print(
|
|
f" {sys_id:35s} ({file_count} file{'s' if file_count != 1 else ''}){mfr_display}"
|
|
)
|
|
|
|
|
|
def build_target_cores_cache(
|
|
platforms: list[str],
|
|
target: str,
|
|
platforms_dir: str,
|
|
is_all: bool = False,
|
|
) -> tuple[dict[str, set[str] | None], list[str]]:
|
|
"""Build target cores cache for a list of platforms.
|
|
|
|
Returns (cache dict, list of platforms to keep after skipping failures).
|
|
"""
|
|
cache: dict[str, set[str] | None] = {}
|
|
skip: list[str] = []
|
|
for p in platforms:
|
|
try:
|
|
cache[p] = load_target_config(p, target, platforms_dir)
|
|
except FileNotFoundError:
|
|
if is_all:
|
|
cache[p] = None
|
|
else:
|
|
raise
|
|
except ValueError as e:
|
|
if is_all:
|
|
print(f"INFO: Skipping {p}: {e}")
|
|
skip.append(p)
|
|
else:
|
|
raise
|
|
kept = [p for p in platforms if p not in skip]
|
|
return cache, kept
|
|
|
|
|
|
# Re-exported so the existing call sites keep working while the
|
|
# modules above become the place to import from.
|
|
from safeparse import ( # noqa: E402,F401
|
|
parse_untrusted_xml,
|
|
require_yaml,
|
|
yaml_load,
|
|
_pick_yaml_loader,
|
|
_YAML_LOADER,
|
|
)
|
|
from hashing import ( # noqa: E402,F401
|
|
compute_hashes,
|
|
md5sum,
|
|
md5_composite,
|
|
parse_md5_list,
|
|
_ALL_ALGORITHMS,
|
|
_md5_composite_cache,
|
|
)
|
|
from ziptools import ( # noqa: E402,F401
|
|
check_inside_zip,
|
|
build_zip_contents_index,
|
|
MAX_ZIP_MEMBERS,
|
|
MAX_ZIP_MEMBER_SIZE,
|
|
MAX_ZIP_TOTAL_SIZE,
|
|
MAX_ZIP_COMPRESSION_RATIO,
|
|
safe_extract_zip,
|
|
_zip_contents_cache,
|
|
_BOUNDED_RATIO_METHODS,
|
|
)
|
|
from artifacts import ( # noqa: E402,F401
|
|
write_if_changed,
|
|
ArtifactLockBusy,
|
|
artifact_lock,
|
|
_TIMESTAMP_PATTERNS,
|
|
_strip_timestamps,
|
|
)
|
|
from largefiles import ( # noqa: E402,F401
|
|
LARGE_FILES_RELEASE,
|
|
LARGE_FILES_REPO,
|
|
LARGE_FILES_CACHE,
|
|
fetch_large_file,
|
|
)
|
|
from dumpcatalog import ( # noqa: E402,F401
|
|
DEFAULT_PROVENANCE_DIR,
|
|
load_provenance_snapshots,
|
|
build_provenance_index,
|
|
annotate_provenance,
|
|
write_provenance_snapshot,
|
|
)
|