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.
This commit is contained in:
Abdessamad Derraz committed 2026-10-18 05:30:10 +02:00
1 parent cc7f9fbdac
commit 88a3bb8953
2 files changed
+90 -38

No files matched your search

+55 -38
View File
@@ -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]}...` |"
)
+35
View File
@@ -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.