chore: merge archive recipes

scripts/archive_build.py

@@ -0,0 +1,538 @@
+"""A recipe for every archive the collection holds, and the build that uses it.
+
+The collection must be able to regenerate each archive it serves from the
+ROMs it holds plus a committed recipe, without any copy of that archive
+from elsewhere. ``recipes/archives.json`` is that recipe list: one record
+per archive in ``database.json``, keyed by its path.
+
+A record names the members in archive order, each with its size, crc32 and
+sha1, and the container's sha1, md5 and size. ``rebuild`` says what the
+build reproduces:
+
+``exact``
+    The archive is TorrentZip and ``build_torrentzip()`` over its members
+    gives its bytes. Proven at capture by rebuilding it.
+``canonical``
+    Same members, other container bytes: the archive was written by another
+    tool. The build writes the TorrentZip form and checks members only.
+
+``sources`` lists the recipes that agree on the member list: profile
+``contents:`` blocks and the MAME and FBNeo DAT snapshots under
+``recipes/``, then ``collection``, the archive itself. A DAT is provenance,
+never the only source. ``bytes_pinned`` marks an archive whose container
+hash a platform or a profile pins, with each pin named.
+
+The build reads each member by sha1: a loose file first, then a member of
+another archive, then the archive's own copy, and verifies what it wrote
+before moving it into place.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import io
+import json
+import os
+import re
+import sys
+import zipfile
+from collections import defaultdict
+from collections.abc import Callable, Iterable
+from concurrent.futures import ProcessPoolExecutor
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import BinaryIO
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+
+from torrentzip import Source as Readable
+from torrentzip import write_torrentzip, zlib_ng_version  # noqa: E402
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+ARCHIVES_FILE = REPO_ROOT / "recipes" / "archives.json"
+# The writer contract: build_torrentzip's layout over stock zlib. A change to
+# the layout is a new builder name and a new proof of every record.
+BUILDER = "torrentzip-1"
+_ARCHIVE_PATH = re.compile(r"\.zip(\.[0-9a-f]{8})?$", re.IGNORECASE)
+_TORRENTZIP_COMMENT = b"TORRENTZIPPED-"
+
+
+def is_archive_path(path: str) -> bool:
+    """A ZIP by its name: ``x.zip`` or a ``.variants/`` copy ``x.zip.1a2b3c4d``."""
+    return bool(_ARCHIVE_PATH.search(path))
+
+
+def database_archives(db: dict) -> dict[str, dict]:
+    """Path -> database entry, for every archive the database lists."""
+    return {
+        entry["path"]: entry
+        for entry in (db.get("files") or {}).values()
+        if is_archive_path(entry.get("path", ""))
+    }
+
+
+# --- the ledger file --------------------------------------------------------
+
+
+def load_ledger(path: Path = ARCHIVES_FILE) -> dict[str, dict]:
+    """Path -> record, from ``recipes/archives.json``, empty when absent."""
+    if not path.is_file():
+        return {}
+    with path.open(encoding="utf-8") as handle:
+        data = json.load(handle)
+    return {record["path"]: record for record in data.get("archives", [])}
+
+
+def ledger_text(records: dict[str, dict]) -> str:
+    """One record per line, so a diff names the archive that changed."""
+    lines = [
+        json.dumps(records[path], ensure_ascii=False, separators=(",", ":"))
+        for path in sorted(records)
+    ]
+    body = ",\n".join(f"  {line}" for line in lines)
+    return f'{{"builder": "{BUILDER}", "archives": [\n{body}\n]}}\n'
+
+
+def write_ledger(records: dict[str, dict], path: Path = ARCHIVES_FILE) -> bool:
+    from artifacts import write_if_changed  # noqa: PLC0415
+
+    path.parent.mkdir(parents=True, exist_ok=True)
+    return write_if_changed(str(path), ledger_text(records))
+
+
+# --- capture ----------------------------------------------------------------
+
+
+class HashSink:
+    """A write target that hashes what it receives, and keeps it if asked."""
+
+    def __init__(self, handle: BinaryIO | None = None) -> None:
+        self.handle = handle
+        self.sha1 = hashlib.sha1()
+        self.md5 = hashlib.md5()
+        self.size = 0
+
+    def write(self, data: bytes) -> int:
+        self.sha1.update(data)
+        self.md5.update(data)
+        self.size += len(data)
+        if self.handle is not None:
+            self.handle.write(data)
+        return len(data)
+
+
+class _Member(io.RawIOBase):
+    """One member of a ZIP, open for reading, closing the archive with it."""
+
+    def __init__(self, archive: Path, name: str) -> None:
+        super().__init__()
+        self._zip = zipfile.ZipFile(archive)
+        self._stream = self._zip.open(name)
+
+    def readable(self) -> bool:
+        return True
+
+    def read(self, size: int = -1) -> bytes:
+        return self._stream.read(size)
+
+    def close(self) -> None:
+        if not self.closed:
+            self._stream.close()
+            self._zip.close()
+        super().close()
+
+
+def _opener(source: Path, member: str | None) -> Callable[[], Readable]:
+    if member is None:
+        return lambda: source.open("rb")
+    return lambda: _Member(source, member)
+
+
+def _file_infos(path: Path) -> tuple[list[zipfile.ZipInfo], list[str], bool]:
+    with zipfile.ZipFile(path) as archive:
+        infos = archive.infolist()
+        torrentzip = archive.comment.startswith(_TORRENTZIP_COMMENT)
+    files = [info for info in infos if not info.is_dir()]
+    directories = [info.filename for info in infos if info.is_dir()]
+    return files, directories, torrentzip
+
+
+def _member_sha1(path: Path, name: str) -> str:
+    digest = hashlib.sha1()
+    with _Member(path, name) as stream:
+        while chunk := stream.read(16 * 1024 * 1024):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def read_archive(path: Path, sha1: str) -> dict:
+    """Members, directories and rebuild class of the archive at *path*.
+
+    A TorrentZip archive is rebuilt from its own members: equal bytes make
+    the recipe ``exact``. A mismatch is rebuilt once more before it is
+    believed, because this host sometimes computes wrongly under load. Any
+    other archive is ``canonical`` and only its members are read.
+    """
+    files, directories, torrentzip = _file_infos(path)
+    rebuild = "canonical"
+    by_name: dict[str, str] = {}
+    if torrentzip:
+        for _attempt in range(2):
+            sink = HashSink()
+            written = write_torrentzip(
+                sink, [(info.filename, _opener(path, info.filename)) for info in files]
+            )
+            by_name = {member.name: member.sha1 for member in written}
+            if sink.sha1.hexdigest() == sha1:
+                rebuild = "exact"
+                break
+    else:
+        by_name = {info.filename: _member_sha1(path, info.filename) for info in files}
+    record: dict = {
+        "rebuild": rebuild,
+        "members": [
+            {
+                "name": info.filename,
+                "size": info.file_size,
+                "crc32": f"{info.CRC:08x}",
+                "sha1": by_name[info.filename],
+            }
+            for info in files
+        ],
+    }
+    if directories:
+        record["directories"] = directories
+    return record
+
+
+def _read_job(job: tuple[str, str]) -> tuple[str, dict | str]:
+    path, sha1 = job
+    try:
+        return path, read_archive(Path(path), sha1)
+    except (OSError, zipfile.BadZipFile, UnicodeEncodeError) as exc:
+        return path, f"{type(exc).__name__}: {exc}"
+
+
+def _hash_values(value: object) -> set[str]:
+    raw = value if isinstance(value, list) else str(value or "").split(",")
+    return {str(item).strip().lower() for item in raw if str(item).strip()}
+
+
+def container_pins(platforms_dir: str, emulators_dir: str) -> dict[str, set[str]]:
+    """Container hash -> who pins it: ``platform:<name>`` or ``profile:<name>``.
+
+    A ``zipped_file`` entry pins a ROM inside the archive, and a Recalbox
+    archive md5 is the composite of its members (``zip_digest: composite``,
+    Bios.cpp:64-68): neither pins the container bytes.
+    """
+    from common import (  # noqa: PLC0415
+        list_registered_platforms,
+        load_emulator_profiles,
+        load_platform_config,
+    )
+
+    pins: dict[str, set[str]] = defaultdict(set)
+    for platform in list_registered_platforms(platforms_dir):
+        config = load_platform_config(platform, platforms_dir)
+        for system in config.get("systems", {}).values():
+            for entry in system.get("files", []) or []:
+                if entry.get("zipped_file") or entry.get("zip_digest") == "composite":
+                    continue
+                for key in ("md5", "sha1", "alt_md5"):
+                    for value in _hash_values(entry.get(key)):
+                        pins[value].add(f"platform:{platform}")
+    for name, profile in load_emulator_profiles(emulators_dir, skip_aliases=False).items():
+        for entry in profile.get("files", []) or []:
+            for key in ("md5", "sha1", "sha256"):
+                for value in _hash_values(entry.get(key)):
+                    pins[value].add(f"profile:{name}")
+    return dict(pins)
+
+
+MemberList = tuple[tuple[str, str], ...]
+
+
+def recipe_index(emulators_dir: str, recipes_dir: str) -> dict[MemberList, set[str]]:
+    """Sorted (name, crc32) -> labels of the recipes with exactly those members.
+
+    Matching on the member list, not the archive name, credits a DAT that
+    knows the set under another name (``aga.zip`` is MAME's ``isa_aga``).
+    """
+    from common import load_emulator_profiles  # noqa: PLC0415
+    from romset_recipes import collect_recipes, load_dat_recipes  # noqa: PLC0415
+
+    index: dict[MemberList, set[str]] = defaultdict(set)
+    profiles = load_emulator_profiles(emulators_dir, skip_aliases=False)
+    for per_label in collect_recipes(profiles).values():
+        for label, recipe in per_label.items():
+            index[tuple(sorted(recipe))].add(f"profile:{label}")
+    for per_label in load_dat_recipes(recipes_dir).values():
+        for label, recipe in per_label.items():
+            index[tuple(sorted(recipe))].add(label)
+    return dict(index)
+
+
+def member_key(members: Iterable[dict]) -> MemberList:
+    return tuple(sorted((m["name"], m["crc32"]) for m in members))
+
+
+def capture(
+    db: dict,
+    previous: dict[str, dict],
+    index: dict[MemberList, set[str]],
+    pins: dict[str, set[str]],
+    workers: int = 4,
+    log: Callable[[str], None] = print,
+) -> tuple[dict[str, dict], dict[str, str]]:
+    """A record for every archive in *db*, and the archives that could not be read.
+
+    A previous record whose sha1 still names the file is kept as read: the
+    container bytes fix the members. Sources and pins are recomputed every
+    time, since profiles, snapshots and platform lists move.
+    """
+    if zlib_ng_version():
+        raise ZlibNgError(zlib_ng_version())
+    archives = database_archives(db)
+    reused: dict[str, dict] = {}
+    jobs: list[tuple[str, str]] = []
+    for path, entry in sorted(archives.items()):
+        old = previous.get(path)
+        if old and old.get("sha1") == entry["sha1"] and old.get("members"):
+            reused[path] = old
+        else:
+            jobs.append((path, entry["sha1"]))
+    read: dict[str, dict] = {}
+    failed: dict[str, str] = {}
+    if jobs:
+        log(f"  reading {len(jobs)} archives ({len(reused)} unchanged)")
+        with ProcessPoolExecutor(max_workers=workers) as pool:
+            for path, result in pool.map(_read_job, jobs, chunksize=4):
+                if isinstance(result, str):
+                    failed[path] = result
+                else:
+                    read[path] = result
+    records: dict[str, dict] = {}
+    for path, entry in sorted(archives.items()):
+        body = reused.get(path) or read.get(path)
+        if body is None:
+            continue
+        record = {
+            "path": path,
+            "sha1": entry["sha1"],
+            "md5": entry["md5"],
+            "size": entry["size"],
+            "rebuild": body["rebuild"],
+            "members": body["members"],
+        }
+        if body.get("directories"):
+            record["directories"] = body["directories"]
+        record["sources"] = sorted(index.get(member_key(body["members"]), set())) + [
+            "collection"
+        ]
+        pinned = set()
+        for value in (entry["md5"], entry["sha1"], entry.get("sha256", "")):
+            pinned |= pins.get(str(value).lower(), set())
+        if pinned:
+            record["bytes_pinned"] = True
+            record["pins"] = sorted(pinned)
+        records[path] = record
+    return records, failed
+
+
+class ZlibNgError(RuntimeError):
+    def __init__(self, version: str | None) -> None:
+        super().__init__(
+            f"this Python deflates with zlib-ng {version}: TorrentZip bytes are "
+            "stock zlib's level-9 output, so an exact rebuild cannot be proven "
+            "or claimed here. Run under a Python linked against stock zlib."
+        )
+
+
+# --- build ------------------------------------------------------------------
+
+
+@dataclass
+class Source:
+    path: Path
+    member: str | None  # None for a loose file
+
+
+@dataclass
+class AtomIndex:
+    """Where each member sha1 can be read in the collection."""
+
+    loose: dict[str, Path] = field(default_factory=dict)
+    inside: dict[str, list[tuple[Path, str]]] = field(default_factory=dict)
+
+    @classmethod
+    def from_collection(cls, ledger: dict[str, dict], db: dict, root: Path) -> AtomIndex:
+        index = cls()
+        for entry in (db.get("files") or {}).values():
+            path = root / entry.get("path", "")
+            if not is_archive_path(entry.get("path", "")) and path.is_file():
+                index.loose.setdefault(entry["sha1"], path)
+        for archive, record in sorted(ledger.items()):
+            path = root / archive
+            if not path.is_file():
+                continue
+            for member in record.get("members", []):
+                index.inside.setdefault(member["sha1"], []).append((path, member["name"]))
+        return index
+
+    def find(self, sha1: str, own: Path) -> Source | None:
+        """A loose copy first, then another archive, then the archive itself."""
+        if sha1 in self.loose:
+            return Source(self.loose[sha1], None)
+        holders = self.inside.get(sha1, [])
+        for path, member in holders:
+            if path != own:
+                return Source(path, member)
+        for path, member in holders:
+            return Source(path, member)
+        return None
+
+    def held(self, sha1: str) -> bool:
+        return sha1 in self.loose or sha1 in self.inside
+
+
+@dataclass
+class BuildResult:
+    path: str
+    status: str  # exact, canonical, impossible, failed
+    output: str = ""
+    detail: str = ""
+    missing: list[dict] = field(default_factory=list)
+    own_members: int = 0
+
+
+def _members_match(built: Path, members: list[dict]) -> str:
+    """Empty when *built* holds exactly *members*, else what differs."""
+    expected = sorted((m["name"], m["size"], m["crc32"], m["sha1"]) for m in members)
+    with zipfile.ZipFile(built) as archive:
+        names = [info.filename for info in archive.infolist() if not info.is_dir()]
+        sizes = {info.filename: info.file_size for info in archive.infolist()}
+        crcs = {info.filename: f"{info.CRC:08x}" for info in archive.infolist()}
+    held = sorted(
+        (name, sizes[name], crcs[name], _member_sha1(built, name)) for name in names
+    )
+    if held == expected:
+        return ""
+    missing = sorted(set(expected) - set(held))
+    extra = sorted(set(held) - set(expected))
+    return f"members differ: expected {missing[:3]}, built {extra[:3]}"
+
+
+def _file_sha1(path: Path) -> str:
+    digest = hashlib.sha1()
+    with path.open("rb") as handle:
+        while chunk := handle.read(16 * 1024 * 1024):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def destination(record_path: str, output_dir: Path) -> Path:
+    """Where a build lands: the archive's path below ``bios/``, under *output_dir*."""
+    parts = Path(record_path).parts
+    relative = Path(*parts[1:]) if parts and parts[0] == "bios" else Path(record_path)
+    return output_dir / relative
+
+
+def build_archive(
+    record: dict,
+    atoms: AtomIndex,
+    output_dir: Path,
+    root: Path = REPO_ROOT,
+    attempts: int = 2,
+) -> BuildResult:
+    """Build *record* from held members into *output_dir*, verified.
+
+    The bytes are written to a scratch file beside the destination and read
+    back from disk: an exact recipe must give its recorded sha1, a canonical
+    one its members. A mismatch is built once more, then refused: nothing
+    wrong is ever moved to the destination. Under zlib-ng an exact recipe is
+    checked by its members only and never reported exact.
+    """
+    path = record["path"]
+    own = root / path
+    sources: list[tuple[str, Source]] = []
+    missing: list[dict] = []
+    for member in record["members"]:
+        source = atoms.find(member["sha1"], own)
+        if source is None:
+            missing.append(member)
+        else:
+            sources.append((member["name"], source))
+    if missing:
+        return BuildResult(path, "impossible", missing=missing)
+    own_members = sum(1 for _name, source in sources if source.path == own)
+    ng = zlib_ng_version()
+    exact = record["rebuild"] == "exact" and not ng
+    target = destination(path, output_dir)
+    target.parent.mkdir(parents=True, exist_ok=True)
+    scratch = target.with_name(f".{target.name}.{os.getpid()}.part")
+    detail = ""
+    try:
+        for _attempt in range(attempts):
+            with scratch.open("wb") as handle:
+                write_torrentzip(
+                    handle,
+                    [(name, _opener(source.path, source.member)) for name, source in sources],
+                )
+            if exact:
+                written = _file_sha1(scratch)
+                detail = "" if written == record["sha1"] else (
+                    f"sha1 {written} is not the recorded {record['sha1']}"
+                )
+            else:
+                detail = _members_match(scratch, record["members"])
+            if not detail:
+                os.replace(scratch, target)
+                status = "exact" if exact else "canonical"
+                note = (
+                    f"zlib-ng {ng}: exact bytes not claimed, members verified"
+                    if ng and record["rebuild"] == "exact" else ""
+                )
+                return BuildResult(path, status, str(target), note, own_members=own_members)
+    finally:
+        if scratch.exists():
+            scratch.unlink()
+    return BuildResult(path, "failed", detail=detail, own_members=own_members)
+
+
+def select(ledger: dict[str, dict], wanted: str, root: Path = REPO_ROOT) -> list[dict]:
+    """Records named by a path (relative or absolute) or by a file name."""
+    candidate = Path(wanted)
+    if candidate.is_absolute():
+        try:
+            candidate = candidate.resolve().relative_to(root.resolve())
+        except ValueError:
+            return []
+    key = candidate.as_posix()
+    if key in ledger:
+        return [ledger[key]]
+    return [record for path, record in sorted(ledger.items()) if Path(path).name == wanted]
+
+
+def missing_members(ledger: dict[str, dict], db: dict) -> list[dict]:
+    """Members held nowhere: neither as a loose file nor inside any archive.
+
+    Read from the committed data alone, so it holds without the files: a
+    member is held when database.json has its sha1 as a file, or when the
+    recipe of an archive database.json lists carries it.
+    """
+    archives = database_archives(db)
+    loose = {
+        sha1 for sha1, entry in (db.get("files") or {}).items()
+        if not is_archive_path(entry.get("path", ""))
+    }
+    inside = {
+        member["sha1"]
+        for path, record in ledger.items() if path in archives
+        for member in record.get("members", [])
+    }
+    return [
+        {"archive": path, "member": member["name"], "sha1": member["sha1"]}
+        for path, record in sorted(ledger.items())
+        for member in record.get("members", [])
+        if member["sha1"] not in loose and member["sha1"] not in inside
+    ]

scripts/generate_site.py

@@ -5985,7 +5985,7 @@ def _romset_line(f: dict) -> str:
     MAME and FBNeo check the members of an archive by CRC32, never its own
     bytes or its member names: a set works with a release when it holds
     every CRC32 the release's recipe lists, extra members or not. The
-    main neogeo.zip, which every MAME from 0.190 to 0.289 loads, read "No
+    main neogeo.zip, which every MAME from 0.177 to 0.289 loads, read "No
     match" when names had to agree and no member could be extra. Empty
     when no recipe names the archive or the file is not on disk.
     """

scripts/pipeline.py

@@ -5,7 +5,8 @@ Steps, numbered as the run prints them:
   0    validate_schemas.py          profiles and platform YAMLs, as PR CI checks them
   1    generate_db.py --force       rebuild database.json from bios/
   1b   provenance_report.py         dump-catalog coverage, derivable.json
-  1c   romset_recipes.py            archive identification, reconstruction targets
+  1c   romset_recipes.py            archive identification, reconstruction targets,
+                                    and --capture: recipes/archives.json
   2    refresh_data_dirs.py         Dolphin Sys, PPSSPP assets and the other trees
   2a   mame_hash_scraper            MAME BIOS hashes, and device sets on a new release
   2a2  fbneo_hash_scraper           FBNeo BIOS hashes
@@ -1216,6 +1217,14 @@ def main() -> None:
         [sys.executable, "scripts/romset_recipes.py"],
         STEP_LABELS["romset_recipes"],
     )
+    # A recipe for every archive database.json lists (recipes/archives.json):
+    # only an archive whose sha1 moved is read again, so the step costs the
+    # archives added since the last run.
+    captured, _ = run(
+        [sys.executable, "scripts/romset_recipes.py", "--capture"],
+        f"{STEP_LABELS['romset_recipes']} (archive recipes)",
+    )
+    ok = ok and captured
     results["romset_recipes"] = ok
     all_ok = all_ok and ok
 

scripts/romset_recipes.py

@@ -20,6 +20,24 @@ Two questions follow, and this module answers both:
 Reconstruction is only possible when the pinned archive is itself TorrentZip.
 An archive whose bytes carry metadata unrelated to its contents cannot be
 derived from ROMs by anyone, and is reported as such rather than guessed at.
+
+Every archive the collection holds also has a recipe of its own in
+``recipes/archives.json`` (scripts/archive_build.py):
+
+``--capture``
+    Write a record for every archive in database.json: members in order with
+    size, crc32 and sha1, container sha1, md5 and size, and whether the
+    TorrentZip build reproduces the bytes (``exact``) or only the members
+    (``canonical``). Archives already recorded under the same sha1 are not
+    read again.
+
+``--build NAME --output-dir DIR``
+    Build an archive from the members the collection holds, loose or inside
+    another archive, and verify it before moving it into place.
+
+``--build-all --output-dir DIR``
+    The same for every recorded archive, with one line per archive: exact,
+    canonical, or impossible with the member nothing holds.
 """
 
 from __future__ import annotations
@@ -43,6 +61,8 @@ from common import (
 )
 from torrentzip import build_torrentzip, is_torrentzip
 
+ARCHIVES_FILE = Path(__file__).resolve().parent.parent / "recipes" / "archives.json"
+
 Recipe = list[tuple[str, str]]
 
 
@@ -290,7 +310,9 @@ def platform_targets(platforms_dir: str) -> dict[str, set[str]]:
     """Container MD5s the platforms pin, per archive name.
 
     ``zipped_file`` entries pin an inner ROM instead of the container, so
-    their MD5 is not an archive hash and is skipped.
+    their MD5 is not an archive hash and is skipped. So is a Recalbox md5,
+    the composite of the members (``zip_digest: composite``, Bios.cpp:64-68):
+    no container build can ever give it.
     """
     targets: dict[str, set[str]] = defaultdict(set)
     for name in list_registered_platforms(platforms_dir):
@@ -298,7 +320,11 @@ def platform_targets(platforms_dir: str) -> dict[str, set[str]]:
         for system in config.get("systems", {}).values():
             for entry in system.get("files", []) or []:
                 archive = entry.get("name", "")
-                if not archive.endswith(".zip") or entry.get("zipped_file"):
+                if (
+                    not archive.endswith(".zip")
+                    or entry.get("zipped_file")
+                    or entry.get("zip_digest") == "composite"
+                ):
                     continue
                 for value in str(entry.get("md5") or "").split(","):
                     value = value.strip().lower()
@@ -453,6 +479,95 @@ def print_missing(gap: dict) -> None:
         print(f"  NO RECIPE: {record['archive']} md5={record['md5'][:8]}")
 
 
+def run_capture(args: argparse.Namespace, db: dict) -> int:
+    """Write recipes/archives.json for every archive the database lists."""
+    import archive_build  # noqa: PLC0415
+
+    try:
+        records, failed = archive_build.capture(
+            db,
+            archive_build.load_ledger(Path(args.archives)),
+            archive_build.recipe_index(args.emulators_dir, args.dat_recipes),
+            archive_build.container_pins(args.platforms_dir, args.emulators_dir),
+        )
+    except archive_build.ZlibNgError as exc:
+        print(f"ERROR: {exc}", file=sys.stderr)
+        return 2
+    for path, reason in sorted(failed.items()):
+        print(f"  UNREADABLE: {path}: {reason}", file=sys.stderr)
+    changed = archive_build.write_ledger(records, Path(args.archives))
+    by_class: dict[str, int] = defaultdict(int)
+    for record in records.values():
+        by_class[record["rebuild"]] += 1
+    agreed = sum(1 for record in records.values() if len(record["sources"]) > 1)
+    pinned = sum(1 for record in records.values() if record.get("bytes_pinned"))
+    print(
+        f"{len(records)} archive recipes: {by_class['exact']} exact, "
+        f"{by_class['canonical']} canonical; {agreed} agree with a profile or DAT "
+        f"recipe, {pinned} pinned as bytes"
+    )
+    print(f"{'Wrote' if changed else 'Unchanged'} {args.archives}")
+    return 1 if failed else 0
+
+
+def run_build(args: argparse.Namespace, db: dict) -> int:
+    """Build one archive, or all of them, from the members the collection holds."""
+    import archive_build  # noqa: PLC0415
+
+    if not args.output_dir:
+        print("ERROR: --build and --build-all need --output-dir", file=sys.stderr)
+        return 2
+    ledger = archive_build.load_ledger(Path(args.archives))
+    if args.build_all:
+        records = [ledger[path] for path in sorted(ledger)]
+    else:
+        records = archive_build.select(ledger, args.build)
+        if not records:
+            print(f"ERROR: no archive recipe named {args.build}", file=sys.stderr)
+            return 2
+    ng = archive_build.zlib_ng_version()
+    if ng:
+        print(
+            f"  zlib-ng {ng} in use: TorrentZip bytes are stock zlib's output, so no "
+            "build is claimed exact here; exact recipes are checked by their members"
+        )
+    root = Path(args.bios_dir).resolve().parent
+    atoms = archive_build.AtomIndex.from_collection(ledger, db, root)
+    counts: dict[str, int] = defaultdict(int)
+    own: list[tuple[str, int]] = []
+    for record in records:
+        result = archive_build.build_archive(record, atoms, Path(args.output_dir), root)
+        counts[result.status] += 1
+        if result.own_members:
+            own.append((result.path, result.own_members))
+        if result.status == "impossible":
+            for member in result.missing:
+                print(
+                    f"  IMPOSSIBLE: {result.path}: missing {member['name']} "
+                    f"({member['sha1']})"
+                )
+        elif result.status == "failed":
+            print(f"  FAILED: {result.path}: {result.detail}")
+        else:
+            note = f" ({result.detail})" if result.detail else ""
+            print(f"  {result.status.upper()}: {result.path}{note}")
+            if args.discard:
+                Path(result.output).unlink()
+    print(
+        f"\n{len(records)} archives: {counts['exact']} rebuilt exact, "
+        f"{counts['canonical']} canonical, {counts['impossible']} impossible, "
+        f"{counts['failed']} failed"
+    )
+    print(
+        f"  {len(own)} of them read {sum(n for _p, n in own)} members from their "
+        "own archive, held nowhere else in the collection"
+    )
+    if counts["failed"]:
+        print("ERROR: a build did not match its recipe and was not kept", file=sys.stderr)
+        return 1
+    return 1 if counts["impossible"] else 0
+
+
 def _resolve_modes(args: argparse.Namespace) -> None:
     """Which reports run: both by default, and --missing whenever --write.
 
@@ -502,11 +617,35 @@ def main() -> int:
         default="recipes",
         help="recipe snapshot file, or a directory of them",
     )
+    parser.add_argument(
+        "--archives", default=str(ARCHIVES_FILE),
+        help="archive recipe file (default: recipes/archives.json)",
+    )
+    parser.add_argument(
+        "--capture", action="store_true",
+        help="write a recipe for every archive in the database",
+    )
+    build = parser.add_mutually_exclusive_group()
+    build.add_argument("--build", metavar="ARCHIVE", help="build one archive, by path or name")
+    build.add_argument("--build-all", action="store_true", help="build every recorded archive")
+    parser.add_argument("--output-dir", help="where --build and --build-all write")
+    parser.add_argument(
+        "--discard", action="store_true",
+        help="delete each build once verified: a full proof without the disk space",
+    )
     args = parser.parse_args()
     require_directory(args.bios_dir)
-    _resolve_modes(args)
+    if (args.output_dir or args.discard) and not (args.build or args.build_all):
+        parser.error("--output-dir and --discard need --build or --build-all")
+    if args.capture and (args.build or args.build_all):
+        parser.error("--capture runs alone")
 
     db = load_database(args.db)
+    if args.capture:
+        return run_capture(args, db)
+    if args.build or args.build_all:
+        return run_build(args, db)
+    _resolve_modes(args)
     profiles = load_emulator_profiles(args.emulators_dir, skip_aliases=False)
     profile_recipes = collect_recipes(profiles)
     dat_recipes = load_dat_recipes(args.dat_recipes)
@@ -560,7 +699,7 @@ def main() -> int:
         )
         print(
             "  The label is the oldest version producing these bytes: a recipe "
-            "unchanged since 0.190 is counted there, not at its own version."
+            "unchanged since 0.177 is counted there, not at its own version."
         )
         by_recipe: dict[str, int] = defaultdict(int)
         for record in ident["identified"]:

scripts/scraper/romset_dat_importer.py

@@ -16,9 +16,12 @@ in its repository, so both are fetched without a browser:
 
 Only sets the collection has a reason to know are kept, so the committed
 snapshot stays small: the archive names that emulator profiles or platform
-BIOS lists reference, and the sets holding a dump-catalog entry the
-collection lacks (provenance_report names the set as where to find it).
-``--all`` keeps everything.
+BIOS lists reference, the archives the collection holds, the sets holding a
+dump-catalog entry the collection lacks (provenance_report names the set as
+where to find it), and every set whose members equal an archive the
+collection keeps under another name. That last entry carries the
+collection's name and keeps the upstream one in ``set``. ``--all`` keeps
+everything.
 
 Members without a CRC32 are undumped. A real romset does not carry them, so
 they are dropped from the recipe rather than disqualifying the whole set.
@@ -80,6 +83,83 @@ def relevant_archives(emulators_dir: str, platforms_dir: str) -> set[str]:
 
 
 Targets = dict[tuple[str, int], set[str]]
+MemberList = tuple[tuple[str, str], ...]
+ByMembers = dict[MemberList, set[str]]
+
+_VARIANT_TAG = re.compile(r"\.[0-9a-f]{8}$")
+
+
+def member_list(members: list[dict]) -> MemberList:
+    """Sorted (name, crc32) of a recipe, the key an archive is matched on."""
+    return tuple(
+        sorted((str(m["name"]), str(m.get("crc32") or "").lower()) for m in members)
+    )
+
+
+def collection_archives(db_path: str) -> tuple[set[str], ByMembers]:
+    """Names and member lists of every archive the collection holds.
+
+    A set the collection keeps under another name is still that set:
+    ``aga.zip`` holds exactly MAME's ``isa_aga`` and ``votrax.zip`` its
+    ``votrsc01a``. Matched by name alone, neither ever entered the snapshot.
+    A ``.variants/`` copy stands for the name its tag is cut from.
+    """
+    with open(db_path, encoding="utf-8") as handle:
+        files = json.load(handle).get("files", {})
+    names: set[str] = set()
+    by_members: ByMembers = {}
+    for entry in files.values():
+        path = Path(entry.get("path", ""))
+        name = _VARIANT_TAG.sub("", path.name)
+        if not name.lower().endswith(".zip") or not path.is_file():
+            continue
+        try:
+            with zipfile.ZipFile(path) as archive:
+                held = [
+                    {"name": info.filename, "crc32": f"{info.CRC:08x}"}
+                    for info in archive.infolist() if not info.is_dir()
+                ]
+        except (OSError, zipfile.BadZipFile) as exc:
+            print(f"  UNREADABLE: {path}: {exc}", file=sys.stderr)
+            continue
+        names.add(name)
+        if held:
+            by_members.setdefault(member_list(held), set()).add(name)
+    return names, by_members
+
+
+def _entry(
+    dat: str, archive: str, set_name: str, description: str, members: list[dict]
+) -> dict:
+    return {
+        "dat": dat,
+        "name": archive,
+        "set": set_name,
+        "description": description,
+        "members": sorted(members, key=lambda member: member["name"]),
+    }
+
+
+def named_entries(
+    dat: str,
+    set_name: str,
+    description: str,
+    members: list[dict],
+    kept: bool,
+    by_members: ByMembers | None,
+) -> list[dict]:
+    """The snapshot entries one upstream set gives.
+
+    Under its own name when kept, and under every collection name whose
+    archive holds exactly its members. ``set`` keeps the upstream name.
+    """
+    archives = {f"{set_name}.zip"} if kept else set()
+    if by_members:
+        archives |= by_members.get(member_list(members), set())
+    return [
+        _entry(dat, archive, set_name, description, members)
+        for archive in sorted(archives)
+    ]
 
 
 def catalog_targets(provenance_dir: str, db_path: str) -> Targets:
@@ -116,8 +196,28 @@ def holds_target(members: list[dict], targets: Targets | None) -> bool:
     return False
 
 
+def one_set_per_archive(entries: list[dict]) -> list[dict]:
+    """One entry per archive and member list, its own set preferred.
+
+    Thirty MAME keyboard sets carry the same ROM, so ``a500kbd_ch.zip``
+    matches all of them. The set named like the archive is the one kept,
+    then the first by name.
+    """
+    chosen: dict[tuple[str, MemberList], dict] = {}
+    for entry in entries:
+        key = (entry["name"], member_list(entry["members"]))
+        current = chosen.get(key)
+        own = entry["name"] == f"{entry['set']}.zip"
+        if current is None or (own and current["name"] != f"{current['set']}.zip"):
+            chosen[key] = entry
+    return sorted(chosen.values(), key=lambda e: (e["name"], e["set"]))
+
+
 def recipe_entries(
-    dat: LogiqxDat, keep: set[str] | None = None, targets: Targets | None = None
+    dat: LogiqxDat,
+    keep: set[str] | None = None,
+    targets: Targets | None = None,
+    by_members: ByMembers | None = None,
 ) -> list[dict]:
     """Group a parsed DAT into one recipe per set."""
     by_set: dict[str, list] = {}
@@ -128,32 +228,22 @@ def recipe_entries(
 
     entries: list[dict] = []
     for set_name, roms in sorted(by_set.items()):
-        archive = f"{set_name}.zip"
-        if keep is not None and archive not in keep and not holds_target(
-            [{"crc32": r.crc32, "size": r.size, "sha1": r.sha1} for r in roms], targets
-        ):
-            continue
-        entries.append(
-            {
-                "dat": dat.name or "unnamed",
-                "name": archive,
-                "set": set_name,
-                "description": roms[0].description,
-                "members": sorted(
-                    (
-                        {
-                            "name": rom.name,
-                            "size": rom.size,
-                            "crc32": rom.crc32,
-                            "sha1": rom.sha1,
-                        }
-                        for rom in roms
-                    ),
-                    key=lambda member: member["name"],
-                ),
-            }
+        members = [
+            {"name": rom.name, "size": rom.size, "crc32": rom.crc32, "sha1": rom.sha1}
+            for rom in roms
+        ]
+        kept = (
+            keep is None
+            or f"{set_name}.zip" in keep
+            or holds_target(members, targets)
         )
-    return entries
+        entries.extend(
+            named_entries(
+                dat.name or "unnamed", set_name, roms[0].description, members,
+                kept, by_members,
+            )
+        )
+    return one_set_per_archive(entries)
 
 
 _MACHINE = re.compile(rb'<machine name="([^"]+)"([^>]*)>')
@@ -240,19 +330,31 @@ def listxml_entries(
     label: str,
     keep: set[str] | None,
     targets: Targets | None = None,
+    by_members: ByMembers | None = None,
 ) -> list[dict]:
     """Recipes from a MAME -listxml document, parents resolved.
 
     A non-merged archive carries its parent set's ROMs as well as its own, so
     a machine with ``romof`` is read as the union, its own members winning a
     name collision. Two passes: the parents a set needs are only known once
-    that set has been seen.
+    that set has been seen. Matching by member list reads every machine in
+    one pass, every parent included.
     """
+    def kept(name: str) -> bool:
+        return keep is None or f"{name}.zip" in keep
+
     with open_stream() as stream:
         wanted = _scan_listxml(
-            stream, lambda name: keep is None or f"{name}.zip" in keep, targets
+            stream, (lambda _name: True) if by_members else kept, targets
         )
-    requested = set(wanted)
+    requested = (
+        {
+            name for name, (_parent, roms) in wanted.items()
+            if kept(name) or holds_target(roms, targets)
+        }
+        if by_members
+        else set(wanted)
+    )
     parents = {
         parent for parent, _roms in wanted.values() if parent and parent not in wanted
     }
@@ -263,23 +365,20 @@ def listxml_entries(
 
     entries: list[dict] = []
     for name, (parent, roms) in sorted(wanted.items()):
-        # A parent read only to complete a child is not itself an entry.
-        if name not in requested:
+        if not by_members and name not in requested:
+            # A parent read only to complete a child is not itself an entry.
             continue
         members = {member["name"]: member for member in wanted.get(parent, (None, []))[1]} if parent else {}
         members.update({member["name"]: member for member in roms})
         if not members:
             continue
-        entries.append(
-            {
-                "dat": label,
-                "name": f"{name}.zip",
-                "set": name,
-                "description": f"parent {parent}" if parent else name,
-                "members": sorted(members.values(), key=lambda m: m["name"]),
-            }
+        entries.extend(
+            named_entries(
+                label, name, f"parent {parent}" if parent else name,
+                list(members.values()), name in requested, by_members,
+            )
         )
-    return entries
+    return one_set_per_archive(entries)
 
 
 def _recipe_key(entry: dict) -> tuple[str, str]:
@@ -609,12 +708,13 @@ def main() -> int:
 
     output = args.output or DEFAULT_OUTPUT.format(source=args.source)
     Path(output).parent.mkdir(parents=True, exist_ok=True)
-    keep = (
-        None
-        if args.all
-        else relevant_archives(args.emulators_dir, args.platforms_dir)
-    )
-    targets = None if args.all else catalog_targets(args.provenance_dir, args.db)
+    keep: set[str] | None = None
+    targets = None
+    by_members: ByMembers | None = None
+    if not args.all:
+        held, by_members = collection_archives(args.db)
+        keep = relevant_archives(args.emulators_dir, args.platforms_dir) | held
+        targets = catalog_targets(args.provenance_dir, args.db)
 
     dats: dict[str, str] = {}
     entries: list[dict] = []
@@ -622,7 +722,9 @@ def main() -> int:
 
     if _is_listxml(pack):
         label = f"MAME {pack.stem.replace('lx', '')}"
-        entries = listxml_entries(_listxml_opener(pack), label, keep, targets)
+        entries = listxml_entries(
+            _listxml_opener(pack), label, keep, targets, by_members
+        )
         dats[label] = pack.stem
         print(f"  {label}: {len(entries)} sets retenus")
         members = sum(len(entry["members"]) for entry in entries)
@@ -645,7 +747,7 @@ def main() -> int:
             continue
         if revision:
             dat.name = f"{dat.name or member} @ {revision}"
-        found = recipe_entries(dat, keep, targets)
+        found = recipe_entries(dat, keep, targets, by_members)
         if not found:
             continue
         dats[dat.name or member] = dat.version

scripts/torrentzip.py

@@ -23,16 +23,140 @@ CLI:
 from __future__ import annotations
 
 import argparse
+import functools
 import hashlib
+import io
 import struct
 import zipfile
 import zlib
+from collections.abc import Callable, Iterable
 from pathlib import Path
+from typing import Any, NamedTuple, Protocol
+
+
+class Source(Protocol):
+    """An open member: read in chunks, closed by its ``with`` block."""
+
+    def read(self, size: int = -1, /) -> bytes: ...
+    def __enter__(self) -> Source: ...
+    def __exit__(self, *exc: Any) -> Any: ...
+
+
+class Sink(Protocol):
+    def write(self, data: bytes, /) -> int: ...
 
 # TorrentZip constants: 1996-12-24 23:32:00, deflate level 9
 _DOS_DATE = ((1996 - 1980) << 9) | (12 << 5) | 24
 _DOS_TIME = (23 << 11) | (32 << 5) | 0
 _COMMENT_PREFIX = "TORRENTZIPPED-"
+# Members are read and deflated this much at a time. Deflate output does not
+# depend on how its input is cut, so a 2 GiB disk image streams through in
+# the same bytes a one-shot compression gives.
+CHUNK = 16 * 1024 * 1024
+# A size or offset that does not fit 32 bits takes a ZIP64 field. No
+# TorrentZip-conformant archive needs one, so their layout is untouched.
+ZIP32_LIMIT = 0xFFFFFFFF
+_ZIP64_MARK = 0xFFFFFFFF
+
+
+class Written(NamedTuple):
+    """A member as the writer read it from its source."""
+
+    name: str
+    size: int
+    crc32: str
+    sha1: str
+
+
+def zlib_ng_version(module: object = zlib) -> str | None:
+    """The zlib-ng version this interpreter deflates with, None for zlib.
+
+    TorrentZip bytes are the output of zlib's level-9 deflate. Stock zlib
+    1.2.13 to 1.3.2 reproduces every TorrentZip archive in the collection.
+    zlib-ng 2.2.4 (python.org 3.14 on Windows) reproduces 479 of 1,561 and
+    2.3.3 545, and the two disagree with each other.
+    """
+    version = getattr(module, "ZLIBNG_VERSION", None)
+    if version:
+        return str(version)
+    runtime = str(getattr(module, "ZLIB_RUNTIME_VERSION", "") or "")
+    return runtime if "zlib-ng" in runtime.lower() else None
+
+
+def _deflate(stream: Source) -> tuple[bytes, Written]:
+    compressor = zlib.compressobj(9, zlib.DEFLATED, -15)
+    blob = bytearray()
+    crc, size, sha1 = 0, 0, hashlib.sha1()
+    while chunk := stream.read(CHUNK):
+        crc = zlib.crc32(chunk, crc)
+        size += len(chunk)
+        sha1.update(chunk)
+        blob += compressor.compress(chunk)
+    blob += compressor.flush()
+    return bytes(blob), Written("", size, f"{crc & 0xFFFFFFFF:08x}", sha1.hexdigest())
+
+
+def write_torrentzip(
+    sink: Sink, members: Iterable[tuple[str, Callable[[], Source]]]
+) -> list[Written]:
+    """Write a TorrentZip archive to *sink*, one member in memory at a time.
+
+    *members* pairs each name with a callable opening its bytes. Members are
+    sorted case-insensitively by name, as the format requires. Returns what
+    was read, in archive order, so a caller can check every member.
+    """
+    ordered = sorted(members, key=lambda m: m[0].lower())
+    central = bytearray()
+    offset = 0
+    written: list[Written] = []
+    for name, opener in ordered:
+        raw = name.encode("ascii")
+        with opener() as stream:
+            blob, read = _deflate(stream)
+        crc = int(read.crc32, 16)
+        written.append(read._replace(name=name))
+        wide = read.size >= ZIP32_LIMIT or len(blob) >= ZIP32_LIMIT
+        far = offset >= ZIP32_LIMIT
+        local_extra = struct.pack("<HHQQ", 1, 16, read.size, len(blob)) if wide else b""
+        central_extra = (
+            struct.pack("<QQ", read.size, len(blob)) if wide else b""
+        ) + (struct.pack("<Q", offset) if far else b"")
+        if central_extra:
+            central_extra = struct.pack("<HH", 1, len(central_extra)) + central_extra
+        sizes = (_ZIP64_MARK, _ZIP64_MARK) if wide else (len(blob), read.size)
+        needed = 45 if wide or far else 20
+        header = struct.pack(
+            "<IHHHHHIIIHH", 0x04034B50, needed, 2, 8, _DOS_TIME, _DOS_DATE,
+            crc, *sizes, len(raw), len(local_extra),
+        ) + raw + local_extra
+        sink.write(header)
+        sink.write(blob)
+        central += struct.pack(
+            "<IHHHHHHIIIHHHHHII", 0x02014B50, 0, needed, 2, 8, _DOS_TIME, _DOS_DATE,
+            crc, *sizes, len(raw), len(central_extra), 0, 0, 0, 0,
+            _ZIP64_MARK if far else offset,
+        ) + raw + central_extra
+        offset += len(header) + len(blob)
+
+    cd_offset = offset
+    sink.write(central)
+    count = len(ordered)
+    if cd_offset >= ZIP32_LIMIT or len(central) >= ZIP32_LIMIT or count >= 0xFFFF:
+        end64 = cd_offset + len(central)
+        sink.write(struct.pack(
+            "<IQHHIIQQQQ", 0x06064B50, 44, 45, 45, 0, 0, count, count,
+            len(central), cd_offset,
+        ))
+        sink.write(struct.pack("<IIQI", 0x07064B50, 0, end64, 1))
+    comment = (
+        f"{_COMMENT_PREFIX}{zlib.crc32(bytes(central)) & 0xFFFFFFFF:08X}"
+    ).encode("ascii")
+    sink.write(struct.pack(
+        "<IHHHHIIH", 0x06054B50, 0, 0, min(count, 0xFFFF), min(count, 0xFFFF),
+        min(len(central), _ZIP64_MARK), min(cd_offset, _ZIP64_MARK), len(comment),
+    ))
+    sink.write(comment)
+    return written
 
 
 def build_torrentzip(members: list[tuple[str, bytes]]) -> bytes:
@@ -41,38 +165,11 @@ def build_torrentzip(members: list[tuple[str, bytes]]) -> bytes:
     Members are sorted case-insensitively by name, as the format requires.
     Returns the complete archive bytes.
     """
-    ordered = sorted(members, key=lambda m: m[0].lower())
-    body = bytearray()
-    central = bytearray()
-
-    for name, data in ordered:
-        raw = name.encode("ascii")
-        compressor = zlib.compressobj(9, zlib.DEFLATED, -15)
-        blob = compressor.compress(data) + compressor.flush()
-        crc = zlib.crc32(data) & 0xFFFFFFFF
-        offset = len(body)
-        body += struct.pack(
-            "<IHHHHHIIIHH", 0x04034B50, 20, 2, 8, _DOS_TIME, _DOS_DATE,
-            crc, len(blob), len(data), len(raw), 0,
-        )
-        body += raw + blob
-        central += struct.pack(
-            "<IHHHHHHIIIHHHHHII", 0x02014B50, 0, 20, 2, 8, _DOS_TIME, _DOS_DATE,
-            crc, len(blob), len(data), len(raw), 0, 0, 0, 0, 0, offset,
-        )
-        central += raw
-
-    cd_offset = len(body)
-    body += central
-    comment = (
-        f"{_COMMENT_PREFIX}{zlib.crc32(bytes(central)) & 0xFFFFFFFF:08X}"
-    ).encode("ascii")
-    body += struct.pack(
-        "<IHHHHIIH", 0x06054B50, 0, 0, len(ordered), len(ordered),
-        len(central), cd_offset, len(comment),
+    sink = io.BytesIO()
+    write_torrentzip(
+        sink, [(name, functools.partial(io.BytesIO, data)) for name, data in members]
     )
-    body += comment
-    return bytes(body)
+    return sink.getvalue()
 
 
 def rebuild_torrentzip(source: str | Path) -> bytes: