feat: round 4 reference and data

This commit is contained in:
Abdessamad Derraz committed 2026-10-23 19:23:58 +02:00
commit 8f50015a0c
15 files changed
+10306 -100

No files matched your search

+19 -18
View File
@@ -14,49 +14,50 @@ from pathlib import Path
API = "https://abdess.github.io/retrobios/api/v1/platforms/"
# The digest each check mode compares. A mode missing here (existence) only
# asks that the file be present.
# The digest each mode compares. Without one (existence) a file need only be present.
DIGESTS = {"md5": "md5", "sha1": "sha1"}
def accepts(entry: dict, path: Path, mode: str) -> bool:
"""Whether the platform's own check passes the file at path."""
digest = DIGESTS.get(mode)
raw = entry.get(mode) or ""
values = raw if isinstance(raw, list) else str(raw).split(",")
declared = [value.strip().lower() for value in values if value.strip()]
if digest is None or not declared:
digest, raw = DIGESTS.get(mode), entry.get(mode) or ""
values = [v.strip().lower() for v in (raw if isinstance(raw, list) else str(raw).split(",")) if v.strip()]
if digest is None or not values:
return path.is_file() # nothing to compare: the platform checks presence
try:
if entry.get("zipped_file"): # the hash of one member, its name casefolded
with zipfile.ZipFile(path) as archive:
names = {name.casefold(): name for name in archive.namelist()}
member = names.get(entry["zipped_file"].casefold())
return bool(member) and hashlib.new(digest, archive.read(member)).hexdigest() in declared
return bool(member) and hashlib.new(digest, archive.read(member)).hexdigest() in values
actual = hashlib.new(digest, path.read_bytes()).hexdigest()
if entry.get("zip_digest") != "composite" and any(actual.startswith(d) for d in declared):
if entry.get("zip_digest") != "composite" and any(actual.startswith(v) for v in values):
return True # a truncated md5 (Batocera) compares by its prefix
if digest == DIGESTS["md5"] and path.suffix.lower() == ".zip": # Recalbox: members in name order
with zipfile.ZipFile(path) as archive:
names = sorted(name for name in archive.namelist() if not name.endswith("/"))
return hashlib.md5(b"".join(map(archive.read, names))).hexdigest() in declared
return hashlib.md5(b"".join(map(archive.read, names))).hexdigest() in values
except (OSError, zipfile.BadZipFile):
return False
return False
def check(platform: str, root: Path, api: str = API) -> list[tuple[str, str]]:
"""(destination, ok | wrong | missing) for each entry of the list."""
"""(destination, ok | wrong | missing) for each entry. Entries with no destination
(EmuDeck) name no file: their system passes when any file in root has their hash."""
with urllib.request.urlopen(f"{api}{platform}.json") as response:
contract = json.load(response)["items"][0]["contract"]
mode = contract["verification_mode"]
results = []
for system in contract["systems"].values():
for entry in system.get("files") or []:
destination = entry.get("destination") or entry["name"]
path = root / destination
mode, results = contract["verification_mode"], []
loose = [path for path in root.iterdir() if path.is_file()] if root.is_dir() else []
for system_id, system in contract["systems"].items():
listed = [e for e in system.get("files") or [] if not e.get("destination")]
for entry in (e for e in system.get("files") or [] if e.get("destination")):
path = root / entry["destination"]
verdict = "ok" if accepts(entry, path, mode) else "wrong" if path.is_file() else "missing"
results.append((destination, verdict))
results.append((entry["destination"], verdict))
if listed:
held = any(accepts(e, path, mode) for e in listed for path in loose)
results.append((f"{system_id}: any file", "ok" if held else "missing"))
return results
+214 -23
View File
@@ -1758,8 +1758,11 @@ def system_index_items(db: dict, coverages: dict, profiles: dict) -> list[dict]:
api/v1/systems/ is keyed by manufacturer and its consoles by display
name: a frontend keying by system id (sony-playstation) had no way in.
The url is the section system_section gives, the one every page links.
An id with no section of its own (tyrian, scummvm) that holds files
takes the card of its first file, else the page of the emulator naming
it: 28 such ids sent their hashes to no page.
"""
from siterender import system_ids, system_section
from siterender import card_page, system_ids, system_section
sha1_by_path = _sha1_by_path(db)
files = db.get("files") or {}
@@ -1795,16 +1798,33 @@ def system_index_items(db: dict, coverages: dict, profiles: dict) -> list[dict]:
name = system_name(sys_id)
except KeyError:
name = sys_id
items.append({
"id": sys_id,
"name": name,
"manufacturer": manufacturers.get(section[0], "") if section else "",
"url": _site_path(console_href(*section)) if section else None,
"sha1": sorted(held.get(sys_id, set()) & files.keys()),
})
sha1s = sorted(held.get(sys_id, set()) & files.keys())
manufacturer = manufacturers.get(section[0], "") if section else ""
url = _site_path(console_href(*section)) if section else None
if url is None and sha1s:
manufacturer, url = _first_file_place(sys_id, sha1s, files, profiles, card_page)
items.append({"id": sys_id, "name": name, "manufacturer": manufacturer,
"url": url, "sha1": sha1s})
return items
def _first_file_place(
sys_id: str, sha1s: list[str], files: dict, profiles: dict, card_page,
) -> tuple[str, str | None]:
"""(manufacturer, url) of the card of an id's first collection file,
else of the page of the first emulator naming the id."""
for sha1 in sha1s:
parts = str(files[sha1].get("path", "")).split("/")
if len(parts) >= 3 and parts[0] == "bios":
page = card_page("/".join(parts), sha1)
return parts[1], _site_path(f"systems/{page}.md#sha1-{sha1[:12]}")
for key in sorted(profiles):
profile = profiles[key]
if profile.get("type") not in ("alias", "test") and sys_id in (profile.get("systems") or []):
return "", _site_path(f"emulators/{key}.md")
return "", None
def core_index_items(profiles: dict) -> list[dict]:
"""Every name a core goes by, with the profile it designates.
@@ -1834,14 +1854,28 @@ def _is_manifest(item: dict) -> bool:
return item["url"].startswith("api/v1/manifests/")
# A URL template written in prose (https://.../<ref>/<repo_path>): escaped,
# the autolinker made a link of it, &lt;ref> on screen and a 404 behind it.
_URL_TEMPLATE = re.compile(r"(?<!`)https?://[^\s`]*<[^\s`]*?(?=[.,;]?(?:\s|$))")
def _schema_field_rows(schema: dict, keys: tuple[str, ...]) -> list[str]:
"""One table row per field, its description read from the schema, so the
page and the published contract say the same thing."""
page and the published contract say the same thing. A URL template is
written as code."""
props = schema.get("properties") or {}
return [
f"| `{key}` | {_escape_tags(str(props[key]['description']))} |"
for key in keys if (props.get(key) or {}).get("description")
]
rows = []
for key in keys:
text = str((props.get(key) or {}).get("description") or "")
if text:
rows.append(f"| `{key}` | {_escape_tags(_URL_TEMPLATE.sub(r'`\g<0>`', text))} |")
return rows
def _other_fields(schema: dict, shown: tuple[str, ...]) -> list[str]:
"""The fields a short table leaves out, every one with its description."""
rest = tuple(key for key in schema.get("properties") or {} if key not in shown)
return ["| Field | Meaning |", "|-------|---------|", *_schema_field_rows(schema, rest)]
# The manifest fields an integrator reads first, in the order they matter.
@@ -1859,12 +1893,34 @@ def _short_sha256(item: dict, catalog: str) -> str:
return f'<a href="{catalog}"><code title="{value}">{value[:12]}</code></a>'
def _platforms_without_a_list(platforms_dir: Path | None = None) -> list[str]:
"""The names of the platforms whose registry says they keep no list."""
platforms_dir = platforms_dir or _REPO_ROOT / "platforms"
names = []
for key, entry in sorted(load_platform_registry(str(platforms_dir)).items()):
if entry.get("native_list") != "none":
continue
config = platforms_dir / str(entry.get("config") or f"{key}.yml")
data = yaml_load(config.read_text(encoding="utf-8")) if config.is_file() else {}
names.append(str((data or {}).get("platform") or key))
return names
def _join_names(names: list[str]) -> str:
return names[0] if len(names) == 1 else ", ".join(names[:-1]) + f" and {names[-1]}"
def generate_data_page(exports: list[dict]) -> str:
manifest_schema = json.loads(
(_REPO_ROOT / "schemas" / "install-manifest.schema.json").read_text(encoding="utf-8"))
snippet = (_REPO_ROOT / "scripts" / "check_platform_folder.py").read_text(encoding="utf-8")
manifests = [item for item in exports if _is_manifest(item)]
no_list = _platforms_without_a_list()
lines = [
"---",
"terms: manifest, schema, check a folder, rate limit, api",
"---",
"",
"# Data and API",
"",
"RetroBIOS publishes the same metadata used by its verifier, pack builder "
@@ -1907,6 +1963,10 @@ def generate_data_page(exports: list[dict]) -> str:
"URL prefix (`/api/v2/`), and fields may only be added compatibly within v1. "
"Each schema's `$id` is the address it is served at.",
"",
"`database.json`, `stats.json`, `release.json` and the manifests are not "
"envelopes. Their revision is the `commit` of `catalog.json` that lists "
"their SHA256.",
"",
"One emulator profile without the whole list: "
"`api/v1/emulators/<id>.json`, the envelope of `emulators.json` holding "
"that one item, its `profile` valid against "
@@ -1923,7 +1983,10 @@ def generate_data_page(exports: list[dict]) -> str:
"`url` of the section holding its files and their SHA1. "
"[`cores.json`](api/v1/cores.json) gives each core name the buildbot or a "
"platform uses (`mednafen_psx`) the profile it designates (`beetle_psx`), "
"whose document is `api/v1/emulators/<profile>.json`.",
"whose document is `api/v1/emulators/<profile>.json`. A name two profiles "
"claim (`pcsx2`) has one item per profile, and `preferred: true` marks the "
"one a platform list naming that core gets: the profile with that very id, "
"else the first id in alphabetical order.",
"",
"One hash without the whole list: `api/v1/lookup/<xx>.json`, where `<xx>` "
"is the first two characters of the hash in lowercase, holds every SHA1, "
@@ -1931,7 +1994,27 @@ def generate_data_page(exports: list[dict]) -> str:
"(the collection, a data directory, a platform's list or an emulator "
"profile), the file `name`, the `url` of the page that shows it and, "
"when one is known, the `system_name` the site gives its system. "
"[Find a file](files/index.md) reads it.",
"[Find or check a file](files/index.md) reads it.",
"",
"In a browser, the SHA1 of a file the reader picks finds its records. Web "
"Crypto has no MD5, and none is needed: a setup that lists a file by MD5 "
"or CRC32 is filed under the SHA1 of the collection file it names too.",
"",
"```js",
"// The records of the file a reader picked, by its SHA1.",
"async function lookUp(file) {",
" const digest = await crypto.subtle.digest(\"SHA-1\", await file.arrayBuffer());",
" const sha1 = [...new Uint8Array(digest)]",
" .map((byte) => byte.toString(16).padStart(2, \"0\"))",
" .join(\"\");",
f" const base = \"{SITE_URL}api/v1/lookup/\";",
" const response = await fetch(`${base}${sha1.slice(0, 2)}.json`);",
" if (!response.ok) return [];",
" const shard = await response.json();",
" // layer: collection, data-directory, platform or emulator. owner: which one.",
" return shard.items.filter((item) => item.value === sha1);",
"}",
"```",
"",
"Two kinds of `url`:",
"",
@@ -1965,6 +2048,15 @@ def generate_data_page(exports: list[dict]) -> str:
"|-------|---------|",
*_schema_field_rows(manifest_schema["$defs"]["file"], MANIFEST_FILE_FIELDS),
"",
*markup.details("note", "All fields", [
"The manifest:",
"",
*_other_fields(manifest_schema, MANIFEST_TOP_FIELDS),
"",
"Each entry of `files`:",
"",
*_other_fields(manifest_schema["$defs"]["file"], MANIFEST_FILE_FIELDS),
]),
"## Check a folder for one platform",
"",
"The job most tools come for: is this BIOS folder what the platform checks? "
@@ -1979,8 +2071,12 @@ def generate_data_page(exports: list[dict]) -> str:
"Recalbox writes several accepted md5 in one field, comma separated.",
"- `zip_digest: composite` (Recalbox): the md5 of a ZIP is that of its members "
"read in name order, never the archive's.",
"- An entry with no `destination` (EmuDeck) names no file: the system "
"passes when any file in the BIOS folder has one of its md5.",
"- `hash_rule: size_and_any_hash` (RomM) wants the size and any one hash. "
"The script below does not apply it.",
*([f"- {_join_names(no_list)} have no list of their own (`native_list: none`). "
"An empty list does not mean nothing is needed."] if no_list else []),
"",
"```python",
snippet.rstrip("\n"),
@@ -2057,13 +2153,17 @@ def generate_data_page(exports: list[dict]) -> str:
"- `curated`, `shared_group` and `mirror_of` mark an entry the platform's "
"own file does not carry.",
"",
"## Hosting and limits",
"## Hosting, rate limits and caching",
"",
"- Static files on GitHub Pages, read with a plain GET. No key, no account.",
"- Any origin may read them (`Access-Control-Allow-Origin: *`), so a "
"browser tool can call them directly.",
"- Responses are cached for 10 minutes (`Cache-Control: max-age=600`). "
"Send the `ETag` back in `If-None-Match` to skip an unchanged download.",
"- GitHub may answer HTTP 429 to a client that sends too many requests, and "
"a Pages site has a soft limit of 100 GB of traffic a month ([GitHub Pages "
"limits](https://docs.github.com/en/pages/getting-started-with-github-pages/"
"github-pages-limits)). Fetch the one shard or document a question needs.",
"- The base is the project's address, "
f"`{SITE_URL}api/v1/`. Compare `commit` to know whether two downloads differ.",
"",
@@ -3855,6 +3955,8 @@ def generate_platform_page(
if no_list:
lines.extend(_core_folders_section(name, config))
if _own_list_only(config):
lines.extend([f"**Cores:** {OWN_LIST_ONLY}", ""])
# System summary table (quick navigation)
lines.extend(
@@ -3885,9 +3987,12 @@ def generate_platform_page(
if len(sys_emus) > 3:
emu_str += f" [+{len(sys_emus) - 3}](#{anchor})"
# The name a reader filters on, the id the list writes beside it.
# The id and the count are no words a reader searches for: indexed,
# every platform page answered a system id before its system's page.
lines.append(
f"| [{_escape_tags(system_name(sys_id))}](#{anchor}) <small>{sys_id}</small> "
f"| {ok_count}/{total} | {status} | {emu_str} |"
f"| [{_escape_tags(system_name(sys_id))}](#{anchor}) "
f"<small data-pagefind-ignore>{sys_id}</small> "
f"| <span data-pagefind-ignore>{ok_count}/{total}</span> | {status} | {emu_str} |"
)
lines.append("")
@@ -3911,7 +4016,13 @@ def generate_files_page() -> str:
address and opens the card directly.
"""
return "\n".join([
"# Find a file",
"---",
"terms: check dump, md5 check, verify bios file, check hash",
"---",
"",
"# Find or check a file",
"",
"Check a BIOS dump: paste its hash or choose the file.",
"",
"A hash is a fingerprint of a file's contents ([Glossary](../wiki/glossary.md)). "
"Tools such as clrmamepro and RomVault print one, and so do emulator logs "
@@ -3944,7 +4055,7 @@ def generate_files_page() -> str:
"## The index behind it",
"",
"The lookup reads one file of [`api/v1/lookup/`](../data.md), named by the "
"first two characters of the hash, about 20 KB compressed. "
"first two characters of the hash: one of 256, never the whole index. "
"[`api/v1/file-locations.json`](../api/v1/file-locations.json) lists every "
"card by SHA1 in one file.",
"",
@@ -4147,11 +4258,46 @@ def file_lookup_records(
"owner": key, "name": name, "url": url,
**({"system_name": entry_label} if entry_label else {}),
})
records += _held_under_every_hash(records, files, db.get("indexes") or {})
unique = {json.dumps(r, sort_keys=True): r for r in records}
return sorted(unique.values(), key=lambda r: (
r["value"], r["layer"], r.get("owner", ""), r["name"], r["url"]))
def _collection_sha1(files: dict, indexes: dict, algorithm: str, value: str) -> str | None:
"""The collection file a declared value designates, None when none does."""
if algorithm == "sha1":
return value if value in files else None
found = indexes.get(f"by_{algorithm}", {}).get(value)
if isinstance(found, list):
found = found[0] if len(found) == 1 else None
return found if found in files else None
def _held_under_every_hash(records: list[dict], files: dict, indexes: dict) -> list[dict]:
"""A platform or emulator record whose value designates a collection
file, written under the file's other hashes too.
Batocera lists scph5501.bin by md5: pasted by its sha1, the file showed
no setup at all. Each record now carries the file's sha1, and
``declared`` names the hash its owner wrote when the record is a copy.
"""
copies: list[dict] = []
for record in records:
if record["layer"] not in ("platform", "emulator"):
continue
sha1 = _collection_sha1(files, indexes, record["algorithm"], record["value"])
if sha1 is None:
continue
record["sha1"] = sha1
for algorithm in HASH_ALGORITHMS:
for value in _hash_values(files[sha1].get(algorithm)):
if value != record["value"]:
copies.append({**record, "value": value, "algorithm": algorithm,
"declared": record["algorithm"]})
return copies
def write_lookup_shards(
api: Path, records: list[dict], generated_at: str, commit: str = "",
) -> int:
@@ -8289,6 +8435,26 @@ def _ratio_cell(done: int, total: int, alarm: int = 0) -> str:
return cell
# GitHub's upload page for a folder of bios/: a browser upload by someone
# without write access lands on a fork and opens a pull request.
UPLOAD_URL = f"{REPO_URL}/upload/main/bios"
# Rows of Files to source on one page: 601 on one page made a phone page
# 274,509 px long.
FILES_TO_SOURCE_PART = 100
def _upload_folder(profile: dict) -> str:
"""The bios/ folder a file the emulator loads is uploaded into: the
console folder of its first system that has one, else none."""
from siterender import _own_section
for sys_id in profile.get("systems") or []:
own = _own_section(str(sys_id))
if own is not None:
return "/".join(own)
return ""
def generate_gap_analysis(
profiles: dict,
coverages: dict,
@@ -8702,9 +8868,11 @@ def generate_gap_analysis(
f"[{_escape_tags(display)}](emulators/{key}.md)"
for display, key in zip(m["emulators"], m["emu_keys"])
)
folder = _upload_folder(profiles[m["emu_keys"][0]])
upload = f"{UPLOAD_URL}/{urllib.parse.quote(folder)}" if folder else UPLOAD_URL
rows.append(
f"| `{m['name']}` | {destination} | {expected} | {emulators} | {req} | "
f"{plat_badges} | "
f"| `{m['name']}` [Have it? Upload]({upload}) | {destination} | {expected} "
f"| {emulators} | {req} | {plat_badges} | "
f"{_source_ref_markdown(profiles[m['emu_keys'][0]], m['source_ref'])} |"
)
lines.extend(["### Files to source", "", counted, ""])
@@ -8716,7 +8884,8 @@ def generate_gap_analysis(
f"{counted} Each is a file an emulator loads that no platform lists and "
"the collection does not hold. Part of "
"[What the packs do not hold](gaps.md#files-to-source).",
head, [[row] for row in rows], moved,
head, [rows[i:i + FILES_TO_SOURCE_PART] for i in range(
0, len(rows), FILES_TO_SOURCE_PART)], moved, budget=1,
parent=("What the packs do not hold", "/gaps/"),
)
lines.extend([f"Listed on {where}.", ""])
@@ -8893,6 +9062,16 @@ def _xref_platform_rows(config: dict, unique: dict) -> tuple[int, list[str]]:
return len(matched), rows
# What a platform that declares no core (cores: []) is told, on its page
# and in the cross-reference.
OWN_LIST_ONLY = "None. Its own list is the whole truth: no core profile adds a file."
def _own_list_only(config: dict) -> bool:
"""The platform declares `cores: []`: its own list decides alone."""
return isinstance(config.get("cores"), list) and not config["cores"]
def _render_xref_by_platform(
coverages: dict, unique: dict, moved: dict[str, str] | None = None,
) -> list[str]:
@@ -8913,6 +9092,12 @@ def _render_xref_by_platform(
config = cov["config"]
count, rows = _xref_platform_rows(config, unique)
stem = f"cross-reference-{pname}"
if _own_list_only(config):
# cores: [] (MiSTer, BizHawk): no profile describes what they
# read, and a page of cores would be an empty table.
lines.append(f"| [{_escape_tags(display)}](platforms/{pname}.md) | 0 "
f"| {OWN_LIST_ONLY} |")
continue
lines.append(
f"| [{_escape_tags(display)}](platforms/{pname}.md) | {count}"
f"{_core_scope_note(config, unique)} | [Cores on {_escape_tags(display)}]({stem}.md) |"
@@ -10034,6 +10219,12 @@ def _generate(args: argparse.Namespace) -> None:
)
for stem, page in xref_pages.items():
write_if_changed(str(docs / f"{stem}.md"), tables_out_of_search(page))
for pname, cov in coverages.items():
if _own_list_only(cov["config"]):
# The empty table it had is gone: its address leads to the
# platform's page, which says its own list decides alone.
sitestarlight.MOVED_PAGES.setdefault(
f"cross-reference-{pname}.md", f"platforms/{pname}.md")
print("Generating gap analysis page...")
gap_pages: dict[str, str] = {}
+4 -1
View File
@@ -31,6 +31,9 @@ from common import (
)
RECORD = "release.json"
# The record's format, which release.schema.json pins: a reader holding the
# file alone can tell which shape it reads.
SCHEMA_VERSION = 1
# Each registered platform has a page at get/<platform>/ on the site: the
# current pack's files, or the command when no release publishes one. The
# address outlives every release, so a link written elsewhere keeps working.
@@ -120,7 +123,7 @@ def _read_record(dist: Path, tag: str, owners: dict[str, list[str]]) -> dict:
"assets": [path.name for path in paths],
"platforms": owners[name],
}
return {"tag": tag, "packs": packs}
return {"schema_version": SCHEMA_VERSION, "tag": tag, "packs": packs}
def load_record(path: str | Path = RECORD) -> dict:
+218 -41
View File
@@ -5,17 +5,21 @@ was written against, so a permalink is pinned, never guessed."""
from __future__ import annotations
import argparse
import functools
import json
import re
import sys
import urllib.parse
from collections.abc import Iterator
from dataclasses import dataclass
from pathlib import Path
import urllib.request
import upstream
from siterender import _by_mode
from sourceref import (
_prose_runs, is_binary_citation, ref_role, source_ref_values, split_source_ref,
_HEX_ADDRESS_RE, _prose_runs, is_binary_citation, ref_role, source_ref_values,
split_source_ref,
)
@@ -154,43 +158,159 @@ def _source_permalink(repo: upstream.Repo, pin: str, path: str,
url += f"-L{end}"
return url
# The tree listings profile_sync caches. Read offline only: a page build
# never calls a forge, and a listing it does not find leaves the link as the
# profile's declarations alone decide it.
# The tree listings profile_sync caches, read only by the command that
# writes SOURCE_PATHS. The pages read that file: the deploy has no tree
# cache, and a citation it could not check was linked into whichever
# repository the profile declared, 30 of them to a 404.
TREE_CACHE = Path(__file__).resolve().parent.parent / upstream.CACHE_DIR
SOURCE_PATHS = (Path(__file__).resolve().parent.parent
/ "website" / "src" / "data" / "source-paths.json")
def tree_key(repo: upstream.Repo, pin: str) -> str:
"""How SOURCE_PATHS names one repository at one revision."""
return f"{repo.host}/{repo.slug}@{pin}"
@functools.lru_cache(maxsize=None)
def _tree_paths(repo: upstream.Repo, pin: str) -> frozenset[str] | None:
"""Every file of a repository at a revision, None when not cached or
truncated (a missing path would then prove nothing)."""
if not repo.api_base:
return None # a browse-only forge: no listing to read
paths, truncated = upstream.list_tree(repo, pin, str(TREE_CACHE), offline=True)
return frozenset(paths) if paths and not truncated else None
def _verified() -> dict[str, frozenset[str]]:
"""The committed record, empty when it has not been written."""
try:
data = json.loads(SOURCE_PATHS.read_text(encoding="utf-8"))
except FileNotFoundError:
return {}
return {key: frozenset(paths) for key, paths in data.items()}
def tree_holds(repo: upstream.Repo, pin: str, path: str) -> bool | None:
"""Whether the repository holds ``path`` at ``pin``: None when its tree
is not cached. trident cites src/core/loader/ncch.cpp, a Panda3DS file,
and the page linked it into 3dsTrident, where it is a 404."""
paths = _tree_paths(repo, pin)
return None if paths is None else path in paths
@functools.lru_cache(maxsize=None)
def _tree_paths(repo: upstream.Repo, pin: str) -> frozenset[str]:
"""The cited paths of a repository at a revision that its tree holds."""
return _verified().get(tree_key(repo, pin), frozenset())
def tree_holds(repo: upstream.Repo, pin: str, path: str) -> bool:
"""Whether ``path`` was found in the repository's tree at ``pin``.
trident cites src/core/loader/ncch.cpp, a Panda3DS file, and the page
linked it into 3dsTrident, where it is a 404. A path no tree confirmed,
on a forge whose tree cannot be read or a revision not cached when the
record was written, is not linked: no link beats a link to nothing.
"""
return path in _tree_paths(repo, pin)
def _holder(
candidates: list[tuple[upstream.Repo, str]], path: str, default: tuple | None,
candidates: list[tuple[upstream.Repo, str]], path: str,
) -> tuple[upstream.Repo, str] | None:
"""The pinned repository that holds ``path``, source before upstream as
profile_sync.select_views orders them. Without every tree cached, the
declarations decide (``default``). A path no cached tree holds links
nowhere: no link beats a link to a file that is not there."""
verdicts = [tree_holds(repo, pin, path) for repo, pin in candidates]
if not candidates or None in verdicts:
if default is not None and tree_holds(*default, path) is False:
profile_sync.select_views orders them."""
return next((cand for cand in candidates if tree_holds(*cand, path)), None)
def _cached_tree(
repo: upstream.Repo, pin: str, cache_dirs: list[Path], fetch: bool = False,
) -> frozenset[str] | None:
"""Every file of a repository at a revision, None when not cached or
truncated (a missing path would then prove nothing). Each cache is read
in turn. With *fetch*, a tree none holds is fetched into the last one."""
if not repo.api_base:
return None # a browse-only forge: no listing to read
for index, cache_dir in enumerate(cache_dirs):
online = fetch and index == len(cache_dirs) - 1
try:
paths, truncated = upstream.list_tree(repo, pin, str(cache_dir), offline=not online)
except upstream.UpstreamError as exc:
print(f" {tree_key(repo, pin)}: {exc}", file=sys.stderr)
return None
return default
return next((cand for cand, held in zip(candidates, verdicts) if held), None)
if paths and not truncated:
return frozenset(paths)
return None
def _walk(value: object) -> Iterator[tuple[str, object]]:
"""Every (key, value) of a profile, nested lists and mappings included."""
if isinstance(value, dict):
for key, item in value.items():
yield str(key), item
yield from _walk(item)
elif isinstance(value, list):
for item in value:
yield from _walk(item)
def cited_paths(profile: dict) -> set[str]:
"""Every path a page of the profile could link: the parts of each
source_ref and the file:line citations of its prose."""
paths: set[str] = set()
for key, value in _walk(profile):
if key == "source_ref" and isinstance(value, (str, dict)):
for _label, refs in source_ref_values(value):
paths.update(ref_role(part.path)[1] for part in split_source_ref(refs))
elif isinstance(value, str) and ":" in value:
for _n, _span, _spans, parts, _run, _wrapped in _prose_runs(value):
paths.update(part.path for part in parts)
return {path for path in paths
if _PATH_RE.fullmatch(path) and not is_binary_citation(path)}
def profile_trees(profile: dict) -> list[tuple[upstream.Repo, str]]:
"""Every repository and revision a page of the profile can link into."""
found: list[tuple[upstream.Repo, str]] = []
labels = {""} | {
str(mode) for field in ("source", "upstream", "source_commit", "upstream_commit")
if isinstance(profile.get(field), dict) for mode in profile[field]}
for label in sorted(labels):
for pair in _forge_sources(profile, label) + list(role_sources(profile, label).values()):
if pair not in found:
found.append(pair)
return found
def _listed_directories(
repo: upstream.Repo, pin: str, paths: set[str], cache_dirs: list[Path], fetch: bool,
) -> frozenset[str] | None:
"""The cited paths found by listing their directories one by one, for a
tree the forge truncates (aps3e vendors RPCS3 whole). A directory that
cannot be listed confirms none of its paths. None off GitHub, the one
forge with that call."""
if repo.family != "github":
return None
found: set[str] = set()
for folder in sorted({path.rpartition("/")[0] for path in paths} - {""}):
names = None
for index, cache_dir in enumerate(cache_dirs):
online = fetch and index == len(cache_dirs) - 1
try:
names = upstream.list_directory(repo, pin, folder, str(cache_dir), offline=not online)
except upstream.UpstreamError:
names = None
if names is not None:
break
found.update(f"{folder}/{name}" for name in names or ())
return frozenset(found)
def build_source_paths(
profiles: dict, cache_dirs: list[Path], fetch: bool = False,
) -> tuple[dict[str, list[str]], list[str]]:
"""The cited paths each cached tree holds, and the trees not cached."""
held: dict[str, set[str]] = {}
unread: set[str] = set()
for _name, profile in sorted(profiles.items()):
paths = cited_paths(profile)
if not paths:
continue
for repo, pin in profile_trees(profile):
tree = _cached_tree(repo, pin, cache_dirs, fetch)
if tree is None:
tree = _listed_directories(repo, pin, paths, cache_dirs, fetch)
if tree is None:
unread.add(tree_key(repo, pin))
continue
found = paths & tree
if found:
held.setdefault(tree_key(repo, pin), set()).update(found)
return {key: sorted(paths) for key, paths in sorted(held.items())}, sorted(unread)
def _commit_url(repo: upstream.Repo, pin: str) -> str:
@@ -279,11 +399,11 @@ def _source_ref_markdown(profile: dict, value: str | dict | None) -> str:
"""Render source_ref values as pinned links when their forge is known.
A profile can declare two repositories: the libretro port in ``source`` and
the original emulator in ``upstream``. Which of the two carries a given
path cannot be decided without reading their trees, and this generator runs
offline. Rather than guess, an unattributable path is rendered as plain
the original emulator in ``upstream``. A path links into the one whose
tree SOURCE_PATHS says holds it. Any other path is rendered as plain
code: a citation with no link still names the file and the lines, while a
link to the wrong repository is a false citation.
link to the wrong repository is a false citation. An address inside a
shipped binary reads as one (offset 0x8094-0x80ac in libnes.so).
"""
rendered_groups: list[str] = []
path_re = re.compile(r"^[A-Za-z0-9_.@/+~-]+$")
@@ -292,7 +412,13 @@ def _source_ref_markdown(profile: dict, value: str | dict | None) -> str:
by_role = role_sources(profile, label)
rendered: list[str] = []
current_role = ""
binary = ""
for part in split_source_ref(refs):
if is_binary_citation(part.path):
text, binary = _binary_citation(part.path, binary)
rendered.append(text)
continue
binary = ""
role, bare = ref_role(part.path)
if bare != part.path or role:
current_role = role
@@ -311,20 +437,14 @@ def _source_ref_markdown(profile: dict, value: str | dict | None) -> str:
# broke 470 links. Two repositories and no tree to read: no link.
selected: tuple[upstream.Repo, str] | None = None
real_path = part.path
# A shipped binary (`DesktopKinect.dll`, `0x10037130-0x100371de`
# continuing a BAM.dll range) is no path of either repository:
# without a cached tree the single declared one took it.
if path_re.fullmatch(part.path) and not is_binary_citation(part.path):
if path_re.fullmatch(part.path):
if current_role:
# Only the repository the part names, or no link at all.
selected = by_role.get(current_role)
if selected is not None and tree_holds(*selected, part.path) is False:
if selected is not None and not tree_holds(*selected, part.path):
selected = None
else:
selected = _holder(
candidates, part.path,
candidates[0] if len(candidates) == 1 else None,
)
selected = _holder(candidates, part.path)
tag = (f" ({role_label(profile, current_role, part.path)})"
if current_role and _two_repositories(profile) else "")
@@ -344,6 +464,22 @@ def _source_ref_markdown(profile: dict, value: str | dict | None) -> str:
return ", ".join(rendered_groups)
def _binary_citation(path: str, previous: str) -> tuple[str, str]:
"""A part citing code inside a shipped binary, and the binary it names.
`Nesoid 2.5.0 libnes.so 0x455a4-0x45694, 0x8094-0x80ac` cites two
ranges of one library: the bare address continues the binary before it.
No revision of a repository holds those bytes, so nothing is linked.
"""
found = _HEX_ADDRESS_RE.search(path)
if found is None:
return f"`{path}`", path
binary = path[:found.start()].rstrip(" :") or previous
if not binary:
return f"`{path}`", ""
return f"offset `{path[found.start():]}` in `{binary}`", binary
def link_prose_citations(profile: dict, text: str) -> str:
"""Prose file:line citations as pinned links, where the target is certain.
@@ -381,7 +517,7 @@ def link_prose_citations(profile: dict, text: str) -> str:
continue
if any(lo <= path_span[0] < hi for lo, hi in guarded):
continue
if tree_holds(repo, pin, path) is False:
if not tree_holds(repo, pin, path):
continue
for index, ((lo, hi), part) in enumerate(zip(spans, parts)):
start = path_span[0] if index == 0 else lo
@@ -494,3 +630,44 @@ def launcher_ref_markdown(source: ListSource | None, ref: str) -> str:
else:
rendered.append(f"[`{display}`]({source.browse(part.path, part.start, part.end)})")
return ", ".join(rendered)
def main() -> int:
parser = argparse.ArgumentParser(
allow_abbrev=False,
description="Record which cited source paths the cached upstream trees hold",
)
parser.add_argument(
"--cache-dir", type=Path, action="append",
help=f"tree listings profile_sync cached, read in the order given (default: {TREE_CACHE})",
)
parser.add_argument(
"--fetch", action="store_true",
help="fetch a tree no cache holds into the last --cache-dir",
)
parser.add_argument(
"--emulators-dir", default="emulators",
help="profiles whose citations are checked (default: %(default)s)",
)
parser.add_argument(
"--output", type=Path, default=SOURCE_PATHS,
help="record the site reads (default: %(default)s)",
)
args = parser.parse_args()
cache_dirs = args.cache_dir or [TREE_CACHE]
if not args.fetch and not any(path.is_dir() for path in cache_dirs):
print(f"Error: no tree cache at {', '.join(map(str, cache_dirs))}", file=sys.stderr)
return 1
from common import load_emulator_profiles, write_if_changed # noqa: PLC0415
profiles = load_emulator_profiles(args.emulators_dir, skip_aliases=False)
held, unread = build_source_paths(profiles, cache_dirs, args.fetch)
args.output.parent.mkdir(parents=True, exist_ok=True)
write_if_changed(str(args.output), json.dumps(held, indent=1, sort_keys=True) + "\n")
print(f"{sum(map(len, held.values()))} paths held in {len(held)} trees, "
f"{len(unread)} trees not cached: {args.output}")
return 0
if __name__ == "__main__":
sys.exit(main())