diff --git a/scripts/livecoding_video_pipeline.py b/scripts/livecoding_video_pipeline.py index 4f96361dd6..48eb24d980 100644 --- a/scripts/livecoding_video_pipeline.py +++ b/scripts/livecoding_video_pipeline.py @@ -1,6 +1,6 @@ """V0 narrow du pipeline livecoding-video (issue #15604, homage a une voix tierce). -**Scope** (V0 c.573 + etape 4 c.580) : +**Scope** (V0 c.573 + etape 2 c.1215 + etape 4 c.580) : Etape 1 (composition Strudel multi-pistes) livree en V0. Etape 4 (capture navigateur) livree ensuite : le REPL strudel.cc est pilote par Playwright headed — pattern injecte par hash d'URL, visuals @@ -8,10 +8,16 @@ par le moteur offline NATIF du REPL (Export to WAV, aucune instrumentation WebAudio, aucun routage systeme VB-Cable), video capturee par ``canvas.captureStream`` + MediaRecorder, mux ffmpeg. -Les etapes restantes (2 narration LLM, 3 TTS, 5 visualizer custom, -6 mixage ffmpeg complet) restent documentation-ONLY (HARD Tell -c.1102 : pas de pipeline squelette qui pretend faire ce qu'il ne -fait pas). +Etape 2 (narration poetique timestampee) livree : segments +``{start_s, end_s, text, intensity}`` generes par moteur template +deterministe (defaut, offline) ou moteur LLM optionnel (client +OpenAI-compat via ``OPENAI_API_KEY``), valides par +:func:`validate_narration` (garde anti-nomination d'artiste incluse — +ce n'est PAS un detecteur de verbatim, cf la limitation documentee +sur :func:`validate_narration`). Les etapes +restantes (3 TTS, 5 visualizer custom, 6 mixage ffmpeg complet) +restent documentation-ONLY (HARD Tell c.1102 : pas de pipeline +squelette qui pretend faire ce qu'il ne fait pas). **Etats cles** (cf issue #15604 et c.446 demucs Phase A deferree) : - Composition Strudel via template parametrable (PAS de LLM libre en @@ -21,8 +27,9 @@ pattern de fade-out coordonne musique + (futur) video. **Hors V0 narrow (c.574+, hand-off cycles suivants)** : -- Etape 2 narration LLM : sous-grain SEPARE avec claim sur - `MyIA.AI.Notebooks/GenAI/Audio/04-Applications/04-5-LiveCoding-LLM-Music.ipynb`. +- Etape 2 narration timestampee : LIVREE (c.1215) — moteur template + deterministe par defaut + moteur LLM optionnel, sortie JSON segments + (unite d'entree de l'etape 3 TTS et des sous-titres de l'etape 6). - Etape 3 TTS Kokoro/FishAudio : `scripts/audiobook_pipeline.py` deja disponible, integration differee a un cycle c.574+. - Etape 4 capture Playwright sur `https://strudel.cc/` : LIVREE (c.580). @@ -43,11 +50,14 @@ from __future__ import annotations import argparse +import json import math +import os +import re import sys -from dataclasses import dataclass +from dataclasses import asdict, dataclass from pathlib import Path -from typing import Dict, List, Optional +from typing import Dict, List, Optional, Tuple # --- V0 narrow : composition Strudel multi-pistes par template ---------------- @@ -175,6 +185,560 @@ def compose_strudel( return "\n".join(lines) +# --- Etape 2 : narration poetique timestampee (c.1215) ----------------------- + + +@dataclass(frozen=True) +class NarrationSegment: + """Segment de voix off timestampe (etape 2, #15604). + + Contrat JSON : ``{"start_s": float, "end_s": float, "text": str, + "intensity": float}`` — unite d'entree de l'etape 3 (TTS expressif, + benchmark #17244) et des sous-titres de l'etape 6. + """ + + start_s: float + end_s: float + text: str + intensity: float + + +# Garde ANTI-NOMINATION d'artiste tiers (voie 3 B.0, #15604) : bloque +# toute casse du nom de la source d'inspiration dans les textes generes. +# LIMITATION explicite : ce n'est PAS un detecteur de verbatim — detecter +# du verbatim exigerait un corpus de reference, lequel reste hors depot +# (bibliography-hygiene §2). Le verbatim est exclu d'une autre maniere : +# template = textes originaux par construction (ecrits dans ce module) ; +# LLM = exigence portee par le system prompt (mitigation d'instruction, +# non enforcement). Tokens construits par concatenation runtime pour +# qu'aucun grep sur la source ne trouve la forme contigue (pattern +# anonymisation c.578/c.634, cf test_no_voice_cloning_legal_proof). +NARRATION_FORBIDDEN_SUBSTRINGS: Tuple[str, ...] = ( + "switchan" + "gel", + "switch" + " angel", +) + +_NARRATION_TEXT_MAX_CHARS = 600 +_NARRATION_TIME_EPS = 1e-9 + + +@dataclass(frozen=True) +class NarrationPlan: + """Fenetre de voix off d'un style (issue #15604, etape 2) : + 60-90 s de narration pour les videos courtes rythmees, jusqu'a + 100 % de la duree pour les ambient lents.""" + + start_offset_s: float # intro musicale avant la premiere parole + span_fraction: float # part de la duree couverte par la voix off + span_max_s: float # plafond de la fenetre + segment_len_s: float # longueur cible d'un segment + + +NARRATION_PLANS: Dict[str, NarrationPlan] = { + "trance": NarrationPlan(4.0, 0.5, 90.0, 8.0), + "techno": NarrationPlan(3.0, 0.5, 90.0, 8.0), + "melancholy": NarrationPlan(6.0, 0.75, 240.0, 12.0), + "ambient": NarrationPlan(2.0, 1.0, 600.0, 18.0), +} + +# Mouvements narratifs, dans l'ordre du morceau. Les mouvements lead/ +# nappe ne sont narrés que si les voix correspondantes existent dans +# le script (>= 3 / >= 4 lignes `$:` composees a l'etape 1) : la +# narration nomme ce qui joue REELLEMENT. +_NARRATION_MOVEMENTS: List[str] = [ + "ouverture", "basse", "lead", "nappe", "modulation", "climax", "fade", +] +_NARRATION_MOVEMENT_MIN_VOICES: Dict[str, int] = {"lead": 3, "nappe": 4} + +# Banques de mouvements : textes FRANCAIS ORIGINAUX, ecrits pour ce +# pipeline, calibres sur le GENRE observe (nommer l'action du code puis +# glisser vers le poetique) SANS aucun verbatim tiers. Les instruments +# cites sont ceux des STYLES ci-dessus (narration ancree au script). +NARRATION_BANK: Dict[str, Dict[str, List[str]]] = { + "trance": { + "ouverture": [ + "Le kick entre le premier — quatre temps par cycle, une mécanique sans appel. La piste s'allume, la nuit commence.", + "Posons la pulsation : la grosse caisse marque chaque temps. Autour d'elle, le silence attend son tour.", + ], + "basse": [ + "La basse sawtooth glisse dessous, ronde et patiente. Quelque chose de lent passe sous la surface.", + "Une basse au filtre serré rejoint le kick — les deux battent ensemble maintenant, comme un seul muscle.", + ], + "lead": [ + "Voici le supersaw : des dents liquides, une mélodie qui cherche sa hauteur. Elle dit ce qu'on n'a pas le temps de dire.", + "Le lead supersaw prend son envol, filtre ouvert. C'est la voix du morceau — elle raconte, le reste accompagne.", + ], + "nappe": [ + "La nappe carrée s'installe derrière, discrète — une pièce qu'on éclaire avant d'y entrer.", + "Un accord tenu en arrière-plan : assez large pour marcher dessus, assez bas pour ne pas gêner.", + ], + "modulation": [ + "Le filtre s'ouvre lentement, le slider monte — la lumière avec. On laisse respirer, on laisse grandir.", + "Chaque tour de slider ajoute un degré. La même phrase, jamais deux fois la même couleur.", + ], + "climax": [ + "Tout joue ensemble : {voices_list}. Le morceau tient debout tout seul — on ne fait qu'assister.", + "Ici, plus rien à ajouter. Reste une seule question : combien de temps peut durer un présent pareil.", + ], + "fade": [ + "Le gain descend, cycle après cycle. Ce qui était plein se vide doucement, sans drame.", + "On éteint les voix une à une, dans l'ordre inverse. La piste rend la lumière.", + ], + }, + "ambient": { + "ouverture": [ + "Un sinus grave tout d'abord, presque rien. Un point dans une pièce vide.", + "La grosse caisse n'entre pas — elle suggère. Une frappe presque effacée, comme un souvenir de rythme.", + ], + "basse": [ + "La basse sinus tient le sol, une note longue par-ci par-là. Le sol respire.", + "Sous la surface, un sinus grave circule. Il ne demande rien, il attend.", + ], + "lead": [ + "Le supersine apparaît, au loin, grandi par sa pièce et son delay. Il faut du temps pour venir jusqu'ici.", + "Une note haute se détache, se répète en écho, puis se tait — le temps de s'approcher.", + ], + "nappe": [ + "La nappe sawtooth, filtre presque fermé, remplit l'espace entre les notes. On ne sait plus si c'est du son ou déjà de l'air.", + "Un accord très large ouvre et ferme son filtre au ralenti. Toute la pièce prend cette respiration.", + ], + "modulation": [ + "Le slowslider avance un paramètre à la fois. Rien ne saute ; tout dérive.", + "Le mouvement est si lent qu'on ne le voit qu'en se retournant : il y a une heure, nous n'étions pas ici.", + ], + "climax": [ + "Ce n'est pas un sommet, c'est une pleine — l'eau monte sans vague. Nous sommes dedans depuis déjà longtemps.", + "Toutes les voix flottent ensemble, aucune ne domine. Le morceau ne finira pas ; il cessera.", + ], + "fade": [ + "Le gain descend sans qu'on sache depuis quand. La dernière note a peut-être déjà eu lieu.", + "On n'éteint pas : on laisse la pièce retrouver son silence d'avant.", + ], + }, + "techno": { + "ouverture": [ + "Quatre kicks par cycle, secs, sans réverbération. Une usine qui ne s'excuse pas.", + "Le kick d'entrée — la seule autorité de ce morceau. Le reste suivra, ou ne suivra pas.", + ], + "basse": [ + "La basse carrée, filtre modulé, cogne à contretemps du mur. Deux machines, un seul rythme.", + "Un carré grave entre en boucle courte. Il tourne, il tourne — c'est exactement le but.", + ], + "lead": [ + "Le lead sawtooth crache ses notes, filtre droit. Un signal plus qu'une mélodie.", + "Des notes en dents de scie passent la porte, une par une. Chacune doit mériter sa place.", + ], + "nappe": [ + "Une nappe sombre tient l'arrière — à peine plus qu'une vibration de sol.", + "Derrière le mur de kick, un accord en scie tend la corde. Il ne cédera pas avant la fin.", + ], + "modulation": [ + "Le filtre de la basse respire au slider. Deux degrés d'ouverture, deux degrés de nuit.", + "On ne change pas le pattern : on change ce qui le traverse.", + ], + "climax": [ + "Pleine puissance : chaque voix à sa place, chaque temps occupé. La seule issue est de danser.", + "Le mur est complet. Rien à ajouter, rien à retirer — le mouvement est la mélodie.", + ], + "fade": [ + "Le gain retire les voix du haut vers le bas. L'usine ralentit ses lignes.", + "Derniers cycles, le mur s'efface par strates. La pièce retrouve son bruit propre.", + ], + }, + "melancholy": { + "ouverture": [ + "Une frappe fatiguée, presque effacée — le rythme d'un souvenir plus que d'un morceau.", + "Le kick entre à demi, comme quelqu'un qui frappe à une porte sans vouloir qu'on ouvre.", + ], + "basse": [ + "La basse triangle, douce et fermée, tient deux notes. Une phrase de deux mots.", + "Un triangle grave pose la seule certitude du morceau. Tout le reste demandera pardon.", + ], + "lead": [ + "Le supersine chante par petites phrases, trempé de delay. Chaque note revient plus loin, comme une lettre relue.", + "Une mélodie en mineur avance, s'arrête, reprend. Elle connaît la fin et y va quand même.", + ], + "nappe": [ + "La nappe sawtooth, filtre bas, garde la chambre à demi-ouverte. Lumière de couloir.", + "Un accord tenu, presque un soupir. La pièce entière penche du même côté.", + ], + "modulation": [ + "Le filtre s'ouvre à peine, un tout petit degré. C'est déjà l'événement de la journée.", + "Le slowslider tourne sur lui-même. Ce qui change ne se voit pas ; ce qui ne change pas s'entend.", + ], + "climax": [ + "Ce n'est pas l'explosion, c'est l'aveu : toutes les voix se le disent en même temps.", + "Un moment de pleine voix, puis la retenue revient. On savait qu'elle reviendrait.", + ], + "fade": [ + "Le gain descend comme une fièvre. Les notes s'éloignent sans se retourner.", + "Les dernières mesures rendent la mélodie au silence. Elle n'était pas à nous.", + ], + }, +} + +# Arc d'intensite par mouvement : intro douce -> climax -> fade (l'unite +# d'entree de la prosodie TTS de l'etape 3). +_NARRATION_INTENSITY_ANCHORS: Dict[str, float] = { + "ouverture": 0.25, "basse": 0.40, "lead": 0.60, "nappe": 0.55, + "modulation": 0.70, "climax": 0.92, "fade": 0.12, +} + + +def parse_strudel_highlights(strudel_script: str) -> Dict[str, object]: + """Extrait du script etape 1 les elements que la narration nomme. + + La narration n'est pas generique : voix, syntheses et effets sont + lus dans le script REELLEMENT compose (nombre de lignes ``$:``), + noms ``s('...')`` / ``.s("...")``, effets connus, BPM du header. + """ + voices = re.findall(r"^\$:", strudel_script, flags=re.MULTILINE) + sounds = sorted(set(re.findall(r'\.s\("([a-z0-9_]+)"\)', strudel_script))) + drums = sorted(set(re.findall(r"s\('([a-z0-9_]+)", strudel_script))) + effects = sorted(set( + re.findall( + r"\.(delay|room|lpf|pan|scope|pianoroll|slowslider|slider)\b", + strudel_script, + ) + )) + bpm_match = re.search(r"BPM=(\d+)", strudel_script) + return { + "voices": len(voices), + "sounds": sounds, + "drums": drums, + "effects": effects, + "bpm": int(bpm_match.group(1)) if bpm_match else None, + "has_fade": "fade-out" in strudel_script, + } + + +def _snap_to_cycle(t: float, cps: float) -> float: + """Aligne un timestamp sur une frontiere de cycle strudel (``cps``). + + La voix off demarre donc en phase avec la grille musicale ; si le + snap produit une valeur non positive, la valeur brute est rendue. + """ + snapped = round(t * cps) / cps + return snapped if snapped > 0.0 else t + + +def _voice_labels(style_name: str, n_voices: int) -> List[str]: + """Noms francais des voix REELLEMENT presentes dans le script. + + Sert a l'enumeration du climax : on ne nomme pas des voix absentes + (la narration suit le script, pas une liste fixe). Le nom du lead + est extrait du pattern du style (``supersaw``, ``supersine``...). + """ + labels = ["kick", "basse"] + if n_voices >= 3: + match = re.search(r'\.s\("([a-z0-9_]+)"\)', STYLES[style_name].pattern_lead) + labels.append(match.group(1) if match else "lead") + if n_voices >= 4: + labels.append("nappe") + return labels + + +def compose_narration_template( + strudel_script: str, + style_name: str, + duration_seconds: int, +) -> List[NarrationSegment]: + """Genere la narration par mouvements deterministes (moteur template). + + Le decoupage suit les voix du script : ouverture (kick), basse, + lead, nappe — seuls les mouvements dont la voix existe sont + narrés — puis modulation, climax, fade-out alignes sur le marqueur + du script. Frontieres alignees aux cycles strudel (cps du style), + intensite en rampe lineaire entre ancres de mouvements. + Deterministe : meme entree -> meme sortie (aucun alea). + """ + plan = NARRATION_PLANS[style_name] + highlights = parse_strudel_highlights(strudel_script) + n_voices = max(int(highlights["voices"]), 2) + + movements = [ + m for m in _NARRATION_MOVEMENTS + if _NARRATION_MOVEMENT_MIN_VOICES.get(m, 2) <= n_voices + ] + window_start = plan.start_offset_s + span = min(plan.span_fraction * duration_seconds, plan.span_max_s) + window_end = min(window_start + span, float(duration_seconds)) + window_len = window_end - window_start + + n_target = max(len(movements), round(window_len / plan.segment_len_s)) + # Chaque mouvement a au moins 1 segment ; les extras vont aux + # phases qui s'etirent (modulation, climax), en round-robin. + stretch = [m for m in ("modulation", "climax") if m in movements] + counts: Dict[str, int] = {m: 1 for m in movements} + extra = n_target - len(movements) + i = 0 + while extra > 0 and stretch: + m = stretch[i % len(stretch)] + counts[m] += 1 + extra -= 1 + i += 1 + + bank = NARRATION_BANK[style_name] + cps = style_cps(style_name) + anchor_list = [_NARRATION_INTENSITY_ANCHORS[m] for m in movements] + total_segs = sum(counts.values()) + voices_list = ", ".join(_voice_labels(style_name, n_voices)) + + segments: List[NarrationSegment] = [] + prev_end = round(window_start, 3) + cursor = window_start + for idx, m in enumerate(movements): + m_len = window_len * counts[m] / total_segs + for j in range(counts[m]): + raw_start = cursor + m_len * j / counts[m] + raw_end = cursor + m_len * (j + 1) / counts[m] + start = max(_snap_to_cycle(raw_start, cps), prev_end) + end = min(_snap_to_cycle(raw_end, cps), window_end) + if end <= start: + end = min(raw_end, window_end) + start_r = round(start, 3) + end_r = round(end, 3) + if start_r < prev_end: + start_r = prev_end + if end_r <= start_r: + end_r = min(start_r + 0.001, round(window_end, 3)) + if end_r <= start_r: + continue # frontiere snappee degeneree : segment omis + nxt = ( + anchor_list[idx + 1] if idx + 1 < len(anchor_list) + else anchor_list[-1] / 2.0 + ) + intensity = anchor_list[idx] + (nxt - anchor_list[idx]) * (j / counts[m]) + raw_text = bank[m][j % len(bank[m])] + text = ( + raw_text.format(voices_list=voices_list) + if "{voices_list}" in raw_text else raw_text + ) + segments.append(NarrationSegment( + start_s=start_r, + end_s=end_r, + text=text, + intensity=round(intensity, 3), + )) + prev_end = end_r + cursor += m_len + return segments + + +NARRATION_SYSTEM_PROMPT = ( + "Tu es l'auteur de la voix off d'une vidéo de livecoding musical " + "Strudel. Écris une narration poétique française slamée, orale et " + "rythmée, qui NOMME l'action réelle du code fourni (les pistes, les " + "synthés, les effets présents dans le script), puis glisse vers le " + "poétique. CONTRAINTES : français exclusivement ; texte 100 % " + "original ; interdiction absolue de citer, imiter ou nommer un " + "artiste ou une œuvre tiers ; interdiction de reprendre des paroles " + "existantes. SORTIE : strictement un tableau JSON d'objets " + '{"start_s": , "end_s": , "text": , ' + '"intensity": } couvrant la fenêtre temporelle demandée sans ' + "chevauchement, timestamps croissants dans [0, durée], intensity " + "dans [0, 1]. Aucun texte hors du JSON." +) + + +def _build_llm_client(): + """Construit le client OpenAI-compat depuis l'environnement. + + La cle est verifier AVANT l'import du package : un appel llm sans + ``OPENAI_API_KEY`` echoue avec un message explicite (pas de + fallback silencieux vers le template — Tell c.1102). + """ + api_key = os.environ.get("OPENAI_API_KEY") + if not api_key: + raise RuntimeError( + "moteur narration llm demande mais OPENAI_API_KEY absente de " + "l'environnement — définir la clé ou utiliser " + "--narration-engine template" + ) + try: + import openai + except ImportError as exc: + raise RuntimeError( + "moteur narration llm demande mais le package 'openai' n'est " + "pas installé (pip install openai) — ou utiliser " + "--narration-engine template" + ) from exc + kwargs = {"api_key": api_key} + base_url = os.environ.get("OPENAI_BASE_URL") + if base_url: + kwargs["base_url"] = base_url + return openai.OpenAI(**kwargs) + + +def compose_narration_llm( + strudel_script: str, + style_name: str, + duration_seconds: int, + llm_client=None, + llm_model: Optional[str] = None, +) -> List[NarrationSegment]: + """Genere la narration par LLM (client OpenAI-compat, etape 2). + + Prompt system dedie, calibre sur le GENRE (nommer l'action puis + glisser vers le poetique), francais exclusif ; l'exigence zero + verbatim / zero nom d'artiste tiers y est portee par INSTRUCTION + (mitigation, non enforcement — cf NARRATION_FORBIDDEN_SUBSTRINGS + pour la seule garde executable, anti-nomination). La sortie est + VALIDEE par :func:`validate_narration` — une reponse invalide + echoue explicitement, elle n'est jamais rafistolee. + """ + client = llm_client if llm_client is not None else _build_llm_client() + model = llm_model or os.environ.get("OPENAI_MODEL", "gpt-4o-mini") + plan = NARRATION_PLANS[style_name] + span = min(plan.span_fraction * duration_seconds, plan.span_max_s) + window_start = plan.start_offset_s + window_end = min(window_start + span, float(duration_seconds)) + n_target = max(3, round((window_end - window_start) / plan.segment_len_s)) + user_prompt = ( + f"Script Strudel (etape 1) :\n```\n{strudel_script}\n```\n" + f"Style : {style_name}. Durée totale : {duration_seconds} s.\n" + f"Fenêtre de voix off : [{window_start:.1f} s, {window_end:.1f} s], " + f"environ {n_target} segments. Nomme les voix, synthés et effets " + f"réellement présents dans le script ci-dessus." + ) + response = client.chat.completions.create( + model=model, + messages=[ + {"role": "system", "content": NARRATION_SYSTEM_PROMPT}, + {"role": "user", "content": user_prompt}, + ], + temperature=0.8, + ) + raw = response.choices[0].message.content or "" + cleaned = re.sub(r"^```(?:json)?\s*|\s*```$", "", raw.strip()) + try: + data = json.loads(cleaned) + except json.JSONDecodeError as exc: + raise ValueError( + f"sortie LLM non parsable en JSON ({exc}) — extrait : {cleaned[:200]!r}" + ) from exc + if not isinstance(data, list): + raise ValueError( + f"sortie LLM : attendu un tableau JSON, reçu {type(data).__name__}" + ) + segments: List[NarrationSegment] = [] + for i, item in enumerate(data): + try: + segments.append(NarrationSegment( + start_s=float(item["start_s"]), + end_s=float(item["end_s"]), + text=str(item["text"]), + intensity=float(item["intensity"]), + )) + except (KeyError, TypeError, ValueError) as exc: + raise ValueError(f"segment LLM #{i} invalide ({exc}) : {item!r}") from exc + return segments + + +def compose_narration( + strudel_script: str, + style_name: str, + duration_seconds: int, + engine: str = "template", + llm_client=None, + llm_model: Optional[str] = None, +) -> List[NarrationSegment]: + """Genere les segments de narration timestampes (etape 2, #15604). + + Moteurs : + - ``template`` (defaut) : deterministe, hors-ligne, mouvements + alignes sur les voix reelles du script et la grille de cycles. + - ``llm`` : client OpenAI-compat (``OPENAI_API_KEY`` obligatoire, + ``OPENAI_BASE_URL`` / ``OPENAI_MODEL`` optionnels) — echec + explicite sans cle, sortie validee, aucun fallback silencieux. + + Retourne des :class:`NarrationSegment` valides pour + ``duration_seconds`` (validateur : :func:`validate_narration`). + """ + if style_name not in STYLES: + raise ValueError( + f"style {style_name!r} inconnu ; styles disponibles : {sorted(STYLES.keys())}" + ) + if engine == "template": + segments = compose_narration_template( + strudel_script, style_name, duration_seconds + ) + elif engine == "llm": + segments = compose_narration_llm( + strudel_script, style_name, duration_seconds, + llm_client=llm_client, llm_model=llm_model, + ) + else: + raise ValueError( + f"moteur de narration inconnu {engine!r} (attendu 'template' ou 'llm')" + ) + validate_narration(segments, duration_seconds) + return segments + + +def validate_narration( + segments: List[NarrationSegment], + duration_seconds: int, +) -> None: + """Valide le contrat des segments (autorite finale etape 2 -> 3). + + Verifie : au moins 2 segments ; timestamps dans [0, duree], + croissants, sans recouvrement ; textes non vides (<= 600 + caracteres) sans nom d'artiste tiers (garde ANTI-NOMINATION, + voie 3 B.0) ; intensite dans [0, 1]. Raise ``ValueError`` + detaillee au premier defaut. + + LIMITATION : la garde de texte est anti-nomination, PAS un + detecteur de verbatim — un detecteur de verbatim exigerait un + corpus de reference (hors depot, bibliography-hygiene §2). Le + verbatim est exclu par provenance : template = textes originaux + par construction ; LLM = exigence du system prompt (instruction, + non enforcement). + """ + if len(segments) < 2: + raise ValueError( + f"narration : au moins 2 segments attendus, reçu {len(segments)}" + ) + prev_end = 0.0 + for i, seg in enumerate(segments): + if not (0.0 <= seg.start_s < seg.end_s <= duration_seconds + _NARRATION_TIME_EPS): + raise ValueError( + f"segment #{i} : timestamps invalides (start={seg.start_s}, " + f"end={seg.end_s}, durée max={duration_seconds})" + ) + if seg.start_s < prev_end - _NARRATION_TIME_EPS: + raise ValueError( + f"segment #{i} : recouvrement (start={seg.start_s} < " + f"fin précédente={prev_end})" + ) + if not seg.text or not seg.text.strip(): + raise ValueError(f"segment #{i} : texte vide") + if len(seg.text) > _NARRATION_TEXT_MAX_CHARS: + raise ValueError( + f"segment #{i} : texte trop long " + f"({len(seg.text)} > {_NARRATION_TEXT_MAX_CHARS})" + ) + lowered = seg.text.lower() + for needle in NARRATION_FORBIDDEN_SUBSTRINGS: + if needle in lowered: + raise ValueError( + f"segment #{i} : nom d'artiste tiers interdit " + "(garde anti-nomination, voie 3 B.0)" + ) + if not (0.0 <= seg.intensity <= 1.0): + raise ValueError( + f"segment #{i} : intensité hors [0, 1] ({seg.intensity})" + ) + prev_end = seg.end_s + + +def narration_to_json(segments: List[NarrationSegment]) -> str: + """Serialise les segments en JSON (entree de l'etape 3 TTS et des + sous-titres de l'etape 6). ``ensure_ascii=False`` : le texte + francais accentue reste lisible.""" + return json.dumps([asdict(s) for s in segments], ensure_ascii=False, indent=2) + + # --- V0 narrow : orchestrateur scaffold --------------------------------------- @@ -197,26 +761,33 @@ def run_pipeline( capture: bool = False, capture_seconds: int = 30, headless: bool = False, -) -> Dict[str, Optional[str]]: - """Orchestrateur V0 narrow : compose le script Strudel et documente - les autres etapes. + narration_engine: str = "template", + narration_json_path: Optional[str] = None, +) -> Dict[str, object]: + """Orchestrateur : compose le script Strudel, genere la narration + timestampee (etape 2) et documente les autres etapes. - V0 narrow retourne un dict avec une cle par etape et la valeur - 'deferred' pour les etapes non livrees en V0. C'est un contrat - HONNETE (Tell c.1102) : pas de pipeline squelette qui pretend - faire la capture navigateur quand il ne fait que composer. + Retourne un dict avec une cle par etape ; les etapes non livrees + portent la valeur 'deferred...'. C'est un contrat HONNETE (Tell + c.1102) : pas de pipeline squelette qui pretend faire ce qu'il ne + fait pas. Parametres : - style_name : style musical (cf :func:`compose_strudel`). - duration_seconds : duree cible en secondes. - - output_path : chemin du fichier .mp4 final (en V0, ce chemin - n'est PAS cree : l'integrateur ffmpeg est deferred). + - output_path : chemin du fichier .mp4 final (sans --capture, ce + chemin n'est PAS cree : l'integrateur ffmpeg est deferred). - tts_voice : voix TTS Kokoro/FishAudio (optionnel, deferred). - playwright_url : URL du navigateur Strudel (deferred). - ffmpeg_loudnorm_lufs : cible loudness (deferred). - - Retourne : dict avec cles 'strudel_script', 'narration', 'tts', - 'browser_capture', 'visualizer', 'final_mix', 'verdict'. + - narration_engine : moteur de l'etape 2 ('template' deterministe + par defaut, ou 'llm' — cf :func:`compose_narration`). + - narration_json_path : si fourni, ecrit les segments valides dans + ce fichier JSON (entree reelle de l'etape 3 TTS). + + Retourne : dict avec cles 'strudel_script', 'narration' (liste de + dicts segments), 'tts', 'browser_capture', 'visualizer', + 'final_mix', 'verdict'. """ strudel = compose_strudel( style_name=style_name, @@ -224,6 +795,19 @@ def run_pipeline( visuals=capture, ) + narration_segments = compose_narration( + strudel_script=strudel, + style_name=style_name, + duration_seconds=duration_seconds, + engine=narration_engine, + ) + if narration_json_path: + json_path = Path(narration_json_path) + json_path.parent.mkdir(parents=True, exist_ok=True) + json_path.write_text( + narration_to_json(narration_segments), encoding="utf-8" + ) + browser_capture: object = f"deferred ({playwright_url})" final_mix: object = f"deferred ({output_path}, loudnorm {ffmpeg_loudnorm_lufs} LUFS)" if capture: @@ -243,15 +827,16 @@ def run_pipeline( return { "strudel_script": strudel, - "narration": "deferred", # Phase B — LLM segments timestampes + "narration": [asdict(s) for s in narration_segments], "tts": "deferred" if tts_voice is None else f"requested={tts_voice}", "browser_capture": browser_capture, "visualizer": "deferred (V1 only — capture Strudel inclut scope/pianoroll)", "final_mix": final_mix, "verdict": ( - "Etape 1 (composition) + etape 4 (capture navigateur : export WAV " - "offline natif + MediaRecorder + mux ffmpeg) livrees ; etapes " - "2/3/5/6 deferred avec claim explicite par phase." + "Etape 1 (composition) + etape 2 (narration timestampee, " + f"moteur {narration_engine}) + etape 4 (capture navigateur : " + "export WAV offline natif + MediaRecorder + mux ffmpeg) livrees ; " + "etapes 3/5/6 deferred avec claim explicite par phase." ), } @@ -435,9 +1020,10 @@ def capture_repl_session( def _build_arg_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="livecoding_video_pipeline", - description="V0 narrow du pipeline livecoding-video (#15604). " - "Compose un script Strudel multi-pistes par template ; les " - "autres etapes sont deferred au cycle c.574+.", + description="Pipeline livecoding-video (#15604). Compose un " + "script Strudel multi-pistes par template (etape 1) et la " + "narration poetique timestampee (etape 2) ; les etapes 3/5/6 " + "restent deferred.", ) parser.add_argument( "--style", @@ -485,26 +1071,60 @@ def _build_arg_parser() -> argparse.ArgumentParser: help="lancer Chromium headless (DECONSEILLE : le canvas REPL ne " "s'anime pas sans fenetre compositée — mesure c.580)", ) + parser.add_argument( + "--narration-engine", + choices=("template", "llm"), + default="template", + help="etape 2 : moteur de narration (default: template, " + "deterministe hors-ligne ; llm = client OpenAI-compat, " + "OPENAI_API_KEY requise, echec explicite sans cle)", + ) + parser.add_argument( + "--narration-json", + default=None, + help="etape 2 : ecrire les segments timestampes valides dans ce " + "fichier JSON (entree de l'etape 3 TTS)", + ) return parser def main(argv: Optional[List[str]] = None) -> int: args = _build_arg_parser().parse_args(argv) - result = run_pipeline( - style_name=args.style, - duration_seconds=args.duration, - output_path=args.output, - tts_voice=args.tts_voice, - capture=args.capture, - capture_seconds=args.capture_seconds, - headless=args.headless, - ) + try: + result = run_pipeline( + style_name=args.style, + duration_seconds=args.duration, + output_path=args.output, + tts_voice=args.tts_voice, + capture=args.capture, + capture_seconds=args.capture_seconds, + headless=args.headless, + narration_engine=args.narration_engine, + narration_json_path=args.narration_json, + ) + except RuntimeError as exc: + # Echec explicite (ex. moteur llm sans cle) — pas de fallback + # silencieux vers template (Tell c.1102). + print(f"ERREUR : {exc}", file=sys.stderr) + return 2 # Verdict explicite — Tell c.1102 anti-stonewall print("=== Strudel script (etape 1, livree) ===") print(result["strudel_script"]) print() - print("=== Status des etapes 2-6 ===") - for key in ("narration", "tts", "browser_capture", "visualizer", "final_mix"): + narration = result["narration"] + first, last = narration[0], narration[-1] + print("=== Narration (etape 2, livree) ===") + print( + f" {len(narration)} segments, fenetre " + f"[{first['start_s']} s, {last['end_s']} s]" + ) + preview = first["text"] + print(f" apercu : {preview[:120]}{'...' if len(preview) > 120 else ''}") + if args.narration_json: + print(f" JSON ecrit : {args.narration_json}") + print() + print("=== Status des etapes 3-6 ===") + for key in ("tts", "browser_capture", "visualizer", "final_mix"): print(f" {key}: {result[key]}") print() print(f"=== Verdict ===\n{result['verdict']}") diff --git a/scripts/tests/test_livecoding_video_pipeline.py b/scripts/tests/test_livecoding_video_pipeline.py index b91b067009..94415e7eef 100644 --- a/scripts/tests/test_livecoding_video_pipeline.py +++ b/scripts/tests/test_livecoding_video_pipeline.py @@ -1,21 +1,25 @@ """Tests du pipeline livecoding-video (#15604). -Etapes couvertes par ces tests : 1 (composition Strudel, heritage V0) -et 4 (capture navigateur : logique PURE — encodage URL, calcul de -cycles, commande ffmpeg, visuals, deferral). La capture REELLE -(navigateur + reseau + ffmpeg) est provee par l'execution documentee -dans le body de la PR, pas par un test qui simulerait le navigateur : -verifier un verdict fake sur une capture qui n'a pas eu lieu = test -menteur (Tell c.1102). +Etapes couvertes par ces tests : 1 (composition Strudel, heritage V0), +2 (narration poetique timestampee : moteur template deterministe, +moteur LLM via client injecte — jamais un vrai appel reseau dans les +tests, la sortie LLM est VALIDEE, pas crue) et 4 (capture navigateur : +logique PURE — encodage URL, calcul de cycles, commande ffmpeg, +visuals, deferral). La capture REELLE (navigateur + reseau + ffmpeg) +est provee par l'execution documentee dans le body de la PR, pas par +un test qui simulerait le navigateur : verifier un verdict fake sur +une capture qui n'a pas eu lieu = test menteur (Tell c.1102). """ from __future__ import annotations import base64 +import json import subprocess import sys import urllib.parse from pathlib import Path +from types import SimpleNamespace import pytest @@ -28,15 +32,25 @@ _FORBIDDEN_VOICE_TOKEN = _FORBIDDEN_PERSONAL + "_voice" from scripts.livecoding_video_pipeline import ( + NARRATION_FORBIDDEN_SUBSTRINGS, + NARRATION_PLANS, + NARRATION_SYSTEM_PROMPT, + NARRATION_BANK, + NarrationSegment, STYLES, StrudelStyle, build_repl_url, + compose_narration, compose_strudel, cycles_for_duration, mux_ffmpeg, + narration_to_json, + parse_strudel_highlights, run_pipeline, style_cps, + validate_narration, ) +from scripts.livecoding_video_pipeline import main as pipeline_main class TestStrudelStyle: @@ -193,12 +207,331 @@ def test_commande_canonique(self, tmp_path): assert cmd[-1].endswith("out.mp4") +# --- Etape 2 : narration poetique timestampee (c.1215) ------------------------ + + +def _seg(start, end, text="Texte original, écrit pour le test.", intensity=0.5): + return NarrationSegment( + start_s=start, end_s=end, text=text, intensity=intensity + ) + + +class TestParseStrudelHighlights: + """La narration est ancrée au script : le parseur extrait ce qui + joue REELLEMENT (voix, synthés, effets, BPM, fade).""" + + def test_extraits_voix_sounds_effets(self): + script = compose_strudel(style_name="trance", duration_seconds=180, voices=4) + hl = parse_strudel_highlights(script) + assert hl["voices"] == 4 + assert "supersaw" in hl["sounds"] + assert "bd" in hl["drums"] + assert "lpf" in hl["effects"] + assert hl["bpm"] == 140 + assert hl["has_fade"] is True + + def test_deux_voix_ne_exposent_pas_le_lead(self): + script = compose_strudel(style_name="ambient", duration_seconds=120, voices=2) + hl = parse_strudel_highlights(script) + assert hl["voices"] == 2 + assert "supersine" not in hl["sounds"] + + +class TestComposeNarrationTemplate: + def test_deterministe(self): + """Meme entree -> meme sortie : aucun alea dans le moteur + template (re-executabilite CI).""" + for style in STYLES: + script = compose_strudel(style, 180) + a = compose_narration(script, style, 180) + b = compose_narration(script, style, 180) + assert a == b, f"{style}: la narration template doit etre deterministe" + + def test_tous_styles_valides(self): + for style in STYLES: + for dur in (60, 180, 600): + script = compose_strudel(style, dur) + segments = compose_narration(script, style, dur) + validate_narration(segments, dur) # ne leve pas + + def test_timestamps_dans_la_duree_sans_recouvrement(self): + script = compose_strudel("trance", 180) + segs = compose_narration(script, "trance", 180) + prev_end = 0.0 + for s in segs: + assert 0.0 <= s.start_s < s.end_s <= 180 + assert s.start_s >= prev_end + prev_end = s.end_s + + def test_trance_fenetre_plafonnee_90s(self): + """Issue #15604 etape 2 : 60-90 s de narration pour les videos + courtes rythmees (trance/techno), pas 100 % de la duree.""" + script = compose_strudel("trance", 180) + segs = compose_narration(script, "trance", 180) + span = segs[-1].end_s - segs[0].start_s + assert span <= 90.5 + assert segs[0].start_s >= NARRATION_PLANS["trance"].start_offset_s + + def test_ambient_couvre_toute_la_duree(self): + """Ambient lent : la voix off peut tenir 100 % de la duree.""" + script = compose_strudel("ambient", 180) + segs = compose_narration(script, "ambient", 180) + assert segs[0].start_s <= 3.0 + assert segs[-1].end_s >= 179.0 + + def test_arc_intensite_non_plat(self): + """L'intensite est l'unite d'entree de la prosodie TTS (etape 3, + cf acceptance 3 'prosodie non plate') : l'arc doit varier.""" + for style in STYLES: + script = compose_strudel(style, 180) + segs = compose_narration(script, style, 180) + distinct = {round(s.intensity, 3) for s in segs} + assert len(distinct) >= 3, f"{style}: intensite plate" + assert max(s.intensity for s in segs) >= 0.85, f"{style}: pas de climax" + assert min(s.intensity for s in segs) <= 0.3, f"{style}: pas de fade" + + def test_narration_nomme_laction_reelle(self): + """Calibration GENRE (issue #15604) : nommer l'action du code. + Le lead REELLEMENT present doit etre nomme (ancrage script).""" + lead_names = { + "trance": "supersaw", + "ambient": "supersine", + "techno": "sawtooth", + "melancholy": "supersine", + } + for style, lead in lead_names.items(): + script = compose_strudel(style, 180, voices=4) + segs = compose_narration(script, style, 180) + assert any(lead in s.text.lower() for s in segs), ( + f"{style}: le lead ({lead}) n'est jamais nomme" + ) + + def test_narration_suit_le_nombre_de_voix(self): + """voices=2 : les mouvements lead/nappe sont absents du script, + la narration ne doit pas nommer leurs instruments — y compris + dans l'enumeration dynamique du climax (_voice_labels).""" + script = compose_strudel("trance", 180, voices=2) + segs = compose_narration(script, "trance", 180) + texts = " ".join(s.text.lower() for s in segs) + assert "supersaw" not in texts + assert "nappe" not in texts + climax = [s.text for s in segs if "Tout joue ensemble" in s.text] + assert climax, "le mouvement climax doit etre present" + assert "kick, basse." in climax[0] # enumeration = voix reelles + + def test_frontieres_alignees_aux_cycles(self): + """Les frontieres tombent sur la grille de cycles (cps du style) + — la voix off demarre en phase avec la musique.""" + script = compose_strudel("trance", 180) + segs = compose_narration(script, "trance", 180) + cps = style_cps("trance") + boundaries = [segs[0].start_s] + [s.end_s for s in segs] + aligned = sum( + 1 for b in boundaries if abs(b * cps - round(b * cps)) < 1e-3 + ) + assert aligned >= len(boundaries) // 2, ( + f"{aligned}/{len(boundaries)} frontieres cycle-alignees" + ) + + def test_banques_sans_placeholder_oublie(self): + """Un placeholder {voices_list} non substitue dans la sortie + serait un bug de formatage (cf c.575 sur le marqueur fade).""" + for style in STYLES: + script = compose_strudel(style, 180) + for s in compose_narration(script, style, 180): + assert "{voices_list}" not in s.text + + def test_banques_completes(self): + """Chaque style porte une banque pour chaque mouvement du + template, avec au moins 2 lignes (variete).""" + for style, bank in NARRATION_BANK.items(): + for movement in ("ouverture", "basse", "lead", "nappe", + "modulation", "climax", "fade"): + assert movement in bank, f"{style}: mouvement {movement} absent" + assert len(bank[movement]) >= 2, ( + f"{style}/{movement}: au moins 2 lignes attendues" + ) + + +class TestValidateNarration: + def test_valide_segments_template(self): + segs = compose_narration(compose_strudel("techno", 120), "techno", 120) + validate_narration(segs, 120) # ne leve pas + + def test_rejette_moins_de_deux_segments(self): + with pytest.raises(ValueError, match="2 segments"): + validate_narration([_seg(0, 10)], 120) + + def test_rejette_recouvrement(self): + with pytest.raises(ValueError, match="recouvrement"): + validate_narration([_seg(0, 20), _seg(19, 40)], 120) + + def test_rejette_hors_duree(self): + with pytest.raises(ValueError, match="timestamps invalides"): + validate_narration([_seg(0, 20), _seg(20, 130)], 120) + + def test_rejette_intensite_hors_range(self): + bad = [_seg(0, 20, intensity=1.5), _seg(20, 40)] + with pytest.raises(ValueError, match="intensité"): + validate_narration(bad, 120) + + def test_rejette_texte_vide(self): + bad = [_seg(0, 20, text=" "), _seg(20, 40)] + with pytest.raises(ValueError, match="texte vide"): + validate_narration(bad, 120) + + def test_rejette_texte_trop_long(self): + bad = [_seg(0, 20, text="a" * 601), _seg(20, 40)] + with pytest.raises(ValueError, match="trop long"): + validate_narration(bad, 120) + + def test_rejette_nom_artiste_tiers(self): + """Garde ANTI-NOMINATION (voie 3 B.0) : toute variation de casse + du nom de la source d'inspiration est rejetee dans les textes. + NB : c'est une garde de NOM, pas un detecteur de verbatim (cf + limitation documentee sur validate_narration — un detecteur de + verbatim exigerait un corpus de reference hors depot).""" + for needle in NARRATION_FORBIDDEN_SUBSTRINGS: + for variant in (needle, needle.upper(), needle.title()): + bad = [ + _seg(0, 20, text=f"Il y avait {variant} un jour."), + _seg(20, 40), + ] + with pytest.raises(ValueError, match="artiste"): + validate_narration(bad, 120) + + +class _FakeCompletions: + def __init__(self, payload): + self._payload = payload + self.messages = None + + def create(self, model, messages, **kwargs): + self.messages = messages + content = ( + self._payload + if isinstance(self._payload, str) + else json.dumps(self._payload, ensure_ascii=False) + ) + return SimpleNamespace( + choices=[ + SimpleNamespace(message=SimpleNamespace(content=content)) + ] + ) + + +class _FakeClient: + """Client LLM de test : double de contrat (aucun appel reseau, la + reponse est VALIDEe par le pipeline, jamais crue).""" + + def __init__(self, payload): + self.chat = SimpleNamespace(completions=_FakeCompletions(payload)) + + +_VALID_LLM_PAYLOAD = [ + {"start_s": 5.0, "end_s": 30.0, + "text": "La pulsation s'installe, patiente, avant tout le reste.", + "intensity": 0.3}, + {"start_s": 30.0, "end_s": 60.0, + "text": "Le supersaw ouvre ses ailes au-dessus de nous.", + "intensity": 0.8}, + {"start_s": 60.0, "end_s": 90.0, + "text": "Puis tout redescend, comme une marée qui rentre.", + "intensity": 0.2}, +] + + +class TestComposeNarrationLLM: + def test_sortie_validee_et_client_appelle(self): + client = _FakeClient(_VALID_LLM_PAYLOAD) + script = compose_strudel("trance", 180) + segs = compose_narration( + script, "trance", 180, engine="llm", llm_client=client + ) + assert len(segs) == 3 + validate_narration(segs, 180) + system, user = client.chat.completions.messages + assert system["role"] == "system" + # prompt system : contrat JSON + originalite + aucun artiste tiers + assert system["content"] == NARRATION_SYSTEM_PROMPT + assert "JSON" in NARRATION_SYSTEM_PROMPT + assert "original" in NARRATION_SYSTEM_PROMPT.lower() + assert "artiste" in NARRATION_SYSTEM_PROMPT + # prompt user : le script reel et la duree sont envoyes + assert "setcps(" in user["content"] + assert "180" in user["content"] + + def test_sortie_non_json_rejetee(self): + client = _FakeClient("ceci n'est pas du JSON") + with pytest.raises(ValueError, match="JSON"): + compose_narration( + compose_strudel("trance", 180), "trance", 180, + engine="llm", llm_client=client, + ) + + def test_sortie_invalide_rejetee(self): + """Une sortie LLM recouvrante echoue explicitement — jamais + rafistolee (Tell c.1102).""" + bad_payload = [ + {"start_s": 0, "end_s": 50, "text": "A.", "intensity": 0.5}, + {"start_s": 40, "end_s": 80, "text": "B.", "intensity": 0.5}, + ] + client = _FakeClient(bad_payload) + with pytest.raises(ValueError, match="recouvrement"): + compose_narration( + compose_strudel("trance", 180), "trance", 180, + engine="llm", llm_client=client, + ) + + def test_sans_cle_echec_explicite(self, monkeypatch): + """Moteur llm sans OPENAI_API_KEY : RuntimeError explicite, PAS + de fallback silencieux vers template (Tell c.1102).""" + monkeypatch.delenv("OPENAI_API_KEY", raising=False) + with pytest.raises(RuntimeError, match="OPENAI_API_KEY"): + compose_narration( + compose_strudel("trance", 180), "trance", 180, engine="llm" + ) + + def test_package_openai_absent_echec_explicite(self, monkeypatch): + """Branche import de _build_llm_client : cle presente mais + package openai absent -> RuntimeError explicite (la cle est + verifiee AVANT l'import, cf _build_llm_client).""" + monkeypatch.setenv("OPENAI_API_KEY", "dummy-key-for-import-branch-test") + # sys.modules['openai'] = None force `import openai` a echouer + monkeypatch.setitem(sys.modules, "openai", None) + with pytest.raises(RuntimeError, match="openai"): + compose_narration( + compose_strudel("trance", 180), "trance", 180, engine="llm" + ) + + def test_moteur_inconnu_rejete(self): + with pytest.raises(ValueError, match="moteur"): + compose_narration( + compose_strudel("trance", 180), "trance", 180, + engine="hallucine", + ) + + +class TestNarrationToJson: + def test_roundtrip_et_accents(self): + segs = compose_narration(compose_strudel("ambient", 240), "ambient", 240) + data = json.loads(narration_to_json(segs)) + assert len(data) == len(segs) + assert set(data[0].keys()) == {"start_s", "end_s", "text", "intensity"} + # ensure_ascii=False : les accents restent lisibles dans le JSON + assert "é" in narration_to_json(segs) + # un dict JSON reconstruit == le dataclass d'origine + first = NarrationSegment(**data[0]) + assert first == segs[0] + + class TestRunPipeline: """L'orchestrateur marque les etapes non livrees comme ``deferred`` plutot que de pretendre les avoir executees (Tell c.1102). Depuis - l'etape 4 (c.580), browser_capture/final_mix ne sont deferred QUE - sans ``capture=True`` — les tests ci-dessous couvrent le cas sans - capture ; la capture reelle est provee dans le body de la PR.""" + l'etape 2 (c.1215), 'narration' est une liste de segments valides ; + depuis l'etape 4 (c.580), browser_capture/final_mix ne sont + deferred QUE sans ``capture=True`` — les tests ci-dessous couvrent + le cas sans capture ; la capture reelle est provee dans le body.""" def test_step1_strudel_returned(self): result = run_pipeline( @@ -210,25 +543,54 @@ def test_step1_strudel_returned(self): assert "setcps(" in result["strudel_script"] assert "BPM=140" in result["strudel_script"] - def test_steps_2_to_6_marked_deferred(self): - """Les etapes non livrees sont deferred et la valeur exacte doit - l'ecrire HONNETEMENT (pas un faux succes).""" + def test_narration_est_une_liste_de_segments(self): + """Depuis l'etape 2 (c.1215), 'narration' n'est plus 'deferred' : + une liste de dicts segments VALIDES (contrat d'entree etape 3).""" + result = run_pipeline( + style_name="ambient", + duration_seconds=240, + output_path="out/test.mp4", + ) + narration = result["narration"] + assert isinstance(narration, list) and len(narration) >= 2 + for item in narration: + assert set(item.keys()) == {"start_s", "end_s", "text", "intensity"} + segments = [NarrationSegment(**item) for item in narration] + validate_narration(segments, 240) + + def test_etapes_3_5_6_marquees_deferred(self): + """Les etapes non livrees restent deferred et la valeur exacte + doit l'ecrire HONNETEMENT (pas un faux succes).""" result = run_pipeline( style_name="ambient", duration_seconds=240, output_path="out/test.mp4", ) - assert result["narration"] == "deferred" assert result["tts"].startswith("deferred") or result["tts"].startswith("requested=") assert result["browser_capture"].startswith("deferred") assert result["visualizer"].startswith("deferred") assert result["final_mix"].startswith("deferred") # Le verdict reste explicite sur ce qui est livre vs deferred - # (adaptation c.580 : etape 4 livree, formulation V0 narrow - # remplacee par l'enumeration des etapes livrees/deferees). + # (adaptation c.1215 : etapes 1/2/4 livrees, 3/5/6 deferees). + assert "etape 2" in result["verdict"].lower() assert "etape 4" in result["verdict"].lower() assert "deferred" in result["verdict"].lower() + def test_narration_json_ecrit(self, tmp_path): + """--narration-json : le fichier est reellement ecrit et + rechargerable (entree de l'etape 3 TTS).""" + target = tmp_path / "narr" / "segments.json" + run_pipeline( + style_name="techno", + duration_seconds=120, + output_path="out/test.mp4", + narration_json_path=str(target), + ) + assert target.exists() + data = json.loads(target.read_text(encoding="utf-8")) + assert len(data) >= 2 + assert set(data[0].keys()) == {"start_s", "end_s", "text", "intensity"} + def test_output_path_is_documented_not_created(self): """Sans capture, le pipeline ne cree PAS le fichier .mp4 final ; le path est documente dans le verdict sans execution. Un test @@ -347,3 +709,55 @@ def test_cli_help_documente_la_capture(self): ) assert "--capture" in result.stdout assert "--capture-seconds" in result.stdout + + def test_cli_help_documente_la_narration(self): + """Les options etape 2 (--narration-engine/--narration-json) + doivent etre documentees dans l'aide.""" + result = subprocess.run( + [sys.executable, "scripts/livecoding_video_pipeline.py", "--help"], + capture_output=True, + text=True, + encoding="utf-8", + ) + assert "--narration-engine" in result.stdout + assert "--narration-json" in result.stdout + + def test_cli_narration_json(self, tmp_path): + """--narration-json ecrit un JSON valide via la commande unique + (critere 1 de l'acceptance, partie etape 2).""" + target = tmp_path / "segments.json" + result = subprocess.run( + [ + sys.executable, "scripts/livecoding_video_pipeline.py", + "--style", "techno", + "--duration", "120", + "--output", "out/cli_narr.mp4", + "--narration-json", str(target), + ], + capture_output=True, + text=True, + encoding="utf-8", + ) + assert result.returncode == 0 + assert "Narration (etape 2" in result.stdout + assert "deferred" in result.stdout # etapes 3/5/6 toujours honnetes + data = json.loads(target.read_text(encoding="utf-8")) + assert len(data) >= 2 + assert set(data[0].keys()) == {"start_s", "end_s", "text", "intensity"} + + def test_cli_llm_sans_cle_echec_explicite(self, monkeypatch, capsys): + """--narration-engine llm sans cle : code retour 2 et message + d'erreur explicite sur stderr — pas de fallback silencieux + vers template (Tell c.1102).""" + monkeypatch.delenv("OPENAI_API_KEY", raising=False) + rc = pipeline_main( + [ + "--style", "trance", + "--duration", "120", + "--output", "out/x.mp4", + "--narration-engine", "llm", + ] + ) + assert rc == 2 + err = capsys.readouterr().err + assert "OPENAI_API_KEY" in err