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"
    "- LANGUE : conserve strictement la langue du contenu fourni (un plan "
    "anglais donne des champs 100 % anglais). Ces instructions sont en "
    "français mais ce n'est JAMAIS la langue du livrable. Ne traduis "
    "rien.\n"
    "- 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."
)

# FORMAT RÉEL du Designer (prompt_the_designer_v3) : le Markdown du
# Narrator repris intégralement, avec une ligne "@layout: nom" ajoutée
# sous chaque titre de slide, et des "@note:" facultatives.
#   ### 3. The CEO's 3 priorities
#   @layout: two_cols_text
#   @note: ...
#   <contenu>
# On découpe donc sur les lignes @layout, et on déduit la position du
# titre qui précède (### 3. …) ou, à défaut, de l'ordre d'apparition.
LAYOUT_RE = re.compile(r"^[ \t]*@layout[ \t]*:[ \t]*([a-z][a-z0-9_]*)[ \t]*$",
                       re.MULTILINE)
# Titre de slide : "### 12. Titre" ou "### [OPT-EXEC] 5. Titre"
TITLE_NUM_RE = re.compile(
    r"^[ \t]*#{1,6}[ \t]*(?:\[[^\]]+\][ \t]*)?(\d+)[.)][ \t]*",
    re.MULTILINE)
# Compat ascendante : ancien format "SLIDE 12 — layout_name"
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 _position_before(plan_md, idx, fallback):
    """Numéro de la dernière ligne de titre '### N.' avant idx."""
    last = None
    for m in TITLE_NUM_RE.finditer(plan_md, 0, idx):
        last = m
    if last:
        try:
            return int(last.group(1))
        except (TypeError, ValueError):
            pass
    return fallback


def split_plan(plan_md):
    """Découpe le plan annoté par le Designer en slides.
    Format principal : lignes '@layout: nom'. Compat : 'SLIDE N — nom'.
    Retourne (segments, errors) ; segment = {position, layout, content}."""
    segments, errors = [], []
    valid = set(known_layouts())

    matches = list(LAYOUT_RE.finditer(plan_md))
    if matches:
        for i, m in enumerate(matches):
            start = m.start()
            end = (matches[i + 1].start() if i + 1 < len(matches)
                   else len(plan_md))
            layout = m.group(1)
            pos = _position_before(plan_md, start, i + 1)
            seg = {"position": pos, "layout": layout,
                   "content": plan_md[start:end].strip()}
            if layout not in valid:
                errors.append("Slide %d : layout '%s' sans schéma (connus"
                              " : %s)" % (pos, layout,
                                          ", ".join(sorted(valid))))
                continue
            segments.append(seg)
        # Positions dupliquées (titre absent) → renumérotation d'ordre
        seen = [s["position"] for s in segments]
        if len(set(seen)) != len(seen):
            for i, s in enumerate(segments, start=1):
                s["position"] = i
        return segments, errors

    # Fallback : ancien format explicite
    matches = list(SLIDE_RE.finditer(plan_md))
    if not matches:
        return [], ["Aucune annotation '@layout: nom' (ni 'SLIDE N — "
                    "layout') détectée dans le plan du Designer."]
    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, lang=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":
                ("LANGUE DU DECK : %s — tous les champs texte restent "
                 "dans cette langue.\n" % lang if lang else "")
                + "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


def detect_lang(text):
    """Heuristique stopwords : 'anglais' / 'français' / None (mixte)."""
    t = " %s " % re.sub(r"\s+", " ", text.lower())
    en = sum(t.count(w) for w in (" the ", " and ", " of ", " to ",
                                  " is ", " for ", " with "))
    fr = sum(t.count(w) for w in (" le ", " la ", " les ", " des ",
                                  " est ", " pour ", " avec ", " une "))
    if en > fr * 1.5:
        return "anglais"
    if fr > en * 1.5:
        return "français"
    return None


TITLE_PREFIX_RE = re.compile(r"^\s*(?:\[[^\]]+\]\s*)?\d+[.)]\s*")


def strip_title_number(content):
    """Retire la numérotation Markdown résiduelle du titre ('13. X' → 'X')."""
    t = content.get("titre")
    if isinstance(t, str):
        content["titre"] = TITLE_PREFIX_RE.sub("", t)
    return content


def trim_trailing_meta(segments):
    """Le DERNIER segment court jusqu'à la fin du texte : il peut embarquer
    les commentaires de conclusion du Designer (après un '---'). On coupe."""
    if segments:
        parts = segments[-1]["content"].split("\n---")
        if parts[0].count("\n") >= 1:
            segments[-1]["content"] = parts[0].rstrip()
    return segments


# ── 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": [...]}
    La langue du plan est détectée et imposée à chaque slide (anti-dérive).
      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)
    segments = trim_trailing_meta(segments)
    lang = detect_lang(plan_md)
    if lang:
        say("Langue du deck : %s (imposée à chaque slide)" % lang)
    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, lang=lang)
        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))
        content = strip_title_number(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()