327 lines
11 KiB
Python
327 lines
11 KiB
Python
#!/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.")
|