mirror of
https://github.com/Abdess/retroarch_system.git
synced 2026-10-24 12:49:03 -05:00
Abdessamad Derraz
a74a56abc4
The acquisition report never read a profile, so XM7's required DICROM.ROM sat among 2 287 targets at the rank of dumps no core reads. A missing entry that a profile reads and the collection lacks is now listed first with the profile, matched by hash or, for a hashless entry with a size, by name and size. 29 entries on the current snapshots.
462 lines
19 KiB
Python
462 lines
19 KiB
Python
#!/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
|
|
|
|
A missing dump that is a file an emulator profile reads and the collection
|
|
lacks is named with that profile: by hash when the profile declares one, by
|
|
name and size when it declares a size and no hash.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import sys
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, os.path.dirname(__file__))
|
|
from common import (
|
|
DEFAULT_PROVENANCE_DIR,
|
|
load_database,
|
|
load_emulator_profiles,
|
|
load_provenance_snapshots,
|
|
write_if_changed,
|
|
)
|
|
|
|
HASH_FIELDS = ("sha1", "md5", "sha256")
|
|
|
|
|
|
def _values(entry: dict, field: str) -> set[str]:
|
|
"""Every value a profile or catalog entry gives for a hash field."""
|
|
value = entry.get(field)
|
|
values = value if isinstance(value, list) else str(value or "").split(",")
|
|
return {str(v).strip().lower() for v in values if str(v).strip()}
|
|
|
|
|
|
def _crc_key(value: str) -> str:
|
|
return value.removeprefix("0x").zfill(8)
|
|
|
|
|
|
def _fits(size: int | None, entry: dict) -> bool:
|
|
"""Whether a size meets what a profile entry states about it."""
|
|
if size is None:
|
|
return False
|
|
declared = entry.get("size")
|
|
if declared is not None:
|
|
accepted = declared if isinstance(declared, list) else [declared]
|
|
return size in {int(value) for value in accepted}
|
|
low, high = entry.get("min_size"), entry.get("max_size")
|
|
return (low is None or size >= int(low)) and (high is None or size <= int(high))
|
|
|
|
|
|
def profile_needs(profiles: dict, db: dict) -> dict:
|
|
"""The files emulator profiles read that the collection does not hold.
|
|
|
|
Keyed by every hash a profile accepts, and, for an entry that declares
|
|
a size and no hash, by its names casefolded. An entry with neither
|
|
cannot be told apart from any same-named dump and is left out.
|
|
"""
|
|
files = db.get("files", {})
|
|
indexes = db.get("indexes", {})
|
|
held_hashes = {
|
|
"sha1": set(files),
|
|
"md5": set(indexes.get("by_md5", {})),
|
|
"sha256": set(indexes.get("by_sha256", {})),
|
|
}
|
|
held_crc = {
|
|
(_crc_key(str(f.get("crc32", "")).lower()), f.get("size"))
|
|
for f in files.values() if f.get("crc32")
|
|
}
|
|
sizes_by_name: dict[str, set[int]] = {}
|
|
for record in files.values():
|
|
sizes_by_name.setdefault(str(record.get("name", "")).casefold(), set()).add(
|
|
record.get("size")
|
|
)
|
|
|
|
by_hash: dict[tuple, list[dict]] = {}
|
|
by_name: dict[str, list[tuple[dict, dict]]] = {}
|
|
for emulator, profile in sorted(profiles.items()):
|
|
for entry in profile.get("files", []) or []:
|
|
name = str(entry.get("name", ""))
|
|
if not name or name.endswith("/"):
|
|
continue
|
|
need = {"emulator": emulator, "name": name,
|
|
"required": bool(entry.get("required"))}
|
|
hashes = {(field, v) for field in HASH_FIELDS for v in _values(entry, field)}
|
|
crcs = {_crc_key(v) for v in _values(entry, "crc32")}
|
|
declared = entry.get("size")
|
|
sizes = [] if declared is None else (
|
|
declared if isinstance(declared, list) else [declared]
|
|
)
|
|
crc_keys = {(crc, int(size)) for crc in crcs for size in sizes}
|
|
if hashes or crc_keys:
|
|
if any(v in held_hashes[field] for field, v in hashes) or (
|
|
crc_keys & held_crc
|
|
):
|
|
continue
|
|
for key in hashes | {("crc32", crc, size) for crc, size in crc_keys}:
|
|
by_hash.setdefault(key, []).append(need)
|
|
continue
|
|
if all(entry.get(k) is None for k in ("size", "min_size", "max_size")):
|
|
continue
|
|
names = [name, *(entry.get("aliases") or [])]
|
|
if any(
|
|
_fits(held, entry)
|
|
for alias in names
|
|
for held in sizes_by_name.get(str(alias).casefold(), ())
|
|
):
|
|
continue
|
|
for alias in names:
|
|
by_name.setdefault(str(alias).casefold(), []).append((need, entry))
|
|
return {"by_hash": by_hash, "by_name": by_name}
|
|
|
|
|
|
def needs_filled(entry: dict, needs: dict) -> list[dict]:
|
|
"""The profile needs a catalog dump answers, required ones first."""
|
|
found: dict[tuple[str, str], dict] = {}
|
|
keys = [(field, v) for field in HASH_FIELDS for v in _values(entry, field)]
|
|
if entry.get("size") is not None:
|
|
keys += [("crc32", _crc_key(v), entry["size"]) for v in _values(entry, "crc32")]
|
|
for key in keys:
|
|
for need in needs["by_hash"].get(key, []):
|
|
found.setdefault((need["emulator"], need["name"]), {**need, "by": key[0]})
|
|
size = entry.get("size")
|
|
name = str(entry.get("name", "")).casefold()
|
|
for need, declared in needs["by_name"].get(name, []):
|
|
if _fits(size, declared):
|
|
found.setdefault(
|
|
(need["emulator"], need["name"]), {**need, "by": "name and size"}
|
|
)
|
|
return sorted(found.values(), key=lambda n: (not n["required"], n["emulator"], n["name"]))
|
|
|
|
|
|
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
|
|
|
|
|
|
RECIPES_DIR = Path(__file__).resolve().parent.parent / "recipes"
|
|
|
|
|
|
def romset_index(recipes_dir: Path = RECIPES_DIR) -> dict[tuple[str, int], list[dict]]:
|
|
"""(crc32, size) of every member of an imported romset -> where it sits.
|
|
|
|
A dump the collection lacks may still ship inside an emulator's romset:
|
|
17 TOSEC ZX Spectrum ROMs are members of MAME's spectrum.zip. Naming the
|
|
set and the versions that carry it turns a dead end into a download.
|
|
"""
|
|
index: dict[tuple[str, int], list[dict]] = {}
|
|
for snapshot in sorted(recipes_dir.glob("*.json")):
|
|
with snapshot.open(encoding="utf-8") as handle:
|
|
data = json.load(handle)
|
|
source = data.get("source", snapshot.stem)
|
|
for entry in data.get("entries", []):
|
|
dats = entry.get("dats") or [entry.get("dat", "")]
|
|
for member in entry.get("members", []) or []:
|
|
crc, size = str(member.get("crc32") or "").lower(), member.get("size")
|
|
if not crc or size is None:
|
|
continue
|
|
index.setdefault((crc.zfill(8), int(size)), []).append({
|
|
"source": source,
|
|
"archive": entry.get("name", ""),
|
|
"member": member.get("name", ""),
|
|
"sha1": member.get("sha1", ""),
|
|
"dats": dats,
|
|
})
|
|
return index
|
|
|
|
|
|
def romsets_holding(entry: dict, index: dict) -> list[dict]:
|
|
"""The imported romsets whose members are this dump.
|
|
|
|
A member and an entry that both carry a sha1 must agree on it: crc32
|
|
and size alone match by accident across 40 000 ROMs.
|
|
"""
|
|
crc, size = str(entry.get("crc32") or "").lower(), entry.get("size")
|
|
if not crc or size is None:
|
|
return []
|
|
sha1 = str(entry.get("sha1") or "").lower()
|
|
# A set whose contents changed between versions is stored once per
|
|
# distinct content: one line per set and member, with every version.
|
|
merged: dict[tuple[str, str, str], dict] = {}
|
|
for hit in index.get((crc.zfill(8), int(size)), []):
|
|
if sha1 and hit["sha1"] and hit["sha1"].lower() != sha1:
|
|
continue
|
|
key = (hit["source"], hit["archive"], hit["member"])
|
|
held = merged.setdefault(key, {**hit, "dats": []})
|
|
held["dats"] = sorted(set(held["dats"]) | set(hit["dats"]))
|
|
return list(merged.values())
|
|
|
|
|
|
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,
|
|
romsets: dict | None = None,
|
|
needs: 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
|
|
romsets = romset_index() if romsets is None else romsets
|
|
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())
|
|
for entry in missing:
|
|
holders = romsets_holding(entry, romsets)
|
|
if holders:
|
|
entry["romsets"] = holders
|
|
filled = needs_filled(entry, needs) if needs else []
|
|
if filled:
|
|
entry["needed_by"] = filled
|
|
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 35 s",
|
|
)
|
|
parser.add_argument("--json", action="store_true", help="Full report as JSON")
|
|
parser.add_argument(
|
|
"--emulators-dir", default="emulators",
|
|
help="profiles whose unheld files a missing dump answers (default: %(default)s)",
|
|
)
|
|
parser.add_argument(
|
|
"--write-derivable",
|
|
metavar="PATH",
|
|
help="With --derive, record the derivable entries for the site "
|
|
"(the pipeline writes derivable.json)",
|
|
)
|
|
args = parser.parse_args()
|
|
if args.write_derivable and not args.derive:
|
|
parser.error("--write-derivable records what --derive finds: pass both")
|
|
|
|
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
|
|
|
|
needs = profile_needs(load_emulator_profiles(args.emulators_dir), db)
|
|
report = build_report(db, snapshots, archive_members(db), needs=needs)
|
|
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.write_derivable:
|
|
# The site cannot spend 80 s reading the collection on every
|
|
# build: the pipeline records what this pass found, and the
|
|
# Wanted page stops asking for dumps the collection can produce.
|
|
write_if_changed(
|
|
args.write_derivable,
|
|
json.dumps(dict(sorted(derived.items())), indent=2, sort_keys=True) + "\n",
|
|
)
|
|
|
|
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:
|
|
# A dump an emulator reads comes first: it fills a hole in a pack.
|
|
for entry in sorted(data["missing"], key=lambda e: not e.get("needed_by")):
|
|
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')}")
|
|
for need in entry.get("needed_by", []):
|
|
print(f" needed by {need['emulator']} as {need['name']} "
|
|
f"({'required' if need['required'] else 'optional'}, "
|
|
f"matched by {need['by']})")
|
|
for hit in entry.get("romsets", []):
|
|
versions = hit["dats"][0] if len(hit["dats"]) == 1 else (
|
|
f"{hit['dats'][0]} to {hit['dats'][-1]}"
|
|
)
|
|
print(f" in {hit['source']} {hit['archive']} as "
|
|
f"{hit['member']} ({versions})")
|
|
|
|
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)
|
|
needed = {
|
|
dump_identity(e)
|
|
for data in report.values() for e in data["missing"] if e.get("needed_by")
|
|
}
|
|
if needed:
|
|
print(f" {len(needed)} of them a profile reads and the collection lacks"
|
|
+ ("" if args.missing else " (--missing names them)"))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
from common import run_cli
|
|
|
|
run_cli(main)
|