refactor: move the collection model to its module

.github/workflows/deploy-site.yml

@@ -47,6 +47,7 @@ on:
       - "scripts/safeparse.py"
       - "scripts/sect233r1.py"
       - "scripts/sitebootstrap.py"
+      - "scripts/sitecollection.py"
       - "scripts/sitediagrams.py"
       - "scripts/siteentries.py"
       - "scripts/siteguides.py"

scripts/fetch_site_images.py

@@ -172,7 +172,8 @@ def fetch_logos() -> None:
 
 
 def fetch_icons() -> None:
-    from generate_site import _group_by_manufacturer, load_database
+    from common import load_database
+    from sitecollection import _group_by_manufacturer
     from siterender import SYSTEM_ICON_BASE, _icon_name
 
     db = load_database(str(ROOT / "database.json"))

scripts/generate_site.py

@@ -39,6 +39,10 @@ from pathlib import Path
 from typing import TypeVar
 
 sys.path.insert(0, os.path.dirname(__file__))
+from sitecollection import (  # noqa: E402
+    _copy_places, _group_by_manufacturer, _profiles_in,
+    _profiles_on_disk, _slug, collection_paths,
+)
 from common import (
     BUCKET_FOLDERS,
     names_manufacturer,
@@ -70,10 +74,6 @@ from cross_reference import FileConsumers, entry_target, held_copy
 from nativemode import check_label, digest_algorithm, reads_file_contents
 
 
-
-
-
-
 yaml = require_yaml()
 from generate_readme import (
     compute_coverage,
@@ -138,8 +138,6 @@ GENERATED_DIRS = ["platforms", "systems", "emulators", "wiki", "api", "downloads
 PUBLIC_DIRS = ("api", "downloads")
 
 
-
-
 def _content_check_ceiling(profiles: dict) -> str:
     """How far content checking can reach across every profiled entry."""
     hashed = sized = neither = 0
@@ -169,9 +167,6 @@ def _content_check_ceiling(profiles: dict) -> str:
     )
 
 
-
-
-
 CLS_LABELS = {
     "official_port": "Official ports",
     "community_fork": "Community forks",
@@ -201,33 +196,6 @@ CLS_LABEL_ONE = {
 }
 
 
-def _copy_places(db: dict) -> dict[str, set[tuple[str, str, str]]]:
-    """sha1 -> every (manufacturer, console, file name) holding a copy.
-
-    The database keeps one path per content, deduplication having left one
-    copy. by_path_suffix still records the other folders a copy is filed in,
-    under the name that folder gives it.
-    """
-    files = db.get("files", {})
-    bios_dir = Path(str(db.get("bios_dir") or "bios"))
-    places: dict[str, set[tuple[str, str, str]]] = {}
-    for sha1, entry in files.items():
-        parts = str(entry.get("path", "")).split("/")
-        if len(parts) >= 3 and parts[0] == "bios":
-            places.setdefault(sha1, set()).add((parts[1], parts[2], parts[-1]))
-    for tail, sha1s in db.get("indexes", {}).get("by_path_suffix", {}).items():
-        parts = tail.split("/")
-        if len(parts) < 3 or parts[1].startswith(".") or len(sha1s) != 1:
-            continue
-        if sha1s[0] in files and (bios_dir / tail).is_file():
-            places.setdefault(sha1s[0], set()).add((parts[0], parts[1], parts[-1]))
-    return places
-
-
-def _slug(manufacturer: str) -> str:
-    return manufacturer.lower().replace(" ", "-")
-
-
 def _copies_by_path(db: dict) -> dict[str, set[tuple[str, str, str]]]:
     """The copy places of each content, keyed by the path the database keeps."""
     files = db.get("files", {})
@@ -413,11 +381,6 @@ def _system_pages_from_data(
     return pages
 
 
-
-
-
-
-
 def _timestamp() -> str:
     """The build time for data files (stats, API records), ISO 8601."""
     return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
@@ -436,8 +399,6 @@ def _date_line() -> str:
     return f'<div class="rb-timestamp">{data_as_of()}.</div>'
 
 
-
-
 # Home page
 
 
@@ -5271,122 +5232,6 @@ def write_lookup_shards(
     return len(shards)
 
 
-def collection_paths(db: dict) -> dict[str, set[str]]:
-    """Every path under bios/ that holds each file, by sha1.
-
-    The database keeps one path per content. The others are the full tails
-    by_path_suffix records that name a file on disk: RPG Maker, engine
-    trees and the copies filed under several consoles keep 2,502 of them.
-    """
-    files = db.get("files", {})
-    bios_dir = Path(str(db.get("bios_dir") or "bios"))
-    paths: dict[str, set[str]] = {}
-    for sha1, entry in files.items():
-        path = str(entry.get("path", ""))
-        if path.startswith("bios/") and path.count("/") >= 2:
-            paths.setdefault(sha1, set()).add(path)
-    for tail, sha1s in db.get("indexes", {}).get("by_path_suffix", {}).items():
-        if tail.count("/") < 2 or len(sha1s) != 1 or sha1s[0] not in files:
-            continue
-        if (bios_dir / tail).is_file():
-            paths.setdefault(sha1s[0], set()).add(f"bios/{tail}")
-    return paths
-
-
-def _group_by_manufacturer(db: dict) -> dict[str, dict[str, list]]:
-    """Group files by manufacturer -> console -> files.
-
-    The database keeps one path per content, and a copy filed elsewhere was
-    on no page of its own: the Channel F BIOS sat under Arcade > FBNeo only,
-    and Fairchild had no system page. Every folder holding a copy lists it,
-    read from the full paths by_path_suffix records. The game engines'
-    files are grouped by engine (_engine_sections).
-    """
-    files = db.get("files", {})
-    manufacturers: dict[str, dict[str, list]] = {}
-    for sha1, where in _copy_places(db).items():
-        for manufacturer, console in sorted({(m, c) for m, c, _ in where}):
-            manufacturers.setdefault(manufacturer, {}).setdefault(console, []).append(files[sha1])
-    return _engine_sections(manufacturers)
-
-
-def _engine_sections(manufacturers: dict[str, dict[str, list]]) -> dict[str, dict[str, list]]:
-    """A game engine's files in one section headed with the engine's name.
-
-    The Systems index listed ".variants", "scummvm.zip" and "wildmidi.cfg"
-    as games, and ScummVM's files sat on three pages. A folder of variants
-    joins the section of its primary, a file kept at a top's root joins
-    the section of the one engine that reads it (else of the engine the
-    top is named after), the folder under Other named after a game-engine
-    top joins that top, and a folder named by a profile key is headed with
-    the engine's name (SECTION_LABELS), its address unchanged.
-    """
-    from siterender import _PROGRAM_TOPS, SECTION_LABELS, record_card_move  # noqa: PLC0415
-
-    profiles = {key: profile for key, profile in _profiles_on_disk().items()
-                if profile.get("type") not in ("alias", "test")}
-    readers: dict[str, set[str]] = {}
-    for key, profile in profiles.items():
-        for entry in profile.get("files") or []:
-            if isinstance(entry, dict):
-                for name in [entry.get("name"), *(entry.get("aliases") or [])]:
-                    if name:
-                        readers.setdefault(str(name).rsplit("/", 1)[-1], set()).add(key)
-
-    def engine(key: str) -> str:
-        return str(profiles[key].get("emulator") or key)
-
-    def profile_of(name: str) -> str | None:
-        return next((k for k in sorted(profiles)
-                     if name.casefold() in (k.casefold(), engine(k).casefold())), None)
-
-    out: dict[str, dict[str, list]] = {}
-    for mfr, consoles in manufacturers.items():
-        if mfr not in GAME_DATA_TOPS:
-            out[mfr] = dict(consoles)
-            continue
-        own = profile_of(mfr)
-        regrouped: dict[str, list] = {}
-        for console, files in sorted(consoles.items(), key=lambda item: item[0].startswith(".")):
-            for f in files:
-                parts = str(f.get("path", "")).split("/")
-                target = console
-                if console.startswith(".") or (len(parts) == 3 and console == parts[-1]):
-                    # A variant answers its primary's name: scummvm.zip.a93f1c4b.
-                    base = parts[-1].rsplit(".", 1)[0] if console.startswith(".") else parts[-1]
-                    keys = sorted(readers.get(base, ()))
-                    key = keys[0] if len(keys) == 1 else own
-                    found = next((c for c, held in regrouped.items()
-                                  if any(h.get("name") == base for h in held)), None)
-                    target = found or (next((c for c in consoles if c == key), None)
-                                       or (engine(key) if key else base))
-                if target != console:
-                    record_card_move(mfr, console, str(f.get("sha1", "")), mfr, target)
-                regrouped.setdefault(target, []).append(f)
-        out[mfr] = regrouped
-    # The folder under Other an engine top is named after: ScummVM's
-    # engine data under Other/ScummVM, its themes under ScummVM/.
-    others = out.get("Other", {})
-    for folder in sorted(others):
-        top = next((t for t in out if t in GAME_DATA_TOPS and t.casefold() == folder.casefold()),
-                   None)
-        if top is None or top in _PROGRAM_TOPS:
-            continue
-        sections = out[top]
-        target = next((c for c in sections if c.casefold() == folder.casefold()), folder)
-        held = {str(f.get("sha1", "")) for f in sections.get(target, [])}
-        for f in others[folder]:
-            record_card_move("Other", folder, str(f.get("sha1", "")), top, target)
-        sections[target] = [*sections.get(target, []),
-                            *(f for f in others.pop(folder) if str(f.get("sha1", "")) not in held)]
-    for mfr in GAME_DATA_TOPS:
-        for console in out.get(mfr, {}):
-            key = profiles.get(console) and console
-            if key and engine(key) != console:
-                SECTION_LABELS[(mfr, console)] = engine(key)
-    return {mfr: consoles for mfr, consoles in out.items() if consoles}
-
-
 def _console_display(manufacturer: str, console: str) -> str:
     """A console section's name as the reader looks for it: the heading
     system_name gives its id, or the folder alone under a program's top
@@ -6701,15 +6546,6 @@ def _reading_pairs(sha1: str, consumers: FileConsumers) -> list[tuple[str, dict]
     return pairs
 
 
-def _profiles_on_disk() -> dict[str, dict]:
-    return _profiles_in(str(siteinputs.current().emulators), skip_aliases=False)
-
-
-@functools.lru_cache(maxsize=None)
-def _profiles_in(emulators_dir: str, skip_aliases: bool) -> dict[str, dict]:
-    return load_emulator_profiles(emulators_dir, skip_aliases=skip_aliases)
-
-
 def _entry_systems(emulator: str, entry: dict) -> list[str]:
     """The systems an entry is read for: its own `system:` or `systems:`
     (a MAME device set several systems load), else its profile's only one."""
@@ -11361,8 +11197,6 @@ def generate_gap_analysis(
     return "\n".join(lines) + "\n"
 
 
-
-
 def _core_scope_note(config: dict, profiles: dict) -> str:
     """" (all libretro)" when the platform resolves every libretro core.
 

scripts/make_og_cards.py

@@ -70,7 +70,7 @@ def system_cards() -> dict[str, tuple[str, str]]:
     manufacturer and listing its consoles. Read from the same grouping the
     system pages are written from."""
     from common import GAME_DATA_TOPS, load_database
-    from generate_site import _group_by_manufacturer, _slug
+    from sitecollection import _group_by_manufacturer, _slug
 
     def lead(manufacturer: str) -> str:
         # Other holds one folder per emulator, a game publisher's page its

scripts/sitecollection.py

@@ -0,0 +1,169 @@
+"""The collection as the site files it: every folder holding each file.
+
+The database keeps one path per content. The folders a copy is also filed
+in, the system sections and the game-engine sections are read here, by the
+page generator and by the tools that draw or fetch per-system images. The
+bios folder comes from siteinputs, never the working directory.
+"""
+
+from __future__ import annotations
+
+import functools
+import os
+import sys
+
+sys.path.insert(0, os.path.dirname(__file__))
+import siteinputs  # noqa: E402
+from common import GAME_DATA_TOPS, load_emulator_profiles  # noqa: E402
+
+
+def _copy_places(db: dict) -> dict[str, set[tuple[str, str, str]]]:
+    """sha1 -> every (manufacturer, console, file name) holding a copy.
+
+    The database keeps one path per content, deduplication having left one
+    copy. by_path_suffix still records the other folders a copy is filed in,
+    under the name that folder gives it.
+    """
+    files = db.get("files", {})
+    bios_dir = siteinputs.bios_dir(db)
+    places: dict[str, set[tuple[str, str, str]]] = {}
+    for sha1, entry in files.items():
+        parts = str(entry.get("path", "")).split("/")
+        if len(parts) >= 3 and parts[0] == "bios":
+            places.setdefault(sha1, set()).add((parts[1], parts[2], parts[-1]))
+    for tail, sha1s in db.get("indexes", {}).get("by_path_suffix", {}).items():
+        parts = tail.split("/")
+        if len(parts) < 3 or parts[1].startswith(".") or len(sha1s) != 1:
+            continue
+        if sha1s[0] in files and (bios_dir / tail).is_file():
+            places.setdefault(sha1s[0], set()).add((parts[0], parts[1], parts[-1]))
+    return places
+
+
+def _slug(manufacturer: str) -> str:
+    return manufacturer.lower().replace(" ", "-")
+
+
+def collection_paths(db: dict) -> dict[str, set[str]]:
+    """Every path under bios/ that holds each file, by sha1.
+
+    The database keeps one path per content. The others are the full tails
+    by_path_suffix records that name a file on disk: RPG Maker, engine
+    trees and the copies filed under several consoles keep 2,502 of them.
+    """
+    files = db.get("files", {})
+    bios_dir = siteinputs.bios_dir(db)
+    paths: dict[str, set[str]] = {}
+    for sha1, entry in files.items():
+        path = str(entry.get("path", ""))
+        if path.startswith("bios/") and path.count("/") >= 2:
+            paths.setdefault(sha1, set()).add(path)
+    for tail, sha1s in db.get("indexes", {}).get("by_path_suffix", {}).items():
+        if tail.count("/") < 2 or len(sha1s) != 1 or sha1s[0] not in files:
+            continue
+        if (bios_dir / tail).is_file():
+            paths.setdefault(sha1s[0], set()).add(f"bios/{tail}")
+    return paths
+
+
+def _group_by_manufacturer(db: dict) -> dict[str, dict[str, list]]:
+    """Group files by manufacturer -> console -> files.
+
+    The database keeps one path per content, and a copy filed elsewhere was
+    on no page of its own: the Channel F BIOS sat under Arcade > FBNeo only,
+    and Fairchild had no system page. Every folder holding a copy lists it,
+    read from the full paths by_path_suffix records. The game engines'
+    files are grouped by engine (_engine_sections).
+    """
+    files = db.get("files", {})
+    manufacturers: dict[str, dict[str, list]] = {}
+    for sha1, where in _copy_places(db).items():
+        for manufacturer, console in sorted({(m, c) for m, c, _ in where}):
+            manufacturers.setdefault(manufacturer, {}).setdefault(console, []).append(files[sha1])
+    return _engine_sections(manufacturers)
+
+
+def _engine_sections(manufacturers: dict[str, dict[str, list]]) -> dict[str, dict[str, list]]:
+    """A game engine's files in one section headed with the engine's name.
+
+    The Systems index listed ".variants", "scummvm.zip" and "wildmidi.cfg"
+    as games, and ScummVM's files sat on three pages. A folder of variants
+    joins the section of its primary, a file kept at a top's root joins
+    the section of the one engine that reads it (else of the engine the
+    top is named after), the folder under Other named after a game-engine
+    top joins that top, and a folder named by a profile key is headed with
+    the engine's name (SECTION_LABELS), its address unchanged.
+    """
+    from siterender import _PROGRAM_TOPS, SECTION_LABELS, record_card_move  # noqa: PLC0415
+
+    profiles = {key: profile for key, profile in _profiles_on_disk().items()
+                if profile.get("type") not in ("alias", "test")}
+    readers: dict[str, set[str]] = {}
+    for key, profile in profiles.items():
+        for entry in profile.get("files") or []:
+            if isinstance(entry, dict):
+                for name in [entry.get("name"), *(entry.get("aliases") or [])]:
+                    if name:
+                        readers.setdefault(str(name).rsplit("/", 1)[-1], set()).add(key)
+
+    def engine(key: str) -> str:
+        return str(profiles[key].get("emulator") or key)
+
+    def profile_of(name: str) -> str | None:
+        return next((k for k in sorted(profiles)
+                     if name.casefold() in (k.casefold(), engine(k).casefold())), None)
+
+    out: dict[str, dict[str, list]] = {}
+    for mfr, consoles in manufacturers.items():
+        if mfr not in GAME_DATA_TOPS:
+            out[mfr] = dict(consoles)
+            continue
+        own = profile_of(mfr)
+        regrouped: dict[str, list] = {}
+        for console, files in sorted(consoles.items(), key=lambda item: item[0].startswith(".")):
+            for f in files:
+                parts = str(f.get("path", "")).split("/")
+                target = console
+                if console.startswith(".") or (len(parts) == 3 and console == parts[-1]):
+                    # A variant answers its primary's name: scummvm.zip.a93f1c4b.
+                    base = parts[-1].rsplit(".", 1)[0] if console.startswith(".") else parts[-1]
+                    keys = sorted(readers.get(base, ()))
+                    key = keys[0] if len(keys) == 1 else own
+                    found = next((c for c, held in regrouped.items()
+                                  if any(h.get("name") == base for h in held)), None)
+                    target = found or (next((c for c in consoles if c == key), None)
+                                       or (engine(key) if key else base))
+                if target != console:
+                    record_card_move(mfr, console, str(f.get("sha1", "")), mfr, target)
+                regrouped.setdefault(target, []).append(f)
+        out[mfr] = regrouped
+    # The folder under Other an engine top is named after: ScummVM's
+    # engine data under Other/ScummVM, its themes under ScummVM/.
+    others = out.get("Other", {})
+    for folder in sorted(others):
+        top = next((t for t in out if t in GAME_DATA_TOPS and t.casefold() == folder.casefold()),
+                   None)
+        if top is None or top in _PROGRAM_TOPS:
+            continue
+        sections = out[top]
+        target = next((c for c in sections if c.casefold() == folder.casefold()), folder)
+        held = {str(f.get("sha1", "")) for f in sections.get(target, [])}
+        for f in others[folder]:
+            record_card_move("Other", folder, str(f.get("sha1", "")), top, target)
+        sections[target] = [*sections.get(target, []),
+                            *(f for f in others.pop(folder) if str(f.get("sha1", "")) not in held)]
+    for mfr in GAME_DATA_TOPS:
+        for console in out.get(mfr, {}):
+            key = profiles.get(console) and console
+            if key and engine(key) != console:
+                SECTION_LABELS[(mfr, console)] = engine(key)
+    return {mfr: consoles for mfr, consoles in out.items() if consoles}
+
+
+def _profiles_on_disk() -> dict[str, dict]:
+    return _profiles_in(str(siteinputs.current().emulators), skip_aliases=False)
+
+
+@functools.lru_cache(maxsize=None)
+def _profiles_in(emulators_dir: str, skip_aliases: bool) -> dict[str, dict]:
+    return load_emulator_profiles(emulators_dir, skip_aliases=skip_aliases)

scripts/siteinputs.py

@@ -30,3 +30,14 @@ def current() -> SiteInputs:
 
 def use(inputs: SiteInputs) -> None:
     _current[0] = inputs
+
+
+def bios_dir(db: dict) -> Path:
+    """The database's bios folder, as the inputs place it.
+
+    database.json names it relative to its own folder ("bios"). Read from the
+    working directory instead, a tool started elsewhere found no copy on disk
+    and dropped five system sections and 2,500 collection paths in silence.
+    """
+    named = Path(str(db.get("bios_dir") or "bios"))
+    return named if named.is_absolute() else current().db.parent / named

scripts/siterender.py

@@ -780,7 +780,7 @@ def _ensure_console_sections() -> None:
     import json
 
     from common import load_emulator_profiles
-    from generate_site import _group_by_manufacturer
+    from sitecollection import _group_by_manufacturer
 
     inputs = siteinputs.current()
     with open(inputs.db, encoding="utf-8") as handle:

tests/test_site_inputs.py

@@ -85,11 +85,48 @@ class TheGivenProfilesAndPlatforms(_Inputs):
         self.assertEqual(sitediagrams.registry()["retroarch"].get("release_pack"), False)
 
 
+class TheCollectionDoesNotDependOnTheWorkingDirectory(unittest.TestCase):
+    """database.json names its bios folder "bios", and the collection model
+    read it from the working directory: started from tmp/, it found no copy
+    on disk and dropped five system sections and 2,500 collection paths."""
+
+    def test_same_sections_and_paths_from_another_directory(self) -> None:
+        import json
+
+        import sitecollection
+
+        db = json.loads((REPO_ROOT / "database.json").read_text(encoding="utf-8"))
+        here = sitecollection.collection_paths(db)
+        places = sitecollection._copy_places(db)
+        elsewhere = Path(tempfile.mkdtemp(dir=REPO_ROOT / "tmp" if (REPO_ROOT / "tmp").is_dir() else None))
+        self.addCleanup(shutil.rmtree, elsewhere)
+        cwd = os.getcwd()
+        os.chdir(elsewhere)
+        self.addCleanup(os.chdir, cwd)
+        self.assertEqual(sitecollection.collection_paths(db), here)
+        self.assertEqual(sitecollection._copy_places(db), places)
+        self.assertGreater(sum(map(len, here.values())), len(here))
+
+    def test_the_bios_folder_sits_beside_the_given_database(self) -> None:
+        import siteinputs
+
+        siteinputs.use(siteinputs.SiteInputs(db=Path("/elsewhere/database.json")))
+        self.addCleanup(siteinputs.use, siteinputs.SiteInputs())
+        self.assertEqual(siteinputs.bios_dir({"bios_dir": "bios"}), Path("/elsewhere/bios"))
+        self.assertEqual(siteinputs.bios_dir({"bios_dir": "/abs/bios"}), Path("/abs/bios"))
+
+    def test_the_image_tools_do_not_load_the_page_generator(self) -> None:
+        for module in ("make_og_cards.py", "fetch_site_images.py", "siterender.py"):
+            with self.subTest(module=module):
+                source = (REPO_ROOT / "scripts" / module).read_text(encoding="utf-8")
+                self.assertNotIn("from generate_site import", source)
+
+
 class NoPageModuleReadsTheRepositoryInputs(unittest.TestCase):
     """A helper that opens the repository's database, profiles or platform
     files itself bypasses the flags. siteinputs.current() is the one source."""
 
-    MODULES = ("generate_site.py", "siterender.py", "sitediagrams.py")
+    MODULES = ("generate_site.py", "siterender.py", "sitediagrams.py", "sitecollection.py")
     INPUTS = {"database.json", "emulators", "platforms"}
 
     def test_no_literal_input_path(self) -> None: