From 88a3bb8953d156316b5a6cf58e842a771b1680a6 Mon Sep 17 00:00:00 2001 From: Abdessamad Derraz <3028866+Abdess@users.noreply.github.com> Date: Sun, 18 Oct 2026 05:30:10 +0200 Subject: [PATCH] fix: let auto_fetch read lists and keep held files auto_fetch stopped on any profile listing several accepted hashes, and filed a fetched file in the first folder sharing one word of its system, overwriting a same-named revision held there. Hash fields are read as sets, a folder must match the whole system slug, and another revision goes to .variants/ beside the file already held. --- scripts/auto_fetch.py | 93 +++++++++++++++++++-------------- tests/test_audit_regressions.py | 35 +++++++++++++ 2 files changed, 90 insertions(+), 38 deletions(-) diff --git a/scripts/auto_fetch.py b/scripts/auto_fetch.py index beee9eb4..f338228a 100644 --- a/scripts/auto_fetch.py +++ b/scripts/auto_fetch.py @@ -23,6 +23,7 @@ import os import subprocess import sys import urllib.error +import urllib.parse import urllib.request from pathlib import Path @@ -115,14 +116,26 @@ def find_missing(config: dict, db: dict) -> list[dict]: return missing +def _accepted(entry: dict, field: str) -> set[str]: + """Every value an entry accepts for a hash field. + + A profile lists several revisions as a YAML list (daedalusx64's + n64.psh), a platform as a comma-separated string; read as one value, + the list stopped the run with an unhashable-type error. + """ + 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 verify_content(data: bytes, expected: dict) -> bool: - """Verify downloaded content matches expected hashes.""" - if expected.get("sha1"): - actual = hashlib.sha1(data).hexdigest() - return actual == expected["sha1"] - if expected.get("md5"): - actual = hashlib.md5(data).hexdigest() - return actual == expected["md5"] + """Verify downloaded content matches one of the expected hashes.""" + sha1s = _accepted(expected, "sha1") + if sha1s: + return hashlib.sha1(data).hexdigest() in sha1s + md5s = _accepted(expected, "md5") + if md5s: + return hashlib.md5(data).hexdigest() in md5s return False @@ -138,17 +151,15 @@ def _read_limited( def step1_crossref_db(entry: dict, db: dict) -> str | None: """Check if file exists under different name/path in database.""" - sha1 = entry.get("sha1") - md5 = entry.get("md5") - - if sha1 and sha1 in db.get("files", {}): - return db["files"][sha1]["path"] - - if md5: - sha1_match = db.get("indexes", {}).get("by_md5", {}).get(md5) - if sha1_match and sha1_match in db["files"]: - return db["files"][sha1_match]["path"] - + files = db.get("files", {}) + for sha1 in sorted(_accepted(entry, "sha1")): + if sha1 in files: + return files[sha1]["path"] + by_md5 = db.get("indexes", {}).get("by_md5", {}) + for md5 in sorted(_accepted(entry, "md5")): + sha1_match = by_md5.get(md5) + if sha1_match and sha1_match in files: + return files[sha1_match]["path"] return None @@ -247,12 +258,15 @@ def step4_search_archive_org(entry: dict) -> bytes | None: except (urllib.error.URLError, urllib.error.HTTPError): continue - sha1 = entry.get("sha1", "") - if not sha1: + sha1s = sorted(_accepted(entry, "sha1")) + if not sha1s: return None + # Each accepted revision is a query of its own; archive.org matches one. + sha1 = " OR ".join(f"sha1:{value}" for value in sha1s) search_url = ( - f"https://archive.org/advancedsearch.php?q=sha1:{sha1}&output=json&rows=1" + "https://archive.org/advancedsearch.php?q=" + f"{urllib.parse.quote(sha1)}&output=json&rows=1" ) try: @@ -289,21 +303,24 @@ def place_file(data: bytes, entry: dict, bios_dir: str, db: dict) -> str: dest_dir = Path(bios_dir) - for manufacturer_dir in dest_dir.iterdir(): - if not manufacturer_dir.is_dir(): - continue - for console_dir in manufacturer_dir.iterdir(): - if not console_dir.is_dir(): - continue - dir_path = f"{manufacturer_dir.name}/{console_dir.name}".lower() - if any(part in dir_path for part in system.split("-") if len(part) > 2): - dest = console_dir / name - dest.parent.mkdir(parents=True, exist_ok=True) - with open(dest, "wb") as f: - f.write(data) - return str(dest) - - dest = dest_dir / "Unknown" / system / name + # Every part of the slug, not any: matching "nintendo" alone filed the + # N64's n64.psh under the first Nintendo folder walked, the SNES one. + parts = [part for part in system.lower().split("-") if len(part) > 1] + matches = sorted( + (len(f"{manufacturer_dir.name}/{console_dir.name}"), str(console_dir), console_dir) + for manufacturer_dir in dest_dir.iterdir() if manufacturer_dir.is_dir() + for console_dir in manufacturer_dir.iterdir() if console_dir.is_dir() + and all( + part in f"{manufacturer_dir.name}/{console_dir.name}".lower() + for part in parts + ) + ) + folder = matches[0][2] if matches else dest_dir / "Unknown" / system + dest = folder / name + if dest.exists() and dest.read_bytes() != data: + # Another revision under the same name: the collection keeps it + # beside, never in place of, the file already there. + dest = folder / ".variants" / f"{name}.{hashlib.sha1(data).hexdigest()[:8]}" dest.parent.mkdir(parents=True, exist_ok=True) with open(dest, "wb") as f: f.write(data) @@ -377,8 +394,8 @@ def generate_issue_body(missing: list[dict], platform: str) -> str: ] for entry in missing: - sha1 = entry.get("sha1") or "N/A" - md5 = entry.get("md5") or "N/A" + sha1 = ", ".join(sorted(_accepted(entry, "sha1"))) or "N/A" + md5 = ", ".join(sorted(_accepted(entry, "md5"))) or "N/A" lines.append( f"| `{entry['name']}` | {entry['system']} | `{sha1[:12]}...` | `{md5[:12]}...` |" ) diff --git a/tests/test_audit_regressions.py b/tests/test_audit_regressions.py index f96059e4..602e4108 100644 --- a/tests/test_audit_regressions.py +++ b/tests/test_audit_regressions.py @@ -602,6 +602,41 @@ class PackReadmesSayWhatTheInstallerDoes(unittest.TestCase): self.assertIn("install.sh | sh", guide) +class AutoFetchReadsListsAndPlacesWithoutClobbering(unittest.TestCase): + """A profile hash list (daedalusx64's n64.psh) stopped auto_fetch with + an unhashable-type error, and a fetched file went to the first folder + sharing one word of the system slug (nintendo-64 into Nintendo/SNES), + overwriting a same-named file there.""" + + def test_a_hash_list_is_every_accepted_revision(self): + import hashlib + + from scripts import auto_fetch + + data = b"second revision" + entry = {"sha1": ["0" * 40, hashlib.sha1(data).hexdigest()]} + self.assertTrue(auto_fetch.verify_content(data, entry)) + db = {"files": {hashlib.sha1(data).hexdigest(): {"path": "bios/x/n64.psh"}}, + "indexes": {"by_md5": {}}} + self.assertEqual(auto_fetch.step1_crossref_db(entry, db), "bios/x/n64.psh") + + def test_the_whole_slug_picks_the_folder_and_nothing_is_overwritten(self): + from scripts import auto_fetch + + with tempfile.TemporaryDirectory(dir=TMP_ROOT) as directory: + bios = Path(directory) + for folder in ("Nintendo/SNES", "Nintendo/N64", "Nintendo/Nintendo 64DD"): + (bios / folder).mkdir(parents=True) + held = bios / "Nintendo" / "N64" / "n64.psh" + held.write_bytes(b"other revision") + placed = Path(auto_fetch.place_file( + b"fetched", {"name": "n64.psh", "system": "nintendo-64"}, str(bios), {} + )) + self.assertEqual(placed.parent.parent, bios / "Nintendo" / "N64") + self.assertEqual(placed.parent.name, ".variants") + self.assertEqual(held.read_bytes(), b"other revision") + + class FaqRegressions(unittest.TestCase): """The FAQ states facts the code owns; these tie it back to the source.