Add TRS-80, RX-78, Sega AI entries; refactor tools

Add many MAME/MESS BIOS entries (TRS-80 family, Bandai RX-78, Sega AI) and update docs/navigation counts (README, mkdocs). Remove empty supplemental file references from database.json and update generated timestamps and totals. Harden and refactor tooling: add MAX_RESPONSE_SIZE limited reader in base_scraper, make target scrapers an abstract base, narrow exception handling in the Batocera targets parser, and switch generate_pack.py and verify.py to use build_target_cores_cache (simplifies target config loading and error handling). verify.py also loads supplemental cross-reference names and accepts them through verify_platform. Update tests to import from updated modules (validation/truth). Misc: small bugfix for case-insensitive path conflict check.
This commit is contained in:
Abdessamad Derraz
2026-03-29 23:04:30 +02:00
parent a08c730805
commit 0c5cde83e1
11 changed files with 658 additions and 92 deletions

View File

@@ -27,9 +27,9 @@ from pathlib import Path
sys.path.insert(0, os.path.dirname(__file__))
from common import (
MANUFACTURER_PREFIXES,
build_zip_contents_index, check_inside_zip, compute_hashes,
fetch_large_file, group_identical_platforms, list_emulator_profiles,
list_platform_system_ids, list_registered_platforms,
build_target_cores_cache, build_zip_contents_index, check_inside_zip,
compute_hashes, fetch_large_file, group_identical_platforms,
list_emulator_profiles, list_platform_system_ids, list_registered_platforms,
filter_systems_by_target, list_system_ids, load_database,
load_data_dir_registry, load_emulator_profiles, load_platform_config,
md5_composite, require_yaml, resolve_local_file,
@@ -950,7 +950,7 @@ def generate_pack(
src = os.path.join(root, fname)
rel = os.path.relpath(src, local_path)
full = f"{dd_prefix}/{rel}"
if full in seen_destinations or full.lower() in seen_lower and case_insensitive:
if full in seen_destinations or (full.lower() in seen_lower and case_insensitive):
continue
if _has_path_conflict(full, seen_destinations, seen_parents):
continue
@@ -1913,25 +1913,13 @@ def main():
target_cores_cache: dict[str, set[str] | None] = {}
if args.target:
from common import load_target_config
skip = []
for p in platforms:
try:
target_cores_cache[p] = load_target_config(p, args.target, args.platforms_dir)
except FileNotFoundError:
if args.all:
target_cores_cache[p] = None
else:
print(f"ERROR: No target config for platform '{p}'", file=sys.stderr)
sys.exit(1)
except ValueError as e:
if args.all:
print(f"INFO: Skipping {p}: {e}")
skip.append(p)
else:
print(f"ERROR: {e}", file=sys.stderr)
sys.exit(1)
platforms = [p for p in platforms if p not in skip]
try:
target_cores_cache, platforms = build_target_cores_cache(
platforms, args.target, args.platforms_dir, is_all=args.all,
)
except (FileNotFoundError, ValueError) as e:
print(f"ERROR: {e}", file=sys.stderr)
sys.exit(1)
groups = group_identical_platforms(platforms, args.platforms_dir,
target_cores_cache if args.target else None)

View File

@@ -47,6 +47,24 @@ class ChangeSet:
return ", ".join(parts) if parts else "no changes"
MAX_RESPONSE_SIZE = 50 * 1024 * 1024 # 50 MB
def _read_limited(resp: object, max_bytes: int = MAX_RESPONSE_SIZE) -> bytes:
"""Read an HTTP response with a size limit to prevent OOM."""
chunks: list[bytes] = []
total = 0
while True:
chunk = resp.read(65536) # type: ignore[union-attr]
if not chunk:
break
total += len(chunk)
if total > max_bytes:
raise ValueError(f"Response exceeds {max_bytes} byte limit")
chunks.append(chunk)
return b"".join(chunks)
class BaseScraper(ABC):
"""Abstract base class for platform BIOS requirement scrapers."""
@@ -63,7 +81,7 @@ class BaseScraper(ABC):
try:
req = urllib.request.Request(self.url, headers={"User-Agent": "retrobios-scraper/1.0"})
with urllib.request.urlopen(req, timeout=30) as resp:
self._raw_data = resp.read().decode("utf-8")
self._raw_data = _read_limited(resp).decode("utf-8")
return self._raw_data
except urllib.error.URLError as e:
raise ConnectionError(f"Failed to fetch {self.url}: {e}") from e

View File

@@ -6,18 +6,20 @@ from __future__ import annotations
import importlib
import pkgutil
from abc import ABC, abstractmethod
from pathlib import Path
class BaseTargetScraper:
class BaseTargetScraper(ABC):
"""Base class for target scrapers."""
def __init__(self, url: str = ""):
self.url = url
@abstractmethod
def fetch_targets(self) -> dict:
"""Fetch targets and their core lists. Returns dict matching target YAML format."""
raise NotImplementedError
...
def write_output(self, data: dict, output_path: str) -> None:
"""Write target data to YAML file."""

View File

@@ -152,7 +152,7 @@ def _condition_holds(condition: str, active: frozenset[str]) -> bool:
try:
result, _ = _check_condition(tokens, 0, active)
return result
except Exception:
except (IndexError, ValueError, TypeError):
return True # conservative: include on parse failure

View File

@@ -30,11 +30,11 @@ from pathlib import Path
sys.path.insert(0, os.path.dirname(__file__))
from common import (
build_zip_contents_index, check_inside_zip, compute_hashes,
filter_systems_by_target, group_identical_platforms, list_emulator_profiles,
list_system_ids, load_data_dir_registry, load_emulator_profiles,
load_platform_config, md5sum, md5_composite, require_yaml, resolve_local_file,
resolve_platform_cores,
build_target_cores_cache, build_zip_contents_index, check_inside_zip,
compute_hashes, filter_systems_by_target, group_identical_platforms,
list_emulator_profiles, list_system_ids, load_data_dir_registry,
load_emulator_profiles, load_platform_config, md5sum, md5_composite,
require_yaml, resolve_local_file, resolve_platform_cores,
)
yaml = require_yaml()
@@ -504,6 +504,7 @@ def verify_platform(
emu_profiles: dict | None = None,
target_cores: set[str] | None = None,
data_dir_registry: dict | None = None,
supplemental_names: set[str] | None = None,
) -> dict:
"""Verify all BIOS files for a platform, including cross-reference gaps."""
mode = config.get("verification_mode", "existence")
@@ -601,10 +602,11 @@ def verify_platform(
status_counts[s] = status_counts.get(s, 0) + 1
# Cross-reference undeclared files
from cross_reference import _build_supplemental_index
data_names = _build_supplemental_index()
if supplemental_names is None:
from cross_reference import _build_supplemental_index
supplemental_names = _build_supplemental_index()
undeclared = find_undeclared_files(config, emulators_dir, db, emu_profiles,
target_cores=target_cores, data_names=data_names)
target_cores=target_cores, data_names=supplemental_names)
exclusions = find_exclusion_notes(config, emulators_dir, emu_profiles, target_cores=target_cores)
# Ground truth coverage
@@ -1260,29 +1262,20 @@ def main():
target_cores_cache: dict[str, set[str] | None] = {}
if args.target:
from common import load_target_config
skip = []
for p in platforms:
try:
target_cores_cache[p] = load_target_config(p, args.target, args.platforms_dir)
except FileNotFoundError:
if args.all:
target_cores_cache[p] = None
else:
print(f"ERROR: No target config for platform '{p}'", file=sys.stderr)
sys.exit(1)
except ValueError as e:
if args.all:
print(f"INFO: Skipping {p}: {e}")
skip.append(p)
else:
print(f"ERROR: {e}", file=sys.stderr)
sys.exit(1)
platforms = [p for p in platforms if p not in skip]
try:
target_cores_cache, platforms = build_target_cores_cache(
platforms, args.target, args.platforms_dir, is_all=args.all,
)
except (FileNotFoundError, ValueError) as e:
print(f"ERROR: {e}", file=sys.stderr)
sys.exit(1)
# Group identical platforms (same function as generate_pack)
groups = group_identical_platforms(platforms, args.platforms_dir,
target_cores_cache if args.target else None)
from cross_reference import _build_supplemental_index
suppl_names = _build_supplemental_index()
all_results = {}
group_results: list[tuple[dict, list[str]]] = []
for group_platforms, representative in groups:
@@ -1291,6 +1284,7 @@ def main():
result = verify_platform(
config, db, args.emulators_dir, emu_profiles,
target_cores=tc, data_dir_registry=data_registry,
supplemental_names=suppl_names,
)
names = [load_platform_config(p, args.platforms_dir).get("platform", p) for p in group_platforms]
group_results.append((result, names))