libretro/scripts/scraper/targets/batocera_targets_scraper.py

"""Scraper for Batocera per-board emulator availability.

Sources (batocera-linux/batocera.linux):
  - configs/batocera-*.board  -- board definitions, each sets BR2_PACKAGE_BATOCERA_TARGET_*
  - package/batocera/core/batocera-system/Config.in -- select PACKAGE if CONDITION lines
  - package/batocera/emulationstation/batocera-es-system/es_systems.yml
    -- emulator requireAnyOf flag mapping
Sources (batocera-linux/buildroot, at the submodule commit the tag pins):
  - arch/Config.in, package/mesa3d/Config.in, package/libglvnd/Config.in
    -- the symbols Batocera's conditions test but Buildroot sets
    (BR2_ARCH_IS_64, BR2_PACKAGE_HAS_LIBGL)
"""

from __future__ import annotations

import argparse
import json
import re
import sys
import urllib.error
import urllib.request
from datetime import datetime, timezone

import yaml

from safeparse import yaml_load_upstream

from . import BaseTargetScraper

PLATFORM_NAME = "batocera"

GITHUB_API = "https://api.github.com/repos/batocera-linux/batocera.linux/contents"
try:
    from ..base_scraper import github_headers
except ImportError:
    from scraper.base_scraper import github_headers

RAW_BASE = "https://raw.githubusercontent.com/batocera-linux/batocera.linux/{tag}"

CONFIG_IN_PATH = "package/batocera/core/batocera-system/Config.in"
ES_SYSTEMS_PATH = "package/batocera/emulationstation/batocera-es-system/es_systems.yml"

_TARGET_FLAG_RE = re.compile(r"^(BR2_PACKAGE_BATOCERA_TARGET_\w+)=y", re.MULTILINE)

# Meta-flag definition: "if COND\n\tconfig DERIVED_FLAG\n\t...\nendif"
_META_BLOCK_RE = re.compile(
    r"^if\s+((?:[^\n]|\\\n)+?)\n"  # condition (may span lines via \)
    r"(?:.*?\n)*?"  # optional lines before the config
    r"\s+config\s+(BR2_PACKAGE_\w+)"  # derived flag name
    r".*?^endif",  # end of block
    re.MULTILINE | re.DOTALL,
)


def _fetch(url: str, headers: dict | None = None) -> str:
    h = headers or {"User-Agent": "retrobios-scraper/1.0"}
    try:
        req = urllib.request.Request(url, headers=h)
        with urllib.request.urlopen(req, timeout=30) as resp:
            return resp.read().decode("utf-8")
    except urllib.error.URLError as e:
        raise RuntimeError(f"cannot fetch {url}: {e}") from e


def _fetch_json(url: str) -> list | dict:
    text = _fetch(url, headers=github_headers())
    try:
        return json.loads(text)
    except json.JSONDecodeError as e:
        raise RuntimeError(f"cannot parse {url}: {e}") from e


def _normalise_condition(raw: str) -> str:
    """Strip backslash-continuations and collapse whitespace."""
    return re.sub(r"\\\n\s*", " ", raw).strip()


def _tokenise(condition: str) -> list[str]:
    """Split a Kconfig condition into tokens: flags, !, &&, ||, (, )."""
    token_re = re.compile(r'&&|\|\||!|\(|\)|BR2_\w+|"[^"]*"')
    return token_re.findall(condition)


def _check_condition(
    tokens: list[str], pos: int, active: frozenset[str]
) -> tuple[bool, int]:
    """Recursive descent check of a Kconfig boolean expression."""
    return _check_or(tokens, pos, active)


def _check_or(tokens: list[str], pos: int, active: frozenset[str]) -> tuple[bool, int]:
    left, pos = _check_and(tokens, pos, active)
    while pos < len(tokens) and tokens[pos] == "||":
        pos += 1
        right, pos = _check_and(tokens, pos, active)
        left = left or right
    return left, pos


def _check_and(tokens: list[str], pos: int, active: frozenset[str]) -> tuple[bool, int]:
    left, pos = _check_not(tokens, pos, active)
    while pos < len(tokens) and tokens[pos] == "&&":
        pos += 1
        right, pos = _check_not(tokens, pos, active)
        left = left and right
    return left, pos


def _check_not(tokens: list[str], pos: int, active: frozenset[str]) -> tuple[bool, int]:
    if pos < len(tokens) and tokens[pos] == "!":
        pos += 1
        val, pos = _check_atom(tokens, pos, active)
        return not val, pos
    return _check_atom(tokens, pos, active)


def _check_atom(
    tokens: list[str], pos: int, active: frozenset[str]
) -> tuple[bool, int]:
    if pos >= len(tokens):
        return True, pos
    tok = tokens[pos]
    if tok == "(":
        pos += 1
        val, pos = _check_or(tokens, pos, active)
        if pos < len(tokens) and tokens[pos] == ")":
            pos += 1
        return val, pos
    if tok.startswith("BR2_"):
        pos += 1
        return tok in active, pos
    if tok.startswith('"'):
        pos += 1
        return True, pos
    # Unknown token -treat as true to avoid false negatives
    pos += 1
    return True, pos


def _condition_holds(condition: str, active: frozenset[str]) -> bool:
    """Return True if a Kconfig boolean condition holds for the given active flags."""
    if not condition:
        return True
    norm = _normalise_condition(condition)
    tokens = _tokenise(norm)
    if not tokens:
        return True
    try:
        result, _ = _check_condition(tokens, 0, active)
        return result
    except (IndexError, ValueError, TypeError):
        return True  # conservative: include on parse failure


def _parse_meta_flags(text: str) -> list[tuple[str, str]]:
    """Return [(derived_flag, condition_str)] from top-level if/endif blocks.

    These define derived flags like BR2_PACKAGE_BATOCERA_TARGET_X86_64_ANY,
    BR2_PACKAGE_BATOCERA_GLES3, etc.
    """
    results: list[tuple[str, str]] = []
    for m in _META_BLOCK_RE.finditer(text):
        cond = _normalise_condition(m.group(1))
        flag = m.group(2)
        results.append((flag, cond))
    return results


# Buildroot files whose selects decide symbols Batocera's Config.in tests:
# the architecture's 64-bit flag and the OpenGL provider. Without them xroar
# ("select BR2_PACKAGE_XROAR if BR2_PACKAGE_HAS_LIBGL") reached no board.
BUILDROOT_KCONFIG = (
    "arch/Config.in",
    "package/mesa3d/Config.in",
    "package/libglvnd/Config.in",
)
BUILDROOT_RAW = "https://raw.githubusercontent.com/batocera-linux/buildroot/{sha}"
SYSTEM_SYMBOL = "BR2_PACKAGE_BATOCERA_SYSTEM"

_CONFIG_LINE_RE = re.compile(r"^\s*(?:menu)?config\s+(BR2_\w+)")
_OWNED_SELECT_RE = re.compile(r"^\s+select\s+(BR2_\w+)(?:\s+if\s+([^#]+))?")
_BOARD_FLAG_RE = re.compile(r"^(BR2_\w+)=y\s*$", re.MULTILINE)
_BOARD_INCLUDE_RE = re.compile(r"^(-?)include\s+(\S+)", re.MULTILINE)


def _parse_owned_selects(text: str) -> list[tuple[str, str, str]]:
    """Return [(owner, selected, condition)] for every select in a Kconfig.

    A select fires only when the config it sits under is on: Batocera's
    XORG7 comes from BATOCERA_XORG, which BATOCERA_SYSTEM turns on for x86_64
    alone. Reading every select as unconditional put X11 on the Raspberry Pi.
    """
    rules: list[tuple[str, str, str]] = []
    owner: str | None = None
    for line in re.sub(r"\\\n", " ", text).splitlines():
        config = _CONFIG_LINE_RE.match(line)
        if config:
            owner = config.group(1)
            continue
        if line[:1] not in ("", " ", "\t", "#"):
            owner = None
            continue
        select = _OWNED_SELECT_RE.match(line)
        if select and owner:
            rules.append(
                (owner, select.group(1), _normalise_condition(select.group(2) or ""))
            )
    return rules


def _active_symbols(
    board_flags: set[str],
    meta_rules: list[tuple[str, str]],
    rules: list[tuple[str, str, str]],
) -> frozenset[str]:
    """Every symbol a board ends up with: its own, the derived, the selected.

    Run to a fixed point, as Kconfig does: a select can turn on a config
    whose own selects then fire, and a derived flag can follow from either.
    """
    active = set(board_flags)
    while True:
        before = len(active)
        frozen = frozenset(active)
        active |= {
            derived for derived, condition in meta_rules
            if _condition_holds(condition, frozen)
        }
        active |= {
            selected
            for owner, selected, condition in rules
            if owner in frozen and _condition_holds(condition, frozen)
        }
        if len(active) == before:
            return frozenset(active)


def _parse_es_systems(text: str) -> dict[str, list[str]]:
    """Parse es_systems.yml: map BR2_PACKAGE_* flag -> list of emulator names.

    The file is a dict keyed by system name. Each system has:
      emulators:
        <emulator_group>:
          <core_name>: {requireAnyOf: [BR2_PACKAGE_FOO]}
    """
    try:
        data = yaml_load_upstream(text)
    except yaml.YAMLError:
        return {}

    if not isinstance(data, dict):
        return {}

    package_to_emulators: dict[str, list[str]] = {}

    for _system_name, system_data in data.items():
        if not isinstance(system_data, dict):
            continue
        emulators = system_data.get("emulators")
        if not isinstance(emulators, dict):
            continue
        for group_name, group_data in emulators.items():
            if not isinstance(group_data, dict):
                continue
            for core_name, core_data in group_data.items():
                if not isinstance(core_data, dict):
                    continue
                require = core_data.get("requireAnyOf", [])
                if not isinstance(require, list):
                    continue
                # A standalone emulator is its own core: amiberry, fsuae and
                # vice list machines under their name, and the package that
                # builds a machine builds the emulator.
                names = [core_name] if group_name == "libretro" else [group_name, core_name]
                for pkg_flag in require:
                    if isinstance(pkg_flag, str):
                        package_to_emulators.setdefault(pkg_flag, []).extend(names)

    return package_to_emulators


def _arch_from_flag(flag: str) -> str:
    """Guess architecture from board flag name."""
    low = flag.lower()
    if "x86_64" in low or "zen3" in low:
        return "x86_64"
    if "x86" in low:
        return "x86"
    return "aarch64"


class Scraper(BaseTargetScraper):
    """Cross-references Batocera boards, Config.in, and es_systems to build target lists."""

    def __init__(
        self, url: str = "https://github.com/batocera-linux/batocera.linux"
    ) -> None:
        super().__init__(url=url)
        self.tag = ""

    def _raw(self, path: str) -> str:
        return f"{RAW_BASE.format(tag=self.tag)}/{path}"

    def _list_boards(self) -> list[str]:
        """List batocera-*.board files from configs/ via GitHub API."""
        data = _fetch_json(f"{GITHUB_API}/configs?ref={self.tag}")
        boards = [
            item["name"]
            for item in data
            if isinstance(item, dict)
            and item.get("name", "").startswith("batocera-")
            and item.get("name", "").endswith(".board")
        ] if isinstance(data, list) else []
        if not boards:
            raise RuntimeError(f"no batocera-*.board file in configs/ at {self.tag}")
        return boards

    def _fetch_board(self, board_name: str) -> tuple[str | None, set[str]]:
        """A board's target flag and every flag it sets, includes followed.

        batocera-board.common turns on BATOCERA_SYSTEM, the config whose
        selects bring in the display stack and every emulator group.
        """
        flags: set[str] = set()
        target: str | None = None
        pending, seen = [(board_name, False)], set()
        while pending:
            name, optional = pending.pop()
            if name in seen:
                continue
            seen.add(name)
            try:
                text = _fetch(self._raw(f"configs/{name}"))
            except RuntimeError:
                if optional:
                    continue
                raise
            flags |= set(_BOARD_FLAG_RE.findall(text))
            if target is None:
                m = _TARGET_FLAG_RE.search(text)
                target = m.group(1) if m else None
            pending += [(inc, bool(dash)) for dash, inc in _BOARD_INCLUDE_RE.findall(text)]
        return target, flags

    def _buildroot_rules(self) -> list[tuple[str, str, str]]:
        """Buildroot's selects for the symbols Batocera tests, at its pin."""
        pin = _fetch_json(f"{GITHUB_API}/buildroot?ref={self.tag}")
        sha = pin.get("sha") if isinstance(pin, dict) else None
        if not sha:
            raise RuntimeError(f"no buildroot submodule commit at {self.tag}")
        rules: list[tuple[str, str, str]] = []
        for path in BUILDROOT_KCONFIG:
            rules += _parse_owned_selects(_fetch(f"{BUILDROOT_RAW.format(sha=sha)}/{path}"))
        return rules

    def fetch_targets(self) -> dict:
        """Build per-board emulator availability map."""
        self.tag = self.pinned_revision(PLATFORM_NAME)
        print("  fetching board list...", file=sys.stderr)
        boards = self._list_boards()

        print("  fetching Config.in...", file=sys.stderr)
        config_in_url = self._raw(CONFIG_IN_PATH)
        config_in_text = _fetch(config_in_url)
        if not config_in_text:
            raise RuntimeError(f"empty Config.in from {config_in_url}")

        meta_rules = _parse_meta_flags(config_in_text)
        rules = _parse_owned_selects(config_in_text)
        print(
            f"  parsed {len(meta_rules)} meta-flag rules, {len(rules)} select lines",
            file=sys.stderr,
        )
        if not rules:
            raise RuntimeError(f"no 'select BR2_PACKAGE_*' lines parsed from {config_in_url}")
        rules += self._buildroot_rules()

        print("  fetching es_systems.yml...", file=sys.stderr)
        es_systems_url = self._raw(ES_SYSTEMS_PATH)
        es_text = _fetch(es_systems_url)
        if not es_text:
            raise RuntimeError(f"empty es_systems.yml from {es_systems_url}")
        package_to_emulators = _parse_es_systems(es_text)
        print(
            f"  parsed {len(package_to_emulators)} package->emulator mappings",
            file=sys.stderr,
        )
        if not package_to_emulators:
            raise RuntimeError(f"no package->emulator mappings parsed from {es_systems_url}")

        targets: dict[str, dict] = {}
        for board_name in sorted(boards):
            target_key = board_name.removeprefix("batocera-").removesuffix(".board")
            print(f"  processing {target_key}...", file=sys.stderr)
            primary_flag, board_flags = self._fetch_board(board_name)
            if primary_flag is None:
                print(f"    no target flag found in {board_name}", file=sys.stderr)
                continue

            selected_packages = _active_symbols(board_flags, meta_rules, rules)
            # x86_wow64 builds the 32-bit Wine package the x86_64 image
            # carries: no batocera-board.common, no system, no target.
            if SYSTEM_SYMBOL not in selected_packages:
                print(f"    {board_name} builds no Batocera image", file=sys.stderr)
                continue

            # Map selected packages to emulator names via es_systems.yml
            emulators: set[str] = set()
            for pkg in selected_packages:
                for emu in package_to_emulators.get(pkg, []):
                    emulators.add(emu)

            arch = _arch_from_flag(primary_flag)
            targets[target_key] = {
                "architecture": arch,
                "cores": sorted(emulators),
            }
            print(
                f"    {len(emulators)} emulators ({len(selected_packages)} packages selected)",
                file=sys.stderr,
            )

        return {
            "platform": "batocera",
            "source": f"{self.url}/tree/{self.tag}",
            "scraped_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
            "targets": targets,
        }


def main() -> None:
    parser = argparse.ArgumentParser(
        allow_abbrev=False,
        description="Scrape Batocera per-board emulator targets"
    )
    parser.add_argument("--dry-run", action="store_true", help="Show target summary")
    parser.add_argument("--output", "-o", help="Output YAML file")
    args = parser.parse_args()

    scraper = Scraper()
    data = scraper.fetch_targets()

    if args.dry_run:
        for name, info in data["targets"].items():
            print(f"  {name} ({info['architecture']}): {len(info['cores'])} emulators")
        return

    if args.output:
        scraper.write_output(data, args.output)
        print(f"Written to {args.output}")
        return

    print(yaml.dump(data, default_flow_style=False, sort_keys=False))


if __name__ == "__main__":
    main()