sliding-automation/encoder_schema.py

#!/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 <plan.txt> "
                         "[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()