kmoody/libretro
1
0 Fork 0
mirror of https://github.com/Abdess/retroarch_system.git synced 2026-10-23 20:29:06 -05:00
Code Issues Packages Projects Releases Wiki Activity
Files
6c076884ed650585600e4b05c15f607be504c9a9
libretro/scripts/auto_fetch.py
T
Abdessamad Derraz 6c076884ed fix: let auto_fetch prove sha256 and crc32 entries 
verify_content read sha1 then md5 only, so an entry pinned by sha256 or
crc32 could be searched and never accepted, and one without any hash was
announced as searched. The strongest declared hash now decides, crc32
counts with its size, the database lookup reads sha256 and crc32, and an
entry with nothing to prove a download by is reported, not searched.
2026-10-21 10:02:05 +02:00

584 lines
20 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""Auto-fetch missing BIOS files from multiple sources.
Pipeline:
1. Cross-reference database.json (already exists under different name/path?)
2. Scan old branches (git show origin/branch:path)
3. Search public BIOS repos on GitHub
4. Search archive.org collections
5. Create GitHub Issue for community help
Usage:
python scripts/auto_fetch.py --platform retroarch [--dry-run]
python scripts/auto_fetch.py --all [--dry-run]
"""
from __future__ import annotations
import argparse
import hashlib
import http.client
import json
import os
import subprocess
import sys
import urllib.error
import urllib.parse
import urllib.request
import zlib
from pathlib import Path
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from common import (
platform_arg,
require_directory,
list_registered_platforms,
load_database,
load_platform_config,
require_platform,
require_yaml,
)
yaml = require_yaml()
DEFAULT_DB = "database.json"
DEFAULT_PLATFORMS_DIR = "platforms"
DEFAULT_BIOS_DIR = "bios"
MAX_DOWNLOAD = 100 * 1024 * 1024 # 100MB per file
LEGACY_BRANCHES = ["libretro", "RetroArch", "RetroPie", "Recalbox", "batocera", "Other"]
PUBLIC_REPOS = [
# archtaurus/RetroPieBIOS - most complete verified collection
"https://raw.githubusercontent.com/archtaurus/RetroPieBIOS/master/BIOS/{name}",
"https://raw.githubusercontent.com/archtaurus/RetroPieBIOS/master/BIOS/pcsx2/bios/{name}",
"https://raw.githubusercontent.com/archtaurus/RetroPieBIOS/master/BIOS/ep128emu/roms/{name}",
"https://raw.githubusercontent.com/archtaurus/RetroPieBIOS/master/BIOS/fuse/{name}",
# prefetchnta/retroarch-bios - alternative verified collection
"https://raw.githubusercontent.com/prefetchnta/retroarch-bios/main/system/{name}",
"https://raw.githubusercontent.com/prefetchnta/retroarch-bios/main/system/pcsx2/bios/{name}",
# BatoceraPLUS - Batocera-specific
"https://raw.githubusercontent.com/BatoceraPLUS/Batocera.PLUS-bios/main/{name}",
]
ARCHIVE_ORG_COLLECTIONS = [
"RetroarchSystemFiles",
"retroarch_bios",
"retroarch-ultimate-bios-pack_20250824",
"system_20240621",
"full-pack-bios-batocera-39",
]
def find_missing(config: dict, db: dict) -> list[dict]:
"""Files the platform's own check or its cores' code would not find.
Read from verify, the platform's native check: a hash or name lookup
of our own called pc6001.zip and fmtowns.zip missing on Batocera, which
checks them inside the zip and holds them, and saw none of the files
the cores need that no platform lists.
"""
from nativemode import reads_file_contents # noqa: PLC0415
from verify import Status, verify_platform # noqa: PLC0415
result = verify_platform(config, db)
hashed = reads_file_contents(config.get("verification_mode", "existence"))
wanted = {
(d.get("system", ""), d.get("name", ""))
for d in result["details"]
if d["status"] == Status.MISSING or (hashed and d["status"] == Status.UNTESTED)
}
missing: list[dict] = []
for sys_id, system in config.get("systems", {}).items():
for file_entry in system.get("files", []):
name = file_entry.get("name", "")
if (sys_id, name) not in wanted:
continue
if file_entry.get("storage", "embedded") != "embedded":
continue
missing.append({
"name": name,
"system": sys_id,
"sha1": file_entry.get("sha1"),
"md5": file_entry.get("md5"),
"sha256": file_entry.get("sha256"),
"crc32": file_entry.get("crc32"),
"size": file_entry.get("size"),
"destination": file_entry.get("destination", name),
})
for u in result["undeclared_files"]:
if u.get("in_repo") or not u.get("required"):
continue
missing.append({
"name": u["name"],
"system": u.get("system") or "",
"sha1": u.get("sha1"),
"md5": u.get("md5"),
"sha256": u.get("sha256"),
"crc32": u.get("crc32"),
"size": u.get("size"),
"destination": u.get("path") or u["name"],
"emulator": u.get("emulator", ""),
})
return missing
def _accepted(entry: dict, field: str) -> set[str]:
"""Every value an entry accepts for a hash field.
A profile lists several revisions as a YAML list (daedalusx64's
n64.psh), a platform as a comma-separated string. Read as one value,
the list stopped the run with an unhashable-type error.
"""
value = entry.get(field)
values = value if isinstance(value, list) else str(value or "").split(",")
return {str(v).strip().lower() for v in values if str(v).strip()}
def _crc32s(entry: dict) -> set[str]:
return {value.removeprefix("0x").zfill(8) for value in _accepted(entry, "crc32")}
def _size(entry: dict) -> int | None:
try:
return int(entry["size"]) if entry.get("size") is not None else None
except (TypeError, ValueError):
return None
def verifiable(entry: dict) -> bool:
"""Whether a download can be proved to be this file.
A sha1, sha256 or md5 proves it, a crc32 only with the size beside it,
the way the resolver accepts one. Without any of them a search can find
a file of that name and never accept it.
"""
if any(_accepted(entry, field) for field in ("sha1", "sha256", "md5")):
return True
return bool(_crc32s(entry)) and _size(entry) is not None
def verify_content(data: bytes, expected: dict) -> bool:
"""Verify downloaded content matches one of the expected hashes.
The strongest hash the entry declares decides.
"""
for field, digest in (
("sha1", hashlib.sha1),
("sha256", hashlib.sha256),
("md5", hashlib.md5),
):
accepted = _accepted(expected, field)
if accepted:
return digest(data).hexdigest() in accepted
crcs, size = _crc32s(expected), _size(expected)
if crcs and size is not None:
return len(data) == size and f"{zlib.crc32(data) & 0xFFFFFFFF:08x}" in crcs
return False
def _read_limited(
resp: http.client.HTTPResponse, limit: int = MAX_DOWNLOAD
) -> bytes | None:
"""Read response up to limit bytes. Returns None if exceeded."""
data = resp.read(limit + 1)
if len(data) > limit:
return None
return data
def step1_crossref_db(entry: dict, db: dict) -> str | None:
"""Check if file exists under different name/path in database."""
files = db.get("files", {})
for sha1 in sorted(_accepted(entry, "sha1")):
if sha1 in files:
return files[sha1]["path"]
indexes = db.get("indexes", {})
for field, index in (("sha256", "by_sha256"), ("md5", "by_md5")):
by_hash = indexes.get(index, {})
for value in sorted(_accepted(entry, field)):
sha1_match = by_hash.get(value)
if sha1_match and sha1_match in files:
return files[sha1_match]["path"]
size = _size(entry)
if size is not None:
by_crc32 = indexes.get("by_crc32", {})
for crc in sorted(_crc32s(entry)):
sha1_match = by_crc32.get(crc)
if sha1_match in files and files[sha1_match].get("size") == size:
return files[sha1_match]["path"]
return None
def step2_scan_branches(entry: dict) -> bytes | None:
"""Search old git branches for the file by hash."""
name = entry["name"]
for branch in LEGACY_BRANCHES:
ref = f"origin/{branch}"
try:
subprocess.run(
["git", "rev-parse", "--verify", ref],
capture_output=True,
check=True,
)
except subprocess.CalledProcessError:
continue
result = subprocess.run(
["git", "ls-tree", "-r", "--name-only", ref],
capture_output=True,
text=True,
)
for filepath in result.stdout.strip().split("\n"):
if os.path.basename(filepath) == name:
try:
blob = subprocess.run(
["git", "show", f"{ref}:{filepath}"],
capture_output=True,
check=True,
)
if verify_content(blob.stdout, entry):
return blob.stdout
except subprocess.CalledProcessError:
continue
return None
# Sources that did not answer during the current file's search. A 404 is an
# answer, the file is not there. Anything else leaves the question open: an
# offline run used to report every file as needing a community contribution.
_UNANSWERED: list[str] = []
def _unanswered(url: str, exc: Exception) -> None:
if isinstance(exc, urllib.error.HTTPError) and exc.code == 404:
return
_UNANSWERED.append(f"{url}: {exc}")
def step3_search_public_repos(entry: dict) -> bytes | None:
"""Search public GitHub BIOS repos."""
name = entry["name"]
destination = entry.get("destination", name)
for url_template in PUBLIC_REPOS:
url = url_template.format(name=name)
try:
req = urllib.request.Request(
url, headers={"User-Agent": "retrobios-fetch/1.0"}
)
with urllib.request.urlopen(req, timeout=30) as resp:
data = _read_limited(resp)
if data is None:
continue
if verify_content(data, entry):
return data
except urllib.error.URLError as exc:
_unanswered(url, exc)
continue
if "/" in destination:
url = url_template.format(name=destination)
try:
req = urllib.request.Request(
url, headers={"User-Agent": "retrobios-fetch/1.0"}
)
with urllib.request.urlopen(req, timeout=30) as resp:
data = _read_limited(resp)
if data is None:
continue
if verify_content(data, entry):
return data
except urllib.error.URLError as exc:
_unanswered(url, exc)
continue
return None
def step4_search_archive_org(entry: dict) -> bytes | None:
"""Search archive.org firmware collections by direct download."""
name = entry["name"]
for collection_id in ARCHIVE_ORG_COLLECTIONS:
for path in [name, f"system/{name}", f"bios/{name}"]:
url = f"https://archive.org/download/{collection_id}/{path}"
try:
req = urllib.request.Request(
url, headers={"User-Agent": "retrobios-fetch/1.0"}
)
with urllib.request.urlopen(req, timeout=30) as resp:
data = _read_limited(resp)
if data is None:
continue
if verify_content(data, entry):
return data
except urllib.error.URLError as exc:
_unanswered(url, exc)
continue
sha1s = sorted(_accepted(entry, "sha1"))
if not sha1s:
return None
# Each accepted revision is a query of its own: archive.org matches one.
sha1 = " OR ".join(f"sha1:{value}" for value in sha1s)
search_url = (
"https://archive.org/advancedsearch.php?q="
f"{urllib.parse.quote(sha1)}&output=json&rows=1"
)
try:
req = urllib.request.Request(
search_url, headers={"User-Agent": "retrobios-fetch/1.0"}
)
with urllib.request.urlopen(req, timeout=30) as resp:
result = json.loads(resp.read())
docs = result.get("response", {}).get("docs", [])
if docs:
identifier = docs[0].get("identifier")
if identifier:
dl_url = f"https://archive.org/download/{identifier}/{name}"
try:
req2 = urllib.request.Request(
dl_url, headers={"User-Agent": "retrobios-fetch/1.0"}
)
with urllib.request.urlopen(req2, timeout=30) as resp2:
data = _read_limited(resp2)
if data is not None and verify_content(data, entry):
return data
except urllib.error.URLError as exc:
_unanswered(dl_url, exc)
except (urllib.error.URLError, json.JSONDecodeError) as exc:
_unanswered(search_url, exc)
return None
def place_file(data: bytes, entry: dict, bios_dir: str, db: dict) -> str:
"""Place a fetched BIOS file in the correct location."""
name = os.path.basename(entry["name"])
system = entry["system"]
dest_dir = Path(bios_dir)
# Every part of the slug, not any: matching "nintendo" alone filed the
# N64's n64.psh under the first Nintendo folder walked, the SNES one.
parts = [part for part in system.lower().split("-") if len(part) > 1]
matches = sorted(
(len(f"{manufacturer_dir.name}/{console_dir.name}"), str(console_dir), console_dir)
for manufacturer_dir in dest_dir.iterdir() if manufacturer_dir.is_dir()
for console_dir in manufacturer_dir.iterdir() if console_dir.is_dir()
and all(
part in f"{manufacturer_dir.name}/{console_dir.name}".lower()
for part in parts
)
)
folder = matches[0][2] if matches else dest_dir / "Unknown" / system
dest = folder / name
if dest.exists() and dest.read_bytes() != data:
# Another revision under the same name: the collection keeps it
# beside, never in place of, the file already there.
dest = folder / ".variants" / f"{name}.{hashlib.sha1(data).hexdigest()[:8]}"
dest.parent.mkdir(parents=True, exist_ok=True)
with open(dest, "wb") as f:
f.write(data)
return str(dest)
def fetch_missing(
missing: list[dict],
db: dict,
bios_dir: str,
dry_run: bool = False,
) -> dict:
"""Run the 5-step auto-fetch pipeline for missing files."""
stats = {"found": 0, "not_found": 0, "errors": 0, "unverifiable": 0}
still_missing = []
for entry in missing:
name = entry["name"]
_UNANSWERED.clear()
print(f"\n Searching: {name} ({entry['system']})")
existing = step1_crossref_db(entry, db)
if existing:
print(f" [1] Found in database at: {existing}")
stats["found"] += 1
continue
if not verifiable(entry):
# Searched, a file of that name could only ever be refused.
print(" [-] No sha1, sha256, md5 or crc32 with size to check a "
"download against: not searched")
still_missing.append(entry)
stats["unverifiable"] += 1
continue
if dry_run:
print(" [DRY RUN] Would search branches, repos, archive.org")
still_missing.append(entry)
stats["not_found"] += 1
continue
data = step2_scan_branches(entry)
if data:
path = place_file(data, entry, bios_dir, db)
print(f" [2] Found in branch, saved to: {path}")
stats["found"] += 1
continue
data = step3_search_public_repos(entry)
if data:
path = place_file(data, entry, bios_dir, db)
print(f" [3] Found in public repo, saved to: {path}")
stats["found"] += 1
continue
data = step4_search_archive_org(entry)
if data:
path = place_file(data, entry, bios_dir, db)
print(f" [4] Found on archive.org, saved to: {path}")
stats["found"] += 1
continue
if _UNANSWERED:
# Not looked for everywhere: no issue asks the community for it.
print(
f" [5] Search incomplete, {len(_UNANSWERED)} source(s) did not "
f"answer (first: {_UNANSWERED[0]})"
)
stats["errors"] += 1
continue
print(" [5] Not found - needs community contribution")
still_missing.append(entry)
stats["not_found"] += 1
return {"stats": stats, "still_missing": still_missing}
def generate_issue_body(missing: list[dict], platform: str) -> str:
"""Generate a GitHub Issue body for missing BIOS files."""
lines = [
f"## Missing BIOS Files for {platform}",
"",
"The following BIOS files are required but not available in the repository.",
"If you have any of these files, please submit a Pull Request!",
"",
"| File | System | SHA1 | MD5 |",
"|------|--------|------|-----|",
]
for entry in missing:
sha1 = ", ".join(sorted(_accepted(entry, "sha1"))) or "N/A"
md5 = ", ".join(sorted(_accepted(entry, "md5"))) or "N/A"
lines.append(
f"| `{entry['name']}` | {entry['system']} | `{sha1[:12]}...` | `{md5[:12]}...` |"
)
lines.extend(
[
"",
"### How to Contribute",
"",
"1. Fork this repository",
"2. Add the BIOS file to `bios/Manufacturer/Console/`",
"3. Create a Pull Request - checksums are verified automatically",
]
)
return "\n".join(lines)
def main() -> None:
parser = argparse.ArgumentParser(allow_abbrev=False, description="Auto-fetch missing BIOS files")
parser.add_argument("--platform", "-p", type=platform_arg, help="Platform to check")
parser.add_argument("--all", action="store_true", help="Check all platforms")
parser.add_argument(
"--dry-run", action="store_true", help="Don't download, just report"
)
parser.add_argument(
"--db", default=DEFAULT_DB,
help="database file to read (default: %(default)s)",
)
parser.add_argument(
"--platforms-dir", default=DEFAULT_PLATFORMS_DIR,
help="platform YAML directory (default: %(default)s)",
)
parser.add_argument(
"--bios-dir", default=DEFAULT_BIOS_DIR,
help="BIOS collection directory (default: %(default)s)",
)
parser.add_argument(
"--create-issues", action="store_true", help="Output GitHub Issue bodies"
)
args = parser.parse_args()
require_directory(args.bios_dir)
if not os.path.exists(args.db):
print(f"Error: {args.db} not found. Run generate_db.py first.", file=sys.stderr)
sys.exit(1)
db = load_database(args.db)
if args.all:
platforms = list_registered_platforms(
args.platforms_dir,
include_archived=True,
)
elif args.platform:
require_platform(args.platform, args.platforms_dir)
platforms = [args.platform]
else:
parser.error("Specify --platform or --all")
return
all_still_missing = {}
for platform in sorted(platforms):
print(f"\n{'=' * 60}")
print(f"Platform: {platform}")
print(f"{'=' * 60}")
try:
config = load_platform_config(platform, args.platforms_dir)
except FileNotFoundError:
print(" Config not found, skipping")
continue
missing = find_missing(config, db)
if not missing:
print(" All BIOS files present!")
continue
print(f" {len(missing)} missing files")
result = fetch_missing(missing, db, args.bios_dir, args.dry_run)
if result["still_missing"]:
all_still_missing[platform] = result["still_missing"]
stats = result["stats"]
print(
f"\n Results: {stats['found']} found, {stats['not_found']} not found, "
f"{stats['errors']} not fully searched, "
f"{stats['unverifiable']} without a hash to search by"
)
if args.create_issues and all_still_missing:
print(f"\n{'=' * 60}")
print("GitHub Issue Bodies")
print(f"{'=' * 60}")
for platform, missing in all_still_missing.items():
print(f"\n--- Issue for {platform} ---\n")
print(generate_issue_body(missing, platform))
if __name__ == "__main__":
from common import run_cli
run_cli(main)
Reference in new issue View Git Blame Copy Permalink