Pipeline Video de Production

Module : 04-Applications
Niveau : Applications
Technologies : OpenAI API, diffusers, moviepy, pydub, Pillow
Duree estimee : 55 minutes
VRAM : ~18 GB (optionnel, CPU possible pour l’essentiel)

Objectifs d’Apprentissage

• Generer un script video a partir d’un sujet avec un LLM
• Produire des images pour chaque scene (DALL-E ou generation locale)
• Animer les images avec des effets cinematographiques (zoom, pan, avec moviepy)
• Generer une narration vocale avec OpenAI TTS
• Créer des sous-titres synchronises a partir du texte de narration
• Assembler le pipeline complet : video + audio + sous-titres
• Evaluer la qualite avec des metriques de production

Prerequis

• Python 3.10+
• Ensemble des notebooks Video 01-1 a 04-3
• Notebooks Audio (recommandes pour la partie TTS et sous-titres)
• Packages : moviepy, Pillow, numpy, matplotlib, openai, pydub
• Optionnel : diffusers, torch pour la generation d’images locale
• Cle API OpenAI pour TTS et generation de contenu

Ce notebook est le point culminant de la serie Video et fait reference aux techniques des series Video et Audio.

Navigation : << 04-3 Sora API | Audio 01-1 >>

# Parametres Papermill - JAMAIS modifier ce commentaire

# Configuration notebook
notebook_mode = "interactive"        # "interactive" ou "batch"
skip_widgets = False               # True pour mode batch MCP
debug_level = "INFO"

# Parametres du pipeline
video_topic = "Intelligence Artificielle"  # Sujet de la video
video_width = 1280                 # Largeur video finale
video_height = 720                 # Hauteur video finale
video_fps = 30                     # FPS de la video finale
scene_duration = 8                 # Duree par scene (secondes)
output_format = "mp4"              # Format de sortie

# Options du pipeline
enable_llm_script = True           # Generer le script avec un LLM
enable_image_gen = True            # Generer les images des scenes
enable_animation = True            # Animer les images
enable_tts = True                  # Generer la narration TTS
enable_subtitles = True            # Generer les sous-titres
enable_assembly = True             # Assembler le tout
use_mock_api = True                # Utiliser des reponses simulees si API indisponible
save_results = True                # Sauvegarder les resultats

On importe ensuite toutes les bibliotheques du pipeline de production : client OpenAI pour le LLM et le TTS, moviepy pour l’assemblage video, et PIL pour la manipulation des images.

# Setup environnement et imports
import os
import sys
import json
import time
import math
import struct
import warnings
from pathlib import Path
from datetime import datetime
from typing import Dict, List, Any, Optional, Tuple
import numpy as np
from PIL import Image, ImageDraw, ImageFont
import matplotlib.pyplot as plt
import logging

warnings.filterwarnings('ignore', category=DeprecationWarning)

# Resolution GENAI_ROOT
# Chargement robuste de la configuration .env
from dotenv import load_dotenv
import os
# Recherche du .env dans tous les parents (pour Papermill qui change le cwd)
current_path = Path.cwd()
# Determine GenAI root directory
GENAI_ROOT = Path.cwd()
while GENAI_ROOT.name != "GenAI" and len(GENAI_ROOT.parts) > 1:
    GENAI_ROOT = GENAI_ROOT.parent

env_loaded = False
while current_path.name != "GenAI" and len(current_path.parts) > 1:
    env_path = current_path / ".env"
    if env_path.exists():
        load_dotenv(env_path)
        print(f".env charge depuis: {env_path.name}")
        env_loaded = True
        break
    current_path = current_path.parent
if not env_loaded:
    print("WARNING: .env non trouve, utilisation variables environnement")

HELPERS_PATH = GENAI_ROOT / 'shared' / 'helpers'
if HELPERS_PATH.exists():
    sys.path.insert(0, str(HELPERS_PATH.parent))
    try:
        from helpers.video_helpers import get_video_info, extract_frames, display_frame_grid
        print("Helpers video importes")
    except ImportError as e:
        print(f"Helpers video non disponibles ({e}) - mode autonome")

# Repertoire de sortie
OUTPUT_DIR = GENAI_ROOT / 'outputs' / 'video'
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
PIPELINE_DIR = OUTPUT_DIR / 'pipeline'
PIPELINE_DIR.mkdir(parents=True, exist_ok=True)

# Configuration logging
logging.basicConfig(level=getattr(logging, debug_level))
logger = logging.getLogger('production_pipeline')

print(f"Pipeline Video de Production")
print(f"Date : {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print(f"Mode : {notebook_mode}")
print(f"Sujet : {video_topic}")
print(f"Video : {video_width}x{video_height} @ {video_fps}fps")
print(f"Sortie : {PIPELINE_DIR}")

WARNING: .env non trouve, utilisation variables environnement
Helpers video importes
Pipeline Video de Production
Date : 2026-09-30 10:28:25
Mode : interactive
Sujet : Intelligence Artificielle
Video : 1280x720 @ 30fps
Sortie : D:\Dev\CoursIA-18529\MyIA.AI.Notebooks\GenAI\outputs\video\pipeline

Les variables d’environnement sont ensuite chargees : cles OpenAI pour le LLM et le TTS, chemins de sortie, et indicateurs d’activation des étapes du pipeline (generation, animation, narration).

# Chargement .env et verification des dependances
from dotenv import load_dotenv
import os

# Determine GenAI root directory
GENAI_ROOT = Path.cwd()
while GENAI_ROOT.name != "GenAI" and len(GENAI_ROOT.parts) > 1:
    GENAI_ROOT = GENAI_ROOT.parent

# Chargement robuste du .env
env_path = GENAI_ROOT / ".env"
if env_path.exists():
    load_dotenv(env_path)
    print(f".env charge depuis: {env_path.name}")
else:
    # Fallback: chercher dans les parents
    current_path = Path.cwd()
    while len(current_path.parts) > 1:
        ep = current_path / ".env"
        if ep.exists():
            load_dotenv(ep)
            print(f".env charge depuis: {ep}")
            break
        current_path = current_path.parent
    else:
        print("WARNING: .env non trouve")

print("\n--- VERIFICATION DES DEPENDANCES ---")
print("=" * 40)

dependencies = {}

# moviepy 2.x : `moviepy.editor` est l'ancienne API 1.x retiree en 2.0.
# On importe le top-level `moviepy` (qui existe toujours en 2.x) et on
# verifie la version pour diagnostic (cf 04-2 Creative Video Workflows).
try:
    import moviepy
    dependencies['moviepy'] = True
    print(f"moviepy {moviepy.__version__} : disponible")
except ImportError:
    dependencies['moviepy'] = False
    print(f"moviepy : NON INSTALLE (pip install moviepy)")

try:
    import imageio
    dependencies['imageio'] = True
    print(f"imageio : disponible")
except ImportError:
    dependencies['imageio'] = False
    print(f"imageio : NON INSTALLE")

try:
    from openai import OpenAI
    dependencies['openai'] = True
    print(f"openai : disponible")
except ImportError:
    dependencies['openai'] = False
    print(f"openai : NON INSTALLE (pip install openai)")

try:
    from pydub import AudioSegment
    dependencies['pydub'] = True
    print(f"pydub : disponible")
except ImportError:
    dependencies['pydub'] = False
    print(f"pydub : NON INSTALLE (pip install pydub)")

try:
    import diffusers
    import torch
    dependencies['diffusers'] = True
    gpu_available = torch.cuda.is_available()
    print(f"diffusers : disponible (GPU: {gpu_available})")
except ImportError:
    dependencies['diffusers'] = False
    print(f"diffusers : NON INSTALLE (optionnel)")

# Verifier cle OpenAI
openai_key = os.environ.get('OPENAI_API_KEY', '')
openai_base = os.environ.get('OPENAI_BASE_URL', '')
api_available = bool(openai_key and dependencies.get('openai', False))

print(f"\nAPI OpenAI : {'disponible' if api_available else 'non disponible'}")
print(f"Mode mock : {use_mock_api or not api_available}")

available_count = sum(dependencies.values())
total_count = len(dependencies)
print(f"Dependances disponibles : {available_count}/{total_count}")

.env charge depuis: .env

--- VERIFICATION DES DEPENDANCES ---
========================================
moviepy 2.1.2 : disponible
imageio : disponible
openai : disponible
pydub : disponible
diffusers : disponible (GPU: True)

API OpenAI : disponible
Mode mock : True
Dependances disponibles : 5/5

1. Generation du script video avec LLM

Le pipeline commence par la generation d’un script structure. Un LLM (GPT-4 ou equivalent) produit une liste de scenes avec : - Description visuelle (pour la generation d’images) - Texte de narration (pour le TTS) - Instructions de camera (pour l’animation)

# Etape 1 : Generation du script
print("\n--- ETAPE 1 : GENERATION DU SCRIPT ---")
print("=" * 45)


def generate_script_with_llm(topic: str) -> List[dict]:
    """Genere un script video a partir d'un sujet en utilisant l'API OpenAI."""
    return None  # TODO étudiant : implémenter generate_script_with_llm
# Generer le script
script_scenes = generate_script_with_llm(video_topic)

if script_scenes is None:
    print("Exercice a completer : generate_script_with_llm -- script vide, "
          "le pipeline deroule ses etapes a vide")
    script_scenes = []

print(f"Script genere : {len(script_scenes)} scenes")
print(f"Duree totale estimee : {len(script_scenes) * scene_duration}s")
print(f"\nResume du script :")
for i, scene in enumerate(script_scenes):
    print(f"  Scene {i+1} : {scene['title']}")
    print(f"    Visuel : {scene['visual_description'][:60]}...")
    print(f"    Narration : {scene['narration'][:60]}...")
    print(f"    Camera : {scene['camera_motion']}")

--- ETAPE 1 : GENERATION DU SCRIPT ---
=============================================
Exercice a completer : generate_script_with_llm -- script vide, le pipeline deroule ses etapes a vide
Script genere : 0 scenes
Duree totale estimee : 0s

Resume du script :

Interpretation : Generation du script

Élément du script	Rôle dans le pipeline
title	Identifiant de la scene, sous-titrage des chapitres
visual_description	Prompt pour la generation d’images (DALL-E ou local)
narration	Texte pour le TTS (voix off) et les sous-titres
camera_motion	Instruction d’animation (zoom, pan, etc.)

Points cles : 1. Le script structure sert de plan directeur pour toutes les étapes suivantes 2. La qualite du prompt LLM influence directement la coherence de la video finale 3. Un bon script equilibre la duree des narrations avec le temps d’affichage des visuels

2. Generation des images de chaque scene

A partir des descriptions visuelles du script, nous generons une image representant chaque scene. En production, on utiliserait DALL-E 3 ou un modèle local comme SDXL. Ici, nous generons des images stylisees avec Pillow.

# Etape 2 : Generation des images
print("\n--- ETAPE 2 : GENERATION DES IMAGES ---")
print("=" * 45)


def generate_scene_image(scene: dict, width: int, height: int,
                         scene_idx: int) -> Image.Image:
    """
    Genere une image representative pour une scene.
    En mode demo, utilise Pillow pour un rendu stylise.
    En production, utiliserait DALL-E 3 ou SDXL.
    """
    return None  # TODO étudiant : implémenter generate_scene_image
# Generer les images
scene_images = []
for i, scene in enumerate(script_scenes):
    img = generate_scene_image(scene, video_width, video_height, i)
    scene_images.append(img)
    
    if img is not None and save_results:
        img_path = PIPELINE_DIR / f"scene_{i+1:02d}.png"
        img.save(str(img_path))
    
    etat = "Image generee" if img is not None else "exercice a completer (generate_scene_image)"
    print(f"  Scene {i+1} : {scene['title']} - {etat}")

print(f"\n{len(scene_images)} images generees ({video_width}x{video_height})")

# Apercu en grille (seulement les images effectivement generees)
valid_images = [(im, script_scenes[j]) for j, im in enumerate(scene_images) if im is not None]
n_scenes = len(valid_images)
cols = min(n_scenes, 5)
if cols == 0:
    print("Apercu : aucune image generee (exercice generate_scene_image a completer)")
else:
    fig, axes = plt.subplots(1, cols, figsize=(16, 4))
    if cols == 1:
        axes = [axes]
    for i in range(cols):
        axes[i].imshow(valid_images[i][0])
        axes[i].set_title(f"Scene {i+1}: {valid_images[i][1]['title']}", fontsize=9)
        axes[i].axis('off')
    plt.suptitle("Images des scenes du pipeline", fontsize=13, fontweight='bold')
    plt.tight_layout()
    plt.show()

--- ETAPE 2 : GENERATION DES IMAGES ---
=============================================

0 images generees (1280x720)
Apercu : aucune image generee (exercice generate_scene_image a completer)

3. Animation des images

Chaque image statique est transformee en sequence animee avec un mouvement de camera cinematographique : zoom avant/arriere, panoramique gauche/droite. Ces effets, connus sous le nom d’“effets Ken Burns”, donnent vie aux images fixes.

# Etape 3 : Animation des images avec effets camera
if enable_animation:
    print("\n--- ETAPE 3 : ANIMATION DES IMAGES ---")
    print("=" * 45)
    
    def animate_image(image: Image.Image, motion: str, duration: float,
                      fps: int) -> List[np.ndarray]:
        """
        Anime une image avec un mouvement de camera.
        
        Mouvements : zoom_in, zoom_out, pan_left, pan_right, slow_zoom
        """
        return None  # TODO étudiant : implémenter animate_image
    # Animer toutes les scenes
    animated_scenes = []
    total_frames = 0
    
    for i, (image, scene) in enumerate(zip(scene_images, script_scenes)):
        motion = scene.get('camera_motion', 'slow_zoom')
        scene_frames = animate_image(image, motion, scene_duration, video_fps)
        animated_scenes.append(scene_frames)
        if scene_frames:
            total_frames += len(scene_frames)
            print(f"  Scene {i+1} : {scene['title']} - {len(scene_frames)} frames ({motion})")
        else:
            print(f"  Scene {i+1} : {scene['title']} - exercice a completer (animate_image)")
    
    print(f"\nTotal frames animees : {total_frames}")
    print(f"Duree totale : {total_frames / video_fps:.1f}s")
else:
    # Pas d'animation : frames statiques
    animated_scenes = []
    for image in scene_images:
        static = [np.array(image)] * (scene_duration * video_fps) if image is not None else []
        animated_scenes.append(static)
    print("Animation desactivee - frames statiques")

--- ETAPE 3 : ANIMATION DES IMAGES ---
=============================================

Total frames animees : 0
Duree totale : 0.0s

4. Generation de la narration TTS

La narration vocale est generee a partir du texte de chaque scene avec OpenAI TTS. En mode demonstration, nous generons un signal audio synthetique (ton sinusoidal) de la bonne duree pour simuler la narration.

# Etape 4 : Generation de la narration TTS
if enable_tts:
    print("\n--- ETAPE 4 : GENERATION NARRATION TTS ---")
    print("=" * 50)
    
    def generate_tts_audio(text: str, scene_idx: int,
                           target_duration: float) -> Tuple[np.ndarray, int]:
        """
        Genere l'audio TTS pour une scene.
        Retourne (audio_samples, sample_rate).
        """
        return None  # TODO étudiant : implémenter generate_tts_audio
    # Generer la narration pour chaque scene
    narration_audio = []
    narration_texts = []
    
    for i, scene in enumerate(script_scenes):
        text = scene['narration']
        narration_texts.append(text)
        
        audio_data, sr = generate_tts_audio(text, i, scene_duration)
        if audio_data is None or sr is None:
            print(f"  Scene {i+1} : exercice a completer (generate_tts_audio)")
            continue
        narration_audio.append((audio_data, sr))
        
        duration_actual = len(audio_data) / sr
        print(f"  Scene {i+1} : {duration_actual:.1f}s audio "
              f"({len(text)} chars, {sr}Hz)")
    
    # Concatener tout l'audio
    combined_audio = np.zeros(0, dtype=np.float32)
    combined_sr = 44100
    if narration_audio:
        combined_sr = narration_audio[0][1]
        combined_audio = np.concatenate([a[0] for a in narration_audio])
    total_audio_duration = len(combined_audio) / combined_sr
    
    print(f"\nAudio total : {total_audio_duration:.1f}s, {combined_sr}Hz")
    print(f"Echantillons : {len(combined_audio)}")
    
    # Sauvegarder l'audio combine en WAV
    if save_results and len(combined_audio) > 0:
        combined_audio_path = PIPELINE_DIR / "narration_combined.wav"
        # Ecriture WAV simple
        audio_int16 = (combined_audio * 32767).astype(np.int16)
        import wave
        with wave.open(str(combined_audio_path), 'w') as wav_file:
            wav_file.setnchannels(1)
            wav_file.setsampwidth(2)
            wav_file.setframerate(combined_sr)
            wav_file.writeframes(audio_int16.tobytes())
        print(f"Audio sauvegarde : {combined_audio_path.name} "
              f"({combined_audio_path.stat().st_size / 1024:.0f} KB)")
else:
    narration_audio = []
    narration_texts = [s['narration'] for s in script_scenes]
    print("TTS desactive")

--- ETAPE 4 : GENERATION NARRATION TTS ---
==================================================

Audio total : 0.0s, 44100Hz
Echantillons : 0

5. Sous-titres et assemblage final

Les sous-titres sont generes a partir du texte de narration avec un timing synchronise sur chaque scene. L’assemblage final combine video + audio + sous-titres en un seul fichier MP4.

# Etape 5 : Sous-titres et assemblage final
print("\n--- ETAPE 5 : SOUS-TITRES ET ASSEMBLAGE ---")
print("=" * 50)

# Generation des sous-titres
if enable_subtitles:
    print("Generation des sous-titres...")
    
    def generate_srt(scenes: List[dict], duration_per_scene: float) -> str:
        """Genere un fichier SRT a partir des textes de narration."""
        srt_content = ""
        for i, scene in enumerate(scenes):
            start_time = i * duration_per_scene
            end_time = (i + 1) * duration_per_scene
            
            # Formater en HH:MM:SS,mmm
            def fmt_time(seconds):
                h = int(seconds // 3600)
                m = int((seconds % 3600) // 60)
                s = int(seconds % 60)
                ms = int((seconds % 1) * 1000)
                return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}"
            
            # Decouper le texte en segments de ~60 caracteres
            text = scene['narration']
            words = text.split()
            lines = []
            current_line = ""
            for word in words:
                if len(current_line) + len(word) + 1 > 60:
                    lines.append(current_line)
                    current_line = word
                else:
                    current_line = f"{current_line} {word}".strip()
            if current_line:
                lines.append(current_line)
            
            # Repartir le temps entre les lignes
            time_per_line = duration_per_scene / max(len(lines), 1)
            for j, line in enumerate(lines):
                sub_start = start_time + j * time_per_line
                sub_end = sub_start + time_per_line
                sub_idx = i * 10 + j + 1  # Index unique
                srt_content += f"{sub_idx}\n"
                srt_content += f"{fmt_time(sub_start)} --> {fmt_time(sub_end)}\n"
                srt_content += f"{line}\n\n"
        
        return srt_content
    
    srt_text = generate_srt(script_scenes, scene_duration)
    
    if save_results:
        srt_path = PIPELINE_DIR / "subtitles.srt"
        srt_path.write_text(srt_text, encoding='utf-8')
        print(f"Sous-titres sauvegardes : {srt_path.name}")
    
    # Afficher un extrait
    print(f"\nExtrait SRT (3 premières entrees) :")
    srt_lines = srt_text.strip().split('\n\n')[:3]
    for block in srt_lines:
        print(f"  {block}")

# Assemblage final : ajouter les sous-titres aux frames
print("\nAssemblage video avec sous-titres incrustes...")

def burn_subtitle(frame: np.ndarray, text: str,
                  width: int, height: int) -> np.ndarray:
    """Incruste un sous-titre en bas de la frame."""
    img = Image.fromarray(frame)
    draw = ImageDraw.Draw(img)
    
    try:
        font = ImageFont.truetype("arial.ttf", 22)
    except (OSError, IOError):
        font = ImageFont.load_default()
    
    # Fond semi-transparent pour le sous-titre
    bbox = draw.textbbox((0, 0), text, font=font)
    tw = bbox[2] - bbox[0]
    th = bbox[3] - bbox[1]
    x = (width - tw) // 2
    y = height - 70
    
    # Rectangle de fond
    padding = 8
    draw.rectangle(
        [x - padding, y - padding, x + tw + padding, y + th + padding],
        fill=(0, 0, 0)
    )
    draw.text((x, y), text, fill=(255, 255, 255), font=font)
    
    return np.array(img)


# Assembler toutes les scenes avec transitions et sous-titres
final_frames = []
transition_frames_count = int(0.5 * video_fps)  # 0.5s de fondu entre scenes

for scene_idx, (scene_frames, scene_data) in enumerate(zip(animated_scenes, script_scenes)):
    if not scene_frames:
        continue  # scene non animee : exercice animate_image a completer
    text = scene_data['narration']
    # Decouper le texte pour les sous-titres
    words = text.split()
    mid = len(words) // 2
    sub_parts = [' '.join(words[:mid]), ' '.join(words[mid:])]
    
    for f_idx, frame in enumerate(scene_frames):
        # Choisir la partie de sous-titre
        part_idx = 0 if f_idx < len(scene_frames) // 2 else 1
        if enable_subtitles and sub_parts[part_idx]:
            # Limiter la longueur du sous-titre
            sub_text = sub_parts[part_idx][:70]
            frame = burn_subtitle(frame, sub_text, video_width, video_height)
        final_frames.append(frame)
    
    # Transition fondu vers la scene suivante (seulement si la scene
    # suivante est aussi animee -- les scenes non animees sont skippees)
    next_frames = animated_scenes[scene_idx + 1] if scene_idx < len(animated_scenes) - 1 else None
    if scene_frames and next_frames:
        last_frame = scene_frames[-1]
        next_first = next_frames[0]
        for t_idx in range(transition_frames_count):
            alpha = t_idx / max(transition_frames_count - 1, 1)
            blended = ((1 - alpha) * last_frame.astype(np.float32) +
                       alpha * next_first.astype(np.float32)).astype(np.uint8)
            final_frames.append(blended)

print(f"Frames finales : {len(final_frames)}")
print(f"Duree video : {len(final_frames) / video_fps:.1f}s")

--- ETAPE 5 : SOUS-TITRES ET ASSEMBLAGE ---
==================================================
Generation des sous-titres...
Sous-titres sauvegardes : subtitles.srt

Extrait SRT (3 premières entrees) :
  

Assemblage video avec sous-titres incrustes...
Frames finales : 0
Duree video : 0.0s

Une fois l’assemblage avec sous-titres termine, on exporte le fichier video final en MP4 avec les paramètres de codec et de qualite adaptes a la distribution.

# Export de la video finale
print("\n--- EXPORT VIDEO FINALE ---")
print("=" * 40)

final_video_path = PIPELINE_DIR / f"production_video.{output_format}"

file_size_mb = 0.0
if dependencies.get('imageio', False) and final_frames:
    start_time = time.time()
    
    writer = imageio.get_writer(
        str(final_video_path), fps=video_fps, codec='libx264',
        output_params=['-crf', '20', '-preset', 'medium']
    )
    for frame in final_frames:
        writer.append_data(frame)
    writer.close()
    
    export_time = time.time() - start_time
    file_size_mb = final_video_path.stat().st_size / (1024 * 1024)
    
    print(f"Video exportee : {final_video_path.name}")
    print(f"  Resolution : {video_width}x{video_height}")
    print(f"  FPS : {video_fps}")
    print(f"  Duree : {len(final_frames) / video_fps:.1f}s")
    print(f"  Taille : {file_size_mb:.2f} MB")
    print(f"  Temps d'export : {export_time:.2f}s")
    
    # Apercu final
    n_preview = 6
    preview_idx = np.linspace(0, len(final_frames) - 1, n_preview, dtype=int)
    fig, axes = plt.subplots(2, 3, figsize=(16, 7))
    for ax, idx in zip(axes.flatten(), preview_idx):
        ax.imshow(final_frames[idx])
        ax.set_title(f"t = {idx / video_fps:.1f}s", fontsize=10)
        ax.axis('off')
    plt.suptitle("Pipeline Video de Production - Apercu final", fontsize=13, fontweight='bold')
    plt.tight_layout()
    plt.show()
else:
    print("Export differe : completer les exercices (generate_scene_image / animate_image) -- ou imageio absent de l environnement")

# Metriques de qualite
print("\n--- METRIQUES DE QUALITE ---")
print("=" * 40)

metrics = {
    "Scenes": len(script_scenes),
    "Duree totale": f"{len(final_frames) / video_fps:.1f}s",
    "Resolution": f"{video_width}x{video_height}",
    "FPS": video_fps,
    "Frames totales": len(final_frames),
    "Taille fichier": f"{file_size_mb:.2f} MB" if dependencies.get('imageio', False) else "N/A",
    "Sous-titres": "Oui" if enable_subtitles else "Non",
    "Narration audio": "Oui" if enable_tts and narration_audio else "Non",
    "Animations camera": "Oui" if enable_animation else "Non",
    "Transitions": f"{len(script_scenes) - 1} fondus",
}

print(f"{'Metrique':<25} {'Valeur':<30}")
print("-" * 55)
for key, value in metrics.items():
    print(f"  {key:<23} {str(value):<30}")

--- EXPORT VIDEO FINALE ---
========================================
Export differe : completer les exercices (generate_scene_image / animate_image) -- ou imageio absent de l environnement

--- METRIQUES DE QUALITE ---
========================================
Metrique                  Valeur                        
-------------------------------------------------------
  Scenes                  0                             
  Duree totale            0.0s                          
  Resolution              1280x720                      
  FPS                     30                            
  Frames totales          0                             
  Taille fichier          0.00 MB                       
  Sous-titres             Oui                           
  Narration audio         Non                           
  Animations camera       Oui                           
  Transitions             -1 fondus                     

Interpretation : Pipeline complet

Étape	Entree	Sortie	Outil
1. Script	Sujet (texte)	Scenes structurees	LLM (GPT-4)
2. Images	Descriptions visuelles	PNG par scene	DALL-E / Pillow
3. Animation	Images + instructions camera	Sequences de frames	Pillow / moviepy
4. Narration	Texte de narration	Audio WAV	OpenAI TTS
5. Sous-titres	Texte de narration	SRT + incrustation	Python
6. Assemblage	Frames + audio + SRT	MP4 final	imageio / moviepy

Points cles : 1. Chaque étape est independante et peut etre amelioree separement 2. Le script LLM est le point de depart : un bon script produit une bonne video 3. La synchronisation audio/video est essentielle pour un résultat professionnel 4. En production, chaque étape utiliserait les meilleurs modèles disponibles

Recapitulatif de la serie Video

Ce notebook conclut la serie Video. Voici un resume des competences acquises :

Tier	Notebooks	Competences
01-Foundation	01-1 a 01-5	Opérations de base, extraction, analyse, amelioration, generation
02-Advanced	02-1 a 02-4	Modèles avances, contrôle fin, evaluation qualite
03-Orchestration	03-1 a 03-3	Multi-modèles, workflows, ComfyUI
04-Applications	04-1 a 04-4	Videos educatives, workflows creatifs, API cloud, pipeline production

# Mode interactif - Choix du sujet
if notebook_mode == "interactive" and not skip_widgets:
    print("\n--- MODE INTERACTIF ---")
    print("=" * 40)
    print("Entrez un sujet pour generer un nouveau pipeline video.")
    print("(Laissez vide pour passer)")
    
    try:
        custom_topic = input("\nSujet (ou vide) : ").strip()
        
        if custom_topic:
            print(f"Pour generer une video sur '{custom_topic}',")
            print(f"modifiez le parametre video_topic et relancez le notebook.")
        else:
            print("Mode interactif ignore")
    
    except (KeyboardInterrupt, EOFError):
        print("Mode interactif interrompu")
    except Exception as e:
        error_type = type(e).__name__
        if "StdinNotImplemented" in error_type:
            print("Mode interactif non disponible (execution automatisee)")
        else:
            print(f"Erreur : {error_type}")
else:
    print("Mode batch - Interface interactive desactivee")

--- MODE INTERACTIF ---
========================================
Entrez un sujet pour generer un nouveau pipeline video.
(Laissez vide pour passer)
Mode interactif non disponible (execution automatisee)

Les statistiques de session enregistrent les temps d’exécution de chaque étape du pipeline, les tailles des fichiers produits et les ressources GPU consommees.

# Statistiques de session
print("\n--- STATISTIQUES DE SESSION ---")
print("=" * 40)

print(f"Date : {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print(f"Mode : {notebook_mode}")
print(f"Sujet : {video_topic}")
print(f"Scenes : {len(script_scenes)}")
print(f"Video : {video_width}x{video_height} @ {video_fps}fps")
print(f"Frames finales : {len(final_frames)}")
print(f"Duree : {len(final_frames) / video_fps:.1f}s")

# Etapes du pipeline
pipeline_steps = [
    ("Script LLM", enable_llm_script),
    ("Images", enable_image_gen),
    ("Animation", enable_animation),
    ("TTS", enable_tts),
    ("Sous-titres", enable_subtitles),
    ("Assemblage", enable_assembly),
]
print(f"\nEtapes du pipeline :")
for step_name, enabled in pipeline_steps:
    status = "activee" if enabled else "desactivee"
    print(f"  {step_name} : {status}")

if save_results and PIPELINE_DIR.exists():
    generated_files = list(PIPELINE_DIR.glob('*'))
    total_size_mb = sum(f.stat().st_size for f in generated_files) / (1024 * 1024)
    print(f"\nFichiers generes ({len(generated_files)}, {total_size_mb:.1f} MB total) :")
    for f in sorted(generated_files):
        size_kb = f.stat().st_size / 1024
        print(f"  {f.name} ({size_kb:.1f} KB)")

print(f"\nDependances utilisees :")
for dep, available in dependencies.items():
    status = "utilisee" if available else "non disponible"
    print(f"  {dep} : {status}")

print(f"\n--- FIN DE LA SERIE VIDEO ---")
print(f"Ce notebook conclut la serie Video (04-Applications).")
print(f"Pour combiner audio et video, consultez la serie Audio :")
print(f"  -> Audio/01-Foundation/01-1-OpenAI-TTS-Intro.ipynb")

print(f"\nNotebook 04-4 Production Video Pipeline termine - {datetime.now().strftime('%H:%M:%S')}")

--- STATISTIQUES DE SESSION ---
========================================
Date : 2026-09-30 10:28:47
Mode : interactive
Sujet : Intelligence Artificielle
Scenes : 0
Video : 1280x720 @ 30fps
Frames finales : 0
Duree : 0.0s

Etapes du pipeline :
  Script LLM : activee
  Images : activee
  Animation : activee
  TTS : activee
  Sous-titres : activee
  Assemblage : activee

Fichiers generes (1, 0.0 MB total) :
  subtitles.srt (0.0 KB)

Dependances utilisees :
  moviepy : utilisee
  imageio : utilisee
  openai : utilisee
  pydub : utilisee
  diffusers : utilisee

--- FIN DE LA SERIE VIDEO ---
Ce notebook conclut la serie Video (04-Applications).
Pour combiner audio et video, consultez la serie Audio :
  -> Audio/01-Foundation/01-1-OpenAI-TTS-Intro.ipynb

Notebook 04-4 Production Video Pipeline termine - 10:28:47

---

Exercice : Pipeline Production End-to-End

Duree estimee : 90-120 minutes

Objectif

Créer un pipeline de production video complet de bout en bout pour un cas d’usage reel (ex: demo produit, contenu social, formation), en integrant generation de script, visuels, audio et assemblage.

Instructions

1. Définir le cas d’usage
   • Type de contenu et public cible
   • Contraintes (duree, budget, plateforme)
   • KPIs de succes (engagement, conversion)
2. Concevoir le pipeline
   • Architecture : étapes, outils, interdependencies
   • Choix des modèles (LLM, TTS, generation image/video)
   • Pipeline de traitement et de validation
3. Implementer et tester
   • Pipeline fonctionnel de bout en bout
   • Tests sur 3 contenus différents
   • Mesure des performances (temps, cout, qualite)
4. Optimiser et documenter
   • Identification des goulots d’etranglement
   • Optimisations (cache, parallelisation)
   • Documentation pour reutilisation

Indices :

• Cas d’usage : demo produit (60s), content social (15s), formation (5min)
• Étapes : script -> visuels -> audio -> assemblage -> export
• Optimisations : cache LLM, parallelisation generations, compression intermediaires

Code de depart

Voir la cellule de code suivante pour le squelette de la classe ProductionPipeline.

Critères de succes

• Cas d’usage défini avec KPIs
• Pipeline complet implemente (6+ étapes)
• Tests fonctionnels sur 3 cas
• Performances mesurees (temps, cout, qualite)
• Optimisations identifiees et appliquees
• Documentation reutilisable
• Video exportee de qualite

---

Exercice Avance : Système de A/B Testing Video

Duree estimee : 180-240 minutes

Objectif

Developper une plateforme de A/B testing pour videos qui permet de generer automatiquement des variations d’une video (différents visuels, narrations, styles) et de mesurer leur impact sur des metriques d’engagement.

Instructions

1. Etudier les principes du A/B testing
   • Variations controlables (visuels, audio, duree, style)
   • Metriques d’engagement (retention, click-through, completion)
   • Signification statistique et taille d’echantillon
2. Concevoir la plateforme
   • Système de generation de variations
   • Integration avec les analytics (ou simulation)
   • Pipeline d’experimentation et de reporting
3. Implementer le moteur
   • Fonction generate_variations(base_video, params)
   • Tracking des metriques (ou simulation)
   • Analyse statistique des résultats
4. Prouver la valeur
   • Cas d’usage reel (ex: video marketing)
   • Experimentation avec 5+ variations
   • Analyse des résultats et recommandations

Indices :

• Variations : visuels (palette, composition), audio (voix, ton), duree (15s, 30s, 60s), style (animations, transitions)
• Metriques : completion rate, engagement rate, click-through rate
• Signification statistique : test du chi-carre, t-test, ANOVA

Code de depart

Voir la cellule de code suivante pour le squelette de la classe VideoABTestPlatform.

Critères de succes

• Étude des principes du A/B testing
• Plateforme concue et documentee
• Generation de variations fonctionnelle
• Tracking des metriques (ou simulation)
• Analyse statistique implementee
• Cas d’usage reel teste
• Résultats avec recommandations
• ROI demontre

---

# TODO: Exercice - Pipeline Production End-to-End
class ProductionPipeline:
    """
    Pipeline de production video de bout en bout.

    Etapes:
    1. Script (LLM)
    2. Visuels (generation image/video)
    3. Animation (effets camera)
    4. Audio (TTS)
    5. Sous-titres
    6. Assemblage
    7. Export
    """

    def __init__(self, config: Dict):
        self.config = config
        self.llm_client = None  # TODO: Initialiser OpenAI client
        self.tts_client = None  # TODO: Initialiser OpenAI client
        self.image_gen = None  # TODO: Initialiser generateur d'images
        self.video_gen = None  # TODO: Initialiser generateur de videos

    def produce(self, topic: str, duration: int, style: str) -> Dict:
        """
        Produit une video complete.

        Args:
            topic: Sujet de la video
            duration: Duree cible (secondes)
            style: Style visuel ("corporate", "social", "creative")

        Returns:
            {"video_path": ..., "metadata": ..., "cost": ..., "time": ...}
        """
        result = {
            "steps": [],
            "video_path": None,
            "cost": 0,
            "time": 0
        }

        # TODO: Implementer chaque etape avec timings
        # 1. Generer le script
        # 2. Generer les visuels
        # 3. Animer les visuels
        # 4. Generer l'audio
        # 5. Creer les sous-titres
        # 6. Assembler le tout
        # 7. Exporter

        return result


# TODO: Tester sur 3 cas d'usage
test_cases = [
    {"topic": "Lancement produit X", "duration": 60, "style": "corporate"},
    {"topic": "Astuce cuisine", "duration": 30, "style": "social"},
    {"topic": "Tuto Python", "duration": 300, "style": "educational"},
]

# TODO: Initialiser la configuration
config = {}
pipeline = ProductionPipeline(config)

for case in test_cases:
    print(f"Production: {case['topic']}")
    # result = pipeline.produce(case['topic'], case['duration'], case['style'])
    # print(f"  Video: {result['video_path']}")
    # print(f"  Cout: ${result['cost']:.2f}, Temps: {result['time']:.0f}s")

Production: Lancement produit X
Production: Astuce cuisine
Production: Tuto Python

Le système de A/B testing permet de comparer deux approches de generation video sur les mêmes prompts et de mesurer objectivement la qualite percue par des metriques automatiques.

# TODO: Exercice Avance - Systeme de A/B Testing Video
class VideoABTestPlatform:
    """
    Plateforme de A/B testing pour videos.

    Fonctionnalites:
    - Generation automatique de variations
    - Tracking des metriques d'engagement
    - Analyse statistique des resultats
    - Recommandations
    """

    def __init__(self, config: Dict):
        self.pipeline = None  # TODO: Initialiser ProductionPipeline
        self.analytics = None  # TODO: ou simulation
        self.variations_cache = {}

    def generate_variations(self, base_request: Dict,
                            variation_params: List[Dict]) -> List[Dict]:
        """
        Genere N variations d'une video.

        Variation params possibles:
        - visual_style: "corporate", "creative", "minimal"
        - narrator_voice: "male", "female", "neutral"
        - duration: 15, 30, 60
        - color_palette: "warm", "cool", "neutral"
        - animation_style: "smooth", "dynamic", "static"
        """
        variations = []

        # TODO: Implementer la generation de variations
        # - Modifier les parametres du pipeline selon variation_params
        # - Paralleliser les generations
        # - Logger les metadonnees

        return variations

    def track_metrics(self, video_id: str, user_actions: List[Dict]) -> Dict:
        """
        Track les metriques d'engagement.

        Actions: "play", "pause", "complete", "click_cta", "share"
        """
        # TODO: Implementer le tracking
        # - Calculer les metriques
        # - Stocker dans une base de donnees
        pass

    def analyze_results(self, experiment_id: str) -> Dict:
        """
        Analyse les resultats du A/B test.

        Retourne:
        - Winning variation
        - Significativite statistique
        - Recommandations
        """
        # TODO: Implementer l'analyse statistique
        # - Tests de significativite
        # - Calcul de l'uplift
        # - Recommandations
        pass


# TODO: Cas d'usage reel
# Ex: Video marketing pour un produit
# Generer 5 variations
# Simuler ou tracker les metriques
# Analyser les resultats

# TODO: Prouver la valeur
# - Improvement du taux de conversion
# - Cout vs benefice
# - Recommandations pour futurs tests

---

Exercice : Automatisation des Metriques Qualite Pipeline

Duree estimee : 40-50 minutes

Objectif

Implementer un système de validation automatique de la qualite a chaque étape du pipeline de production, generant un rapport de qualite global avec des scores et des alertes.

Contexte

En production, la qualite de chaque étape doit etre verifiee automatiquement avant de passer a la suivante. Par exemple : - Script : verifier la coherence narrative, la duree estimee vs cible, le nombre de scenes - Images : verifier la resolution, le contraste, l’absence d’artefacts - Assemblage : verifier la synchronisation audio/video, la duree totale, la taille du fichier

Un rapport de qualite automatique permet de detecter les problemes sans relecture manuelle.

Instructions

1. Implementer les fonctions de validation par étape
   • validate_script(script_scenes, target_duration) : coherence, duree, structure
   • validate_images(images, min_resolution) : resolution, contraste, luminosite
   • validate_assembly(frames, audio, fps) : sync, duree, artefacts
2. Implementer le rapport de qualite
   • generate_quality_report(validations) : compiler les résultats en rapport structure
   • Score global (0-100) pondere par étape
   • Alertes pour les problemes critiques (resolution insuffisante, duree non conforme)
3. Tester sur le pipeline existant
   • Valider les sorties du pipeline de production déjà execute
   • Injecter des problemes (image basse resolution, scene trop courte) et verifier la detection
   • Generer et afficher le rapport de qualite

Indices :

• Pour le contraste : calculer l’ecart-type des pixels de l’image ( faible = image plate)
• Pour la luminosite : calculer la moyenne des pixels (trop sombre/clair = problème)
• Pour la sync audio/video : comparer la duree de l’audio avec le nombre de frames / fps
• Le score peut etre pondere : script (20%), images (30%), animation (20%), assemblage (30%)

# TODO: Implementer les fonctions de validation qualite

def validate_script(script_scenes: List[Dict], target_duration: float) -> Dict:
    """
    Valide la qualite du script video.
    
    Checks:
    - Nombre de scenes (3-10 recommande)
    - Duree estimee vs duree cible (tolerance 20%)
    - Chaque scene a les champs requis (title, narration, visual_description)
    - Longueur des narrations (ni trop courte ni trop longue)
    
    Args:
        script_scenes: Liste des scenes du script
        target_duration: Duree cible en secondes
    
    Returns:
        Dict avec 'score' (0-100), 'checks' (list), 'alerts' (list)
    """
    result = {"score": 0, "checks": [], "alerts": []}
    
    # TODO: Verifier le nombre de scenes
    # TODO: Verifier les champs requis de chaque scene
    # TODO: Verifier la duree estimee vs cible
    # TODO: Verifier la longueur des narrations
    
    return result


def validate_images(images: List[Image.Image], min_width: int = 640, min_height: int = 360) -> Dict:
    """
    Valide la qualite des images generees.
    
    Checks:
    - Resolution minimale (640x360 pour du web, 1280x720 pour HD)
    - Contraste (ecart-type des pixels > seuil)
    - Luminosite moyenne (entre 30 et 230)
    - Pas d'images vides ou uniformes
    
    Args:
        images: Liste d'images PIL
        min_width, min_height: Resolution minimale attendue
    
    Returns:
        Dict avec 'score' (0-100), 'checks' (list), 'alerts' (list)
    """
    result = {"score": 0, "checks": [], "alerts": []}
    
    # TODO: Verifier la resolution de chaque image
    # TODO: Calculer le contraste (ecart-type des pixels)
    # TODO: Verifier la luminosite moyenne
    # TODO: Detecter les images vides/uniformes
    
    return result


def validate_assembly(frames: List[np.ndarray], audio_duration: float,
                      target_fps: int, target_duration: float) -> Dict:
    """
    Valide la qualite de l'assemblage final.
    
    Checks:
    - Synchronisation audio/video (difference < 1 seconde)
    - Duree totale vs cible (tolerance 10%)
    - FPS coherent avec la cible
    - Pas de frames noires ou corrompues
    
    Args:
        frames: Liste des frames finales (numpy arrays)
        audio_duration: Duree de l'audio en secondes
        target_fps: FPS cible
        target_duration: Duree cible en secondes
    
    Returns:
        Dict avec 'score' (0-100), 'checks' (list), 'alerts' (list)
    """
    result = {"score": 0, "checks": [], "alerts": []}
    
    # TODO: Verifier la synchronisation audio/video
    # video_duration = len(frames) / target_fps
    # sync_diff = abs(video_duration - audio_duration)
    
    # TODO: Verifier la duree totale vs cible
    # TODO: Verifier les frames noires/corrompues
    
    return result


def generate_quality_report(validations: Dict[str, Dict]) -> Dict:
    """
    Genere un rapport de qualite global a partir des validations par etape.
    
    Pondération:
    - Script: 20%
    - Images: 30%
    - Animation: 20%
    - Assemblage: 30%
    
    Args:
        validations: Dict avec les resultats de chaque validateur
            {"script": {...}, "images": {...}, "assembly": {...}}
    
    Returns:
        Dict avec 'global_score', 'details', 'alerts'
    """
    weights = {"script": 0.2, "images": 0.3, "animation": 0.2, "assembly": 0.3}
    
    report = {
        "global_score": 0,
        "details": {},
        "alerts": []
    }
    
    # TODO: Calculer le score global pondere
    # TODO: Collecter toutes les alertes
    # TODO: Ajouter un verdict (PASS/WARN/FAIL)
    
    return report


# TODO: Tester les validateurs sur les donnees du pipeline
# script_validation = validate_script(script_scenes, target_duration=40)
# images_validation = validate_images(scene_images, min_width=640, min_height=360)
# assembly_validation = validate_assembly(final_frames, total_audio_duration, video_fps, 40)

# TODO: Generer le rapport
# quality_report = generate_quality_report({
#     "script": script_validation,
#     "images": images_validation,
#     "assembly": assembly_validation
# })

# TODO: Afficher le rapport
# print(f"Score global : {quality_report['global_score']}/100")
# print(f"Verdict : {quality_report.get('verdict', 'N/A')}")
# for alert in quality_report['alerts']:
#     print(f"  ALERTE : {alert}")

print("Exercice a completer : automatisation des metriques qualite pipeline")

Exercice a completer : automatisation des metriques qualite pipeline