#!/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 # ── Forçage UTF-8 des entrées/sorties ──────────────────────────────────────── # Évite les UnicodeDecodeError sur les accents en session SSH dont la locale # n'est pas en UTF-8. Indépendant de LANG/LC_ALL du système. for _stream in (sys.stdin, sys.stdout, sys.stderr): try: _stream.reconfigure(encoding="utf-8", errors="replace") except (AttributeError, ValueError): pass # flux non reconfigurable (rare) — on continue 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") ENCODER_MODE = os.getenv("ENCODER_MODE", "agent") # agent | schema (C1) 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") PREVIEW_SCRIPT = os.getenv("PREVIEW_SCRIPT", "./preview.sh") # C2 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", "/afficher": "Réafficher la dernière réponse du Narrator en entier (sans troncature)", "/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) except UnicodeDecodeError: # Faute d'encodage (locale SSH non-UTF-8) : on ne tue pas la session. warn("Caractère mal encodé ignoré. Relance ta saisie " "(ou préfixe la commande par PYTHONUTF8=1 au prochain lancement).") return "" 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 except UnicodeDecodeError: warn("Ligne mal encodée ignorée.") continue 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_plan_compact": "", "plan_modifie_le": "", # horodatage dernière modif du plan "formalise_le": "", # horodatage dernière formalisation "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 save_plan_compact(self, plan: str): """Sauvegarde continue du plan compact (source de vérité de la structure).""" state = self.load_state() state["dernier_plan_compact"] = plan state["plan_modifie_le"] = datetime.now().isoformat(timespec="seconds") self.save_state(state) def mark_formalised(self, markdown: str): """Enregistre le Markdown formalisé + l'horodatage de formalisation.""" state = self.load_state() state["dernier_markdown"] = markdown state["formalise_le"] = datetime.now().isoformat(timespec="seconds") self.save_state(state) def plan_diverge(self) -> bool: """Vrai si le plan compact a été modifié APRÈS la dernière formalisation.""" st = self.load_state() pm, fm = st.get("plan_modifie_le", ""), st.get("formalise_le", "") if not pm: return False if not fm: return bool(st.get("dernier_plan_compact")) return pm > fm def has_state(self) -> bool: """Vrai si le projet a déjà un plan compact OU un narratif formalisé.""" st = self.load_state() return bool(st.get("dernier_plan_compact") or 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 avec le contexte d'un projet déjà travaillé. Priorité au PLAN COMPACT persisté (source de vérité) : on le RECHARGE à l'identique et on le réinjecte comme état de départ — l'agent ne réinvente rien. Repli sur la reconstruction depuis le Markdown uniquement si aucun plan compact n'a été sauvegardé (projets antérieurs à la persistance du plan). """ state = proj.load_state() plan = state.get("dernier_plan_compact", "") last_md = state.get("dernier_markdown", "") summary = proj.journal_summary() if plan: # RECHARGEMENT EXACT — pas de régénération reprime = ( "Nous reprenons un projet de présentation déjà travaillé. " "Voici le PLAN COMPACT EXACT sur lequel nous nous étions arrêtés. " "C'est l'état de référence : tu le conserves tel quel et tu " "travailles dessus. Ne le reformule pas, ne le réinvente pas, " "ne renumérotes que si je te demande une modification de flux.\n\n" "=== PLAN COMPACT ACTUEL ===\n\n" f"{plan}\n\n" ) if summary: reprime += f"=== JOURNAL DES DÉCISIONS ===\n\n{summary}\n\n" reprime += ( "Confirme en réaffichant ce plan compact À L'IDENTIQUE, puis " "attends mes instructions (modes f: flux / d: deep dive). " "N'ajoute aucun commentaire." ) info("Réamorçage — plan compact rechargé à l'identique.") out = session.start(reprime) display_narrator(out) return # Repli : pas de plan compact persisté → reconstruction depuis le Markdown if not last_md: return reprime = ( "Nous reprenons un projet déjà travaillé, mais seul le narratif " "formalisé est disponible (pas de plan compact sauvegardé).\n\n" "=== DERNIER NARRATIF FORMALISÉ ===\n\n" f"{last_md}\n\n" ) if summary: reprime += f"=== JOURNAL DES DÉCISIONS ===\n\n{summary}\n\n" reprime += ( "Reconstruis le PLAN COMPACT à partir de ce narratif (chapitres, " "slides numérotées en continu, tags core/OPT-XXX et audience). " "Affiche uniquement ce plan compact. On repartira de lui (modes f:/d:)." ) warn("Aucun plan compact sauvegardé — reconstruction depuis le narratif " "(peut différer légèrement).") 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() # Précharge le plan compact persisté (reprise) — source de vérité exacte last_plan: str = proj.load_state().get("dernier_plan_compact", "") 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 if last_plan and not session.started: info("Plan compact rechargé depuis la session précédente :") print("\n" + textwrap.indent(last_plan, " ")) if proj.plan_diverge(): warn("Le plan a été modifié après la dernière formalisation — " "pense à /formalise avant de générer.") 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) assets_info = list_assets(proj) if assets_info: context = ((context or "").strip() + "\n\n" + assets_info).strip() ok(f"{len(assets_info.splitlines()) - 1} image(s) dans assets/.") if not new_files and not assets_info: info("Aucun nouveau document dans inputs/ ni image dans assets/.") 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 proj.save_plan_compact(last_plan) 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 == "/afficher": if last_response: print("\n" + textwrap.indent(last_response, " ")) else: warn("Pas encore de réponse à afficher.") 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 proj.mark_formalised(formalized_md) 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 proj.save_plan_compact(last_plan) 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 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) def merge_revision(proj: "Project", partial_yaml_path: Path): """Fusion YAML des révisions ciblées (chantier C3). Remplace dans le dernier YAML complet les slides régénérées (appariement par position ; positions inconnues ajoutées en fin). Retourne le chemin du YAML complet fusionné, ou None si fusion impossible (l'appelant conserve alors le flux partiel actuel).""" state = proj.load_state() last = state.get("dernier_yaml") or "" if not last or not Path(last).exists(): warn("Fusion : pas de YAML complet précédent — PPTX partiel " "conservé.") return None try: full = yaml.safe_load(Path(last).read_text(encoding="utf-8")) part = yaml.safe_load( partial_yaml_path.read_text(encoding="utf-8")) except yaml.YAMLError as e: warn(f"Fusion : YAML illisible ({e}).") return None if not isinstance(full, dict) or not full.get("slides"): warn("Fusion : le YAML précédent ne contient pas de slides.") return None news = {} for s in (part or {}).get("slides", []): if isinstance(s, dict) and s.get("position"): news[int(s["position"])] = s if not news: warn("Fusion : aucune slide positionnée dans la révision.") return None merged, replaced = [], 0 for i, s in enumerate(full["slides"]): pos = int(s.get("position", i + 1)) if isinstance(s, dict) else i + 1 if pos in news: merged.append(news.pop(pos)) replaced += 1 else: merged.append(s) for pos in sorted(news): merged.append(news[pos]) full["slides"] = merged out = proj.out("revision_fusion", "yaml") out.write_text( yaml.safe_dump(full, allow_unicode=True, sort_keys=False, default_flow_style=False, width=100), encoding="utf-8") ok(f"Fusion : {replaced} slide(s) remplacée(s), " f"{len(merged)} au total → {out.name}") return out def list_assets(proj: "Project"): """Inventaire des images de projets//assets/ (chantier C4). Crée le dossier au premier appel. Retourne un bloc texte destiné au contexte Narrator, ou une chaîne vide si aucune image.""" assets = proj.root / "assets" assets.mkdir(exist_ok=True) exts = {".png", ".jpg", ".jpeg", ".webp"} files = sorted(p for p in assets.iterdir() if p.suffix.lower() in exts and p.is_file()) if not files: return "" lines = ["IMAGES DISPONIBLES DANS assets/ (utilisables dans les " "layouts image_split / image_full et le bloc freeform " "image, par leur nom de fichier) :"] for p in files: dims = "" try: from PIL import Image with Image.open(p) as im: dims = f" ({im.size[0]}×{im.size[1]})" except Exception: pass lines.append(f"- {p.name}{dims}") return "\n".join(lines) # ───────────────────────────────────────────────────────────────────────────── # 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", "image", "shape"} VALID_FREE_SHAPES = {"chevron", "arrow", "triangle", "pill", "donut", "bracket_left", "bracket_right", "oval", "diamond", "hexagon", "parallelogram", "moon"} FREE_TOKEN_RE = re.compile( r"^(navy|navy_light|coral|glacier|slate|card|card_alt|white|body|muted)" r"(@\d{1,3})?$") def validate_freeform(raw: str): """Valide un YAML freeform — palier 4 (C6), deux niveaux. Retourne (ok, message, data). Les avertissements n'empêchent pas le rendu ; ils sont intégrés au message.""" 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 def color_ok(c): return bool(FREE_TOKEN_RE.match(str(c).strip().lower())) errors, warns = [], [] 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", None): errors.append(f"Slide {i} : mode doit être 'light' ou " f"'dark'") bg = s.get("background") if bg and not color_ok(bg): errors.append(f"Slide {i} : background '{bg}' hors charte " f"(token ou token@NN)") blocks = s.get("blocks", []) if not blocks: errors.append(f"Slide {i} : aucun bloc") if len(blocks) > 15: errors.append(f"Slide {i} : {len(blocks)} blocs (max 15)") elif len(blocks) > 10: warns.append(f"Slide {i} : {len(blocks)} blocs — pense " f"respiration (10 max conseillé)") for j, b in enumerate(blocks, 1): loc = f"Slide {i} bloc {j}" bt = (b.get("type") or "text").lower() if bt not in VALID_BLOCK_TYPES: errors.append(f"{loc} : type '{bt}' invalide") if bt == "shape": sk = str(b.get("shape") or "").lower() if sk and sk not in VALID_FREE_SHAPES: errors.append( f"{loc} : shape '{sk}' inconnue " f"({', '.join(sorted(VALID_FREE_SHAPES))})") 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"{loc} : déborde " f"(col+w={col + w:.1f}>12)") if row + h > 12.01: errors.append(f"{loc} : déborde " f"(row+h={row + h:.1f}>12)") for key in ("color", "text_color"): c = b.get(key) if c and not color_ok(c): errors.append(f"{loc} : {key} '{c}' hors charte " f"(token ou token@NN, jamais de hex)") border = b.get("border") if border is not None: if not isinstance(border, dict): errors.append(f"{loc} : border doit être " f"{{color, weight}}") elif border.get("color") and not color_ok(border["color"]): errors.append(f"{loc} : border.color " f"'{border['color']}' hors charte") alpha = b.get("alpha") if alpha is not None: try: if not 0 <= float(alpha) <= 100: warns.append(f"{loc} : alpha {alpha} hors " f"0-100 — sera clampé") except (TypeError, ValueError): errors.append(f"{loc} : alpha '{alpha}' non " f"numérique") rot = b.get("rotation") if rot is not None: try: if float(rot) % 15 != 0: warns.append(f"{loc} : rotation {rot}° — sera " f"arrondie au pas de 15°") except (TypeError, ValueError): errors.append(f"{loc} : rotation '{rot}' non " f"numérique") radius = b.get("radius") if radius is not None: try: if not 0 <= float(radius) <= 0.5: warns.append(f"{loc} : radius {radius} hors " f"0-0.5 — sera clampé") except (TypeError, ValueError): errors.append(f"{loc} : radius '{radius}' non " f"numérique") if errors: return False, "\n ".join(errors), data msg = f"{len(slides)} slides freeform valides." if warns: msg += "\n ⚠ " + "\n ⚠ ".join(warns) return True, msg, 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, assets_dir: Optional[Path] = None) -> 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}") if assets_dir is None: # Déduction depuis PROJECTS_DIR si le YAML y vit (C4b) try: rel = yaml_path.resolve().relative_to( Path(PROJECTS_DIR).resolve()) assets_dir = Path(PROJECTS_DIR) / rel.parts[0] / "assets" except ValueError: assets_dir = None # hors projet : défaut du moteur cmd = [sys.executable, RENDER_ENGINE_PATH, str(yaml_path), str(pptx_out), "--theme", THEME_PATH, "--components", COMPONENTS_PATH, "--layouts", LAYOUTS_PATH] if assets_dir is not None: cmd += ["--assets", str(assets_dir)] info(f"Assets : {assets_dir}") result = subprocess.run(cmd, 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 def run_preview(pptx_path: Path, wait: bool = False) -> bool: """Aperçus PNG par slide via preview.sh (chantier C2). wait=False : lancement en arrière-plan (le DS218 est lent), sortie consignée dans _preview.log. wait=True : bloquant (CLI).""" script = Path(PREVIEW_SCRIPT).resolve() if not script.exists(): warn(f"{PREVIEW_SCRIPT} introuvable — aperçus indisponibles.") return False out_dir = pptx_path.parent / f"{pptx_path.stem}_previews" if wait: info("Génération des aperçus (quelques minutes sur le NAS)...") r = subprocess.run([str(script), str(pptx_path)], capture_output=True, text=True) if r.returncode == 0: ok(f"Aperçus : {out_dir}") ok(f"Galerie : {out_dir / 'index.html'}") return True warn("Échec de la génération des aperçus :") print(textwrap.indent((r.stderr or r.stdout or "?").strip(), " ")) return False log = pptx_path.parent / f"{pptx_path.stem}_preview.log" with open(log, "w", encoding="utf-8") as lf: subprocess.Popen([str(script), str(pptx_path)], stdout=lf, stderr=subprocess.STDOUT) info(f"Aperçus en arrière-plan → {out_dir}") info(f"Suivi : {log}") return True def maybe_preview(pptx_path) -> None: """Propose la génération des aperçus après une sortie PPTX.""" if not pptx_path or not Path(PREVIEW_SCRIPT).exists(): return if ask("Générer les aperçus PNG ? (o/N) :").lower() in ( "o", "oui", "y", "yes"): run_preview(pptx_path, wait=False) # ───────────────────────────────────────────────────────────────────────────── # 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)) if ENCODER_MODE == "schema": yaml_str, yaml_data, yaml_path, tries = run_encoder_schema( plan, layouts, proj) else: 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 merged = False if suffix == "revision_ciblee" and yaml_path: fused = merge_revision(proj, yaml_path) if fused: yaml_path, merged = fused, True info("Rendu du deck COMPLET fusionné.") manifest.file(yaml_path) pptx_path = run_render(yaml_path, proj.root / "assets") 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": if merged: persist_generation( proj, markdown=markdown, yaml_path=yaml_path, pptx_path=pptx_path, journal_entry="Révision ciblée fusionnée — deck complet régénéré.") else: 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.") info("(Si la fusion YAML a réussi — voir ci-dessus — le PPTX est déjà le deck complet.)") maybe_preview(pptx_path) 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}") maybe_preview(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, proj.root / "assets") 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") maybe_preview(pptx_path) 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)) if ENCODER_MODE == "schema" and not express: yaml_str, yaml_data, yaml_path, tries = run_encoder_schema( plan, layouts, proj) else: if ENCODER_MODE == "schema" and express: info("Mode express → Encoder agent (plan non annoté).") 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, proj.root / "assets") 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") maybe_preview(pptx_path) 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", "") # On est déjà en révision (choix explicite ci-dessus) : on lance # directement la session de travail, sans repasser par un menu # "veux-tu réviser ?" redondant. 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") # Boucle de suite : une fois cette révision faite, proposer d'itérer # encore ou de terminer — ce menu-ci n'est pas redondant. 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") parser.add_argument("--preview", metavar="PPTX", help="Aperçus PNG d'un PPTX existant (C2)") args = parser.parse_args() if args.preview: p = Path(args.preview) if not p.exists(): warn(f"Fichier introuvable : {p}") sys.exit(1) sys.exit(0 if run_preview(p, wait=True) else 1) 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}") local_assets = yaml_path.parent / "assets" assets_dir = local_assets if local_assets.is_dir() else None sys.exit(0 if run_render(yaml_path, assets_dir) else 1) run_pipeline() if __name__ == "__main__": main()