mirror of
https://github.com/Abdess/retroarch_system.git
synced 2026-10-21 11:19:11 -05:00
Abdessamad Derraz
4231902a6c
Profiles, the wiki, comments, docstrings, command output and schema descriptions carried em-dashes, clause-joining semicolons, unicode arrows, box-drawing banners and check-mark glyphs. Each sentence was rewritten rather than having its character swapped. Hashes, paths, source_ref values and every cited file:line are unchanged, which a comparison of each parsed profile against the previous revision confirms. validate_pr comments now say PASS and FAIL in words, and source_ref values in a profile's structured blocks render as code. test_prose_style fails if any of these comes back, in tracked files or in the generated site.
107 lines
3.7 KiB
Python
107 lines
3.7 KiB
Python
"""Dump catalogues joined onto the collection.
|
|
|
|
Redump, No-Intro and TOSEC annotate what the repo holds. They are an
|
|
opinion on provenance, never an authority: the emulator source is."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from artifacts import write_if_changed
|
|
|
|
|
|
DEFAULT_PROVENANCE_DIR = "provenance"
|
|
|
|
def load_provenance_snapshots(provenance_dir: str = DEFAULT_PROVENANCE_DIR) -> dict:
|
|
"""Load dump-catalog snapshots from provenance/*.json.
|
|
|
|
Returns {source_name: snapshot} where snapshot holds the normalized
|
|
entries written by the redump scraper or the pack importer. Missing
|
|
directory means no snapshots: returns an empty dict.
|
|
"""
|
|
snapshots = {}
|
|
prov_path = Path(provenance_dir)
|
|
if not prov_path.is_dir():
|
|
return snapshots
|
|
for path in sorted(prov_path.glob("*.json")):
|
|
with open(path) as f:
|
|
snapshot = json.load(f)
|
|
source = snapshot.get("source")
|
|
if source and snapshot.get("entries"):
|
|
snapshots[source] = snapshot
|
|
return snapshots
|
|
|
|
def build_provenance_index(snapshots: dict) -> dict:
|
|
"""Index snapshot entries by sha1 and by (md5, size) per source.
|
|
|
|
First entry wins on hash collisions within a source. Entries are
|
|
pre-sorted at snapshot write time so the outcome is deterministic.
|
|
"""
|
|
index = {}
|
|
for source, snapshot in snapshots.items():
|
|
by_sha1 = {}
|
|
by_md5_size = {}
|
|
for entry in snapshot["entries"]:
|
|
sha1 = entry.get("sha1", "")
|
|
md5 = entry.get("md5", "")
|
|
if sha1 and sha1 not in by_sha1:
|
|
by_sha1[sha1] = entry
|
|
if md5 and entry.get("size"):
|
|
key = (md5, entry["size"])
|
|
if key not in by_md5_size:
|
|
by_md5_size[key] = entry
|
|
index[source] = {"by_sha1": by_sha1, "by_md5_size": by_md5_size}
|
|
return index
|
|
|
|
def annotate_provenance(files: dict, snapshots: dict) -> dict[str, int]:
|
|
"""Attach a provenance field to database file entries.
|
|
|
|
Matches by SHA1 first, then MD5 + size. Returns per-source match
|
|
counts. Files without any catalog match keep no provenance field.
|
|
"""
|
|
index = build_provenance_index(snapshots)
|
|
counts = dict.fromkeys(index, 0)
|
|
for sha1, entry in files.items():
|
|
matches = {}
|
|
for source in sorted(index):
|
|
src_index = index[source]
|
|
hit = src_index["by_sha1"].get(sha1) or src_index["by_md5_size"].get(
|
|
(entry.get("md5", ""), entry.get("size", 0))
|
|
)
|
|
if hit:
|
|
matches[source] = {
|
|
"dat": hit.get("dat", ""),
|
|
"name": hit.get("name", ""),
|
|
"description": hit.get("description", ""),
|
|
}
|
|
counts[source] += 1
|
|
if matches:
|
|
entry["provenance"] = matches
|
|
else:
|
|
entry.pop("provenance", None)
|
|
return counts
|
|
|
|
def build_snapshot(
|
|
source: str, imported_at: str, dats: dict, entries: list[dict]
|
|
) -> dict:
|
|
"""A provenance snapshot, sorted for determinism."""
|
|
return {
|
|
"source": source,
|
|
"imported_at": imported_at,
|
|
"dats": dict(sorted(dats.items())),
|
|
"entries": sorted(entries, key=lambda e: (e["dat"], e["name"])),
|
|
}
|
|
|
|
|
|
def write_snapshot(path: str, snapshot: dict) -> bool:
|
|
"""Write a snapshot. Timestamps alone never count as a change."""
|
|
return write_if_changed(path, json.dumps(snapshot, indent=2) + "\n")
|
|
|
|
|
|
def write_provenance_snapshot(
|
|
path: str, source: str, imported_at: str, dats: dict, entries: list[dict]
|
|
) -> bool:
|
|
"""Write a normalized provenance snapshot."""
|
|
return write_snapshot(path, build_snapshot(source, imported_at, dats, entries))
|