diff --git a/scripts/gen_optimizer_reference.py b/scripts/gen_optimizer_reference.py new file mode 100644 index 0000000..05dee86 --- /dev/null +++ b/scripts/gen_optimizer_reference.py @@ -0,0 +1,709 @@ +#!/usr/bin/env python3 +""" +gen_optimizer_reference.py — génère la référence du protocole optimiseur ↔ moteur +depuis le contrat machine du moteur (OpenRPC 1.3). + +Source de vérité : `docs/etm-powersync-optimizer-protocol.openrpc.json` du dépôt du +moteur `etm-powersync-energy-plugin-etm` (la CI en clone la branche par défaut ; +en local, `--src` pointe sur une copie). La prose du protocole +(`docs/OPTIMIZER_PROTOCOL.md` du moteur) n'est PAS reprise : la section Optimiseur +du site a sa propre prose courte, traduite, et renvoie au document du moteur. + +Produit : + - docs/optimiseur/etm-powersync-optimizer-protocol.openrpc.json — copie octet + pour octet du contrat, servie telle quelle par le site ; + - le contenu des marqueurs `` + de toutes les pages de docs/ (tableaux de messages, listes fermées, règles + nommées, machine de repli, transport, fiche du contrat) ; + - docs/optimiseur/protocole*.md et contrat*.md, s'ils sont absents (échafaudage + minimal par langue — une page existante n'est JAMAIS réécrite hors marqueurs). + +Garde-fou : échoue si ce qu'il publie (blocs générés, copie du contrat, pages de +docs/optimiseur/) contient une adresse IPv4 privée (10/8, 172.16/12, 192.168/16) +hors des exemples nommés dans CONTRACT_IP_EXAMPLES, ou les mots « cloud » / « Héos ». + +Bibliothèque standard seule. Aucun accès réseau. + +Usage : + python3 scripts/gen_optimizer_reference.py --src .engine-src/etm-powersync-energy-plugin-etm --docs docs + python3 scripts/gen_optimizer_reference.py --src --docs docs --check +""" +from __future__ import annotations + +import argparse +import hashlib +import ipaddress +import json +import re +import sys +from pathlib import Path + +CONTRACT_NAME = "etm-powersync-optimizer-protocol.openrpc.json" +OUT_DIR = "optimiseur" + +MARKER_RE = re.compile( + r"()(?P.*?)()", + re.DOTALL, +) + +# Motif « instant ISO 8601 avec décalage » du contrat : affiché en clair, pas en regex. +INSTANT_PATTERN = ( + r"^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}(:[0-9]{2}(\.[0-9]+)?)?" + r"(Z|[+-][0-9]{2}:[0-9]{2})$" +) + + +# ═════════════════════════════════════════════════════════════════════════════ +# VOCABULAIRE DU BLOC DE RÉFÉRENCE — EN ANGLAIS, UNE SEULE FOIS +# +# Même règle que gen_api_reference.py (README, décision du 2026-10-12) : le +# contenu entre marqueurs est servi IDENTIQUE dans les trois langues. Le cadre +# (titres, colonnes, oui/non) est en anglais ; les `description` viennent du +# contrat TELLES QUELLES — il les écrit en français, et les traduire ici ferait +# une seconde vérité. La colonne le dit (« from the contract, French »). +# ═════════════════════════════════════════════════════════════════════════════ + +REF = { + "field_cols": ["Field", "Type", "Required", "Values", "Description (contract, French)"], + "yes": "yes", + "no": "no", + "any": "any", + "instant": "instant with offset", + "closed": "Undeclared fields: **rejected** (engine-emitted form, closed).", + "open": "Undeclared fields: **ignored** (optimizer-emitted form, open).", + "at_least_one": "At least one of:", + "one_of": "One of:", + "variant": "variant", + "params": "params", + "result": "result", + "errors": "Errors", + "no_params": "_No parameters._", + "keys": "keys", + "map_of": "map of", +} + +# Exemples d'usage du contrat qui portent une adresse privée — admis UNIQUEMENT à +# ce pointeur JSON et avec cette valeur. Si le contrat change la valeur, l'exemption +# tombe d'elle-même et le garde-fou échoue. +CONTRACT_IP_EXAMPLES = { + ("/servers/1/variables/host/default", "192.168.1.10"), +} + +PRIVATE_NETS = [ipaddress.ip_network(n) for n in ("10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16")] +DOC_NET = ipaddress.ip_network("192.0.2.0/24") # RFC 5737, exemples documentaires +IPV4_RE = re.compile(r"(? Path: + if src.is_file(): + return src + for cand in (src / "docs" / CONTRACT_NAME, src / CONTRACT_NAME): + if cand.is_file(): + return cand + sys.exit(f"ERREUR : {CONTRACT_NAME} introuvable sous {src}") + + +def load_contract(path: Path) -> tuple[bytes, dict]: + raw = path.read_bytes() + try: + d = json.loads(raw.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + sys.exit(f"ERREUR : {path} non parsable : {exc}") + for key in ("openrpc", "info", "methods", "components"): + if key not in d: + sys.exit(f"ERREUR : clé « {key} » absente de {path}") + return raw, d + + +# ── Rendu des schémas ───────────────────────────────────────────────────────── + +def cell(text: object) -> str: + s = " ".join(str(text).split()) + return s.replace("|", "\\|") + + +def ref_name(ref: str) -> str: + return ref.rsplit("/", 1)[-1] + + +def anchor(name: str) -> str: + return "schema-" + name.lower() + + +def link(name: str) -> str: + return f"[`{name}`](#{anchor(name)})" + + +def json_lit(v: object) -> str: + return "`" + json.dumps(v, ensure_ascii=False) + "`" + + +def type_str(s: dict) -> str: + if "$ref" in s: + return link(ref_name(s["$ref"])) + for comb in ("oneOf", "anyOf"): + if comb in s and "type" not in s: + return " \\| ".join(type_str(x) for x in s[comb]) + t = s.get("type") + if t is None: + if "const" in s: + t = {bool: "boolean", str: "string", int: "integer", float: "number"}.get( + type(s["const"]), REF["any"]) + elif "enum" in s: + return "enum" + else: + return REF["any"] + types = t if isinstance(t, list) else [t] + out = [] + for one in types: + if one == "array": + items = s.get("items", {}) + out.append("[" + type_str(items) + "]") + elif one == "object" and isinstance(s.get("additionalProperties"), dict) \ + and "properties" not in s: + out.append(f"{REF['map_of']} " + type_str(s["additionalProperties"])) + else: + out.append(one) + return " \\| ".join(out) + + +def values_str(s: dict) -> str: + v = [] + if "const" in s: + v.append(json_lit(s["const"])) + if "enum" in s: + v.append(" · ".join(f"`{x}`" for x in s["enum"])) + if s.get("type") == "array" or (isinstance(s.get("type"), list) and "array" in s["type"]): + items = s.get("items", {}) + if "enum" in items: + v.append(" · ".join(f"`{x}`" for x in items["enum"])) + if "minimum" in items: + v.append(f"≥ {items['minimum']}") + if "maximum" in items: + v.append(f"≤ {items['maximum']}") + if "minimum" in s: + v.append(f"≥ {s['minimum']}") + if "exclusiveMinimum" in s: + v.append(f"> {s['exclusiveMinimum']}") + if "maximum" in s: + v.append(f"≤ {s['maximum']}") + if "minLength" in s: + v.append(f"≥ {s['minLength']} char.") + if "minItems" in s: + v.append(f"≥ {s['minItems']} items") + if "pattern" in s: + v.append(REF["instant"] if s["pattern"] == INSTANT_PATTERN else f"pattern `{s['pattern']}`") + if isinstance(s.get("propertyNames"), dict) and "enum" in s["propertyNames"]: + v.append(REF["keys"] + " " + " · ".join(f"`{x}`" for x in s["propertyNames"]["enum"])) + return " ; ".join(v) + + +def property_rows(props: dict, required: list, prefix: str = "") -> list[list[str]]: + rows = [] + for name, ps in props.items(): + full = prefix + name + rows.append([ + f"`{full}`", + type_str(ps), + REF["yes"] if name in required else REF["no"], + cell(values_str(ps)), + cell(ps.get("description", "")), + ]) + if ps.get("type") == "object" and "properties" in ps: + rows.extend(property_rows(ps["properties"], ps.get("required", []), full + ".")) + return rows + + +def table(rows: list[list[str]], cols: list[str]) -> str: + lines = ["| " + " | ".join(cols) + " |", "|" + "---|" * len(cols)] + lines += ["| " + " | ".join(r) + " |" for r in rows] + return "\n".join(lines) + + +def object_body(s: dict) -> list[str]: + """Tableau d'un objet JSON Schema, suivi de ses règles d'ouverture.""" + parts = [table(property_rows(s.get("properties", {}), s.get("required", [])), + REF["field_cols"]), ""] + ap = s.get("additionalProperties") + if ap is False: + parts += [REF["closed"], ""] + elif ap is True: + parts += [REF["open"], ""] + if "anyOf" in s and all(set(x) == {"required"} for x in s["anyOf"]): + names = ", ".join(f"`{n}`" for x in s["anyOf"] for n in x["required"]) + parts += [f"{REF['at_least_one']} {names}.", ""] + return parts + + +def render_schema(name: str, s: dict) -> str: + parts = [f"#### `{name}` {{#{anchor(name)}}}", ""] + if s.get("description"): + parts += [cell(s["description"]), ""] + if "properties" in s: + parts += object_body(s) + elif "oneOf" in s or "anyOf" in s: + variants = s.get("oneOf") or s.get("anyOf") + if all("$ref" in x for x in variants): + parts += [f"{REF['one_of']} " + " · ".join(type_str(x) for x in variants) + ".", ""] + else: + for i, x in enumerate(variants, 1): + head = f"**{REF['variant']} {i}**" + if x.get("description"): + head += " — " + cell(x["description"]) + parts += [head, ""] + parts += object_body(x) if "properties" in x else [f"`{type_str(x)}`", ""] + else: + extra = values_str(s) + parts += [f"Type: {type_str(s)}" + (f" — {extra}" if extra else ""), ""] + return "\n".join(parts) + + +def refs_in(o: object) -> list[str]: + """Noms de schémas référencés, dans l'ordre de lecture, sans doublon.""" + seen: list[str] = [] + + def walk(x: object) -> None: + if isinstance(x, dict): + if "$ref" in x: + n = ref_name(x["$ref"]) + if n not in seen: + seen.append(n) + for v in x.values(): + walk(v) + elif isinstance(x, list): + for v in x: + walk(v) + walk(o) + return seen + + +def reach(roots: list[str], schemas: dict) -> list[str]: + out: list[str] = [] + todo = list(roots) + while todo: + n = todo.pop(0) + if n in out or n not in schemas: + continue + out.append(n) + todo += [r for r in refs_in(schemas[n]) if r not in out] + return out + + +def is_closed_list(s: dict) -> bool: + return ("enum" in s or "const" in s) and "properties" not in s + + +# ── Le modèle : quels schémas vont dans quel bloc ───────────────────────────── + +class Model: + def __init__(self, raw: bytes, d: dict): + self.raw = raw + self.d = d + self.schemas: dict = d["components"].get("schemas", {}) + self.methods = {m["name"]: m for m in d["methods"]} + for needed in ("Handshake", "GetPlan", "UpdateLearned"): + if needed not in self.methods: + sys.exit(f"ERREUR : méthode {needed} absente du contrat") + + closed = [n for n, s in self.schemas.items() if is_closed_list(s)] + placed = set(closed) + self.blocks: dict[str, list[str]] = {"closed_lists": closed} + + def take(key: str, roots: list[str]) -> None: + names = [n for n in reach(roots, self.schemas) if n not in placed] + placed.update(names) + self.blocks[key] = names + + hs, gp, ul = (self.methods[k] for k in ("Handshake", "GetPlan", "UpdateLearned")) + take("handshake", refs_in(hs.get("params", [])) + refs_in(hs.get("result", {}))) + take("context", refs_in(gp.get("params", []))) + take("plan", refs_in(gp.get("result", {}))) + take("update_learned", refs_in(ul.get("params", []))) + orphans = [n for n in self.schemas if n not in placed] + if orphans: + sys.exit("ERREUR : schémas du contrat rattachés à aucun bloc : " + ", ".join(orphans) + + " — étendre Model dans gen_optimizer_reference.py") + + # ── blocs ── + + def method_block(self, name: str, with_result: bool = True) -> str: + m = self.methods[name] + parts = [f"#### `{name}` — {REF['params']}", ""] + if m.get("summary"): + parts += [cell(m["summary"]), ""] + if m.get("description"): + parts += [cell(m["description"]), ""] + params = m.get("params", []) + if params: + rows = [] + for p in params: + ps = p.get("schema", {}) + rows.append([f"`{p['name']}`", type_str(ps), + REF["yes"] if p.get("required") else REF["no"], + cell(values_str(ps)), + cell(p.get("description", ps.get("description", "")))]) + parts += [table(rows, REF["field_cols"]), ""] + else: + parts += [REF["no_params"], ""] + if with_result and "result" in m: + parts += [f"**{REF['result']}** (`{m['result']['name']}`): " + + type_str(m["result"].get("schema", {})), ""] + if m.get("errors"): + rows = [[f"`{e['code']}`", cell(e.get("message", "")), cell(e.get("data", ""))] + for e in m["errors"]] + parts += [f"**{REF['errors']}**", "", table(rows, ["Code", "Message", "Note (contract)"]), ""] + return "\n".join(parts) + + def schemas_block(self, key: str) -> str: + return "\n".join(render_schema(n, self.schemas[n]) for n in self.blocks[key]) + + def render(self, key: str) -> str | None: + k = key.removeprefix("__optimizer_").removesuffix("__") + fn = getattr(self, "block_" + k, None) + return fn().rstrip() + "\n" if fn else None + + def block_transport(self) -> str: + t = self.d.get("x-etm-transport", {}) + bo = t.get("backoffS", {}) + roles = t.get("roles", {}) + rows = [ + ["Protocol version", f"`{self.d['info'].get('version', '')}`"], + ["Application protocol", f"JSON-RPC `{t.get('jsonrpc', '')}`"], + ["Framing", f"`{t.get('framing', '')}` (one message per line)"], + ["Encoding", f"`{t.get('encoding', '')}`"], + ["Endpoint schemes", " · ".join(f"`{s}://`" for s in t.get("schemes", []))], + ["Default endpoint", f"`{t.get('defaultEndpoint', '')}`"], + ["Client", f"`{roles.get('client', '')}`"], + ["Server", f"`{roles.get('server', '')}`"], + ["Max line size", f"{t.get('maxLineBytes', '')} bytes"], + ["Handshake timeout", f"{t.get('handshakeTimeoutS', '')} s"], + ["Request timeout", f"{t.get('requestTimeoutS', '')} s"], + ["Reconnect backoff", f"{bo.get('initial', '')} s → {bo.get('max', '')} s, ×{bo.get('factor', '')}"], + ["Requests in flight", f"{t.get('requestsInFlight', '')}"], + ] + for k, label in (("ids", "Request ids"), ("lineErrors", "Line errors"), + ("tokenMasking", "Token masking")): + if k in t: + rows.append([label, cell(t[k])]) + parts = [table(rows, ["Property", "Value (contract)"]), ""] + + srv = self.d.get("servers", []) + if srv: + rows = [] + for s in srv: + var = s.get("variables", {}) + defaults = ", ".join(f"`{k}` = `{v.get('default', '')}`" for k, v in var.items()) + rows.append([f"`{s.get('name', '')}`", f"`{s.get('url', '')}`", + cell(s.get("summary", "")), defaults]) + parts += ["**Servers**", "", table(rows, ["Name", "URL", "Summary (contract)", "Example values"]), ""] + + rows = [] + for m in self.d["methods"]: + kind = "notification (no id)" if m.get("x-etm-notification") else "request" + to = f"{m['x-etm-timeout-s']} s" if "x-etm-timeout-s" in m else "—" + rows.append([f"`{m['name']}`", f"`{m.get('x-etm-direction', '')}`", kind, to, + cell(m.get("summary", ""))]) + parts += ["**Methods**", "", table(rows, ["Method", "Direction", "Kind", "Timeout", "Summary (contract)"]), ""] + return "\n".join(parts) + + def block_handshake(self) -> str: + return self.method_block("Handshake") + "\n" + self.schemas_block("handshake") + + def block_context(self) -> str: + return self.method_block("GetPlan", with_result=False) + "\n" + self.schemas_block("context") + + def block_plan(self) -> str: + m = self.methods["GetPlan"] + parts = [f"#### `GetPlan` — {REF['result']}", "", + f"`{m['result']['name']}`: " + type_str(m["result"].get("schema", {})), ""] + if m.get("errors"): + rows = [[f"`{e['code']}`", cell(e.get("message", "")), cell(e.get("data", ""))] + for e in m["errors"]] + parts += [f"**{REF['errors']}**", "", table(rows, ["Code", "Message", "Note (contract)"]), ""] + return "\n".join(parts) + "\n" + self.schemas_block("plan") + + def block_update_learned(self) -> str: + return self.method_block("UpdateLearned") + "\n" + self.schemas_block("update_learned") + + def block_reason_codes(self) -> str: + rp = self.d.get("x-etm-reason-params", {}) + codes = self.schemas.get("ReasonCode", {}).get("enum", sorted(rp)) + rows = [[f"`{c}`", ", ".join(f"`{k}`" for k in rp.get(c, [])) or "—"] for c in codes] + return table(rows, ["`reasonCode`", "Accepted `reasonParams` keys"]) + "\n" + + def block_fallback(self) -> str: + fm = self.d.get("x-etm-fallback-machine", {}) + parts = [] + if fm.get("description"): + parts += [cell(fm["description"]), ""] + if "failuresBeforeDead" in fm: + parts += [f"Consecutive failed requests before `optimizerAlive = false`: " + f"**{fm['failuresBeforeDead']}**.", ""] + rows = [[cell(r.get("situation", "")), f"`{r.get('optimizerAlive', '')}`", + f"`{r.get('activeStrategy', '')}`"] for r in fm.get("rows", [])] + parts += [table(rows, ["Situation (contract)", "`optimizerAlive`", "`activeStrategy`"]), ""] + return "\n".join(parts) + + def block_rules(self) -> str: + sr = self.d.get("x-etm-semantic-rules", {}) + parts = [] + if sr.get("description"): + parts += [cell(sr["description"]), ""] + rows = [] + for i, r in enumerate(sr.get("rules", []), 1): + rows.append([ + str(i), f"`{r['name']}`", f"`{r.get('method', '')}`", f"`{r.get('issue', '')}`", + f"`{r.get('fallbackCode', '')}`", + f"`{r['lastErrorCode']}`" if r.get("lastErrorCode") else "—", + REF["yes"] if r.get("countsAsFailure") else REF["no"], + ", ".join(f"`{n}`" for n in r.get("needs", [])) or "—", + REF["yes"] if r.get("schemaAlsoCatches") else REF["no"], + cell(r.get("description", "")), + ]) + parts += [table(rows, ["#", "Rule", "Method", "Outcome", "`fallbackCode`", "`lastErrorCode`", + "Counts as failure", "Also needs", "Schema catches it too", + "What is rejected (contract)"]), ""] + return "\n".join(parts) + + def block_tolerances(self) -> str: + tol = self.d.get("x-etm-tolerances", {}) + parts = [] + if tol.get("description"): + parts += [cell(tol["description"]), ""] + parts += [f"- {cell(x)}" for x in tol.get("items", [])] + return "\n".join(parts) + "\n" + + def block_closed_lists(self) -> str: + origin = {} + for it in self.d.get("x-etm-generated", {}).get("items", []): + m = re.fullmatch(r"/components/schemas/([^/]+)/(?:enum|const)", it.get("pointer", "")) + if m: + origin[m.group(1)] = it.get("from", "") + parts = [] + for n in self.blocks["closed_lists"]: + s = self.schemas[n] + parts += [f"#### `{n}` {{#{anchor(n)}}}", ""] + if s.get("description"): + parts += [cell(s["description"]), ""] + vals = s["enum"] if "enum" in s else [s["const"]] + parts += [" · ".join(f"`{v}`" for v in vals), ""] + if n in origin: + parts += [f"Regenerated from the engine code: `{origin[n]}`.", ""] + return "\n".join(parts) + + def block_contract(self) -> str: + info = self.d["info"] + lic = info.get("license", {}) + ext = sorted(k for k in self.d if k.startswith("x-etm-")) + ext_m = sorted({k for m in self.d["methods"] for k in m if k.startswith("x-etm-")}) + rows = [ + ["Title", cell(info.get("title", ""))], + ["Protocol version", f"`{info.get('version', '')}`"], + ["Format", f"OpenRPC `{self.d.get('openrpc', '')}`, JSON Schema in `components.schemas`"], + ["License", cell(lic.get("name", ""))], + ["File", f"[`{CONTRACT_NAME}`]({CONTRACT_NAME})"], + ["Size", f"{len(self.raw)} bytes"], + ["SHA-256", f"`{hashlib.sha256(self.raw).hexdigest()}`"], + ["Methods", " · ".join(f"`{m['name']}`" for m in self.d["methods"])], + ["Schemas", str(len(self.schemas))], + ["Semantic rules", str(len(self.d.get("x-etm-semantic-rules", {}).get("rules", [])))], + ["Extensions (document)", " · ".join(f"`{e}`" for e in ext)], + ["Extensions (methods)", " · ".join(f"`{e}`" for e in ext_m)], + ] + return table(rows, ["", "Contract"]) + "\n" + + +# ── Échafaudage des pages créées si absentes ────────────────────────────────── + +SCAFFOLD = { + "protocole": { + "fr": "Protocole", "de": "Protokoll", "en": "Protocol", + "keys": ["transport", "handshake", "context", "plan", "reason_codes", + "update_learned", "fallback", "rules", "tolerances", "closed_lists"], + }, + "contrat": { + "fr": "Contrat", "de": "Vertrag", "en": "Contract", + "keys": ["contract"], + }, +} + + +def page_path(docs: Path, stem: str, loc: str, default: str) -> Path: + suffix = "" if loc == default else f".{loc}" + return docs / OUT_DIR / f"{stem}{suffix}.md" + + +def ensure_pages(docs: Path, locales: list[str], default: str, check: bool) -> int: + rc = 0 + for stem, sc in SCAFFOLD.items(): + for loc in locales: + path = page_path(docs, stem, loc, default) + if path.exists(): + continue + if check: + print(f" MANQUANT : {path}") + rc = 1 + continue + blocks = "\n\n".join(f"\n" + for k in sc["keys"]) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(f"# {sc.get(loc, sc['en'])}\n\n{blocks}\n", encoding="utf-8") + print(f" CRÉÉ : {path}") + return rc + + +# ── Copie servie du contrat ─────────────────────────────────────────────────── + +def copy_contract(model: Model, docs: Path, check: bool) -> int: + dest = docs / OUT_DIR / CONTRACT_NAME + if dest.exists() and dest.read_bytes() == model.raw: + return 0 + if check: + print(f" PÉRIMÉ : {dest}") + return 1 + dest.parent.mkdir(parents=True, exist_ok=True) + dest.write_bytes(model.raw) + print(f" COPIÉ : {dest}") + return 0 + + +# ── Injection dans les marqueurs ────────────────────────────────────────────── + +def process(docs: Path, model: Model, check: bool) -> tuple[int, dict[Path, str]]: + """Retourne (rc, {page: texte publié}) pour le garde-fou.""" + changed: list[Path] = [] + published: dict[Path, str] = {} + for md in sorted(docs.rglob("*.md")): + text = md.read_text(encoding="utf-8") + if "__optimizer_" not in text: + continue + gen_parts: list[str] = [] + + def repl(m: re.Match, _md: Path = md) -> str: + key = m.group("key") + if not key.startswith("__optimizer_"): + return m.group(0) + gen = model.render(key) + if gen is None: + sys.exit(f"ERREUR : {_md}: clé inconnue « {key} »") + gen_parts.append(gen) + return f"{m.group(1)}\n\n{gen}\n{m.group(4)}" + + new = MARKER_RE.sub(repl, text) + published[md] = new if md.parent.name == OUT_DIR else "\n".join(gen_parts) + if new != text: + changed.append(md) + if not check: + md.write_text(new, encoding="utf-8") + if check: + if changed: + print("Doc optimiseur PAS à jour :", ", ".join(str(c) for c in changed)) + return 1, published + print("Doc optimiseur à jour.") + return 0, published + print(f"Régénéré : {len(changed)} page(s).") + for c in changed: + print(f" - {c.relative_to(docs)}") + return 0, published + + +# ── Garde-fou de contenu ────────────────────────────────────────────────────── + +def json_strings(o: object, ptr: str = ""): + if isinstance(o, dict): + for k, v in o.items(): + yield from json_strings(v, f"{ptr}/{k}") + elif isinstance(o, list): + for i, v in enumerate(o): + yield from json_strings(v, f"{ptr}/{i}") + elif isinstance(o, str): + yield ptr, o + + +def private_ips(text: str) -> list[str]: + out = [] + for m in IPV4_RE.finditer(text): + try: + ip = ipaddress.ip_address(m.group(0)) + except ValueError: + continue + if ip in DOC_NET: + continue + if any(ip in n for n in PRIVATE_NETS): + out.append(m.group(0)) + return out + + +def guard(model: Model, published: dict[Path, str]) -> int: + allowed = set() + for ptr, val in CONTRACT_IP_EXAMPLES: + if dict(json_strings(model.d)).get(ptr) == val: + allowed.add(val) + problems: list[str] = [] + notes: list[str] = [] + + for ptr, s in json_strings(model.d): + for ip in private_ips(s): + if (ptr, ip) in CONTRACT_IP_EXAMPLES: + notes.append(f"{CONTRACT_NAME}{ptr} : {ip} (exemple d'usage du contrat, admis)") + else: + problems.append(f"{CONTRACT_NAME}{ptr} : adresse privée {ip}") + for w in WORD_RES: + if w.search(s): + problems.append(f"{CONTRACT_NAME}{ptr} : mot interdit « {w.search(s).group(0)} »") + + for path, text in published.items(): + for ip in private_ips(text): + if ip in allowed: + notes.append(f"{path} : {ip} (rendu de l'exemple du contrat, admis)") + else: + problems.append(f"{path} : adresse privée {ip}") + for w in WORD_RES: + for m in w.finditer(text): + problems.append(f"{path} : mot interdit « {m.group(0)} »") + + for n in sorted(set(notes)): + print(f" garde-fou : {n}") + if problems: + print("ERREUR garde-fou — contenu à ne pas publier :", file=sys.stderr) + for p in problems: + print(f" - {p}", file=sys.stderr) + return 1 + print("Garde-fou : rien à signaler.") + return 0 + + +# ── Point d'entrée ──────────────────────────────────────────────────────────── + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--src", required=True, type=Path, + help=f"Clone du moteur, dossier contenant {CONTRACT_NAME}, ou le fichier lui-même") + ap.add_argument("--docs", default="docs", type=Path, help="Dossier docs/ (défaut : docs)") + ap.add_argument("--locales", default="fr,de,en", + help="Langues des pages échafaudées ; la première est la langue par défaut") + ap.add_argument("--check", action="store_true", help="CI : exit 1 si la doc n'est pas à jour") + a = ap.parse_args() + + locales = [x.strip() for x in a.locales.split(",") if x.strip()] + path = find_contract(a.src) + raw, d = load_contract(path) + model = Model(raw, d) + print(f"Contrat {d['info'].get('version', '?')} : {len(model.methods)} méthodes · " + f"{len(model.schemas)} schémas · " + f"{len(d.get('x-etm-semantic-rules', {}).get('rules', []))} règles sémantiques") + + rc_pages = ensure_pages(a.docs, locales, locales[0], a.check) + rc_copy = copy_contract(model, a.docs, a.check) + rc_proc, published = process(a.docs, model, a.check) + for md in sorted((a.docs / OUT_DIR).glob("*.md")): + published.setdefault(md, md.read_text(encoding="utf-8")) + rc_guard = guard(model, published) + return max(rc_pages, rc_copy, rc_proc, rc_guard) + + +if __name__ == "__main__": + raise SystemExit(main())