libretro/scripts/provenance_report.py

#!/usr/bin/env python3
"""Report dump-catalog coverage of the collection.

Reads provenance/*.json snapshots and database.json, then reports per
source how many catalog entries the collection holds and which are
missing. Missing entries are acquisition targets: catalog-verified
dumps the collection does not have yet.

Usage:
    python scripts/provenance_report.py
    python scripts/provenance_report.py --missing
    python scripts/provenance_report.py --json
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import sys
import zipfile

sys.path.insert(0, os.path.dirname(__file__))
from common import DEFAULT_PROVENANCE_DIR, load_database, load_provenance_snapshots


def archive_members(db: dict) -> dict[tuple[str, int], list[tuple[str, str]]]:
    """(crc32, size) of every member of the collection's ZIPs -> (zip, member).

    Read from the central directories alone: a romset member is a dump the
    collection already holds, and listing it as an acquisition target sent
    searches after astrocdw.zip's bioswhit.bin and the Gamate BIOS.
    """
    members: dict[tuple[str, int], list[tuple[str, str]]] = {}
    for entry in db.get("files", {}).values():
        path = entry.get("path", "")
        if not path.endswith(".zip") or not os.path.exists(path):
            continue
        try:
            with zipfile.ZipFile(path) as archive:
                for info in archive.infolist():
                    if info.is_dir():
                        continue
                    key = (f"{info.CRC:08x}", info.file_size)
                    members.setdefault(key, []).append((path, info.filename))
        except (zipfile.BadZipFile, OSError) as exc:
            print(f"  WARNING: {path}: {exc}", file=sys.stderr)
    return members


def _held_in_archive(entry: dict, members: dict) -> bool:
    """Whether a catalog entry is a member of one of the collection's ZIPs.

    crc32 and size only nominate candidates; a declared sha1 must match the
    member's bytes.
    """
    crc = str(entry.get("crc32") or "").lower()
    size = entry.get("size")
    if not crc or size is None:
        return False
    candidates = members.get((crc.zfill(8), int(size)), [])
    sha1 = str(entry.get("sha1") or "").lower()
    if not sha1:
        return bool(candidates)
    for path, name in candidates:
        try:
            with zipfile.ZipFile(path) as archive:
                if hashlib.sha1(archive.read(name)).hexdigest() == sha1:
                    return True
        except (zipfile.BadZipFile, OSError, KeyError) as exc:
            print(f"  WARNING: {path}:{name}: {exc}", file=sys.stderr)
    return False


def dump_identity(entry: dict) -> tuple:
    """What makes two catalog listings one dump: the strongest hash stated."""
    if entry.get("sha1"):
        return ("sha1", str(entry["sha1"]).lower())
    if entry.get("md5"):
        return ("md5", str(entry["md5"]).lower(), entry.get("size"))
    if entry.get("crc32"):
        return ("crc32", str(entry["crc32"]).lower(), entry.get("size"))
    return ("name", entry.get("dat", ""), entry.get("name", ""))


def distinct_missing(report: dict) -> int:
    """Missing dumps across every catalog, each counted once."""
    return len({dump_identity(e) for data in report.values() for e in data["missing"]})


def build_report(db: dict, snapshots: dict, members: dict | None = None) -> dict:
    """Compare each snapshot against the collection.

    A DAT counts as covered when the collection holds at least one of
    its entries. Missing entries from covered DATs are acquisition
    targets; missing entries from DATs the collection does not cover at
    all are out of scope and only counted. No-Intro tags every non-game
    dump "[BIOS]", including digital title distribution, so without this
    split the target list is swamped by content the project never ships.
    """
    by_sha1 = db.get("files", {})
    # The members a romset holds are held too, whoever asks: the site built
    # its Wanted page without them and asked the public for the Astrocade
    # and Gamate BIOS that astrocdw.zip and gamate.zip already carry.
    members = archive_members(db) if members is None else members
    by_md5_size = {
        (entry.get("md5", ""), entry.get("size", 0)) for entry in by_sha1.values()
    }
    # Redump lists some dumps by crc32 alone (ps2-0101jd-20030110, DTL-H10100):
    # collected as a loose file it would stay MISSING for good without this.
    by_crc32_size = {
        (str(entry.get("crc32", "")).lower(), entry.get("size", 0))
        for entry in by_sha1.values()
        if entry.get("crc32")
    }

    report = {}
    for source, snapshot in sorted(snapshots.items()):
        matched: set = set()
        covered_dats = set()
        unmatched = []
        for entry in snapshot["entries"]:
            if (
                entry.get("sha1") in by_sha1
                or (entry.get("md5"), entry.get("size")) in by_md5_size
                or (str(entry.get("crc32") or "").lower(), entry.get("size")) in by_crc32_size
                or _held_in_archive(entry, members)
            ):
                matched.add(dump_identity(entry))
                covered_dats.add(entry.get("dat", ""))
            else:
                unmatched.append(entry)
        # One dump listed by several DATs (No-Intro's three FM-7 DATs share
        # DICROM.rom, KANJI.rom...) is one dump: counted per listing, the
        # report announced 2 485 missing for 2 284 distinct ones.
        merged: dict = {}
        for entry in unmatched:
            if entry.get("dat", "") not in covered_dats:
                continue
            held = merged.get(dump_identity(entry))
            if held is None:
                merged[dump_identity(entry)] = {**entry, "dats": [entry.get("dat", "")]}
            elif entry.get("dat", "") not in held["dats"]:
                held["dats"].append(entry.get("dat", ""))
        missing = list(merged.values())
        out_of_scope = len({
            dump_identity(e) for e in unmatched if e.get("dat", "") not in covered_dats
        })
        matched = len(matched)
        report[source] = {
            "imported_at": snapshot.get("imported_at", ""),
            "dats": snapshot.get("dats", {}),
            "total": len(snapshot["entries"]),
            "matched": matched,
            "covered_dats": sorted(covered_dats),
            "missing": missing,
            "out_of_scope": out_of_scope,
        }
    return report


def main() -> int:
    parser = argparse.ArgumentParser(allow_abbrev=False, description="Dump-catalog coverage report")
    parser.add_argument("--db", default="database.json", help="Database path")
    parser.add_argument(
        "--provenance-dir",
        default=DEFAULT_PROVENANCE_DIR,
        help="Directory with dump-catalog snapshots",
    )
    parser.add_argument("--missing", action="store_true", help="List missing entries")
    parser.add_argument(
        "--derive",
        action="store_true",
        help="Find missing entries a held dump contains (halves, word splits, "
        "byte swaps, concatenations); reads the collection, about 90 s",
    )
    parser.add_argument("--json", action="store_true", help="Full report as JSON")
    args = parser.parse_args()

    db = load_database(args.db)
    snapshots = load_provenance_snapshots(args.provenance_dir)
    if not snapshots:
        print(f"No provenance snapshots in {args.provenance_dir}/")
        return 0

    report = build_report(db, snapshots, archive_members(db))
    derived: dict[str, dict] = {}
    if args.derive:
        from derivations import derivable  # noqa: PLC0415

        derived = derivable((e for d in report.values() for e in d["missing"]), db)
        for data in report.values():
            data["derivable"] = {
                e["sha1"]: derived[e["sha1"]] for e in data["missing"]
                if e.get("sha1") in derived
            }

    if args.json:
        print(json.dumps(report, indent=2))
        return 0

    for source, data in report.items():
        in_scope = data["matched"] + len(data["missing"])
        pct = 100 * data["matched"] / in_scope if in_scope else 0
        print(
            f"  {source} ({data['imported_at']}): "
            f"{data['matched']}/{in_scope} in collection ({pct:.0f}%) "
            f"across {len(data['covered_dats'])} covered DATs"
        )
        if data["out_of_scope"]:
            print(
                f"    {data['out_of_scope']} entries in DATs the collection "
                f"does not cover, not counted as targets"
            )
        if args.missing:
            for entry in data["missing"]:
                label = entry.get("description") or entry["name"]
                source = derived.get(entry.get("sha1", ""))
                if source:
                    print(f"    DERIVABLE {entry['name']} ({label}): {source['transform']} "
                          f"of {' + '.join(source['from'])}")
                else:
                    print(f"    MISSING {entry['name']} ({label}) sha1={entry.get('sha1')}")

    total_missing = distinct_missing(report)
    if total_missing:
        line = f"  {total_missing} catalogued dumps missing from collection"
        if args.derive:
            line += f", {len(derived)} of them derivable from held dumps"
        print(line)
    return 0


if __name__ == "__main__":
    from common import run_cli

    run_cli(main)