diff --git a/encoder_schema.py b/encoder_schema.py new file mode 100644 index 0000000..d2b7050 --- /dev/null +++ b/encoder_schema.py @@ -0,0 +1,268 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +encoder_schema.py — Sliding Pipeline v11 · Chantier C1 (Encoder structuré) +========================================================================== +Remplace l'agent Encoder par des appels chat/completions en structured +outputs stricts (response_format: json_schema, strict: true), slide par +slide. La sortie ne peut structurellement pas être invalide ; plus de +pagination PAUSE, plus d'extract_yaml par regex, échecs isolés par slide. + +Entrée : le plan Markdown annoté produit par le Designer + (blocs "SLIDE N — layout_name" suivis du contenu). +Sortie : dict {"titre_presentation": ..., "slides": [...]} + YAML. + +Usage module (depuis le facilitator) : + from encoder_schema import encode_plan, to_yaml + data, usage, errors = encode_plan(plan, api_key=API_KEY) + yaml_str = to_yaml(data) + +Usage CLI (test autonome sur le NAS, single-line) : + python3 encoder_schema.py plan_designer.txt > input.yaml + +Config .env : ENCODER_MODEL (défaut mistral-small-latest). +Python 3.9. Dépendances : requests, pyyaml (déjà dans le venv pipeline). +""" + +import json +import os +import re +import sys +import time + +import requests + +from schemas import get_schema, known_layouts + +API_URL = "https://api.mistral.ai/v1/chat/completions" +DEFAULT_MODEL = os.getenv("ENCODER_MODEL", "mistral-small-latest") +MAX_RETRY = 4 +TIMEOUT = 120 + +SYSTEM_PROMPT = ( + "Tu es un transcripteur de contenu de slide. On te donne la description " + "d'UNE slide et tu produis le JSON de son contenu, conforme au schéma " + "imposé. Règles absolues :\n" + "- Fidélité totale au contenu fourni : ne rien inventer, ne rien omettre " + "d'important ; condenser légèrement si nécessaire.\n" + "- Texte brut uniquement : aucun balisage Markdown (pas de **, __, #, `, " + "ni tirets de liste dans les valeurs).\n" + "- Français, ton affirmatif, guillemets typographiques évités.\n" + "- Champs optionnels sans contenu correspondant : null.\n" + "- Les nombres des champs numériques (start, end, x, y, numero) sont des " + "entiers, pas des chaînes.\n" + "- Le champ notes est réservé aux notes du présentateur explicitement " + "marquées (« Notes : ... ») ; sinon notes = null.\n" + "- Les lignes de justification du Designer (commençant par → ou " + "expliquant le choix du layout) ne sont NI du contenu NI des notes : " + "ignore-les." +) + +# "SLIDE 12 — layout_name" (tiret cadratin, demi-cadratin ou simple) +SLIDE_RE = re.compile( + r"^\s*SLIDE\s+(\d+)\s*[—–\-]+\s*([a-z][a-z0-9_]*)\s*$", + re.MULTILINE) + +MD_PATTERNS = [ + (re.compile(r"\*\*(.+?)\*\*"), r"\1"), + (re.compile(r"__(.+?)__"), r"\1"), + (re.compile(r"`([^`]*)`"), r"\1"), +] + + +# ── Découpage du plan ──────────────────────────────────────────────────────── + +def split_plan(plan_md): + """Découpe le plan Designer en slides. + Retourne (segments, errors) ; segment = {position, layout, content}.""" + matches = list(SLIDE_RE.finditer(plan_md)) + segments, errors = [], [] + if not matches: + return [], ["Aucune ligne 'SLIDE N — layout' détectée dans le plan."] + valid = set(known_layouts()) + for i, m in enumerate(matches): + end = matches[i + 1].start() if i + 1 < len(matches) else len(plan_md) + layout = m.group(2) + seg = { + "position": int(m.group(1)), + "layout": layout, + "content": plan_md[m.start():end].strip(), + } + if layout not in valid: + errors.append("Slide %d : layout '%s' sans schéma (connus : %s)" + % (seg["position"], layout, + ", ".join(sorted(valid)))) + continue + segments.append(seg) + return segments, errors + + +# ── Appel API ──────────────────────────────────────────────────────────────── + +def _post_with_retry(payload, api_key): + headers = {"Authorization": "Bearer %s" % api_key, + "Content-Type": "application/json"} + last_err = None + for attempt in range(1, MAX_RETRY + 1): + try: + resp = requests.post(API_URL, headers=headers, json=payload, + timeout=TIMEOUT) + if resp.status_code == 429 or resp.status_code >= 500: + wait = 2 ** attempt * 3 + time.sleep(wait) + last_err = "HTTP %d" % resp.status_code + continue + resp.raise_for_status() + return resp.json() + except requests.RequestException as e: + last_err = e + time.sleep(2 ** attempt * 2) + raise RuntimeError("API Mistral injoignable après %d tentatives : %s" + % (MAX_RETRY, last_err)) + + +def encode_slide(content, layout, api_key, model=None): + """Encode UNE slide. Retourne (dict_contenu, usage_dict). + Lève RuntimeError/ValueError en cas d'échec (isolé par l'appelant).""" + schema = get_schema(layout) + payload = { + "model": model or DEFAULT_MODEL, + "temperature": 0, + "messages": [ + {"role": "system", "content": SYSTEM_PROMPT}, + {"role": "user", "content": + "Layout : %s\n\nContenu de la slide :\n%s" + % (layout, content)}, + ], + "response_format": { + "type": "json_schema", + "json_schema": { + "name": "slide_%s" % layout, + "strict": True, + "schema": schema, + }, + }, + } + data = _post_with_retry(payload, api_key) + raw = data["choices"][0]["message"]["content"] + usage = data.get("usage", {}) or {} + return json.loads(raw), usage + + +# ── Post-traitement ────────────────────────────────────────────────────────── + +def drop_nulls(node): + """Retire récursivement les clés à None (les optionnels non remplis).""" + if isinstance(node, dict): + return {k: drop_nulls(v) for k, v in node.items() if v is not None} + if isinstance(node, list): + return [drop_nulls(v) for v in node] + return node + + +def strip_md(node): + """Défense en profondeur : retire le gras/italique/backticks résiduels. + (La règle définitive côté moteur arrive au chantier C3.)""" + if isinstance(node, dict): + return {k: strip_md(v) for k, v in node.items()} + if isinstance(node, list): + return [strip_md(v) for v in node] + if isinstance(node, str): + out = node + for pat, rep in MD_PATTERNS: + out = pat.sub(rep, out) + return out.strip() + return node + + +def _accumulate(total, usage): + for k in ("prompt_tokens", "completion_tokens", "total_tokens"): + total[k] = total.get(k, 0) + int(usage.get(k, 0) or 0) + return total + + +# ── Orchestration ──────────────────────────────────────────────────────────── + +def encode_plan(plan_md, api_key, model=None, only_positions=None, + progress=None): + """Encode tout le plan slide par slide. + only_positions : iterable de positions à encoder (mode ciblé), None = tout. + progress : callable(str) pour l'affichage (info du facilitator). + Retourne (data, usage, errors) : + data = {"titre_presentation": str|None, "slides": [...]} + usage = tokens cumulés {"prompt_tokens","completion_tokens","total_tokens"} + errors = liste de messages (slides en échec — absentes de data). + """ + def say(msg): + if progress: + progress(msg) + + segments, errors = split_plan(plan_md) + if only_positions is not None: + wanted = set(int(p) for p in only_positions) + segments = [s for s in segments if s["position"] in wanted] + slides, usage = [], {} + titre_presentation = None + for seg in segments: + say("Slide %d (%s)..." % (seg["position"], seg["layout"])) + try: + content, u = encode_slide(seg["content"], seg["layout"], + api_key, model) + except (RuntimeError, ValueError, KeyError, + json.JSONDecodeError) as e: + errors.append("Slide %d (%s) : %s" + % (seg["position"], seg["layout"], e)) + continue + _accumulate(usage, u) + content = strip_md(drop_nulls(content)) + if seg["layout"] == "cover_split" and titre_presentation is None: + titre_presentation = content.get("titre") + slide = {"position": seg["position"], "layout": seg["layout"]} + slide.update(content) + slides.append(slide) + slides.sort(key=lambda s: s["position"]) + data = {"slides": slides} + if titre_presentation: + data = {"titre_presentation": titre_presentation, "slides": slides} + return data, usage, errors + + +def to_yaml(data): + import yaml + return yaml.safe_dump(data, allow_unicode=True, sort_keys=False, + default_flow_style=False, width=100) + + +# ── CLI de test autonome ───────────────────────────────────────────────────── + +def _main(): + from pathlib import Path + try: + from dotenv import load_dotenv + load_dotenv() + except ImportError: + pass + api_key = os.getenv("MISTRAL_API_KEY") + if not api_key: + sys.stderr.write("MISTRAL_API_KEY manquant (.env)\n") + sys.exit(1) + if len(sys.argv) < 2: + sys.stderr.write("Usage : python3 encoder_schema.py " + "[positions ex 3,5,7]\n") + sys.exit(1) + plan = Path(sys.argv[1]).read_text(encoding="utf-8") + only = None + if len(sys.argv) > 2: + only = [int(p) for p in sys.argv[2].split(",") if p.strip()] + data, usage, errors = encode_plan( + plan, api_key, only_positions=only, + progress=lambda m: sys.stderr.write(" %s\n" % m)) + sys.stderr.write("Tokens : %s\n" % json.dumps(usage)) + for e in errors: + sys.stderr.write(" ! %s\n" % e) + sys.stdout.write(to_yaml(data)) + sys.exit(2 if errors else 0) + + +if __name__ == "__main__": + _main() diff --git a/patch_facilitator_c1.py b/patch_facilitator_c1.py new file mode 100644 index 0000000..21fe259 --- /dev/null +++ b/patch_facilitator_c1.py @@ -0,0 +1,182 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +patch_facilitator_c1.py — Chantier C1 (Encoder structuré) +========================================================= +Patch strict de facilitator_v9.py : + P1. Config : ajoute ENCODER_MODE (.env, défaut 'agent' — rien ne change + tant que ENCODER_MODE=schema n'est pas posé). + P2. Insère run_encoder_schema() avant la section FREE DESIGNER + (même contrat de retour que run_encoder, fallback agent intégré). + P3. Dispatch dans run_full_pipeline_pass (pipeline + révisions). + P4. Dispatch dans le flux standard (fallback agent en mode express : + le plan Narrator n'y est pas annoté SLIDE N — layout). + +Usage (dossier du pipeline, single-line) : + python3 patch_facilitator_c1.py +Vérifie chaque ancre (présence + unicité), écrit facilitator_v9.py.bak, +compile le résultat. Idempotent : refuse de patcher deux fois. +""" + +import py_compile +import shutil +import sys +from pathlib import Path + +TARGET = Path("facilitator_v9.py") + +FUNC = ''' +def run_encoder_schema(plan: str, layouts: dict, proj: "Project"): + """Encoder structuré (chantier C1) : chat/completions + json_schema + strict Mistral, slide par slide. Même contrat de retour que + run_encoder : (yaml_str, data, path, attempts) — attempts = nb de + slides en échec. Fallback automatique sur l'Encoder agent si le + module manque ou si aucune slide n'est encodée.""" + section("ÉTAPE 3 — THE ENCODER (structured outputs)") + try: + import encoder_schema as enc + except ImportError as e: + warn(f"encoder_schema.py indisponible ({e}) — bascule mode agent.") + return run_encoder(AgentSession(ENCODER_ID), plan, layouts, proj) + try: + import schemas as sch + for issue in (sch.verify_against_layouts(layouts) if layouts else []): + warn(f"Schéma vs layouts_v2 : {issue}") + except ImportError: + warn("schemas.py absent — vérification de cohérence sautée.") + info(f"Encodage slide par slide ({enc.DEFAULT_MODEL}, temp 0)...") + data, usage, errors = enc.encode_plan(plan, API_KEY, progress=info) + nb = len(data.get("slides", [])) + ok(f"{nb} slides encodées — tokens : {usage.get('total_tokens', 0)} " + f"(prompt {usage.get('prompt_tokens', 0)} / " + f"completion {usage.get('completion_tokens', 0)})") + for e in errors: + warn(e) + if not nb: + warn("Aucune slide encodée — bascule mode agent.") + return run_encoder(AgentSession(ENCODER_ID), plan, layouts, proj) + yaml_str = enc.to_yaml(data) + is_valid, message, _ = validate_yaml(yaml_str, layouts) + if is_valid: + ok(f"YAML valide : {message}") + else: + warn(f"Validation : {message}") + if errors: + print("\\n [1] Continuer sans les slides en échec") + print(" [2] Basculer sur l'Encoder agent (deck complet)") + print(" [0] Abandonner") + choix = ask("Votre choix :") + if choix == "2": + return run_encoder(AgentSession(ENCODER_ID), plan, layouts, proj) + if choix == "0": + return None, None, None, len(errors) + path = save_text(yaml_str, proj.out("input", "yaml")) + return yaml_str, data, path, len(errors) + +''' + +PATCHES = [ + # P1 — config + ( + 'ENCODER_ID = os.getenv("ENCODER_AGENT_ID")\n', + 'ENCODER_ID = os.getenv("ENCODER_AGENT_ID")\n' + 'ENCODER_MODE = os.getenv("ENCODER_MODE", "agent")' + ' # agent | schema (C1)\n', + ), + # P3 — dispatch pipeline pass (révisions comprises) + ( + ' encoder = AgentSession(ENCODER_ID)\n' + ' yaml_str, yaml_data, yaml_path, tries = run_encoder(\n' + ' encoder, plan, layouts, proj, scope_encoder)\n', + ' if ENCODER_MODE == "schema":\n' + ' yaml_str, yaml_data, yaml_path, tries = ' + 'run_encoder_schema(\n' + ' plan, layouts, proj)\n' + ' else:\n' + ' encoder = AgentSession(ENCODER_ID)\n' + ' yaml_str, yaml_data, yaml_path, tries = run_encoder(\n' + ' encoder, plan, layouts, proj, scope_encoder)\n', + ), + # P4 — dispatch flux standard (express → agent) + ( + ' encoder = AgentSession(ENCODER_ID)\n' + ' yaml_str, yaml_data, yaml_path, tries = run_encoder(\n' + ' encoder, plan, layouts, proj)\n' + ' manifest.step("encoder", tentatives_correction=tries)\n' + ' if yaml_str is None:\n' + ' if express:\n', + ' if ENCODER_MODE == "schema" and not express:\n' + ' yaml_str, yaml_data, yaml_path, tries = ' + 'run_encoder_schema(\n' + ' plan, layouts, proj)\n' + ' else:\n' + ' if ENCODER_MODE == "schema" and express:\n' + ' info("Mode express → Encoder agent ' + '(plan non annoté).")\n' + ' encoder = AgentSession(ENCODER_ID)\n' + ' yaml_str, yaml_data, yaml_path, tries = run_encoder(\n' + ' encoder, plan, layouts, proj)\n' + ' manifest.step("encoder", tentatives_correction=tries)\n' + ' if yaml_str is None:\n' + ' if express:\n', + ), +] + +MARKER = "run_encoder_schema" +ANCHOR_SECTION = "# FLUX LIBRE — THE FREE DESIGNER" + + +def fail(msg): + print(" ! %s" % msg) + sys.exit(1) + + +def main(): + if not TARGET.exists(): + fail("%s introuvable — lancer depuis le dossier du pipeline." + % TARGET) + for dep in ("schemas.py", "encoder_schema.py"): + if not Path(dep).exists(): + fail("%s manquant à côté du facilitator." % dep) + content = TARGET.read_text(encoding="utf-8") + if MARKER in content: + fail("Déjà patché (run_encoder_schema présent) — rien à faire.") + + # Vérification de toutes les ancres AVANT toute écriture + for i, (old, _) in enumerate(PATCHES, 1): + n = content.count(old) + if n == 0: + fail("Ancre du patch %d introuvable — facilitator inattendu." + % i) + if n > 1: + fail("Ancre du patch %d non unique (%d occurrences)." % (i, n)) + if content.count(ANCHOR_SECTION) != 1: + fail("Ancre de section FREE DESIGNER introuvable ou non unique.") + + shutil.copy2(TARGET, TARGET.with_suffix(".py.bak")) + print(" + Sauvegarde : %s.bak" % TARGET) + + for old, new in PATCHES: + content = content.replace(old, new) + + # P2 — insertion de la fonction avant le séparateur de la section + lines = content.split("\n") + idx = next(i for i, l in enumerate(lines) if ANCHOR_SECTION in l) + ins = idx - 1 if lines[idx - 1].startswith("# ───") else idx + lines[ins:ins] = FUNC.split("\n") + content = "\n".join(lines) + + TARGET.write_text(content, encoding="utf-8") + print(" + 4 patchs appliqués.") + try: + py_compile.compile(str(TARGET), doraise=True) + print(" + Compilation OK.") + except py_compile.PyCompileError as e: + shutil.copy2(TARGET.with_suffix(".py.bak"), TARGET) + fail("Erreur de compilation — fichier restauré :\n%s" % e) + print("\n Activer : ajouter ENCODER_MODE=schema dans .env") + print(" Retour arrière : ENCODER_MODE=agent (ou supprimer la ligne).") + + +if __name__ == "__main__": + main() diff --git a/schemas.py b/schemas.py new file mode 100644 index 0000000..35b6b35 --- /dev/null +++ b/schemas.py @@ -0,0 +1,326 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +schemas.py — Sliding Pipeline v11 · Chantier C1 (Encoder structuré) +=================================================================== +JSON Schemas stricts par layout pour les structured outputs Mistral +(response_format: json_schema, strict: true). + +Principes : +- Profil "strict" : additionalProperties=false partout, TOUS les champs + en required ; les champs optionnels sont nullables (type: [X, "null"]). + Les null sont retirés en post-traitement (encoder_schema.drop_nulls). +- Pas de minItems/maxItems dans les schémas (support incertain en mode + strict) : les bornes de densité restent des recommandations de prompt, + le fitter (chantier C3) gère la dégradation gracieuse. +- Les types par champ sont déclarés ici (layouts_v2.yaml ne porte que les + noms). verify_against_layouts() contrôle la cohérence au chargement. + Cible R13 : migrer ces types dans layouts_v2.yaml (bloc schema:) pour + une source unique — ce module deviendra alors un pur générateur. +- org_chart : récursion dépliée sur 3 niveaux (root → children → + grandchildren), profondeur exacte gérée par render_engine_v2. +- matrix_2x2 : axis_x/axis_y/quadrants requis dans le schéma d'encodage + (bien qu'optionnels côté moteur) — un libellé vaut mieux qu'un trou. + +Usage : + from schemas import get_schema, verify_against_layouts, LAYOUT_SCHEMAS + schema = get_schema("kpi_grid") # → dict json_schema strict +Python 3.9 compatible. Stdlib uniquement. +""" + +import copy + +# ── Helpers de construction ────────────────────────────────────────────────── + +S = {"type": "string"} +S_OPT = {"type": ["string", "null"]} +I = {"type": "integer"} +I_OPT = {"type": ["integer", "null"]} +B_OPT = {"type": ["boolean", "null"]} + + +def arr(item): + return {"type": "array", "items": item} + + +def obj(props): + """Objet strict : toutes les clés en required, aucune clé libre.""" + return { + "type": "object", + "properties": props, + "required": list(props.keys()), + "additionalProperties": False, + } + + +def _bullets(): + return arr(obj({"texte": S})) + + +def _org_leaf(): + return obj({"label": S}) + + +def _org_mid(): + return obj({"label": S, "children": arr(_org_leaf())}) + + +def _org_root(): + return obj({"label": S, "children": arr(_org_mid())}) + + +# ── Schémas de contenu par layout (hors clés layout/position, injectées) ──── + +LAYOUT_SCHEMAS = { + "cover_split": obj({ + "titre": S, + "sous_titre": S_OPT, + }), + "executive_summary": obj({ + "titre": S, + "situation": S, + "complication": S, + "resolution": S, + }), + "section_divider": obj({ + "titre": S, + }), + "big_stat": obj({ + "titre": S, + "valeur": S, + "description": S_OPT, + "source": S_OPT, + }), + "kpi_grid": obj({ + "titre": S, + "items": arr(obj({"label": S, "valeur": S, "description": S_OPT})), + }), + "two_cols_text": obj({ + "titre": S, + "left": obj({"titre": S, "bullets": _bullets()}), + "right": obj({"titre": S, "bullets": _bullets()}), + }), + "comparison_table": obj({ + "titre": S, + "headers": arr(S), + "rows": arr(obj({"label": S, "values": arr(S)})), + }), + "key_message": obj({ + "message": S, + "detail": S_OPT, + }), + "circular_diagram": obj({ + "titre": S, + "segments": arr(obj({"label": S, "description": S_OPT})), + }), + "default_bullets": obj({ + "titre": S, + "bullets": arr(obj({"texte": S, "niveau": I_OPT})), + }), + "numbered_steps": obj({ + "titre": S, + "steps": arr(obj({"numero": I_OPT, "titre": S, + "description": S_OPT})), + }), + "process_arrow": obj({ + "titre": S, + "steps": arr(obj({"titre": S, "description": S_OPT})), + }), + "phases_timeline": obj({ + "titre": S, + "phases": arr(obj({"label": S, "periode": S})), + }), + "gantt_timeline": obj({ + "titre": S, + "periods": arr(S), + "workstreams": arr(obj({ + "label": S, + "tasks": arr(obj({"label": S, "start": I, "end": I})), + })), + }), + "recommendation_card": obj({ + "numero": I_OPT, + "titre": S, + "headline": S, + "cta": S_OPT, + "bullets": _bullets(), + }), + "end_slide": obj({ + "message": S, + }), + "from_to_pairs": obj({ + "titre": S, + "label_from": S_OPT, + "label_to": S_OPT, + "pairs": arr(obj({"from": S, "to": S})), + }), + "yearly_timeline": obj({ + "titre": S, + "milestones": arr(obj({"annee": S, "label": S, "actif": B_OPT})), + }), + "raci_table": obj({ + "titre": S, + "roles": arr(S), + "tasks": arr(obj({"label": S, "raci": arr(S)})), + }), + "org_chart": obj({ + "titre": S, + "root": _org_root(), + }), + "matrix_2x2": obj({ + "titre": S, + "axis_x": obj({"label": S, "low": S, "high": S}), + "axis_y": obj({"label": S, "low": S, "high": S}), + "quadrants": obj({"top_left": S, "top_right": S, + "bottom_left": S, "bottom_right": S}), + "items": arr(obj({"label": S, "x": I, "y": I})), + }), + "image_split": obj({ + "titre": S, + "image": S, + "bullets": _bullets(), + "side": {"type": ["string", "null"], "enum": ["left", "right", + None]}, + "legende": S_OPT, + }), + "image_full": obj({ + "titre": S, + "image": S, + "sous_titre": S_OPT, + }), + "bar_chart": obj({ + "titre": S, + "categories": arr(S), + "series": arr(obj({"label": S, + "values": arr({"type": "number"})})), + "unite": S_OPT, + "source": S_OPT, + "horizontal": B_OPT, + }), + "line_chart": obj({ + "titre": S, + "points_x": arr(S), + "series": arr(obj({"label": S, + "values": arr({"type": "number"})})), + "unite": S_OPT, + "source": S_OPT, + }), + "donut_split": obj({ + "titre": S, + "segments": arr(obj({"label": S, "valeur": {"type": "number"}})), + "valeur_centrale": S_OPT, + "source": S_OPT, + }), + "waterfall": obj({ + "titre": S, + "depart": obj({"label": S, "valeur": {"type": "number"}}), + "marches": arr(obj({"label": S, "delta": {"type": "number"}})), + "arrivee": obj({"label": S, "valeur": {"type": "number"}}), + "unite": S_OPT, + "source": S_OPT, + }), + "heatmap_table": obj({ + "titre": S, + "headers": arr(S), + "rows": arr(obj({"label": S, "scores": arr(I)})), + "legende": S_OPT, + }), + "funnel": obj({ + "titre": S, + "etapes": arr(obj({"label": S, "valeur": {"type": "number"}, + "description": S_OPT})), + "source": S_OPT, + }), + "agenda": obj({ + "titre": S, + "sections": arr(obj({"numero": I_OPT, "label": S, + "duree": S_OPT, "actif": B_OPT})), + }), + "pyramid": obj({ + "titre": S, + "niveaux": arr(obj({"label": S, "description": S_OPT})), + }), +} + +# Champ transverse (chantier C3) : notes du présentateur, toujours +# optionnel, transcrit par l'Encoder si le plan contient « Notes : ... » +# et écrit dans la zone notes PowerPoint par le moteur. +for _sch in LAYOUT_SCHEMAS.values(): + _sch["properties"]["notes"] = {"type": ["string", "null"]} + _sch["required"].append("notes") + + +# ── API publique ───────────────────────────────────────────────────────────── + +def get_schema(layout): + """Schéma JSON strict du contenu d'une slide pour le layout donné. + Copie profonde : l'appelant peut annoter sans polluer le référentiel.""" + if layout not in LAYOUT_SCHEMAS: + raise KeyError("Layout sans schéma : %s" % layout) + return copy.deepcopy(LAYOUT_SCHEMAS[layout]) + + +def known_layouts(): + return sorted(LAYOUT_SCHEMAS.keys()) + + +def _nullable(field_schema): + t = field_schema.get("type") + return isinstance(t, list) and "null" in t + + +def verify_against_layouts(layouts): + """Croise ce module avec layouts_v2.yaml (dict 'layouts' chargé). + Retourne une liste de divergences (vide = cohérent). Non bloquant : + l'appelant décide (le facilitator affiche en warning).""" + issues = [] + for name, cfg in layouts.items(): + if name not in LAYOUT_SCHEMAS: + issues.append("layouts_v2 déclare '%s' sans schéma ici" % name) + continue + declared = set(cfg.get("champs", [])) - {"notes"} + here = set(LAYOUT_SCHEMAS[name]["properties"].keys()) - {"notes"} + if declared != here: + missing = declared - here + extra = here - declared + if missing: + issues.append("%s : champs layouts_v2 absents du schéma : %s" + % (name, ", ".join(sorted(missing)))) + if extra: + issues.append("%s : champs du schéma absents de layouts_v2 : %s" + % (name, ", ".join(sorted(extra)))) + # Un champ requis côté layouts_v2 ne doit pas être nullable ici + for req in cfg.get("champs_requis", []): + prop = LAYOUT_SCHEMAS[name]["properties"].get(req) + if prop is not None and _nullable(prop): + issues.append("%s : champ requis '%s' nullable dans le schéma" + % (name, req)) + for name in LAYOUT_SCHEMAS: + if name not in layouts: + issues.append("Schéma '%s' sans entrée layouts_v2" % name) + return issues + + +if __name__ == "__main__": + import json + import sys + if len(sys.argv) > 1: + print(json.dumps(get_schema(sys.argv[1]), indent=2, + ensure_ascii=False)) + else: + print("Layouts couverts (%d) : %s" + % (len(LAYOUT_SCHEMAS), ", ".join(known_layouts()))) + try: + import yaml + with open("layouts_v2.yaml", encoding="utf-8") as f: + layouts = yaml.safe_load(f)["layouts"] + issues = verify_against_layouts(layouts) + if issues: + print("\nDivergences avec layouts_v2.yaml :") + for i in issues: + print(" ! %s" % i) + sys.exit(1) + print("Cohérence layouts_v2.yaml : OK (%d layouts)" + % len(layouts)) + except FileNotFoundError: + print("layouts_v2.yaml introuvable ici — vérification sautée.")