libretro/scripts/scraper/mame_hash_scraper.py

"""Fetch MAME BIOS hashes from mamedev/mame source and merge into profiles.

Sparse clones the MAME repo, parses the source tree for BIOS root sets,
caches results to data/mame-hashes.json, and optionally merges into
emulator profiles that reference mamedev/mame upstream.
"""

from __future__ import annotations

import argparse
import json
import os
import logging
import shutil
import subprocess
import sys
import urllib.error
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import yaml

from common import yaml_load

from ._hash_merge import compute_diff, merge_mame_profile
from .mame_parser import parse_mame_source_tree

log = logging.getLogger(__name__)

_ROOT = Path(__file__).resolve().parents[2]
_CACHE_PATH = _ROOT / "data" / "mame-hashes.json"
_CLONE_DIR = _ROOT / "tmp" / "mame"
_EMULATORS_DIR = _ROOT / "emulators"
_REPO_URL = "https://github.com/mamedev/mame.git"
_STALE_HOURS = 24


# ── Cache ────────────────────────────────────────────────────────────


def _load_cache() -> dict[str, Any] | None:
    if not _CACHE_PATH.exists():
        return None
    try:
        with open(_CACHE_PATH, encoding="utf-8") as f:
            return json.load(f)
    except (json.JSONDecodeError, OSError):
        return None


def _is_stale(cache: dict[str, Any] | None) -> bool:
    if cache is None:
        return True
    fetched_at = cache.get("fetched_at")
    if not fetched_at:
        return True
    try:
        ts = datetime.fromisoformat(fetched_at)
        age = datetime.now(timezone.utc) - ts
        return age.total_seconds() > _STALE_HOURS * 3600
    except (ValueError, TypeError):
        return True


def _write_cache(data: dict[str, Any]) -> None:
    _CACHE_PATH.parent.mkdir(parents=True, exist_ok=True)
    with open(_CACHE_PATH, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2, ensure_ascii=False)
    log.info("cache written to %s", _CACHE_PATH)


# ── Git operations ───────────────────────────────────────────────────


def _run_git(
    args: list[str], cwd: Path | None = None
) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        ["git", *args],
        cwd=cwd,
        check=True,
        capture_output=True,
        text=True,
    )


def _sparse_clone(tag: str) -> None:
    """Clone the release the data will be labelled with, not master.

    The version comes from the latest release; cloning master gave line
    numbers and hashes of a later revision under that release's label, and
    the profiles cited lines no declared revision holds.
    """
    if _CLONE_DIR.exists():
        shutil.rmtree(_CLONE_DIR)
    _CLONE_DIR.parent.mkdir(parents=True, exist_ok=True)

    log.info("sparse cloning mamedev/mame %s into %s", tag, _CLONE_DIR)
    _run_git(
        [
            "clone",
            "--depth",
            "1",
            "--branch",
            tag,
            "--filter=blob:none",
            "--sparse",
            _REPO_URL,
            str(_CLONE_DIR),
        ]
    )
    _run_git(
        ["sparse-checkout", "set", "src/mame", "src/devices"],
        cwd=_CLONE_DIR,
    )


def _get_version() -> str:
    """The latest MAME release version, such as 0.289."""
    return _parse_version_tag(_get_release_tag())


def _get_release_tag() -> str:
    """The latest MAME release tag, from the GitHub API.

    version.cpp is generated at build time, not in the repo. A failed lookup
    raises: "unknown" was cached for a day and written as core_version.
    """
    headers = {
        "User-Agent": "retrobios-scraper/1.0",
        "Accept": "application/vnd.github.v3+json",
    }
    token = os.environ.get("GITHUB_TOKEN")
    if token:
        headers["Authorization"] = f"Bearer {token}"
    url = "https://api.github.com/repos/mamedev/mame/releases/latest"
    try:
        req = urllib.request.Request(url, headers=headers)
        with urllib.request.urlopen(req, timeout=10) as resp:
            tag = json.loads(resp.read()).get("tag_name", "")
    except (urllib.error.URLError, json.JSONDecodeError, OSError) as exc:
        raise RuntimeError(f"cannot read the MAME release from {url}: {exc}") from exc
    if not tag:
        raise RuntimeError(f"no tag_name in {url}")
    return tag


def _parse_version_tag(tag: str) -> str:
    prefix = "mame"
    raw = tag.removeprefix(prefix) if tag.startswith(prefix) else tag
    if raw.isdigit() and len(raw) >= 4:
        return f"{raw[0]}.{raw[1:]}"
    return raw


def _get_commit() -> str:
    try:
        result = _run_git(["rev-parse", "HEAD"], cwd=_CLONE_DIR)
        return result.stdout.strip()
    except subprocess.CalledProcessError:
        return ""


def _cleanup() -> None:
    if _CLONE_DIR.exists():
        log.info("cleaning up %s", _CLONE_DIR)
        shutil.rmtree(_CLONE_DIR)


# ── Profile discovery ────────────────────────────────────────────────


_MAME_TREES = frozenset({
    "https://github.com/mamedev/mame",
    "https://github.com/libretro/mame",
})


def _find_mame_profiles() -> list[Path]:
    profiles: list[Path] = []
    for path in sorted(_EMULATORS_DIR.glob("*.yml")):
        if path.name.endswith(".old.yml"):
            continue
        try:
            with open(path, encoding="utf-8") as f:
                data = yaml_load(f)
            if not isinstance(data, dict):
                continue
            upstream = data.get("upstream", "")
            source = data.get("source", "")
            # Only match profiles tracking current MAME (not frozen snapshots
            # which have upstream like "mamedev/mame/tree/mame0139"), and only
            # those whose own tree is MAME's. GroovyMAME, A7800 and SAME CDi
            # declare mamedev/mame upstream but build from their own fork at
            # another release: refreshing their refs from current MAME moves
            # them onto paths their tree does not have, such as the 0.289 move
            # of neogeo.cpp from src/mame/neogeo/ to src/mame/snk/.
            if (
                isinstance(upstream, str)
                and upstream.rstrip("/") == "https://github.com/mamedev/mame"
                and isinstance(source, str)
                and source.rstrip("/") in _MAME_TREES
            ):
                profiles.append(path)
        except (yaml.YAMLError, OSError):
            continue
    return profiles


# ── Diff formatting ──────────────────────────────────────────────────


def _format_diff(
    profile_path: Path,
    diff: dict[str, Any],
    hashes: dict[str, Any],
    show_added: bool = True,
) -> list[str]:
    lines: list[str] = []
    name = profile_path.stem

    added = diff.get("added", [])
    updated = diff.get("updated", [])
    removed = diff.get("removed", [])
    unchanged = diff.get("unchanged", 0)

    if not added and not updated and not removed:
        lines.append(f"  {name}:")
        lines.append("    no changes")
        return lines

    lines.append(f"  {name}:")

    if show_added:
        bios_sets = hashes.get("bios_sets", {})
        for set_name in added:
            rom_count = len(bios_sets.get(set_name, {}).get("roms", []))
            source_file = bios_sets.get(set_name, {}).get("source_file", "")
            source_line = bios_sets.get(set_name, {}).get("source_line", "")
            ref = f"{source_file}:{source_line}" if source_file else ""
            lines.append(f"    + {set_name}.zip ({ref}, {rom_count} ROMs)")
    elif added:
        lines.append(f"    + {len(added)} new sets available (main profile only)")

    for set_name in updated:
        lines.append(f"    ~ {set_name}.zip (refreshed)")

    oos = diff.get("out_of_scope", 0)
    lines.append(f"    = {unchanged} unchanged")
    if oos:
        lines.append(f"    . {oos} out of scope (not BIOS root sets)")
    return lines


# ── Main ─────────────────────────────────────────────────────────────


def _fetch_hashes(force: bool) -> dict[str, Any]:
    cache = _load_cache()
    if not force and not _is_stale(cache):
        log.info("using cached data from %s", cache.get("fetched_at", ""))
        return cache  # type: ignore[return-value]

    try:
        tag = _get_release_tag()
        _sparse_clone(tag)
        bios_sets = parse_mame_source_tree(str(_CLONE_DIR))
        version = _parse_version_tag(tag)
        commit = _get_commit()

        data: dict[str, Any] = {
            "source": "mamedev/mame",
            "version": version,
            "commit": commit,
            "fetched_at": datetime.now(timezone.utc).isoformat(),
            "bios_sets": bios_sets,
        }
        _write_cache(data)
        return data
    finally:
        _cleanup()


def _run(args: argparse.Namespace) -> None:
    hashes = _fetch_hashes(args.force)

    total_sets = len(hashes.get("bios_sets", {}))
    version = hashes.get("version", "unknown")
    commit = hashes.get("commit", "")[:12]

    if args.json:
        json.dump(hashes, sys.stdout, indent=2, ensure_ascii=False)
        sys.stdout.write("\n")
        return

    print(
        f"mame-hashes: {total_sets} BIOS root sets from mamedev/mame"
        f" @ {version} ({commit})"
    )
    print()

    profiles = _find_mame_profiles()
    if not profiles:
        print("  no profiles with mamedev/mame upstream found")
        return

    for profile_path in profiles:
        is_main = profile_path.name == "mame.yml"
        diff = compute_diff(str(profile_path), str(_CACHE_PATH), mode="mame")
        lines = _format_diff(profile_path, diff, hashes, show_added=is_main)
        for line in lines:
            print(line)

        if not args.dry_run:
            updated = diff.get("updated", [])
            added = diff.get("added", []) if is_main else []
            if added or updated:
                merge_mame_profile(
                    str(profile_path),
                    str(_CACHE_PATH),
                    write=True,
                    add_new=is_main,
                )
                log.info("merged into %s", profile_path.name)

    print()
    if args.dry_run:
        print("(dry run, no files modified)")


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="mame_hash_scraper",
        description="Fetch MAME BIOS hashes from source and merge into profiles.",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="show diff only, do not modify profiles",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        help="output raw JSON to stdout",
    )
    parser.add_argument(
        "--force",
        action="store_true",
        help="re-fetch even if cache is fresh",
    )
    return parser


def main() -> None:
    logging.basicConfig(
        level=logging.INFO,
        format="%(levelname)s: %(message)s",
    )
    parser = build_parser()
    args = parser.parse_args()
    _run(args)


if __name__ == "__main__":
    main()