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

792 lines
32 KiB
Python
Raw Permalink Normal View History

Pipeline Sliding : archive v1 + pipeline v2 opérationnel (21 layouts)
2026-06-15 22:09:49 +02:00
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
facilitator_v3.py — Sliding Pipeline v5 · Pernod Ricard
========================================================
Orchestre les 3 agents Mistral + render_engine_v2, avec une
**organisation par projet**.
Arborescence :
projets/
<nom_du_projet>/
inputs/ ← documents de contexte (PDF, txt, md, docx) fournis par Bastien
outputs/ ← Markdown, plan, YAML, PPTX, manifest générés
Workflow :
Lancement → nouveau projet ou existant ?
→ documents dans inputs/ à analyser ?
→ brief
The Narrator (Agent 1) → Markdown [+ contexte documentaire injecté]
The Designer (Agent 2) → Plan [optionnel : mode express]
The Encoder (Agent 3) → YAML validé (contre layouts_v2)
render_engine_v2.py → PPTX
Nouveautés v5 :
- Organisation par projet (projets/<nom>/inputs|outputs)
- Sélection nouveau / existant au lancement
- Lecture des documents de contexte (inputs/) injectés au Narrator
- FIX : double 'output/output/' du manifest (séparation dir / nom de fichier)
Conserve de la v4 : retry réseau, garde pagination, historique cumulatif,
retour Narrator complet, yaml sort_keys=False, mode --render, mode express,
manifest JSON, archivage Trilium optionnel.
Variables .env :
MISTRAL_API_KEY (requis sauf mode --render)
NARRATOR_AGENT_ID / DESIGNER_AGENT_ID / ENCODER_AGENT_ID
PROJECTS_DIR (défaut : ./projets)
RENDER_ENGINE_PATH (défaut : render_engine_v2.py)
THEME_PATH / COMPONENTS_PATH / LAYOUTS_PATH
CONTEXT_MAX_CHARS (défaut : 12000 — plafond du contexte documentaire)
TRILIUM_API_URL / TRILIUM_API_KEY / TRILIUM_PROJET
"""
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
# ─────────────────────────────────────────────────────────────────────────────
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")
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 = "5.0"
MAX_PAGES = 10
MAX_RETRY = 3
PROJECTS_DIR.mkdir(parents=True, exist_ok=True)
# ─────────────────────────────────────────────────────────────────────────────
# 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 display_markdown(text: str, max_lines: int = 60):
lines = text.splitlines()
if len(lines) <= max_lines:
print(textwrap.indent(text, " "))
else:
print(textwrap.indent("\n".join(lines[:max_lines]), " "))
print(f"\n ... [{len(lines) - max_lines} lignes supplémentaires]")
def display_plan(text: str, max_lines: int = 80):
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_markdown(text, max_lines)
def slugify(name: str) -> str:
s = re.sub(r"[^a-zA-Z0-9]+", "-", name.lower()).strip("-")
return s[:40]
# ─────────────────────────────────────────────────────────────────────────────
# GESTION DE PROJET (nouveau v5)
# ─────────────────────────────────────────────────────────────────────────────
class Project:
"""Encapsule un projet : dossiers inputs/ et outputs/."""
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"
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}"
@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]:
"""Menu nouveau / existant au lancement."""
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 DES DOCUMENTS DE CONTEXTE (nouveau v5)
# ─────────────────────────────────────────────────────────────────────────────
def _read_pdf(path: Path) -> str:
try:
from pypdf import PdfReader
reader = PdfReader(str(path))
return "\n".join((pg.extract_text() or "") for pg in reader.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
doc = docx.Document(str(path))
return "\n".join(p.text for p in doc.paragraphs)
except Exception as e:
warn(f"DOCX illisible ({path.name}) : {e}")
return ""
def read_context_documents(proj: Project) -> str:
"""Lit tous les documents de inputs/ et retourne un bloc de contexte."""
if not proj.inputs.is_dir():
return ""
files = [f for f in sorted(proj.inputs.iterdir())
if f.is_file() and not f.name.startswith(".")]
if not files:
return ""
print()
info(f"{len(files)} document(s) trouvé(s) dans inputs/ :")
for f in files:
print(f" - {f.name}")
rep = ask("Analyser ces documents comme contexte ? (O/n) :").lower()
if rep in ("n", "non", "no"):
return ""
chunks = []
for f in files:
ext = f.suffix.lower()
if ext == ".pdf":
txt = _read_pdf(f)
elif ext == ".docx":
txt = _read_docx(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)")
if not chunks:
return ""
context = "\n\n".join(chunks)
if len(context) > CONTEXT_MAX_CHARS:
warn(f"Contexte tronqué à {CONTEXT_MAX_CHARS} caractères "
f"(total {len(context)}).")
context = context[:CONTEXT_MAX_CHARS] + "\n[...contexte tronqué...]"
return context
# ─────────────────────────────────────────────────────────────────────────────
# MANIFEST DE SESSION
# ─────────────────────────────────────────────────────────────────────────────
class Manifest:
def __init__(self, brief: str, proj: Project, has_context: bool):
self.data = {
"version": VERSION,
"demarrage": datetime.now().isoformat(timespec="seconds"),
"projet": proj.name,
"brief": brief,
"contexte_documentaire": has_context,
"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):
"""path = chemin COMPLET du fichier manifest (fix double-output)."""
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 — retry + pagination
# ─────────────────────────────────────────────────────────────────────────────
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_response = ""
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_response += content
if "PAUSE" in content and "FIN —" not in content:
info(f"Pagination détectée (page {page + 1}) → continuation...")
current.append({"role": "assistant", "content": content})
current.append({"role": "user", "content": "continue"})
else:
return full_response
warn(f"Garde pagination atteinte ({MAX_PAGES} pages) — "
"sortie tronquée possible.")
return full_response
class AgentSession:
def __init__(self, agent_id: str):
self.agent_id = agent_id
self.messages: list = []
def start(self, initial: str) -> str:
self.messages = [{"role": "user", "content": initial}]
out = call_agent(self.agent_id, self.messages)
self.messages.append({"role": "assistant", "content": out})
return out
def feedback(self, fb: str) -> str:
self.messages.append({"role": "user", "content": fb})
out = call_agent(self.agent_id, self.messages)
self.messages.append({"role": "assistant", "content": out})
return out
def reset(self, initial: str) -> str:
return self.start(initial)
# ─────────────────────────────────────────────────────────────────────────────
# VALIDATION YAML
# ─────────────────────────────────────────────────────────────────────────────
def extract_yaml(raw: str) -> str:
match = re.search(r"```ya?ml\s*(.*?)```", raw, re.DOTALL | re.IGNORECASE)
content = match.group(1).strip() if match else raw.strip()
return re.sub(r"[─]+", "-", content)
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, "Le YAML 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} : clé 'layout' manquante")
continue
if layout not in valid:
errors.append(f"Slide {pos} : layout '{layout}' inconnu "
f"(valides : {', '.join(sorted(valid))})")
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}) : champ requis "
f"'{field}' manquant ou vide")
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 save_text(content: str, path: Path) -> Path:
path.write_text(content, encoding="utf-8")
ok(f"Sauvegardé : {path}")
return path
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", {})
# ─────────────────────────────────────────────────────────────────────────────
# ARCHIVAGE TRILIUM (non bloquant)
# ─────────────────────────────────────────────────────────────────────────────
def archive_to_trilium(brief, pptx_path, manifest_path, proj):
if not (TRILIUM_API_URL and TRILIUM_API_KEY):
return
try:
detail = f"Projet : {proj.name} | Brief : {brief[:200]}"
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("Génération archivée dans Trilium.") if resp.ok else \
warn(f"Trilium : réponse {resp.status_code} (non bloquant)")
except requests.RequestException as e:
warn(f"Trilium injoignable ({e}) — archivage ignoré.")
# ─────────────────────────────────────────────────────────────────────────────
# ÉTAPE 1 — NARRATOR (avec contexte documentaire)
# ─────────────────────────────────────────────────────────────────────────────
def build_narrator_input(brief: str, context: str) -> str:
if context:
return (f"{brief}\n\n---\nCONTEXTE DOCUMENTAIRE FOURNI "
f"(à utiliser pour enrichir et cadrer la présentation) :\n\n"
f"{context}")
return brief
def run_narrator(session, brief, context, proj, first_output=None):
section("ÉTAPE 1 — THE NARRATOR")
if first_output is None:
if context:
info("Génération du Markdown narratif (avec contexte documentaire)...")
else:
info("Génération du Markdown narratif...")
output = session.start(build_narrator_input(brief, context))
ok(f"Markdown généré ({len(output)} caractères)")
else:
output = first_output
while True:
print()
display_markdown(output)
print("\n [1] Valider → étape suivante")
print(" [2] Donner du feedback au Narrator")
print(" [3] Régénérer complètement")
print(" [4] Afficher le Markdown complet")
print(" [5] Sauvegarder le Markdown")
choix = ask("Votre choix :")
if choix == "1":
ok("Markdown approuvé.")
return output
elif choix == "2":
fb = ask("Votre feedback :")
if fb:
info("Envoi du feedback au Narrator...")
output = session.feedback(fb)
ok(f"Nouveau Markdown ({len(output)} caractères)")
elif choix == "3":
info("Régénération depuis le brief...")
output = session.reset(build_narrator_input(brief, context))
ok(f"Nouveau Markdown ({len(output)} caractères)")
elif choix == "4":
print("\n" + textwrap.indent(output, " "))
elif choix == "5":
save_text(output, proj.out("narrator_draft", "md"))
else:
warn("Choix invalide. Entrez 1 à 5.")
# ─────────────────────────────────────────────────────────────────────────────
# ÉTAPE 2 — DESIGNER
# ─────────────────────────────────────────────────────────────────────────────
def run_designer(session, markdown, proj):
section("ÉTAPE 2 — THE DESIGNER")
info("Génération du plan de présentation...")
output = session.start(markdown)
ok(f"Plan généré ({len(output)} caractères)")
while True:
print()
display_plan(output)
print("\n [1] Valider → passer à l'Encoder")
print(" [2] Donner du feedback au Designer")
print(" [3] Afficher le plan complet")
print(" [4] Sauvegarder le plan")
print(" [0] Revenir au Narrator (modifier le Markdown)")
choix = ask("Votre choix :")
if choix == "1":
ok("Plan approuvé. Passage à l'Encoder...")
save_text(output, proj.out("designer", "txt"))
return output
elif choix == "2":
fb = ask("Votre feedback :")
if fb:
info("Envoi du feedback au Designer...")
output = session.feedback(fb)
ok(f"Nouveau plan ({len(output)} caractères)")
elif choix == "3":
print("\n" + textwrap.indent(output, " "))
elif choix == "4":
save_text(output, proj.out("designer_draft", "txt"))
elif choix == "0":
return None
else:
warn("Choix invalide. Entrez 0 à 4.")
# ─────────────────────────────────────────────────────────────────────────────
# ÉTAPE 3 — ENCODER
# ─────────────────────────────────────────────────────────────────────────────
def run_encoder(session, plan, layouts, proj):
section("ÉTAPE 3 — THE ENCODER")
info("Encodage du plan en YAML...")
digest = layouts_digest(layouts) if layouts else ""
initial = f"{plan}\n\n---\n{digest}" if digest else plan
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"Le YAML que tu as généré contient des erreurs :\n\n{message}\n\n"
f"Corrige ces erreurs et renvoie le YAML COMPLET corrigé, "
f"sans aucun texte avant ou après.")
ok(f"YAML corrigé reçu ({len(output)} caractères)")
print("\n [1] Accepter le YAML tel quel (avec erreurs)")
print(" [2] Donner un feedback manuel à l'Encoder")
print(" [0] Abandonner et 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
# ─────────────────────────────────────────────────────────────────────────────
# É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 path in [THEME_PATH, COMPONENTS_PATH, LAYOUTS_PATH]:
if not os.path.exists(path):
warn(f"Fichier de config manquant : {path}")
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 dans render_engine_v2.py :")
print(textwrap.indent(result.stderr or result.stdout, " "))
return None
# ─────────────────────────────────────────────────────────────────────────────
# 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")
# Sélection du projet (une fois par lancement)
section("PROJET")
proj = select_project()
if proj is None:
print(" Au revoir.")
return
while True:
# Contexte documentaire (inputs/)
context = read_context_documents(proj)
brief = ask("Votre brief (ou 'projet' pour changer, 'quit') :")
if brief.lower() in ("quit", "exit", "q"):
print(" Au revoir.")
break
if brief.lower() == "projet":
section("PROJET")
np = select_project()
if np:
proj = np
continue
if not brief:
continue
express = ask("Mode express — sans Designer ? (o/N) :").lower() \
in ("o", "oui", "y", "yes")
if express:
info("Mode express : Narrator → Encoder direct.")
manifest = Manifest(brief, proj, bool(context))
narrator = AgentSession(NARRATOR_ID)
markdown = run_narrator(narrator, brief, context, proj)
if not markdown:
continue
save_text(markdown, proj.out("narrator", "md"))
manifest.step("narrator", caracteres=len(markdown),
contexte=bool(context))
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:
markdown = run_narrator(narrator, brief, context, proj,
first_output=markdown)
if not markdown:
break
save_text(markdown, proj.out("narrator", "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.")
break
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}")
# FIX : chemin complet du manifest (plus de double output/)
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(brief, pptx_path, manifest_path, proj)
print()
if ask("Nouvelle présentation ? (o/n) :").lower() \
not in ("o", "oui", "y", "yes"):
print(" Au revoir.")
break
# ─────────────────────────────────────────────────────────────────────────────
# POINT D'ENTRÉE
# ─────────────────────────────────────────────────────────────────────────────
def main():
parser = argparse.ArgumentParser(
description="Sliding facilitator v3 — pipeline par projet + render v2")
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