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 source_yaml_paths, write_text_atomic, yaml_load

from ._hash_merge import _build_contents, _build_source_ref, 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)
    write_text_atomic(
        str(_CACHE_PATH), json.dumps(data, 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:
    """The commit the clone holds, or an error: never an empty pin.

    Returned empty, it was cached for a day and the merge advanced
    core_version and the refs to the new release while upstream_commit
    stayed on the old one, a profile describing two revisions at once.
    """
    try:
        result = _run_git(["rev-parse", "HEAD"], cwd=_CLONE_DIR)
    except subprocess.CalledProcessError as exc:
        raise RuntimeError(
            f"cannot read the commit of the MAME clone in {_CLONE_DIR}: {exc}"
        ) from exc
    commit = result.stdout.strip()
    if not commit:
        raise RuntimeError(f"the MAME clone in {_CLONE_DIR} names no commit")
    return commit


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 source_yaml_paths(_EMULATORS_DIR):
        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 release_profile(
    release: str,
    bios_sets: dict[str, Any],
    upstream_commit: str,
    source_commit: str,
    cores: list[str],
    reference: dict[str, Any],
    built_by: list[str],
) -> dict[str, Any]:
    """A profile for one MAME release a platform builds, from its own source.

    Recalbox ships libretro/mame twice, built at 0.258 and at 0.278; the
    BIOS root sets and their ROMs change between releases, so the current
    MAME profile does not describe them. Each set keeps the system and the
    required flag the current profile gives it; the contents and the line
    are the release's own.
    """
    version = release.removeprefix("mame")
    version = f"{version[0]}.{version[1:]}" if version.isdigit() else version
    known = {
        str(f.get("name")): f for f in reference.get("files") or []
        if f.get("category") == "bios_zip"
    }
    files = []
    for set_name, data in sorted(bios_sets.items()):
        name = f"{set_name}.zip"
        entry: dict[str, Any] = {
            "name": name,
            "required": known.get(name, {}).get("required", True),
            "category": "bios_zip",
        }
        if known.get(name, {}).get("system"):
            entry["system"] = known[name]["system"]
        entry["source_ref"] = _build_source_ref(data)
        entry["contents"] = _build_contents(data.get("roms", []))
        files.append(entry)
    return {
        "emulator": f"MAME {version}",
        "type": "libretro",
        "core_classification": "frozen_snapshot",
        "source": "https://github.com/libretro/mame",
        "upstream": f"https://github.com/mamedev/mame/tree/{release}",
        "profiled_date": datetime.now(timezone.utc).strftime("%Y-%m-%d"),
        "source_commit": source_commit,
        "upstream_commit": upstream_commit,
        "core_version": version,
        "display_name": f"Arcade (MAME {version})",
        "cores": cores,
        "built_by": built_by,
        "systems": list(reference.get("systems") or []),
        "notes": (
            f"libretro/mame built at the {version} release. BIOS root sets "
            f"read from mamedev/mame at {release}: the sets and their ROMs "
            "are that release's, not the current one's."
        ),
        "files": files,
    }


def _write_release(args: argparse.Namespace) -> None:
    """Write the profile of one MAME release, read from its own tag."""
    try:
        _sparse_clone(args.release)
        bios_sets = parse_mame_source_tree(str(_CLONE_DIR))
        commit = _get_commit()
    finally:
        _cleanup()
    if not bios_sets or not commit:
        raise RuntimeError(f"no BIOS root set read at {args.release}")
    reference = yaml_load((_EMULATORS_DIR / "mame.yml").read_text(encoding="utf-8"))
    profile = release_profile(
        args.release, bios_sets, commit, args.source_commit, args.core, reference,
        args.built_by,
    )
    text = yaml.safe_dump(profile, sort_keys=False, allow_unicode=True, width=100)
    write_text_atomic(str(args.profile), text)
    print(f"{args.profile}: {len(profile['files'])} BIOS root sets at {args.release}")


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",
    )
    release = parser.add_argument_group(
        "release profile",
        "write the profile of one MAME release a platform builds "
        "(e.g. --release mame0278 --profile emulators/mame0278.yml "
        "--source-commit <libretro/mame sha> --core mame0278 --built-by recalbox)",
    )
    release.add_argument("--release", help="mamedev/mame tag, such as mame0278")
    release.add_argument("--profile", type=Path, help="profile file to write")
    release.add_argument("--source-commit", help="libretro/mame commit the platform builds")
    release.add_argument("--core", action="append", default=[], help="core name, repeatable")
    release.add_argument(
        "--built-by", action="append", default=[],
        help="platform that compiles this build, repeatable",
    )
    return parser


def main() -> None:
    logging.basicConfig(
        level=logging.INFO,
        format="%(levelname)s: %(message)s",
    )
    parser = build_parser()
    args = parser.parse_args()
    if args.release or args.profile:
        missing = [
            flag for flag, value in (
                ("--release", args.release), ("--profile", args.profile),
                ("--source-commit", args.source_commit), ("--core", args.core),
                ("--built-by", args.built_by),
            ) if not value
        ]
        if missing:
            parser.error(f"a release profile needs {', '.join(missing)}")
        if args.dry_run or args.json:
            parser.error("--release writes a profile; --dry-run and --json do not apply")
        _write_release(args)
        return
    _run(args)


if __name__ == "__main__":
    main()