libretro/scripts/auto_fetch.py

#!/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
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"),
                "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"),
            "size": None,
            "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 verify_content(data: bytes, expected: dict) -> bool:
    """Verify downloaded content matches one of the expected hashes."""
    sha1s = _accepted(expected, "sha1")
    if sha1s:
        return hashlib.sha1(data).hexdigest() in sha1s
    md5s = _accepted(expected, "md5")
    if md5s:
        return hashlib.md5(data).hexdigest() in md5s
    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"]
    by_md5 = db.get("indexes", {}).get("by_md5", {})
    for md5 in sorted(_accepted(entry, "md5")):
        sha1_match = by_md5.get(md5)
        if sha1_match and sha1_match in files:
            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}
    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 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"
        )

    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)