"""Controleren, publiceren en uitpakken van de gegevens voor data.mealieshopper.nl. Hoe het werkt (de uitleg voor mensen staat in README.md): - **`data/` is de bron.** Wat in `main` onder `data/` staat, is goedgekeurd. Een bestand is een `.json`-document of een `.zip` met `.json`-documenten. Een zip is doorzichtig: `data/recepten/bundel.zip` met `a.json` erin publiceert `recepten/a.json`, alsof je hem ter plekke had uitgepakt. - **Een collectie** is de eerste map onder `data/` (`recepten` in `data/recepten/...`). Staat er een JSON Schema in `schemas/.schema.json`, dan moet elk document van die collectie eraan voldoen; zonder schema controleren we alleen dat het geldige JSON is. - **`check`** controleert alles in `data/` en `schemas/` (en optioneel dat een pull request met gegevens geen code wijzigt), en schrijft een samenvatting van wat er op de site verandert. Zo ziet wie de pull request goedkeurt ook wat er in een zip zit. - **`build`** maakt de site: elk document opnieuw opgemaakt als JSON, de schema's, `index.json` (de catalogus), `index.html` en `CNAME`. De workflow `publiceren.yml` zet dat op de branch `publicatie` en op GitHub Pages. - **`unpack`** pakt een zip in `data/` uit tot losse `.json`-bestanden en verwijdert de zip, voor wie de gegevens in de repository liever als losse bestanden heeft. Alleen de standaardbibliotheek en `jsonschema` (de referentie-implementatie van JSON Schema). """ from __future__ import annotations import argparse import hashlib import html import json import os import re import shutil import stat import string import sys import zipfile import zlib from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path, PurePosixPath from typing import Any, Iterable, Iterator from jsonschema import exceptions as schema_exceptions from jsonschema import validators ROOT = Path(__file__).resolve().parent.parent DATA_DIR = "data" SCHEMA_DIR = "schemas" SITE_DIR = "site" SITE_DOMAIN = "data.mealieshopper.nl" SITE_URL = f"https://{SITE_DOMAIN}" SCHEMA_SUFFIX = ".schema.json" MAX_JSON_BYTES = 10 * 1024 * 1024 # één document, ook in een zip MAX_ZIP_BYTES = 25 * 1024 * 1024 # een zip zoals hij geüpload wordt (de grens van GitHub voor een upload in de browser) MAX_UNPACKED_BYTES = 100 * 1024 * 1024 # alle documenten in één zip samen MAX_ZIP_ENTRIES = 2000 MAX_SCHEMA_ERRORS = 5 # per document; de rest tellen we alleen MAX_SUMMARY_ROWS = 200 COLLECTION_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,62}$") # Namen die de site zelf gebruikt: een collectie met zo'n naam zou de schema's overschrijven. RESERVED_COLLECTIONS = {SCHEMA_DIR} SEGMENT_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$") # Bestanden die in data/ of schemas/ mogen staan zonder gegevens te zijn. NOTES = {"README.md", ".gitkeep"} # Wat zip-programma's zelf toevoegen; dat slaan we stil over. ZIP_NOISE = {".DS_Store", "Thumbs.db", "desktop.ini"} # Sleutels die op een geheim wijzen. Alles in deze repository wordt openbaar, dus een document met zo'n sleutel komt er niet in. # We vergelijken zonder hoofdletters, '-', '_' en spaties: `api_key`, `apiKey` en `API-Key` zijn dezelfde sleutel. SENSITIVE_KEYS = { "password", "passwd", "wachtwoord", "secret", "secrets", "clientsecret", "token", "accesstoken", "refreshtoken", "idtoken", "apikey", "apitoken", "authorization", "auth", "bearer", "cookie", "cookies", "sessionid", "session", "privatekey", "passphrase", "otp", "totp", "recoverycodes", } SENSITIVE_SUFFIXES = ("password", "secret", "token", "apikey", "privatekey") # Wat een pull request met gegevens (data/, schemas/) niet tegelijk mag wijzigen: alles wat de controle zelf uitvoert. # Zo kan een inzending de controle niet afzwakken zonder dat het een aparte, eigen pull request is. DATA_PREFIXES = (f"{DATA_DIR}/", f"{SCHEMA_DIR}/") CODE_PREFIXES = ("scripts/", "tests/", ".github/", f"{SITE_DIR}/") CODE_FILES = {"requirements.txt"} @dataclass(frozen=True) class Problem: """Iets dat de publicatie tegenhoudt. `file` is het pad in de repository, `member` het bestand in een zip.""" file: str message: str member: str | None = None def __str__(self) -> str: where = f"{self.file} → {self.member}" if self.member else self.file return f"{where}: {self.message}" @dataclass class Document: """Een JSON-document zoals het gepubliceerd wordt. `path` is het pad op de site (`recepten/a.json`).""" path: str source: str member: str | None data: Any @property def collection(self) -> str: return self.path.split("/", 1)[0] @property def body(self) -> bytes: return serialize(self.data) @property def sha256(self) -> str: return hashlib.sha256(self.body).hexdigest() @property def items(self) -> int | None: return len(self.data) if isinstance(self.data, list) else None @property def origin(self) -> str: return f"{self.source} → {self.member}" if self.member else self.source @dataclass class Collection: name: str schema: dict[str, Any] | None @property def title(self) -> str: if self.schema and isinstance(self.schema.get("title"), str): return self.schema["title"] return self.name @property def description(self) -> str: if self.schema and isinstance(self.schema.get("description"), str): return self.schema["description"] return "" @dataclass class Store: """Alles wat in een checkout staat: de documenten, de collecties met hun schema, en wat er mis is.""" documents: list[Document] schemas: dict[str, dict[str, Any]] problems: list[Problem] warnings: list[Problem] def collections(self) -> list[Collection]: names = sorted({doc.collection for doc in self.documents} | set(self.schemas)) return [Collection(name, self.schemas.get(name)) for name in names] # --- JSON ---------------------------------------------------------------------------------------------------------- def serialize(data: Any) -> bytes: """Zo staat elk document op de site: UTF-8, twee spaties inspringen, de volgorde van de inzender, en een regeleinde.""" return (json.dumps(data, ensure_ascii=False, indent=2) + "\n").encode("utf-8") def _no_duplicates(pairs: list[tuple[str, Any]]) -> dict[str, Any]: result: dict[str, Any] = {} for key, value in pairs: if key in result: raise ValueError(f"de sleutel {json.dumps(key, ensure_ascii=False)} staat twee keer in hetzelfde object") result[key] = value return result def _no_constant(name: str) -> Any: raise ValueError(f"{name} is geen geldige JSON-waarde") def parse_json(raw: bytes) -> Any: """Leest een JSON-document streng: UTF-8 (een BOM mag), geen dubbele sleutels, geen NaN of Infinity, en bovenaan een object of een lijst. Geeft een `ValueError` met een Nederlandse uitleg.""" if raw.startswith(b"\xef\xbb\xbf"): raw = raw[3:] try: text = raw.decode("utf-8") except UnicodeDecodeError as exc: raise ValueError(f"het bestand is geen UTF-8 (byte {exc.start}); sla het op als UTF-8") from None try: data = json.loads(text, object_pairs_hook=_no_duplicates, parse_constant=_no_constant) except json.JSONDecodeError as exc: raise ValueError(f"geen geldige JSON: {exc.msg} (regel {exc.lineno}, kolom {exc.colno})") from None except RecursionError: raise ValueError("de JSON is te diep genest") from None if not isinstance(data, (dict, list)): raise ValueError("bovenaan moet een object ({...}) of een lijst ([...]) staan") return data def _pointer(parts: Iterable[Any]) -> str: text = "".join(f"[{part}]" if isinstance(part, int) else f".{part}" for part in parts) return text.lstrip(".") or "(het hele document)" def find_sensitive_keys(data: Any, limit: int = 5) -> list[str]: """De plekken (`a.b[0].token`) van sleutels die op een geheim wijzen, hoogstens `limit`.""" found: list[str] = [] stack: list[tuple[Any, tuple[Any, ...]]] = [(data, ())] while stack and len(found) < limit: value, path = stack.pop() if isinstance(value, dict): for key, child in reversed(list(value.items())): folded = re.sub(r"[\s_-]", "", key).lower() if folded in SENSITIVE_KEYS or folded.endswith(SENSITIVE_SUFFIXES): found.append(_pointer((*path, key))) if len(found) >= limit: break stack.append((child, (*path, key))) elif isinstance(value, list): stack.extend((child, (*path, index)) for index, child in reversed(list(enumerate(value)))) return found # --- Paden --------------------------------------------------------------------------------------------------------- def published_path_problem(path: PurePosixPath) -> str | None: """Waarom `path` (relatief aan data/) geen geldig pad op de site is, of None als het goed is.""" parts = path.parts if len(parts) < 2: return "een document hoort in de map van een collectie, bijvoorbeeld data/recepten/mijn-bestand.json" if parts[0] in RESERVED_COLLECTIONS: return f"'{parts[0]}' is een naam die de site zelf gebruikt; kies een andere naam voor de collectie" if not COLLECTION_RE.match(parts[0]): return ( f"'{parts[0]}' is geen geldige naam voor een collectie: gebruik kleine letters, cijfers en '-' " "(hoogstens 63 tekens)" ) for part in parts[1:]: if not SEGMENT_RE.match(part): return ( f"'{part}' is geen geldige naam: gebruik letters, cijfers, '.', '-' en '_', zonder spaties, " "en begin met een letter of cijfer" ) if path.suffix != ".json": return "alleen .json-bestanden (of een .zip met .json-bestanden) kunnen gepubliceerd worden" return None def _is_zip_noise(name: str) -> bool: parts = name.split("/") return parts[0] == "__MACOSX" or parts[-1] in ZIP_NOISE def _member_problem(info: zipfile.ZipInfo) -> str | None: name = info.filename if "\\" in name: return "een pad met '\\' kan niet; maak de zip opnieuw met '/' als scheidingsteken" if name.startswith("/") or re.match(r"^[A-Za-z]:", name): return "een absoluut pad in een zip kan niet" if any(part in ("", ".", "..") for part in name.split("/")): return "een pad met '..', '.' of '//' in een zip kan niet" mode = info.external_attr >> 16 if mode and stat.S_ISLNK(mode): return "een symbolische koppeling in een zip kan niet" if info.flag_bits & 0x1: return "een zip met een wachtwoord kan niet; alles hier wordt openbaar" return None # --- Inlezen ------------------------------------------------------------------------------------------------------- def _rel(root: Path, path: Path) -> str: return path.relative_to(root).as_posix() def _walk(directory: Path) -> Iterator[Path]: """Alle bestanden en koppelingen onder `directory`, gesorteerd, zonder symbolische koppelingen te volgen.""" for entry in sorted(directory.iterdir(), key=lambda p: p.name): if entry.is_dir() and not entry.is_symlink(): yield from _walk(entry) else: yield entry def load_schemas(root: Path) -> tuple[dict[str, dict[str, Any]], list[Problem]]: schemas: dict[str, dict[str, Any]] = {} problems: list[Problem] = [] directory = root / SCHEMA_DIR if directory.is_symlink(): return schemas, [Problem(SCHEMA_DIR, "schemas/ is een symbolische koppeling; dat kan niet")] if not directory.is_dir(): return schemas, problems for path in _walk(directory): rel = _rel(root, path) if path.name in NOTES and not path.is_symlink(): continue if path.is_symlink() or not path.is_file(): problems.append(Problem(rel, "een symbolische koppeling of een speciaal bestand kan niet")) continue if path.parent != directory or not path.name.endswith(SCHEMA_SUFFIX): problems.append(Problem(rel, f"een schema heet schemas/{SCHEMA_SUFFIX}, direct in schemas/")) continue name = path.name[: -len(SCHEMA_SUFFIX)] if not COLLECTION_RE.match(name): problems.append(Problem(rel, f"'{name}' is geen geldige naam voor een collectie: kleine letters, cijfers en '-'")) continue try: schema = parse_json(path.read_bytes()) except ValueError as exc: problems.append(Problem(rel, str(exc))) continue if not isinstance(schema, dict): problems.append(Problem(rel, "een JSON Schema is een object")) continue try: validators.validator_for(schema).check_schema(schema) except schema_exceptions.SchemaError as exc: problems.append(Problem(rel, f"geen geldig JSON Schema: {exc.message} (bij {_pointer(exc.absolute_path)})")) continue schemas[name] = schema return schemas, problems def _read_json_file(root: Path, path: Path) -> tuple[list[Document], list[Problem]]: rel = _rel(root, path) target = PurePosixPath(path.relative_to(root / DATA_DIR).as_posix()) reason = published_path_problem(target) if reason: return [], [Problem(rel, reason)] if path.stat().st_size > MAX_JSON_BYTES: return [], [Problem(rel, f"groter dan {MAX_JSON_BYTES // (1024 * 1024)} MB; splits het op in meer bestanden")] try: data = parse_json(path.read_bytes()) except ValueError as exc: return [], [Problem(rel, str(exc))] return [Document(target.as_posix(), rel, None, data)], [] def read_zip(root: Path, path: Path) -> tuple[list[Document], list[Problem]]: """De documenten in een zip onder data/. Een zip is doorzichtig: zijn map in data/ is de map waarin hij uitpakt.""" rel = _rel(root, path) if path.stat().st_size > MAX_ZIP_BYTES: return [], [Problem(rel, f"de zip is groter dan {MAX_ZIP_BYTES // (1024 * 1024)} MB; splits hem op")] base = PurePosixPath(path.parent.relative_to(root / DATA_DIR).as_posix()) documents: list[Document] = [] problems: list[Problem] = [] try: archive = zipfile.ZipFile(path) except (zipfile.BadZipFile, OSError) as exc: return [], [Problem(rel, f"geen geldige zip ({exc})")] with archive: entries = [info for info in archive.infolist() if not info.is_dir() and not _is_zip_noise(info.filename)] if len(entries) > MAX_ZIP_ENTRIES: return [], [Problem(rel, f"meer dan {MAX_ZIP_ENTRIES} bestanden in één zip; splits hem op")] if not entries: return [], [Problem(rel, "de zip bevat geen .json-bestanden")] total = 0 for info in entries: name = info.filename reason = _member_problem(info) target = base / name if reason is None: if name.lower().endswith(".zip"): reason = "een zip in een zip kan niet; zet de bestanden er los in" else: reason = published_path_problem(target) if reason: problems.append(Problem(rel, reason, name)) continue try: with archive.open(info) as handle: raw = handle.read(MAX_JSON_BYTES + 1) except (zipfile.BadZipFile, NotImplementedError, RuntimeError, zlib.error, EOFError, OSError) as exc: problems.append(Problem(rel, f"niet te lezen ({exc})", name)) continue if len(raw) > MAX_JSON_BYTES: problems.append(Problem(rel, f"groter dan {MAX_JSON_BYTES // (1024 * 1024)} MB uitgepakt", name)) continue total += len(raw) if total > MAX_UNPACKED_BYTES: problems.append(Problem(rel, f"uitgepakt samen groter dan {MAX_UNPACKED_BYTES // (1024 * 1024)} MB; splits de zip op")) break try: data = parse_json(raw) except ValueError as exc: problems.append(Problem(rel, str(exc), name)) continue documents.append(Document(target.as_posix(), rel, name, data)) return documents, problems def load(root: Path) -> Store: """Leest en controleert alles in data/ en schemas/ van een checkout.""" schemas, problems = load_schemas(root) warnings: list[Problem] = [] documents: list[Document] = [] directory = root / DATA_DIR if directory.is_symlink(): problems.append(Problem(DATA_DIR, "data/ is een symbolische koppeling; dat kan niet")) elif directory.is_dir(): for path in _walk(directory): rel = _rel(root, path) if path.name in NOTES and not path.is_symlink(): continue if path.is_symlink() or not path.is_file(): problems.append(Problem(rel, "een symbolische koppeling of een speciaal bestand kan niet")) continue if path.suffix == ".zip": found, trouble = read_zip(root, path) elif path.suffix == ".json": found, trouble = _read_json_file(root, path) else: found, trouble = [], [Problem(rel, "alleen .json- en .zip-bestanden kunnen in data/")] documents.extend(found) problems.extend(trouble) problems.extend(_check_collisions(documents)) for doc in documents: keys = find_sensitive_keys(doc.data) if keys: problems.append(Problem( doc.source, "dit lijkt een geheim te bevatten (" + ", ".join(keys) + "); alles hier wordt openbaar, haal het eruit", doc.member, )) schema = schemas.get(doc.collection) if schema is not None: problems.extend(_check_schema(doc, schema)) for name in sorted({doc.collection for doc in documents} - set(schemas)): warnings.append(Problem( f"{DATA_DIR}/{name}", f"de collectie '{name}' heeft geen schema ({SCHEMA_DIR}/{name}{SCHEMA_SUFFIX}); we controleren alleen of het geldige JSON is", )) documents.sort(key=lambda doc: doc.path) return Store(documents, schemas, problems, warnings) def _check_collisions(documents: list[Document]) -> list[Problem]: problems: list[Problem] = [] seen: dict[str, Document] = {} folders = { "/".join(doc.path.casefold().split("/")[:depth]): doc for doc in documents for depth in range(1, doc.path.count("/") + 1) } for doc in documents: folder = folders.get(doc.path.casefold()) if folder is not None: problems.append(Problem( doc.source, f"{doc.path} is ook een map, voor {folder.path} uit {folder.origin}; kies een andere naam", doc.member, )) key = doc.path.casefold() earlier = seen.get(key) if earlier is None: seen[key] = doc continue what = "hetzelfde pad" if earlier.path == doc.path else "hetzelfde pad op hoofdletters na" problems.append(Problem( doc.source, f"{doc.path} bestaat al ({what}) uit {earlier.origin}; elk pad op de site komt uit één bestand", doc.member, )) return problems def _short(message: str, limit: int = 160) -> str: """De melding van jsonschema noemt vaak de hele waarde; in het logboek is het begin genoeg.""" message = " ".join(message.split()) return message if len(message) <= limit else message[: limit - 1] + "…" def _check_schema(doc: Document, schema: dict[str, Any]) -> list[Problem]: validator = validators.validator_for(schema)(schema) errors = sorted(validator.iter_errors(doc.data), key=lambda error: [str(part) for part in error.absolute_path]) problems = [ Problem(doc.source, f"voldoet niet aan het schema bij {_pointer(error.absolute_path)}: {_short(error.message)}", doc.member) for error in errors[:MAX_SCHEMA_ERRORS] ] if len(errors) > MAX_SCHEMA_ERRORS: problems.append(Problem(doc.source, f"en nog {len(errors) - MAX_SCHEMA_ERRORS} fouten tegen het schema", doc.member)) return problems # --- Wijzigingen in een pull request ------------------------------------------------------------------------------- def scope_problems(changed: Iterable[str]) -> list[Problem]: """Een pull request met gegevens wijzigt geen code: anders zou een inzending de controle zelf kunnen veranderen.""" changed = sorted(set(changed)) data = [path for path in changed if path.startswith(DATA_PREFIXES)] code = [path for path in changed if path.startswith(CODE_PREFIXES) or path in CODE_FILES] if not data or not code: return [] return [ Problem(path, "een pull request met gegevens (data/, schemas/) wijzigt geen code; zet deze wijziging in een eigen pull request") for path in code ] @dataclass class Change: kind: str # "nieuw", "gewijzigd", "verwijderd" path: str document: Document | None def compare(before: list[Document], after: list[Document]) -> list[Change]: old = {doc.path: doc for doc in before} new = {doc.path: doc for doc in after} changes: list[Change] = [] for path in sorted(old.keys() | new.keys()): if path not in old: changes.append(Change("nieuw", path, new[path])) elif path not in new: changes.append(Change("verwijderd", path, old[path])) elif old[path].sha256 != new[path].sha256: changes.append(Change("gewijzigd", path, new[path])) return changes # --- Samenvatting -------------------------------------------------------------------------------------------------- def _items(doc: Document | None) -> str: if doc is None or doc.items is None: return "object" return str(doc.items) def _cell(text: str) -> str: return text.replace("|", "\\|").replace("\n", " ") def summary(store: Store, changes: list[Change] | None) -> str: lines = ["## Controle van de gegevens", ""] total = len(store.documents) collections = store.collections() if store.problems: lines.append(f"❌ **{len(store.problems)} probleem/problemen** gevonden. Zo kan het niet gepubliceerd worden.") else: lines.append(f"✅ Alles in orde: {total} document(en) in {len(collections)} collectie(s).") lines.append("") if changes is not None: lines += ["### Wat verandert er op " + SITE_DOMAIN, ""] if not changes: lines += ["Niets: na het samenvoegen staat er hetzelfde op de site.", ""] else: icon = {"nieuw": "➕ nieuw", "gewijzigd": "✏️ gewijzigd", "verwijderd": "➖ verwijderd"} lines += ["| | Bestand op de site | Items | Uit |", "|---|---|---|---|"] for change in changes[:MAX_SUMMARY_ROWS]: origin = "" if change.kind == "verwijderd" or change.document is None else f"`{_cell(change.document.origin)}`" lines.append(f"| {icon[change.kind]} | `{_cell(change.path)}` | {_items(change.document)} | {origin} |") if len(changes) > MAX_SUMMARY_ROWS: lines.append(f"| | … en nog {len(changes) - MAX_SUMMARY_ROWS} | | |") lines.append("") if store.problems: lines += ["### Problemen", ""] lines += [f"- `{_cell(problem.file)}`" + (f" → `{_cell(problem.member)}`" if problem.member else "") + f": {_cell(problem.message)}" for problem in store.problems[:MAX_SUMMARY_ROWS]] if len(store.problems) > MAX_SUMMARY_ROWS: lines.append(f"- … en nog {len(store.problems) - MAX_SUMMARY_ROWS}") lines.append("") if store.warnings: lines += ["### Let op", ""] lines += [f"- {_cell(warning.message)}" for warning in store.warnings] lines.append("") if collections: lines += ["### Collecties", "", "| Collectie | Schema | Documenten |", "|---|---|---|"] for collection in collections: count = sum(1 for doc in store.documents if doc.collection == collection.name) has_schema = "✅" if collection.schema is not None else "—" lines.append(f"| `{collection.name}` {_cell(collection.title) if collection.title != collection.name else ''} " f"| {has_schema} | {count} |") lines.append("") return "\n".join(lines) def _annotate(problem: Problem, level: str) -> str: """Een regel voor GitHub Actions, zodat het probleem bij het bestand in de pull request staat.""" message = (f"{problem.member}: " if problem.member else "") + problem.message message = message.replace("%", "%25").replace("\r", "%0D").replace("\n", "%0A") return f"::{level} file={problem.file}::{message}" # --- De site ------------------------------------------------------------------------------------------------------- def catalog(store: Store, commit: str, generated_at: str) -> dict[str, Any]: """`index.json`: welke collecties en documenten er zijn, met hun adres, grootte en SHA-256.""" result: dict[str, Any] = {"site": SITE_URL, "generated_at": generated_at, "commit": commit, "collections": []} for collection in store.collections(): files = [] for doc in store.documents: if doc.collection != collection.name: continue body = doc.body files.append({ "path": doc.path, "url": f"{SITE_URL}/{doc.path}", "bytes": len(body), "sha256": hashlib.sha256(body).hexdigest(), "items": doc.items, }) result["collections"].append({ "name": collection.name, "title": collection.title, "description": collection.description, "schema": f"{SITE_URL}/{SCHEMA_DIR}/{collection.name}{SCHEMA_SUFFIX}" if collection.schema is not None else None, "files": files, }) return result def _count(number: int, one: str, many: str) -> str: return f"{number} {one if number == 1 else many}" def _size(count: int) -> str: if count < 1024: return f"{count} B" if count < 1024 * 1024: return f"{count / 1024:.1f} kB".replace(".", ",") return f"{count / (1024 * 1024):.1f} MB".replace(".", ",") def _page(index: dict[str, Any], template: str) -> str: sections = [] for collection in index["collections"]: rows = "\n".join( f'
  • {html.escape(item["path"].split("/", 1)[1])}' f'{"" if item["items"] is None else _count(item["items"], "item", "items") + " · "}{_size(item["bytes"])}
    • ' for item in collection["files"] ) or '
  • Nog geen documenten.
    • ' schema = ( f' · schema' if collection["schema"] else "" ) description = f'

    {html.escape(collection["description"])}

    ' if collection["description"] else "" sections.append( f'
    \n' f'

    {html.escape(collection["title"])}

    \n' f'

    {html.escape(collection["name"])}/{schema}

    \n' f"{description}\n
      \n{rows}\n
    \n
    " ) body = "\n".join(sections) or '

    Er is nog niets gepubliceerd.

    ' files = sum(len(collection["files"]) for collection in index["collections"]) return string.Template(template).substitute( collections=body, counts=_count(len(index["collections"]), "collectie", "collecties") + " · " + _count(files, "document", "documenten"), commit_short=html.escape(index["commit"][:7] or "lokaal"), generated_at=html.escape(index["generated_at"]), ) PUBLICATION_README = """# Gepubliceerd: {domain} Deze branch maakt de workflow `publiceren.yml` uit `main` (commit `{commit}`). Wijzig hier niets met de hand: de volgende publicatie overschrijft het. Een wijziging aan de gegevens gaat via een pull request naar `main`; zie de README daar. """ def build(root: Path, out: Path, commit: str = "", generated_at: str | None = None) -> Store: """Maakt de site in `out` (die eerst leeg wordt gemaakt). Geeft de `Store` terug; met problemen schrijft hij niets.""" store = load(root) if store.problems: return store out = out.resolve() if out == root.resolve() or root.resolve().is_relative_to(out) or out.is_relative_to(root.resolve() / DATA_DIR): raise ValueError(f"{out} kan niet de uitvoermap zijn: hij valt samen met de repository of data/") if out.exists(): shutil.rmtree(out) out.mkdir(parents=True) generated_at = generated_at or datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") for doc in store.documents: target = out / doc.path target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(doc.body) if store.schemas: (out / SCHEMA_DIR).mkdir() for name, schema in sorted(store.schemas.items()): (out / SCHEMA_DIR / f"{name}{SCHEMA_SUFFIX}").write_bytes(serialize(schema)) index = catalog(store, commit, generated_at) (out / "index.json").write_bytes(serialize(index)) site = root / SITE_DIR template = (site / "index.html").read_text(encoding="utf-8") (out / "index.html").write_text(_page(index, template), encoding="utf-8") for asset in sorted(site.iterdir()): if asset.is_file() and asset.name != "index.html": shutil.copyfile(asset, out / asset.name) (out / "CNAME").write_text(SITE_DOMAIN + "\n", encoding="utf-8") (out / ".nojekyll").write_text("", encoding="utf-8") (out / "README.md").write_text(PUBLICATION_README.format(domain=SITE_DOMAIN, commit=commit or "lokaal"), encoding="utf-8") return store # --- Uitpakken ----------------------------------------------------------------------------------------------------- def unpack(root: Path, path: Path) -> list[Problem]: """Pakt een zip in data/ uit tot losse, opgemaakte .json-bestanden naast hem, en verwijdert de zip. Doet niets als er een probleem is, of als een van de bestanden al bestaat.""" path = path.resolve() data_dir = (root / DATA_DIR).resolve() if not path.is_relative_to(data_dir) or path.suffix != ".zip" or not path.is_file(): return [Problem(str(path), "geef een .zip-bestand in data/")] documents, problems = read_zip(root.resolve(), path) if problems: return problems for doc in documents: target = data_dir / doc.path if target.exists(): problems.append(Problem(_rel(root.resolve(), path), f"{DATA_DIR}/{doc.path} bestaat al", doc.member)) if problems: return problems for doc in documents: target = data_dir / doc.path target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(doc.body) path.unlink() return [] # --- Opdrachtregel ------------------------------------------------------------------------------------------------- def _report(store: Store, extra: list[Problem]) -> None: in_actions = os.environ.get("GITHUB_ACTIONS") == "true" for problem in [*extra, *store.problems]: print(_annotate(problem, "error") if in_actions else f"FOUT {problem}") for warning in store.warnings: print(_annotate(warning, "warning") if in_actions else f"LET OP {warning}") def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) parser.add_argument("--root", type=Path, default=ROOT, help="de checkout (standaard: deze repository)") commands = parser.add_subparsers(dest="command", required=True) check = commands.add_parser("check", help="controleer data/ en schemas/") check.add_argument("--compare", type=Path, help="een checkout van de basisbranch, om de wijzigingen te tonen") check.add_argument("--changed-files", type=Path, metavar="BESTAND", help="de gewijzigde paden van een pull request, één per regel: gegevens en code wijzigen niet samen") check.add_argument("--summary", type=Path, help="schrijf een samenvatting in Markdown (zoals $GITHUB_STEP_SUMMARY)") make = commands.add_parser("build", help="maak de site") make.add_argument("--out", type=Path, required=True) make.add_argument("--commit", default="", help="de commit waaruit de gegevens komen") make.add_argument("--generated-at", help="het moment van die gegevens (ISO 8601); standaard nu") extract = commands.add_parser("unpack", help="pak een zip in data/ uit tot losse .json-bestanden") extract.add_argument("zips", type=Path, nargs="+") args = parser.parse_args(argv) root: Path = args.root.resolve() if args.command == "check": store = load(root) extra = [] if args.changed_files: extra = scope_problems(line.strip() for line in args.changed_files.read_text(encoding="utf-8").splitlines() if line.strip()) changes = compare(load(args.compare.resolve()).documents, store.documents) if args.compare else None _report(store, extra) if args.summary: shown = Store(store.documents, store.schemas, [*extra, *store.problems], store.warnings) with args.summary.open("a", encoding="utf-8") as handle: handle.write(summary(shown, changes) + "\n") failed = bool(extra or store.problems) print(("Niet in orde." if failed else f"In orde: {len(store.documents)} document(en).")) return 1 if failed else 0 if args.command == "build": store = build(root, args.out, args.commit, args.generated_at) _report(store, []) if store.problems: print("Niet gebouwd: los eerst de problemen op.") return 1 print(f"Site gebouwd in {args.out}: {len(store.documents)} document(en).") return 0 failed = False for path in args.zips: problems = unpack(root, path) for problem in problems: print(f"FOUT {problem}") if problems: failed = True else: print(f"Uitgepakt: {path}") return 1 if failed else 0 if __name__ == "__main__": sys.exit(main())