Master/sliding-automation
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
fcd2180c6bc283a473b8e490a768310b271042fd
sliding-automation/facilitator_v9.py
T

1510 lines
64 KiB
Python
Raw Normal View History

feat: Narrator plan compact - modes f:/d: persistants, /coller multiligne
2026-06-21 01:15:39 +02:00
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
facilitator_v9.py — Sliding Pipeline v11 · Pernod Ricard
========================================================
Orchestre les agents Mistral + render_engine_v2, avec une
organisation par projet et un Narrator conversationnel.
Changements v6 vs v5 :
- Narrator en mode conversation libre (plus de menu numéroté)
- Commandes /slash dans la boucle Narrator :
/lire → rescanner inputs/ et injecter les nouveaux docs
/formalise → demander au Narrator de structurer le Markdown
/valider → valider le Markdown formalisé et passer au Designer
/sauvegarder → sauvegarder l'état actuel de la conversation
/afficher → afficher le dernier message du Narrator en entier
/aide → revoir les commandes disponibles
/quitter → abandonner et revenir au menu projet
- Chargement incrémental des docs : /lire charge uniquement
les nouveaux fichiers pas encore injectés
- Transition naturelle : le Markdown formalisé est auto-détecté
ou demandé explicitement via /formalise
Tout le reste (Designer, Encoder, Render, projet, manifest) est
identique au facilitator_v3.
Variables .env : identiques à v3. NARRATOR_AGENT_ID pointe vers
le nouvel agent configuré avec prompt_the_narrator_v3.md.
"""
import argparse
import json
import os
import re
import subprocess
import sys
import textwrap
import time
from datetime import datetime
from pathlib import Path
from typing import Optional
import requests
import yaml
from dotenv import load_dotenv
load_dotenv()
# ─────────────────────────────────────────────────────────────────────────────
# CONFIGURATION (identique v3)
# ─────────────────────────────────────────────────────────────────────────────
API_KEY = os.getenv("MISTRAL_API_KEY")
NARRATOR_ID = os.getenv("NARRATOR_AGENT_ID")
DESIGNER_ID = os.getenv("DESIGNER_AGENT_ID")
ENCODER_ID = os.getenv("ENCODER_AGENT_ID")
FREE_DESIGNER_ID = os.getenv("FREE_DESIGNER_AGENT_ID")
PROJECTS_DIR = Path(os.getenv("PROJECTS_DIR", "./projets"))
RENDER_ENGINE_PATH = os.getenv("RENDER_ENGINE_PATH", "render_engine_v2.py")
THEME_PATH = os.getenv("THEME_PATH", "theme_v2.yaml")
COMPONENTS_PATH = os.getenv("COMPONENTS_PATH", "components_v2.yaml")
LAYOUTS_PATH = os.getenv("LAYOUTS_PATH", "layouts_v2.yaml")
CONTEXT_MAX_CHARS = int(os.getenv("CONTEXT_MAX_CHARS", "12000"))
TRILIUM_API_URL = os.getenv("TRILIUM_API_URL", "")
TRILIUM_API_KEY = os.getenv("TRILIUM_API_KEY", "")
TRILIUM_PROJET = os.getenv("TRILIUM_PROJET", "SlidingAutomation")
HEADERS = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json",
}
VERSION = "11.0"
MAX_PAGES = 10
MAX_RETRY = 3
PROJECTS_DIR.mkdir(parents=True, exist_ok=True)
# Commandes slash reconnues
SLASH_CMDS = {
"/lire": "Charger les documents de inputs/ (incrémental)",
"/coller": "Coller un texte multiligne (termine par une ligne '.')",
"/formalise": "Transformer le plan compact validé en Markdown pour le Designer",
"/valider": "Valider le Markdown formalisé → passer au Designer",
"/plan": "Réafficher le dernier plan compact en entier",
"/aide": "Afficher cette aide",
"/quitter": "Abandonner et revenir au menu projet",
}
# Préfixes de mode du Narrator (plan compact)
MODE_FLUX = "flux"
MODE_DEEP = "deep"
# ─────────────────────────────────────────────────────────────────────────────
# AFFICHAGE
# ─────────────────────────────────────────────────────────────────────────────
def banner():
print("\n" + "=" * 62)
print(f" SLIDING PIPELINE v{VERSION} — Pernod Ricard (design system v2)")
print(f" {datetime.now().strftime('%d/%m/%Y %H:%M')}")
print("=" * 62 + "\n")
def section(title): print(f"\n{'-'*62}\n {title}\n{'-'*62}\n")
def info(msg): print(f" i {msg}")
def ok(msg): print(f" + {msg}")
def warn(msg): print(f" ! {msg}")
def ask(prompt: str) -> str:
try:
return input(f"\n {prompt} ").strip()
except (EOFError, KeyboardInterrupt):
print("\n Session interrompue.")
sys.exit(0)
def ask_multiline(end_marker: str = ".") -> str:
"""Lit plusieurs lignes jusqu'à une ligne contenant uniquement end_marker.
Évite les déclenchements de commandes parasites lors d'un copier-coller."""
print(f"\n Mode collage — colle ton texte, puis une ligne avec '{end_marker}' "
f"seul pour terminer :")
lines = []
while True:
try:
line = input()
except (EOFError, KeyboardInterrupt):
break
if line.strip() == end_marker:
break
lines.append(line)
return "\n".join(lines).strip()
def display_narrator(text: str, max_lines: int = 80):
"""Affiche la réponse du Narrator avec indentation et troncature douce."""
lines = text.splitlines()
if len(lines) <= max_lines:
print("\n" + textwrap.indent(text, " "))
else:
print("\n" + textwrap.indent("\n".join(lines[:max_lines]), " "))
print(f"\n ... [{len(lines) - max_lines} lignes supplémentaires"
f" — /afficher pour tout voir]")
def display_help():
print("\n PRÉFIXES DE MODE (au début de ton message) :")
print(" f: FLUX — structure : ordre, ajout, suppression de slides")
print(" → le Narrator réaffiche le plan compact complet")
print(" d: DEEP DIVE — contenu d'un point précis")
print(" → réponse ciblée, sans réafficher le plan")
print(" (sans préfixe : reste dans le dernier mode utilisé)")
print("\n COMMANDES :\n")
for cmd, desc in SLASH_CMDS.items():
print(f" {cmd:<12} {desc}")
print()
def display_plan(text: str):
slides = [l for l in text.splitlines() if l.strip().startswith("SLIDE")]
if slides:
print(" Résumé du plan :\n")
for s in slides:
print(f" {s.strip()}")
print(f"\n ({len(slides)} slides au total)")
else:
display_narrator(text)
def slugify(name: str) -> str:
s = re.sub(r"[^a-zA-Z0-9]+", "-", name.lower()).strip("-")
return s[:40]
# ─────────────────────────────────────────────────────────────────────────────
# GESTION DE PROJET
# ─────────────────────────────────────────────────────────────────────────────
class Project:
def __init__(self, name: str):
self.name = name
self.slug = slugify(name)
self.root = PROJECTS_DIR / self.slug
self.inputs = self.root / "inputs"
self.outputs = self.root / "outputs"
self.state_file = self.root / "project_state.json"
self.journal_file = self.root / "journal.md"
def ensure(self):
self.inputs.mkdir(parents=True, exist_ok=True)
self.outputs.mkdir(parents=True, exist_ok=True)
def exists(self) -> bool:
return self.root.is_dir()
def out(self, suffix: str, ext: str) -> Path:
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
return self.outputs / f"{self.slug}_{ts}_{suffix}.{ext}"
# ── État persistant (Solution A) ────────────────────────────────────────
def load_state(self) -> dict:
"""Charge project_state.json, ou un état vide si absent."""
if self.state_file.is_file():
try:
return json.loads(self.state_file.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError):
warn("project_state.json illisible — état réinitialisé.")
return {
"nom": self.name,
"cree_le": datetime.now().isoformat(timespec="seconds"),
"dernier_markdown": "",
"dernier_yaml": "",
"dernier_pptx": "",
"nb_sessions": 0,
"derniere_session": "",
}
def save_state(self, state: dict):
state["derniere_session"] = datetime.now().isoformat(timespec="seconds")
self.state_file.write_text(
json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8")
def has_state(self) -> bool:
"""Vrai si le projet a déjà produit au moins un narratif formalisé."""
st = self.load_state()
return bool(st.get("dernier_markdown"))
# ── Journal de décisions (Solution D) ───────────────────────────────────
def log(self, entry: str):
"""Ajoute une entrée horodatée au journal.md du projet."""
ts = datetime.now().strftime("%d/%m/%Y %H:%M")
line = f"- **{ts}** — {entry}\n"
if not self.journal_file.is_file():
header = f"# Journal — {self.name}\n\n"
self.journal_file.write_text(header + line, encoding="utf-8")
else:
with self.journal_file.open("a", encoding="utf-8") as f:
f.write(line)
def journal_summary(self, max_entries: int = 15) -> str:
"""Retourne les dernières entrées du journal pour réamorçage."""
if not self.journal_file.is_file():
return ""
lines = [l for l in self.journal_file.read_text(
encoding="utf-8").splitlines() if l.startswith("- ")]
return "\n".join(lines[-max_entries:])
@staticmethod
def list_existing() -> list:
if not PROJECTS_DIR.is_dir():
return []
return sorted([p.name for p in PROJECTS_DIR.iterdir() if p.is_dir()])
def select_project() -> Optional[Project]:
existing = Project.list_existing()
print(" [1] Nouveau projet")
if existing:
print(" [2] Projet existant")
print(" [0] Quitter")
choix = ask("Votre choix :")
if choix == "0":
return None
if choix == "1":
name = ask("Nom du nouveau projet :")
if not name:
warn("Nom vide — annulé.")
return None
proj = Project(name)
if proj.exists():
warn(f"Le projet '{proj.slug}' existe déjà — il sera réutilisé.")
proj.ensure()
ok(f"Projet créé : {proj.root}")
info(f"Déposez vos documents de contexte dans : {proj.inputs}")
return proj
if choix == "2" and existing:
print()
for i, name in enumerate(existing, 1):
print(f" [{i}] {name}")
sel = ask("Numéro du projet :")
try:
name = existing[int(sel) - 1]
except (ValueError, IndexError):
warn("Sélection invalide.")
return None
proj = Project(name)
proj.ensure()
ok(f"Projet ouvert : {proj.root}")
return proj
warn("Choix invalide.")
return None
# ─────────────────────────────────────────────────────────────────────────────
# LECTURE DE DOCUMENTS — chargement incrémental
# ─────────────────────────────────────────────────────────────────────────────
def _read_pdf(path: Path) -> str:
try:
from pypdf import PdfReader
return "\n".join((pg.extract_text() or "")
for pg in PdfReader(str(path)).pages)
except Exception:
try:
import pdfplumber
with pdfplumber.open(str(path)) as pdf:
return "\n".join((p.extract_text() or "") for p in pdf.pages)
except Exception as e:
warn(f"PDF illisible ({path.name}) : {e}")
return ""
def _read_docx(path: Path) -> str:
try:
import docx
return "\n".join(p.text for p in docx.Document(str(path)).paragraphs)
except Exception as e:
warn(f"DOCX illisible ({path.name}) : {e}")
return ""
def _read_pptx(path: Path) -> str:
"""Extrait le texte (titres, corps, notes) d'un PPTX via python-pptx."""
try:
from pptx import Presentation
prs = Presentation(str(path))
chunks = []
for i, slide in enumerate(prs.slides, 1):
texts = []
for shape in slide.shapes:
if shape.has_text_frame and shape.text_frame.text.strip():
texts.append(shape.text_frame.text.strip())
elif shape.has_table:
for row in shape.table.rows:
cells = [c.text.strip() for c in row.cells]
if any(cells):
texts.append(" | ".join(cells))
if slide.has_notes_slide:
notes = slide.notes_slide.notes_text_frame.text.strip()
if notes:
texts.append(f"[Notes] {notes}")
if texts:
chunks.append(f"-- Slide {i} --\n" + "\n".join(texts))
return "\n\n".join(chunks)
except Exception as e:
warn(f"PPTX illisible ({path.name}) : {e}")
return ""
def load_documents(proj: Project, already_loaded: set) -> tuple[str, set]:
"""
Charge les fichiers de inputs/ non encore injectés.
Retourne (bloc_contexte, nouveaux_fichiers_chargés).
already_loaded = ensemble des noms de fichiers déjà injectés.
"""
if not proj.inputs.is_dir():
return "", set()
files = [f for f in sorted(proj.inputs.iterdir())
if f.is_file()
and not f.name.startswith(".")
and f.name not in already_loaded]
if not files:
return "", set()
info(f"{len(files)} nouveau(x) document(s) détecté(s) :")
for f in files:
print(f" - {f.name}")
chunks = []
loaded = set()
for f in files:
ext = f.suffix.lower()
if ext == ".pdf":
txt = _read_pdf(f)
elif ext == ".docx":
txt = _read_docx(f)
elif ext == ".pptx":
txt = _read_pptx(f)
elif ext in (".txt", ".md", ".markdown", ".csv"):
txt = f.read_text(encoding="utf-8", errors="replace")
else:
warn(f"Format non supporté, ignoré : {f.name}")
continue
txt = txt.strip()
if txt:
chunks.append(f"### Document : {f.name}\n{txt}")
ok(f"Lu : {f.name} ({len(txt)} caractères)")
loaded.add(f.name)
if not chunks:
return "", set()
context = "\n\n".join(chunks)
if len(context) > CONTEXT_MAX_CHARS:
warn(f"Contexte tronqué à {CONTEXT_MAX_CHARS} caractères.")
context = context[:CONTEXT_MAX_CHARS] + "\n[...contexte tronqué...]"
return context, loaded
# ─────────────────────────────────────────────────────────────────────────────
# MANIFEST
# ─────────────────────────────────────────────────────────────────────────────
class Manifest:
def __init__(self, proj: Project):
self.data = {
"version": VERSION,
"demarrage": datetime.now().isoformat(timespec="seconds"),
"projet": proj.name,
"etapes": [],
"fichiers": [],
}
self._t0 = time.time()
def step(self, nom, **extra):
self.data["etapes"].append(
{"nom": nom, "duree_s": round(time.time() - self._t0, 1), **extra})
self._t0 = time.time()
def file(self, path: Path):
self.data["fichiers"].append(str(path))
def save(self, path: Path):
self.data["fin"] = datetime.now().isoformat(timespec="seconds")
path.write_text(json.dumps(self.data, ensure_ascii=False, indent=2),
encoding="utf-8")
ok(f"Manifest : {path}")
return path
# ─────────────────────────────────────────────────────────────────────────────
# APPEL AGENT
# ─────────────────────────────────────────────────────────────────────────────
def _post_with_retry(payload: dict) -> dict:
last_err = None
for attempt in range(1, MAX_RETRY + 1):
try:
resp = requests.post(
"https://api.mistral.ai/v1/agents/completions",
headers=HEADERS, json=payload, timeout=120)
if resp.status_code == 429:
wait = 2 ** attempt * 5
warn(f"Rate limit (429) — attente {wait}s "
f"(tentative {attempt}/{MAX_RETRY})")
time.sleep(wait)
continue
resp.raise_for_status()
return resp.json()
except requests.RequestException as e:
last_err = e
wait = 2 ** attempt * 2
warn(f"Erreur réseau : {e} — retry dans {wait}s "
f"(tentative {attempt}/{MAX_RETRY})")
time.sleep(wait)
raise RuntimeError(f"API Mistral injoignable après {MAX_RETRY} "
f"tentatives : {last_err}")
def call_agent(agent_id: str, messages: list) -> str:
full, current = "", messages.copy()
for page in range(MAX_PAGES):
data = _post_with_retry({"agent_id": agent_id, "messages": current})
content = data["choices"][0]["message"]["content"]
full += content
if "PAUSE" in content and "FIN —" not in content:
info(f"Pagination (page {page + 1}) → continuation...")
current.append({"role": "assistant", "content": content})
current.append({"role": "user", "content": "continue"})
else:
return full
warn(f"Garde pagination atteinte ({MAX_PAGES} pages).")
return full
class AgentSession:
def __init__(self, agent_id: str):
self.agent_id = agent_id
self.messages: list = []
self.started = False
def start(self, msg: str) -> str:
self.messages = [{"role": "user", "content": msg}]
out = call_agent(self.agent_id, self.messages)
self.messages.append({"role": "assistant", "content": out})
self.started = True
return out
def send(self, msg: str) -> str:
"""Premier message ou suite — gère automatiquement start vs feedback."""
if not self.started:
return self.start(msg)
return self.feedback(msg)
def feedback(self, msg: str) -> str:
self.messages.append({"role": "user", "content": msg})
out = call_agent(self.agent_id, self.messages)
self.messages.append({"role": "assistant", "content": out})
return out
def reset(self) -> None:
self.messages = []
self.started = False
# ─────────────────────────────────────────────────────────────────────────────
# ÉTAPE 1 — NARRATOR CONVERSATIONNEL
# ─────────────────────────────────────────────────────────────────────────────
def reprime_narrator(session: AgentSession, proj: Project) -> None:
"""
Réamorce une session Narrator neuve avec le contexte d'un projet
déjà travaillé : dernier Markdown formalisé + journal des décisions.
(Solution C — on ne restaure pas la conversation exacte, on réinforme.)
"""
state = proj.load_state()
last_md = state.get("dernier_markdown", "")
if not last_md:
return
summary = proj.journal_summary()
reprime = (
"Nous reprenons un projet de présentation déjà travaillé. "
"Voici l'état actuel pour te remettre en contexte.\n\n"
"=== DERNIER NARRATIF FORMALISÉ (état actuel du deck) ===\n\n"
f"{last_md}\n\n"
)
if summary:
reprime += (
"=== JOURNAL DES DÉCISIONS (historique du projet) ===\n\n"
f"{summary}\n\n"
)
reprime += (
"Reconstruis à partir de ce narratif le PLAN COMPACT de la présentation "
"(format hiérarchique : chapitres, slides numérotées en continu, tags "
"core/OPT-XXX et audience). Affiche uniquement ce plan compact, sans "
"autre commentaire. On repartira de ce plan pour la révision, avec les "
"mêmes modes f: (flux) et d: (deep dive)."
)
info("Réamorçage du Narrator — reconstruction du plan compact...")
out = session.start(reprime)
display_narrator(out)
def run_narrator(session: AgentSession, proj: Project) -> Optional[str]:
"""
Boucle Narrator « plan compact ».
Modes explicites par préfixe : f: (FLUX) / d: (DEEP DIVE).
Le mode persiste tant qu'aucun nouveau préfixe n'est donné.
Retourne le Markdown formalisé validé, ou None si abandonné.
"""
section("ÉTAPE 1 — THE NARRATOR (plan compact)")
has_docs = any(
f.is_file() and not f.name.startswith(".")
for f in proj.inputs.iterdir()
) if proj.inputs.is_dir() else False
print(" Premier message = brief initial → le Narrator produit le plan compact.")
print(" Préfixes : f: (structure/flux) · d: (contenu/deep dive).")
print(" Le mode reste actif tant que tu ne changes pas de préfixe.")
print(" /aide pour tout voir.")
if has_docs:
print("\n Documents présents dans inputs/ — /lire pour les injecter.")
loaded_docs: set = set()
last_plan: str = "" # dernier plan compact (mode flux)
last_response: str = "" # dernière réponse quelconque
formalized_md: Optional[str] = None
pending_context: str = ""
mode: Optional[str] = None # MODE_FLUX / MODE_DEEP / None
def send_with_mode(msg: str, current_mode: Optional[str]) -> str:
"""Préfixe le message d'une consigne de mode pour l'agent."""
if current_mode == MODE_FLUX:
tag = ("[MODE FLUX] Applique la demande de structure et RÉAFFICHE "
"le plan compact complet, numéros recalculés.\n\n")
elif current_mode == MODE_DEEP:
tag = ("[MODE DEEP DIVE] Réponds de façon ciblée sur ce point, "
"NE réaffiche PAS le plan.\n\n")
else:
tag = ""
return tag + msg
while True:
raw = ask("Vous :")
if not raw:
continue
# ── Commandes slash ────────────────────────────────────────────────
if raw.startswith("/"):
cmd = raw.strip().lower().split()[0]
if cmd == "/aide":
display_help(); continue
elif cmd == "/lire":
context, new_files = load_documents(proj, loaded_docs)
if not new_files:
info("Aucun nouveau document dans inputs/.")
continue
loaded_docs |= new_files
if not session.started:
pending_context = (pending_context + "\n\n" + context).strip()
ok(f"{len(new_files)} doc(s) en attente — joints à ton prochain message.")
else:
info("Injection des documents...")
last_response = session.feedback(
f"[Documents chargés depuis inputs/]\n\n{context}\n\n"
f"Intègre-les. Reste bref, ne résume pas les documents.")
display_narrator(last_response)
continue
elif cmd == "/coller":
pasted = ask_multiline()
if not pasted:
warn("Rien collé."); continue
# Le texte collé devient le message courant, dans le mode actif
if not session.started and pending_context:
pasted = (f"{pasted}\n\n---\nCONTEXTE DOCUMENTAIRE :\n\n"
f"{pending_context}")
pending_context = ""
last_response = session.start(pasted)
else:
last_response = session.send(send_with_mode(pasted, mode))
if mode == MODE_FLUX or mode is None:
last_plan = last_response
display_narrator(last_response)
continue
elif cmd == "/plan":
if last_plan:
print("\n" + textwrap.indent(last_plan, " "))
else:
warn("Aucun plan compact pour l'instant.")
continue
elif cmd == "/formalise":
if not session.started:
warn("Commence par établir le plan compact."); continue
info("Formalisation du plan compact en Markdown...")
last_response = session.feedback(
"Formalise maintenant le plan compact validé en Markdown "
"structuré complet pour le Designer (format #, ##, ###), "
"en développant le contenu de chaque slide. Inclus les "
"slides tiroir en les marquant [OPT-XXX].")
formalized_md = last_response
display_narrator(last_response)
ok("Formalisé. /valider pour passer au Designer, ou continue à ajuster.")
continue
elif cmd == "/valider":
if formalized_md:
ok("Narratif validé. Passage au Designer.")
return formalized_md
warn("Pas encore formalisé — utilise /formalise d'abord.")
continue
elif cmd in ("/quitter", "/exit", "/q"):
if ask("Abandonner la session Narrator ? (o/N) :").lower() in ("o","oui","y","yes"):
return None
continue
else:
warn(f"Commande inconnue : {cmd} — /aide pour la liste.")
continue
# ── Détection de préfixe de mode ───────────────────────────────────
stripped = raw
low = raw.lower()
if low.startswith("f:"):
mode = MODE_FLUX
stripped = raw[2:].strip()
elif low.startswith("d:"):
mode = MODE_DEEP
stripped = raw[2:].strip()
# sinon : on garde le mode courant (persistance)
if not stripped:
# juste un changement de mode sans contenu
info(f"Mode : {'FLUX' if mode==MODE_FLUX else 'DEEP DIVE' if mode==MODE_DEEP else 'libre'}")
continue
# ── Envoi au Narrator ──────────────────────────────────────────────
if not session.started:
msg = stripped
if pending_context:
msg = (f"{stripped}\n\n---\nCONTEXTE DOCUMENTAIRE :\n\n"
f"{pending_context}")
pending_context = ""
last_response = session.start(msg)
else:
last_response = session.send(send_with_mode(stripped, mode))
# Mémoriser le plan si on est en flux (ou tout premier plan)
if mode == MODE_FLUX or (mode is None and not last_plan):
last_plan = last_response
display_narrator(last_response)
def save_text(content: str, path: Path) -> Path:
path.write_text(content, encoding="utf-8")
ok(f"Sauvegardé : {path}")
return path
def persist_generation(proj: Project, markdown: str = None,
yaml_path: Path = None, pptx_path: Path = None,
journal_entry: str = None):
"""Met à jour project_state.json et journal.md après une génération."""
state = proj.load_state()
if markdown is not None:
state["dernier_markdown"] = markdown
if yaml_path is not None:
state["dernier_yaml"] = str(yaml_path)
if pptx_path is not None:
state["dernier_pptx"] = str(pptx_path)
state["nb_sessions"] = state.get("nb_sessions", 0) + 1
proj.save_state(state)
if journal_entry:
proj.log(journal_entry)
# ─────────────────────────────────────────────────────────────────────────────
# VALIDATION YAML
# ─────────────────────────────────────────────────────────────────────────────
def extract_yaml(raw: str) -> str:
m = re.search(r"```ya?ml\s*(.*?)```", raw, re.DOTALL | re.IGNORECASE)
return re.sub(r"[─]+", "-", (m.group(1).strip() if m else raw.strip()))
def validate_yaml(raw: str, layouts: dict):
try:
data = yaml.safe_load(extract_yaml(raw))
except yaml.YAMLError as e:
return False, f"Erreur de syntaxe YAML : {e}", None
if not isinstance(data, dict):
return False, "Doit être un dictionnaire à la racine.", None
slides = data.get("slides")
if not slides:
return False, "Clé 'slides' manquante ou vide.", None
errors, valid = [], set(layouts.keys())
for i, slide in enumerate(slides):
pos = slide.get("position", i + 1)
layout = slide.get("layout", "")
if not layout:
errors.append(f"Slide {pos} : layout manquant")
continue
if layout not in valid:
errors.append(f"Slide {pos} : layout '{layout}' inconnu")
continue
for field in layouts[layout].get("champs_requis", []):
if field not in slide or slide.get(field) in (None, "", []):
errors.append(f"Slide {pos} ({layout}) : '{field}' requis")
if errors:
return False, "\n ".join(errors), data
return True, f"{len(slides)} slides valides.", data
def layouts_digest(layouts: dict) -> str:
lines = ["Layouts valides et champs requis (design system v2) :"]
for name, cfg in layouts.items():
req = ", ".join(cfg.get("champs_requis", []))
lines.append(f"- {name} : {req}")
return "\n".join(lines)
def load_layouts() -> dict:
if not os.path.exists(LAYOUTS_PATH):
warn(f"{LAYOUTS_PATH} introuvable — validation désactivée")
return {}
with open(LAYOUTS_PATH, encoding="utf-8") as f:
return yaml.safe_load(f).get("layouts", {})
# ─────────────────────────────────────────────────────────────────────────────
# ÉTAPE 2 — DESIGNER
# ─────────────────────────────────────────────────────────────────────────────
def display_annotated(md: str, max_lines: int = 60):
"""Affiche le Markdown annoté en mettant en évidence les @layout."""
lines = md.splitlines()
shown = lines[:max_lines]
for l in shown:
if l.strip().startswith("@layout:"):
print(f" \033[1m{l.strip()}\033[0m" if sys.stdout.isatty()
else f" {l.strip()}")
elif l.startswith("#"):
print(f" {l}")
else:
print(f" {l}")
if len(lines) > max_lines:
print(f"\n ... [{len(lines)-max_lines} lignes — option 3 pour tout voir]")
def layout_summary(md: str) -> str:
"""Extrait la liste des layouts annotés pour un résumé rapide."""
out = []
n = 0
for l in md.splitlines():
s = l.strip()
if s.startswith("@layout:"):
n += 1
out.append(f" {n}. {s.replace('@layout:', '').strip()}")
return "\n".join(out) if out else " (aucune annotation @layout détectée)"
def run_designer(session: AgentSession, markdown: str,
proj: Project, scope_instruction: str = "") -> Optional[str]:
section("ÉTAPE 2 — THE DESIGNER (annotation des layouts)")
info("Annotation du Markdown avec les layouts proposés...")
initial = markdown
if scope_instruction:
initial = f"{markdown}\n\n---\n{scope_instruction}"
info("Mode ciblé : le Designer ne traitera que les slides indiquées.")
output = session.start(initial)
ok(f"Markdown annoté ({len(output)} caractères)")
while True:
print()
print(" Layouts proposés :")
print(layout_summary(output))
print("\n [1] Valider → passer à l'Encoder")
print(" [2] Donner du feedback au Designer")
print(" [3] Afficher le Markdown annoté complet")
print(" [4] Sauvegarder le Markdown annoté")
print(" [0] Revenir au Narrator")
choix = ask("Votre choix :")
if choix == "1":
ok("Annotation validée.")
save_text(output, proj.out("designer_annote", "md"))
return output
elif choix == "2":
fb = ask("Votre feedback (ex : 'slide 4 en big_stat') :")
if fb:
info("Envoi au Designer...")
output = session.feedback(fb)
ok(f"Markdown annoté mis à jour ({len(output)} caractères)")
elif choix == "3":
print()
display_annotated(output, max_lines=200)
elif choix == "4":
save_text(output, proj.out("designer_annote_draft", "md"))
elif choix == "0":
return None
else:
warn("Choix invalide.")
# ─────────────────────────────────────────────────────────────────────────────
# ÉTAPE 3 — ENCODER
# ─────────────────────────────────────────────────────────────────────────────
def run_encoder(session: AgentSession, plan: str, layouts: dict,
proj: Project, scope_instruction: str = ""):
section("ÉTAPE 3 — THE ENCODER")
info("Encodage du plan en YAML...")
digest = layouts_digest(layouts) if layouts else ""
parts = [plan]
if scope_instruction:
parts.append(scope_instruction)
info("Mode ciblé : l'Encoder ne produira que les slides indiquées.")
if digest:
parts.append(digest)
initial = "\n\n---\n".join(parts)
output = session.start(initial)
ok(f"YAML brut reçu ({len(output)} caractères)")
attempts, max_attempts = 0, 3
while attempts < max_attempts:
is_valid, message, data = validate_yaml(output, layouts)
if is_valid:
ok(f"YAML valide : {message}")
path = save_text(extract_yaml(output), proj.out("input", "yaml"))
return extract_yaml(output), data, path, attempts
attempts += 1
warn(f"YAML invalide (tentative {attempts}/{max_attempts}) :")
print(f" {message}\n")
if attempts >= max_attempts:
break
info("Correction automatique...")
output = session.feedback(
f"Erreurs dans le YAML :\n\n{message}\n\n"
f"Renvoie le YAML COMPLET corrigé, sans texte avant ou après.")
ok(f"YAML corrigé ({len(output)} caractères)")
print("\n [1] Accepter le YAML tel quel")
print(" [2] Feedback manuel à l'Encoder")
print(" [0] Revenir au Designer")
choix = ask("Votre choix :")
if choix == "1":
path = save_text(extract_yaml(output),
proj.out("input_unvalidated", "yaml"))
try:
data = yaml.safe_load(extract_yaml(output)) or {}
except Exception:
data = {}
return extract_yaml(output), data, path, attempts
elif choix == "2":
fb = ask("Votre feedback :")
output = session.feedback(fb)
path = save_text(extract_yaml(output), proj.out("input_manual", "yaml"))
try:
data = yaml.safe_load(extract_yaml(output)) or {}
except Exception:
data = {}
return extract_yaml(output), data, path, attempts
return None, None, None, attempts
# ─────────────────────────────────────────────────────────────────────────────
# FLUX LIBRE — THE FREE DESIGNER
# ─────────────────────────────────────────────────────────────────────────────
VALID_TOKENS = {"navy", "navy_light", "coral", "glacier", "slate",
"card", "white", "body", "muted"}
VALID_BLOCK_TYPES = {"title", "heading", "text", "stat", "circle",
"badge", "card", "rect", "line"}
def validate_freeform(raw: str):
"""Valide la structure d'un YAML freeform. Retourne (ok, message, data)."""
try:
data = yaml.safe_load(extract_yaml(raw))
except yaml.YAMLError as e:
return False, f"Erreur de syntaxe YAML : {e}", None
if not isinstance(data, dict) or "slides" not in data:
return False, "Clé 'slides' manquante.", None
slides = data["slides"]
if not slides:
return False, "Aucune slide.", None
errors = []
for i, s in enumerate(slides, 1):
if s.get("layout") != "freeform":
errors.append(f"Slide {i} : layout doit être 'freeform'")
continue
if s.get("mode") not in ("light", "dark"):
errors.append(f"Slide {i} : mode doit être 'light' ou 'dark'")
blocks = s.get("blocks", [])
if not blocks:
errors.append(f"Slide {i} : aucun bloc")
if len(blocks) > 8:
errors.append(f"Slide {i} : {len(blocks)} blocs (max 8)")
for j, b in enumerate(blocks, 1):
bt = (b.get("type") or "text").lower()
if bt not in VALID_BLOCK_TYPES:
errors.append(f"Slide {i} bloc {j} : type '{bt}' invalide")
col = float(b.get("col", 0)); w = float(b.get("w", 4))
row = float(b.get("row", 0)); h = float(b.get("h", 1))
if col + w > 12.01:
errors.append(f"Slide {i} bloc {j} : déborde (col+w={col+w:.1f}>12)")
if row + h > 12.01:
errors.append(f"Slide {i} bloc {j} : déborde (row+h={row+h:.1f}>12)")
color = b.get("color")
if color and color not in VALID_TOKENS:
errors.append(f"Slide {i} bloc {j} : couleur '{color}' hors charte")
if errors:
return False, "\n ".join(errors), data
return True, f"{len(slides)} slides freeform valides.", data
def run_free_designer(session: AgentSession, markdown: str, proj: Project):
"""Flux libre : le Free Designer produit directement le YAML freeform."""
section("FLUX LIBRE — THE FREE DESIGNER")
info("Composition libre des slides (charte PR imposée)...")
output = session.start(markdown)
ok(f"YAML freeform reçu ({len(output)} caractères)")
attempts, max_attempts = 0, 3
while attempts < max_attempts:
is_valid, message, data = validate_freeform(output)
if is_valid:
ok(f"Freeform valide : {message}")
break
attempts += 1
warn(f"Freeform invalide (tentative {attempts}/{max_attempts}) :")
print(f" {message}\n")
if attempts >= max_attempts:
break
info("Correction automatique...")
output = session.feedback(
f"Le YAML freeform contient des erreurs :\n\n{message}\n\n"
f"Corrige et renvoie le YAML freeform COMPLET, sans texte autour. "
f"Rappel : couleurs = tokens de charte uniquement, "
f"col+w ≤ 12, row+h ≤ 12, max 8 blocs par slide.")
ok(f"Freeform corrigé ({len(output)} caractères)")
# Boucle de validation utilisateur
while True:
is_valid, message, data = validate_freeform(output)
print()
n = len(data["slides"]) if data and "slides" in data else 0
info(f"Composition libre : {n} slide(s).")
status = "valide" if is_valid else "INVALIDE — " + message.split(chr(10))[0]
print(f" Statut : {status}")
print("\n [1] Valider → rendu")
print(" [2] Donner du feedback au Free Designer")
print(" [3] Afficher le YAML complet")
print(" [4] Sauvegarder le YAML")
print(" [0] Revenir au Narrator")
choix = ask("Votre choix :")
if choix == "1":
if not is_valid:
warn("Le YAML a des erreurs — le rendu peut échouer.")
if ask("Forcer le rendu ? (o/N) :").lower() not in ("o", "oui", "y"):
continue
path = save_text(extract_yaml(output), proj.out("freeform", "yaml"))
return extract_yaml(output), data, path
elif choix == "2":
fb = ask("Votre feedback :")
if fb:
info("Envoi au Free Designer...")
output = session.feedback(fb)
ok(f"Composition mise à jour ({len(output)} caractères)")
elif choix == "3":
print("\n" + textwrap.indent(extract_yaml(output), " "))
elif choix == "4":
save_text(extract_yaml(output), proj.out("freeform_draft", "yaml"))
elif choix == "0":
return None, None, None
else:
warn("Choix invalide.")
# ─────────────────────────────────────────────────────────────────────────────
# ÉTAPE 4 — RENDER ENGINE
# ─────────────────────────────────────────────────────────────────────────────
def run_render(yaml_path: Path) -> Optional[Path]:
section("ÉTAPE 4 — RENDER ENGINE V2")
if not Path(RENDER_ENGINE_PATH).exists():
warn(f"{RENDER_ENGINE_PATH} introuvable.")
return None
for p in [THEME_PATH, COMPONENTS_PATH, LAYOUTS_PATH]:
if not os.path.exists(p):
warn(f"Config manquante : {p}")
return None
pptx_out = yaml_path.with_suffix(".pptx")
info("Lancement de render_engine_v2.py...")
info(f"Sortie : {pptx_out}")
result = subprocess.run(
[sys.executable, RENDER_ENGINE_PATH, str(yaml_path), str(pptx_out),
"--theme", THEME_PATH, "--components", COMPONENTS_PATH,
"--layouts", LAYOUTS_PATH],
capture_output=True, text=True)
if result.returncode == 0:
ok(result.stdout.strip() or f"PPTX généré : {pptx_out}")
return pptx_out
warn("Erreur render_engine_v2.py :")
print(textwrap.indent(result.stderr or result.stdout, " "))
return None
# ─────────────────────────────────────────────────────────────────────────────
# ARCHIVAGE TRILIUM
# ─────────────────────────────────────────────────────────────────────────────
def archive_to_trilium(pptx_path, manifest_path, proj):
if not (TRILIUM_API_URL and TRILIUM_API_KEY):
return
try:
detail = f"Projet : {proj.name}"
if pptx_path:
detail += f" | PPTX : {pptx_path}"
resp = requests.post(
f"{TRILIUM_API_URL.rstrip('/')}/api/historique",
headers={"Authorization": TRILIUM_API_KEY,
"Content-Type": "application/json"},
json={"projet": TRILIUM_PROJET, "type": "Fait etabli",
"enonce": f"Génération PPTX — projet {proj.name}",
"detail": detail},
timeout=15)
ok("Archivé dans Trilium.") if resp.ok \
else warn(f"Trilium : {resp.status_code} (non bloquant)")
except requests.RequestException as e:
warn(f"Trilium injoignable ({e}).")
# ─────────────────────────────────────────────────────────────────────────────
# BOUCLE PRINCIPALE
# ─────────────────────────────────────────────────────────────────────────────
# ─────────────────────────────────────────────────────────────────────────────
# MODE RÉVISION (ciblé / structurant)
# ─────────────────────────────────────────────────────────────────────────────
def parse_slide_numbers(raw: str) -> list:
"""'4, 7, 9' ou '4-6' → [4,7,9] ou [4,5,6]. Retourne liste triée unique."""
nums = set()
for token in re.split(r"[,\s]+", raw.strip()):
if not token:
continue
if "-" in token:
try:
a, b = token.split("-")
nums.update(range(int(a), int(b) + 1))
except ValueError:
pass
else:
try:
nums.add(int(token))
except ValueError:
pass
return sorted(nums)
def run_full_pipeline_pass(narrator, markdown, layouts, proj, manifest,
scope_designer="", scope_encoder="",
suffix="input"):
"""
Exécute Designer → Encoder → Render sur le Markdown fourni.
scope_* restreignent le périmètre (révision ciblée).
suffix nomme les fichiers de sortie (ex 'revision').
Retourne le chemin du PPTX généré, ou None.
"""
while True:
designer = AgentSession(DESIGNER_ID)
plan = run_designer(designer, markdown, proj, scope_designer)
if plan is None:
section("RETOUR AU NARRATOR")
info("La session Narrator est toujours active.")
markdown = run_narrator(narrator, proj)
if not markdown:
return None, markdown
save_text(markdown, proj.out("narrator_rev", "md"))
continue
manifest.step("designer", caracteres=len(plan))
encoder = AgentSession(ENCODER_ID)
yaml_str, yaml_data, yaml_path, tries = run_encoder(
encoder, plan, layouts, proj, scope_encoder)
manifest.step("encoder", tentatives_correction=tries)
if yaml_str is None:
info("Retour au Designer...")
continue
# Renommer la sortie selon le suffixe demandé
if suffix != "input" and yaml_path:
new_path = proj.out(suffix, "yaml")
yaml_path.rename(new_path)
yaml_path = new_path
manifest.file(yaml_path)
pptx_path = run_render(yaml_path)
manifest.step("render", succes=bool(pptx_path))
# Persister l'état du projet (seulement en révision complète :
# le ciblé ne représente pas l'état complet du deck)
if suffix == "revision_complete":
persist_generation(
proj, markdown=markdown, yaml_path=yaml_path,
pptx_path=pptx_path,
journal_entry="Révision structurante — deck complet régénéré.")
elif suffix == "revision_ciblee":
persist_generation(
proj, markdown=markdown,
journal_entry="Révision ciblée — slides régénérées séparément.")
return pptx_path, markdown
def run_revision(narrator, last_markdown, layouts, proj, manifest):
"""
Boucle de révision. Repasse TOUJOURS par le Narrator (contexte complet),
puis propose le mode ciblé ou structurant.
Retourne le dernier markdown (pour chaîner d'autres révisions).
"""
section("MODE RÉVISION")
info("La session Narrator a conservé tout le contexte de la présentation.")
info("Retravaillez une ou plusieurs slides, puis /formalise pour")
info("régénérer le Markdown complet du deck.")
# 1. Narrator — conversation reprise, formalisation complète
new_markdown = run_narrator(narrator, proj)
if not new_markdown:
info("Révision annulée.")
return last_markdown
save_text(new_markdown, proj.out("narrator_rev", "md"))
manifest.step("narrator_revision", caracteres=len(new_markdown))
# 2. Portée de la révision
section("PORTÉE DE LA RÉVISION")
print(" [1] Ciblé — quelques slides modifiées, le reste est inchangé")
print(" → seules ces slides sont régénérées (à coller dans ton deck)")
print(" [2] Structurant — la logique d'ensemble a changé")
print(" → tout le deck est régénéré")
print(" [0] Annuler")
choix = ask("Votre choix :")
if choix == "0":
return new_markdown
if choix == "1":
nums_raw = ask("Numéros des slides à régénérer (ex : 4, 7 ou 4-6) :")
slides = parse_slide_numbers(nums_raw)
if not slides:
warn("Aucun numéro valide — révision annulée.")
return new_markdown
slist = ", ".join(str(n) for n in slides)
ok(f"Régénération ciblée des slides : {slist}")
scope_designer = (
f"RÉVISION CIBLÉE. Le Markdown ci-dessus est le deck COMPLET, "
f"fourni pour que tu comprennes le contexte et la cohérence. "
f"Tu ne dois produire le plan QUE pour les slides suivantes : "
f"{slist}. Numérote-les avec leur position réelle dans le deck "
f"complet (ex : SLIDE {slides[0]}). Ignore toutes les autres slides."
)
scope_encoder = (
f"RÉVISION CIBLÉE. Encode UNIQUEMENT les slides {slist}. "
f"Conserve leur position réelle dans le champ 'position'. "
f"Le YAML produit ne contiendra donc que {len(slides)} slide(s)."
)
pptx_path, new_markdown = run_full_pipeline_pass(
narrator, new_markdown, layouts, proj, manifest,
scope_designer=scope_designer, scope_encoder=scope_encoder,
suffix="revision_ciblee")
if pptx_path and pptx_path.exists():
section("SLIDES RÉVISÉES GÉNÉRÉES")
ok(f"PPTX des slides révisées : {pptx_path}")
info("Ouvre ce fichier et copie-colle les slides dans ton deck maître.")
else:
warn("Les slides révisées n'ont pas pu être générées.")
return new_markdown
if choix == "2":
ok("Régénération complète du deck.")
pptx_path, new_markdown = run_full_pipeline_pass(
narrator, new_markdown, layouts, proj, manifest,
suffix="revision_complete")
if pptx_path and pptx_path.exists():
section("DECK COMPLET RÉGÉNÉRÉ")
ok(f"Nouveau PPTX complet : {pptx_path}")
else:
warn("Le deck n'a pas pu être régénéré.")
return new_markdown
warn("Choix invalide — révision annulée.")
return new_markdown
# ─────────────────────────────────────────────────────────────────────────────
# FLUX DE CRÉATION (nouvelle présentation)
# ─────────────────────────────────────────────────────────────────────────────
def run_creation(proj, layouts):
"""Crée une nouvelle présentation : Narrator → Designer → Encoder → Render.
Persiste l'état du projet à la génération. Gère ensuite la boucle révision."""
manifest = Manifest(proj)
narrator = AgentSession(NARRATOR_ID)
markdown = run_narrator(narrator, proj)
if not markdown:
return
save_text(markdown, proj.out("narrator", "md"))
manifest.step("narrator", caracteres=len(markdown))
proj.log(f"Narratif formalisé ({len(markdown)} car.).")
# ── Choix du flux de mise en forme ──────────────────────────────────────
section("FLUX DE MISE EN FORME")
print(" [1] Standard — layouts prédéfinis (Designer annote, Encoder, rendu)")
print(" → présentations régulières, structure cadrée et reproductible")
print(" [2] Libre — composition sur mesure (Free Designer)")
print(" → présentations stratégiques, charte PR imposée")
flux = ask("Votre choix :")
# ── FLUX LIBRE ──────────────────────────────────────────────────────────
if flux == "2":
if not FREE_DESIGNER_ID:
warn("FREE_DESIGNER_AGENT_ID absent du .env — flux libre indisponible.")
info("Bascule sur le flux standard.")
else:
manifest.step("choix_flux", flux="libre")
proj.log("Flux libre sélectionné.")
yaml_path = yaml_data = None
while True:
free = AgentSession(FREE_DESIGNER_ID)
yaml_str, yaml_data, yaml_path = run_free_designer(
free, markdown, proj)
if yaml_str is None:
section("RETOUR AU NARRATOR")
markdown = run_narrator(narrator, proj)
if not markdown:
return
save_text(markdown, proj.out("narrator_v2", "md"))
continue
manifest.step("free_designer", caracteres=len(yaml_str))
break
if yaml_path and yaml_data:
manifest.file(yaml_path)
pptx_path = run_render(yaml_path)
manifest.step("render", succes=bool(pptx_path))
if pptx_path and pptx_path.exists():
manifest.file(pptx_path)
section("PRÉSENTATION GÉNÉRÉE (flux libre)")
ok(f"Fichier PPTX : {pptx_path}")
ok(f"Taille : {pptx_path.stat().st_size/1024:.1f} Ko")
else:
warn("Le PPTX n'a pas pu être généré.")
persist_generation(
proj, markdown=markdown, yaml_path=yaml_path,
pptx_path=pptx_path,
journal_entry="Présentation générée (flux libre).")
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
manifest_path = manifest.save(
proj.outputs / f"{proj.slug}_{ts}_manifest.json")
archive_to_trilium(pptx_path, manifest_path, proj)
revision_loop(narrator, markdown, layouts, proj, manifest)
return
# ── FLUX STANDARD ───────────────────────────────────────────────────────
manifest.step("choix_flux", flux="standard")
express = ask("Mode express — sans Designer ? (o/N) :").lower() \
in ("o", "oui", "y", "yes")
if express:
info("Mode express : Narrator → Encoder direct.")
yaml_path = yaml_data = None
while True:
if express:
plan = markdown
else:
designer = AgentSession(DESIGNER_ID)
plan = run_designer(designer, markdown, proj)
if plan is None:
section("RETOUR AU NARRATOR")
info("La session Narrator est toujours active.")
markdown = run_narrator(narrator, proj)
if not markdown:
return
save_text(markdown, proj.out("narrator_v2", "md"))
continue
manifest.step("designer", caracteres=len(plan))
encoder = AgentSession(ENCODER_ID)
yaml_str, yaml_data, yaml_path, tries = run_encoder(
encoder, plan, layouts, proj)
manifest.step("encoder", tentatives_correction=tries)
if yaml_str is None:
if express:
warn("Échec Encoder en mode express.")
return
info("Retour au Designer...")
continue
break
if yaml_path and yaml_data:
manifest.file(yaml_path)
pptx_path = run_render(yaml_path)
manifest.step("render", succes=bool(pptx_path))
if pptx_path and pptx_path.exists():
manifest.file(pptx_path)
section("PRÉSENTATION GÉNÉRÉE")
ok(f"Fichier PPTX : {pptx_path}")
ok(f"Taille : {pptx_path.stat().st_size/1024:.1f} Ko")
else:
warn("Le PPTX n'a pas pu être généré.")
info(f"Le YAML est dans : {proj.outputs}")
# Persister l'état complet du projet
persist_generation(proj, markdown=markdown, yaml_path=yaml_path,
pptx_path=pptx_path,
journal_entry="Présentation générée (deck complet).")
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
manifest_path = manifest.save(
proj.outputs / f"{proj.slug}_{ts}_manifest.json")
archive_to_trilium(pptx_path, manifest_path, proj)
# Boucle de révision immédiate
revision_loop(narrator, markdown, layouts, proj, manifest)
def revision_loop(narrator, markdown, layouts, proj, manifest):
"""Menu de révision réutilisable (après création ou à la reprise)."""
while True:
print()
print(" [1] Réviser la présentation (repasse par le Narrator)")
print(" [2] Terminer")
rev = ask("Votre choix :")
if rev == "1":
markdown = run_revision(narrator, markdown, layouts, proj, manifest)
manifest.save(
proj.outputs /
f"{proj.slug}_{datetime.now():%Y%m%d_%H%M%S}_manifest.json")
continue
break
# ─────────────────────────────────────────────────────────────────────────────
# REPRISE D'UN PROJET EXISTANT (Solution B + C)
# ─────────────────────────────────────────────────────────────────────────────
def run_reprise(proj, layouts):
"""Ouvre un projet déjà travaillé : révision par défaut, ou questions."""
state = proj.load_state()
section(f"REPRISE — {proj.name}")
ok(f"Dernière session : {state.get('derniere_session', 'inconnue')}")
ok(f"Sessions précédentes : {state.get('nb_sessions', 0)}")
last_pptx = state.get("dernier_pptx", "")
if last_pptx:
info(f"Dernier PPTX : {last_pptx}")
print("\n [1] Réviser la présentation existante")
print(" [2] Nouvelle présentation dans ce projet")
print(" [3] Poser des questions (sans rien régénérer)")
print(" [0] Retour au menu projet")
choix = ask("Votre choix :")
if choix == "1":
manifest = Manifest(proj)
narrator = AgentSession(NARRATOR_ID)
reprime_narrator(narrator, proj) # réamorçage avec contexte
proj.log("Reprise du projet pour révision.")
markdown = state.get("dernier_markdown", "")
revision_loop(narrator, markdown, layouts, proj, manifest)
elif choix == "2":
proj.log("Nouvelle présentation dans un projet existant.")
run_creation(proj, layouts)
elif choix == "3":
manifest = Manifest(proj)
narrator = AgentSession(NARRATOR_ID)
reprime_narrator(narrator, proj)
section("MODE QUESTIONS")
info("Pose tes questions sur la présentation. Tape /quitter pour sortir.")
while True:
q = ask("Question :")
if q.strip().lower() in ("/quitter", "/q", "quit", "/exit"):
break
if not q:
continue
ans = narrator.send(q)
display_narrator(ans)
# choix 0 ou autre : retour
# ─────────────────────────────────────────────────────────────────────────────
# BOUCLE PRINCIPALE
# ─────────────────────────────────────────────────────────────────────────────
def run_pipeline():
banner()
missing = [k for k, v in {
"MISTRAL_API_KEY": API_KEY, "NARRATOR_AGENT_ID": NARRATOR_ID,
"DESIGNER_AGENT_ID": DESIGNER_ID, "ENCODER_AGENT_ID": ENCODER_ID,
}.items() if not v]
if missing:
for m in missing: warn(f"Variable .env manquante : {m}")
sys.exit(1)
layouts = load_layouts()
if layouts:
ok(f"{LAYOUTS_PATH} chargé — {len(layouts)} layouts")
while True:
section("PROJET")
proj = select_project()
if proj is None:
print(" Au revoir.")
break
# Reprise intelligente : projet avec état → menu révision par défaut
if proj.has_state():
run_reprise(proj, layouts)
else:
run_creation(proj, layouts)
# Après création ou reprise : continuer ou quitter
cont = ask("Revenir au menu projet ? (o/n) :").lower()
if cont not in ("o", "oui", "y", "yes"):
print(" Au revoir.")
break
# ─────────────────────────────────────────────────────────────────────────────
# POINT D'ENTRÉE
# ─────────────────────────────────────────────────────────────────────────────
def main():
parser = argparse.ArgumentParser(
description="Sliding facilitator v8 — flux libre (Free Designer)")
parser.add_argument("--render", metavar="YAML",
help="Rendu direct d'un YAML existant, sans agents")
args = parser.parse_args()
if args.render:
yaml_path = Path(args.render)
if not yaml_path.exists():
warn(f"Fichier introuvable : {yaml_path}")
sys.exit(1)
layouts = load_layouts()
is_valid, message, _ = validate_yaml(
yaml_path.read_text(encoding="utf-8"), layouts)
(ok if is_valid else warn)(f"Validation : {message}")
sys.exit(0 if run_render(yaml_path) else 1)
run_pipeline()
if __name__ == "__main__":
main()
Reference in New Issue Copy Permalink