#!/usr/bin/env python3 """ Radio Juridique FO — Générateur d'émission radio à 3 voix Utilisé par la skill radio-juridique : - Moteur TTS primaire : XTTS v2 (via TTS[audio]) — voix clonées depuis ~/.hermes/voices_tts/ - Fallback : edge-tts (Microsoft Edge TTS) — bascule automatiquement si XTTS échoue - 3 voix distinctes : Présentateur (Henri) + Expert brèves (Marc) + Expert focus (Sophie) - Format dialogué présentateur/experts - 3 parties : Actualités législatives + Brèves de jurisprudence + Focus juridique Usage (lancer avec le venv XTTS) : COQUI_TOS_AGREED=1 OMP_NUM_THREADS=24 MKL_NUM_THREADS=24 \\ OPENBLAS_NUM_THREADS=24 PYTORCH_CPU_THREAD_POOL_SIZE=24 \\ TORCH_NUM_THREADS=24 \\ ~/.hermes/xtts-venv/bin/python /tmp/run_radio_juridique_YYYY_MM_DD.py """ import asyncio import json import os import re import subprocess import sys import tempfile from pathlib import Path # === XTTS CPU PERFORMANCE (24 cœurs) === XTTS_CPU_CORES = 24 os.environ.setdefault("OMP_NUM_THREADS", str(XTTS_CPU_CORES)) os.environ.setdefault("MKL_NUM_THREADS", str(XTTS_CPU_CORES)) os.environ.setdefault("OPENBLAS_NUM_THREADS", str(XTTS_CPU_CORES)) os.environ.setdefault("PYTORCH_CPU_THREAD_POOL_SIZE", str(XTTS_CPU_CORES)) os.environ.setdefault("TORCH_NUM_THREADS", str(XTTS_CPU_CORES)) os.environ.setdefault("COQUI_TOS_AGREED", "1") # === CONFIG === AUDIO_DIR = Path.home() / ".hermes" / "audio_cache" AUDIO_DIR.mkdir(exist_ok=True) VOICES_DIR = Path.home() / ".hermes" / "voices_tts" # 3 voix FR pour edge-tts fallback EDGE_VOICES = { "presentateur": "fr-FR-HenriNeural", "expert_breves": "fr-FR-RemyMultilingualNeural", "expert_focus": "fr-FR-VivienneMultilingualNeural", "jingle": "fr-FR-HenriNeural", } # Mapping XTTS (références vocales) — 3 voix distinctes XTTS_REFERENCES = { "presentateur": str(VOICES_DIR / "presentateur_radio.wav"), "expert_breves": str(VOICES_DIR / "expert_breves_marc.wav"), "expert_focus": str(VOICES_DIR / "expert_focus_sophie.wav"), "jingle": str(VOICES_DIR / "presentateur_radio.wav"), } # Rôles TTS valides VALID_ROLES = set(EDGE_VOICES.keys()) # Taille max de texte par appel XTTS XTTS_MAX_CHARS = 250 # Pauses entre segments (ms) PAUSES = { "between_speaker": 350, "between_rubrique": 800, "after_jingle": 500, } # === CHEMINS VENVS === XTTS_VENV_PYTHON = Path.home() / ".hermes" / "xtts-venv" / "bin" / "python" EDGE_VENV_PYTHON = Path.home() / ".hermes" / "tts-venv" / "bin" / "python" EDGE_VENV_PYTHON_FALLBACK = Path.home() / "tts-venv" / "bin" / "python" # === DÉTECTION MOTEUR TTS === _xtts_available = None _edge_available = None _xtts_model = None def _running_in_xtts_venv(): return sys.executable.startswith(str(XTTS_VENV_PYTHON.parent.parent)) def _check_xtts(): global _xtts_available if _xtts_available is None: try: import TTS.api # noqa: F401 import torch # noqa: F401 _xtts_available = True except ImportError: _xtts_available = False return _xtts_available def _check_edge_tts(): global _edge_available if _edge_available is None: # Try configured venv first, then fallback for venv_path in [EDGE_VENV_PYTHON, EDGE_VENV_PYTHON_FALLBACK]: if venv_path.exists(): try: r = subprocess.run( [str(venv_path), "-c", "import edge_tts"], capture_output=True, text=True, timeout=10, ) if r.returncode == 0: _edge_available = True return True except Exception: pass # System Python fallback try: import edge_tts # noqa: F401 _edge_available = True except ImportError: _edge_available = False return _edge_available def _get_engine_name(): if _check_xtts(): return "XTTS v2 (singleton, processus direct, 24 cœurs CPU)" elif _check_edge_tts(): return "edge-tts (venv subprocess)" else: raise RuntimeError( "Aucun moteur TTS disponible. " "Installer TTS[audio] dans ~/.hermes/xtts-venv ou edge-tts dans ~/.hermes/tts-venv." ) # === MOTEUR XTTS (SINGLETON) === def _get_xtts_model(): global _xtts_model if _xtts_model is not None: return _xtts_model import torch from TTS.api import TTS device = "cpu" print(f" [XTTS] Chargement du modèle xtts_v2 sur {device} ({XTTS_CPU_CORES} cœurs CPU)...") _xtts_model = TTS( model_name="tts_models/multilingual/multi-dataset/xtts_v2", progress_bar=False, gpu=False, ) print(f" [XTTS] Modèle chargé. Prêt à générer.") return _xtts_model def _split_text_for_xtts(text: str, max_chars: int = XTTS_MAX_CHARS) -> list: sentences = re.split(r'(?<=[.!?,])\s+', text.strip()) chunks = [] current = "" for sent in sentences: sent = sent.strip() if not sent: continue if len(current) + len(sent) + 1 <= max_chars: current = f"{current} {sent}".strip() if current else sent else: if current: chunks.append(current) if len(sent) > max_chars: while len(sent) > max_chars: cut = sent[:max_chars].rfind(' ') if cut <= 0: cut = max_chars chunks.append(sent[:cut].strip()) sent = sent[cut:].strip() if sent: current = sent else: current = sent if current: chunks.append(current) return chunks if chunks else [text] def _validate_role(role: str) -> str: if role not in VALID_ROLES: raise ValueError( f"Rôle TTS invalide : '{role}'. Rôles valides : {sorted(VALID_ROLES)}" ) return role def clean_tts_text(text: str) -> str: """Nettoie automatiquement un texte pour TTS.""" # Remove all periods (TTS doesn't handle them well) text = text.replace(".", "") # Replace bullet points with commas text = text.replace("•", ", ") # Remove ► arrows used in FO PDF text = text.replace("►", "") # JORF → Journal Officiel text = re.sub(r'\bJORF\b', 'Journal Officiel', text) text = re.sub(r'\bJO\b', 'Journal Officiel', text) # FO → F.O. (proper pronunciation) text = text.replace(" FO ", " F.O. ") text = text.replace(" FO.", " F.O.") # Clean up hyphenated line breaks from PDF extraction text = text.replace("-\n", "") text = text.replace("- ", " ") # Remove URLs (not readable aloud) text = re.sub(r'https?://\S+', '', text) text = re.sub(r'\S+\.fr/\S+', '', text) # Pronounce French dates in legislative text (e.g., "28 mars 2026", "1er avril") text = _pronounce_french_dates(text) return text def _pronounce_french_dates(text: str) -> str: """Convert French dates like '28 mars 2026', '1er avril 2026' to pronunciation-friendly form. Handles: '28 mars 2026', '1er avril 2026', 'du 28 mars 2026', 'au 30 mars 2026' """ mois = { "janvier": "janvier", "février": "fevrier", "fevrier": "fevrier", "mars": "mars", "avril": "avril", "mai": "mai", "juin": "juin", "juillet": "juillet", "août": "aout", "aout": "aout", "septembre": "septembre", "octobre": "octobre", "novembre": "novembre", "décembre": "decembre", "decembre": "decembre" } def _replace_date(m): # Regex groups: 0=full, 1=optional "du ", 2=day, 3=month, 4=year prefix = m.group(1) or "" # "du " or "" day = m.group(2) month = m.group(3).lower() year = m.group(4) if month not in mois: return m.group(0) # Normalize year if len(year) == 2: year = f"20{year}" # Pronounce day if day == "1": day_text = "1er" else: try: day_text = prononcer_numero(day) except: day_text = day month_text = mois[month] year_text = prononcer_numero(year) # If "du " prefix exists, use it directly (no "le" needed) # Otherwise, add "le" for standalone dates if prefix: return f"{prefix}{day_text} {month_text} {year_text}" else: return f"le {day_text} {month_text} {year_text}" # Match dates like "28 mars 2026", "1er avril 2026", "30 mars 2026" # Also matches "du 28 mars 2026", "au 30 mars 2026" — prefix kept text = re.sub( r'(du\s+|au\s+)?(\d{1,2})(?:er)?\s+(janvier|février|fevrier|mars|avril|mai|juin|juillet|août|aout|septembre|octobre|novembre|décembre|decembre)\s+(\d{2,4})', _replace_date, text, flags=re.IGNORECASE ) return text def xtts_generate(text: str, role: str, output_path: str): """Génération via XTTS singleton.""" text = clean_tts_text(text) _validate_role(role) tts = _get_xtts_model() ref_wav = XTTS_REFERENCES.get(role, XTTS_REFERENCES["presentateur"]) if not os.path.exists(ref_wav): raise FileNotFoundError(f"Fichier de référence XTTS manquant : {ref_wav}") chunks = _split_text_for_xtts(text) if len(chunks) == 1: tts.tts_to_file( text=chunks[0], file_path=output_path, speaker_wav=ref_wav, language="fr", ) else: tmp_files = [] try: for i, chunk in enumerate(chunks): tmp_path = output_path.replace(".wav", f"_chunk_{i}.wav") tts.tts_to_file( text=chunk, file_path=tmp_path, speaker_wav=ref_wav, language="fr", ) tmp_files.append(tmp_path) with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f: concat_list = f.name for tf in tmp_files: f.write(f"file '{tf}'\n") try: subprocess.run( ["ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", concat_list, "-acodec", "pcm_s16le", output_path], capture_output=True, check=True, ) finally: os.unlink(concat_list) finally: for tf in tmp_files: if os.path.exists(tf): os.unlink(tf) # === MOTEUR EDGE-TTS (fallback) === async def edge_generate(text: str, role: str, output_path: str, max_attempts: int = 3): text = clean_tts_text(text) _validate_role(role) voice = EDGE_VOICES.get(role, "fr-FR-HenriNeural") # Use longer timeout for long texts (focus body can be 5000+ chars) timeout = max(60, len(text) // 20) code = f""" import asyncio, json, sys import edge_tts text = json.loads(sys.argv[1]) voice = sys.argv[2] output_path = sys.argv[3] async def main(): comm = edge_tts.Communicate(text, voice) await comm.save(output_path) asyncio.run(main()) """ last_err = None for attempt in range(1, max_attempts + 1): try: result = subprocess.run( [str(EDGE_VENV_PYTHON or EDGE_VENV_PYTHON_FALLBACK), "-c", code, json.dumps(text), voice, output_path], capture_output=True, text=True, timeout=timeout, ) if result.returncode == 0: return last_err = RuntimeError(f"edge-tts subprocess failed:\n{result.stderr}") except Exception as e: last_err = e if attempt < max_attempts: await asyncio.sleep(2 * attempt) raise last_err # === DISPATCHER === def generate_segment_sync(text: str, role: str, output_path: str): _validate_role(role) if _check_xtts(): try: xtts_generate(text, role, output_path) return except Exception as xtts_err: if _check_edge_tts(): print(f" XTTS échoué ({xtts_err}), fallback edge-tts") asyncio.run(edge_generate(text, role, output_path)) else: raise elif _check_edge_tts(): asyncio.run(edge_generate(text, role, output_path)) else: raise RuntimeError("Aucun moteur TTS disponible.") async def generate_segment(text: str, role: str, output_path: str): if _check_xtts(): try: import concurrent.futures loop = asyncio.get_running_loop() with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool: await loop.run_in_executor(pool, xtts_generate, text, role, output_path) return except Exception as xtts_err: if _check_edge_tts(): print(f" XTTS échoué ({xtts_err}), fallback edge-tts") else: raise if _check_edge_tts(): await edge_generate(text, role, output_path) return raise RuntimeError("Aucun moteur TTS disponible.") # === VALIDATION === def validate_tts_text(text: str, segment_name: str = "") -> list: issues = [] if "." in text: issues.append(f"Point(s) détecté(s) dans '{segment_name}'") if "•" in text: issues.append(f"Symbole '•' dans '{segment_name}'") return issues def validate_script_tts(script: dict) -> list: all_issues = [] for name, segment in script.items(): text = segment.get("text", "") issues = validate_tts_text(text, name) all_issues.extend(issues) return all_issues # === UTILS AUDIO === def get_pause_type(current_key: str, next_key: str) -> str: curr_prefix = current_key.split("_")[0] if "_" in current_key else current_key next_prefix = next_key.split("_")[0] if "_" in next_key else next_key rubrique_order = ["intro", "partie", "outro"] try: curr_idx = rubrique_order.index(curr_prefix) next_idx = rubrique_order.index(next_prefix) except ValueError: return "between_speaker" if next_idx > curr_idx: return "between_rubrique" return "between_speaker" def create_silence_wav(duration_ms: int, output_path: str): subprocess.run( [ "ffmpeg", "-y", "-f", "lavfi", "-i", f"anullsrc=r=24000:cl=mono", "-t", f"{duration_ms / 1000:.3f}", "-acodec", "pcm_s16le", output_path, ], capture_output=True, check=True, ) def concatenate_to_mp3(segment_files: list, output_file: str): with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f: concat_list = f.name for seg in segment_files: if os.path.exists(seg): f.write(f"file '{seg}'\n") try: subprocess.run( [ "ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", concat_list, "-acodec", "libmp3lame", "-b:a", "192k", output_file, ], capture_output=True, check=True, ) finally: os.unlink(concat_list) # === GÉNÉRATION ÉMISSION === async def generate_radio_emission(script: dict, output_file: str, pause_config: dict = None): if pause_config is None: pause_config = PAUSES engine = _get_engine_name() print(f" Moteur TTS : {engine}") if _check_xtts(): print(" [Préchargement du modèle XTTS...]") _get_xtts_model() print(" [Modèle prêt — démarrage de la génération]") tts_issues = validate_script_tts(script) if tts_issues: print(" Problèmes TTS détectés :") for issue in tts_issues: print(f" - {issue}") segments_dir = AUDIO_DIR / "radio_juridique_segments" segments_dir.mkdir(exist_ok=True) segment_files = [] items = list(script.items()) try: for f in segments_dir.glob("*"): f.unlink() failed_segments = [] for i, (name, segment) in enumerate(items): role = segment.get("voice_role", "presentateur") seg_path = str(segments_dir / f"seg_{i:03d}_{name}.wav") print(f" [{i+1}/{len(items)}] {name} — {role}") try: await generate_segment(segment["text"], role, seg_path) segment_files.append(seg_path) except Exception as seg_err: print(f" Segment '{name}' échoué ({role}) : {seg_err}") failed_segments.append(name) if i < len(items) - 1: next_name = items[i + 1][0] pause_type = get_pause_type(name, next_name) pause_duration = pause_config.get(pause_type, 350) pause_path = str(segments_dir / f"pause_{i:03d}_{pause_type}.wav") try: create_silence_wav(pause_duration, pause_path) segment_files.append(pause_path) except Exception: pass if failed_segments: print(f" {len(failed_segments)} segment(s) échoué(s) : {failed_segments}") if not segment_files: raise RuntimeError("Aucun segment audio généré — émission vide.") print(" Assemblage MP3...") concatenate_to_mp3(segment_files, output_file) probe = subprocess.run( ["ffprobe", "-v", "quiet", "-show_entries", "format=duration", "-of", "csv=p=0", output_file], capture_output=True, text=True, ) duration_min = float(probe.stdout.strip()) / 60 file_size_mb = os.path.getsize(output_file) / (1024 * 1024) print(f" Émission générée : {output_file}") print(f" Durée : {duration_min:.1f} min — Taille : {file_size_mb:.1f} Mo") return output_file except Exception as e: print(f" ERREUR : {e}") if os.path.exists(output_file): try: os.unlink(output_file) except OSError: pass raise finally: cleaned = 0 for f in segments_dir.glob("*"): try: f.unlink() cleaned += 1 except OSError: pass if cleaned: print(f" Nettoyage : {cleaned} fichiers temporaires supprimés.") # === DATE === def _date_fr(): from datetime import datetime jours = ["lundi", "mardi", "mercredi", "jeudi", "vendredi", "samedi", "dimanche"] mois = ["janvier", "février", "mars", "avril", "mai", "juin", "juillet", "août", "septembre", "octobre", "novembre", "décembre"] now = datetime.now() return f"{jours[now.weekday()]} {now.day} {mois[now.month-1]} {now.year}" # === PRONONCIATION DES RÉFÉRENCES JURIDIQUES === def prononcer_numero(num: str) -> str: """Convertit un numéro d'arrêt en prononciation française. Exemples : "24-13599" → "vingt-quatre tiret treize mille cinq cent quatre-vingt-dix-neuf" "25-12049" → "vingt-cinq tiret douze mille quarante-neuf" "2022-679" → "deux mille vingt-deux tiret six cent soixante-dix-neuf" """ num = num.strip() # Gérer les tirets (n° d'arrêt : 24-13599) if "-" in num and len(num.split("-")) == 2: parts = num.split("-") return prononcer_numero(parts[0]) + " tiret " + prononcer_numero(parts[1]) # Gérer les virgules (n° : 24.13599) if "." in num and len(num.split(".")) == 2: parts = num.split(".") return prononcer_numero(parts[0]) + " virgule " + prononcer_numero(parts[1]) # Nombre pur try: n = int(num) if n == 0: return "zéro" return _int_to_french(n) except ValueError: return num # Retourne tel quel si pas un nombre def _int_to_french(n: int) -> str: """Convertit un entier en texte français pour la prononciation.""" if n == 0: return "zéro" unites = [ "", "un", "deux", "trois", "quatre", "cinq", "six", "sept", "huit", "neuf", "dix", "onze", "douze", "treize", "quatorze", "quinze", "seize", "dix-sept", "dix-huit", "dix-neuf" ] dizaines = [ "", "", "vingt", "trente", "quarante", "cinquante", "soixante", "", "", "quatre-vingt" ] if n < 0: return "moins " + _int_to_french(-n) if n < 20: return unites[n] if n < 100: d = n // 10 u = n % 10 base = dizaines[d] if d == 7: # 70-79 if u == 0: return "soixante" return "soixante-" + unites[u] if d == 8: # 80-89 if u == 0: return "quatre-vingts" return "quatre-vingt-" + unites[u] if u == 0: return base if u == 1 and d not in (8,): # 21, 31, 41, 51, 61, 71 return base + "-" + unites[u] return base + "-" + unites[u] if n < 1000: c = n // 100 r = n % 100 if c == 1: base = "cent" else: base = unites[c] + " cent" if r == 0: return base + ("s" if c > 1 else "") return base + " " + _int_to_french(r) if n < 1000000: m = n // 1000 r = n % 1000 if m == 1: base = "mille" else: base = _int_to_french(m) + " mille" if r == 0: return base return base + " " + _int_to_french(r) if n < 1000000000: m = n // 1000000 r = n % 1000000 if m == 1: base = "un million" else: base = _int_to_french(m) + " millions" if r == 0: return base return base + " " + _int_to_french(r) return str(n) def prononcer_date_cass(date_str: str) -> str: """Prononce une date d'arrêt de la Cour de cassation. Exemples : "6-5-26" → "le 6 mai deux mille vingt-six" "1-4-26" → "le 1er avril deux mille vingt-six" "9-4-26" → "le 9 avril deux mille vingt-six" """ date_str = date_str.strip() # Normalize: remove spaces around dashes, normalize separators date_str = re.sub(r'\s*[-/]\s*', '-', date_str) parts = date_str.split("-") if len(parts) == 3: jour = parts[0].strip() mois_num = int(parts[1].strip()) annee = int(parts[2].strip()) mois = [ "", "janvier", "février", "mars", "avril", "mai", "juin", "juillet", "août", "septembre", "octobre", "novembre", "décembre" ] annee_text = _int_to_french(2000 + annee) if annee < 100 else _int_to_french(annee) # Handle ordinal for day 1 if jour == "1": jour_text = "1er" else: jour_text = prononcer_numero(jour) return f"le {jour_text} {mois[mois_num]} {annee_text}" return date_str def prononcer_numero_arret(num_arret: str) -> str: """Prononce un numéro d'arrêt de la Cour de cassation. Exemples : "24-13599" → "numéro vingt-quatre tiret treize mille cinq cent quatre-vingt-dix-neuf" "25-12049" → "numéro vingt-cinq tiret douze mille quarante-neuf" """ num_arret = num_arret.strip() return f"numéro {prononcer_numero(num_arret)}" def prononcer_article_loi(article: str) -> str: """Prononce un article de loi (sans préfixe "article"). Exemples : "L 7112-4" → "L sept mille cent douze tiret quatre" "L. 3251-1" → "L trois mille deux cent cinquante et un" "L 1235-2" → "L douze cent trente-cinq tiret deux" """ article = article.strip() # Enlever les points article = article.replace(".", "").replace(" ", "") # Séparer la lettre du numéro match = re.match(r'^([A-Z]+)\s*(.*)', article) if match: lettre = match.group(1) num = match.group(2) return f"{lettre} {prononcer_numero(num)}" return f"{prononcer_numero(article)}" def prononcer_reference_juridique(text: str) -> str: """Remplace les références juridiques par leur prononciation. Exemples : "Cass. soc., 6-5-26, n°24-13599" → "Cour de cassation, chambre sociale, le 6 mai deux mille vingt-six, numéro vingt-quatre tiret treize mille cinq cent quatre-vingt-dix-neuf" "article L 7112-4" → "article L sept mille cent douze tiret quatre" "article L. 3251-1" → "article L trois mille deux cent cinquante et un" "décret n°2022-679" → "décret numéro deux mille vingt-deux tiret six cent soixante-dix-neuf" """ # Normalize dates: remove spaces around dashes in date patterns # This handles "18-3- 26", "25- 3-26", "1-4-26" etc. text = re.sub(r'(\d+)\s*-\s*(\d+)\s*-\s*(\d{2,4})', r'\1-\2-\3', text) # Cass. soc., DATE, n°NUMERO text = re.sub( r'Cass\.?\s*soc\.?,\s*(\d+-\d+-\d+),\s*n°(\d+-\d+)', lambda m: f"Cour de cassation, chambre sociale, {prononcer_date_cass(m.group(1))}, {prononcer_numero_arret(m.group(2))}", text ) # Cass. soc. DATE, n°NUMERO (sans virgule après soc.) text = re.sub( r'Cass\.?\s*soc\.?\s+(\d+-\d+-\d+),\s*n°(\d+-\d+)', lambda m: f"Cour de cassation, chambre sociale, {prononcer_date_cass(m.group(1))}, {prononcer_numero_arret(m.group(2))}", text ) # Cass. soc. DATE (without n°) text = re.sub( r'Cass\.?\s*soc\.?\s+(\d+-\d+-\d+)', lambda m: f"Cour de cassation, chambre sociale, {prononcer_date_cass(m.group(1))}", text ) # Cass. crim. DATE, n°NUMERO text = re.sub( r'Cass\.?\s*crim\.?,\s*(\d+-\d+-\d+),\s*n°(\d+-\d+)', lambda m: f"Cour de cassation, chambre criminelle, {prononcer_date_cass(m.group(1))}, {prononcer_numero_arret(m.group(2))}", text ) # Cass. crim. DATE (without n°) text = re.sub( r'Cass\.?\s*crim\.?\s+(\d+-\d+-\d+)', lambda m: f"Cour de cassation, chambre criminelle, {prononcer_date_cass(m.group(1))}", text ) # Multiple cassation refs in one string (e.g., "Cass. soc., 24-3-10, n°08-43996 ; Cass. soc., 2-4-14, n°13-10569") # Already handled by the regex above which matches each occurrence # article L XXX-XX (with or without point, replace entire "article L..." match) # Handles both "article L" and "articles L" text = re.sub( r'(\barticles?\s+)([A-Z][.]?\s*\d+-\d+)', lambda m: f"article {prononcer_article_loi(m.group(2))}", text, flags=re.IGNORECASE ) # décret n°XXXX-XXX text = re.sub( r'(décret\s+n°)(\d+-\d+)', lambda m: f"{m.group(1)}{prononcer_numero(m.group(2))}", text, flags=re.IGNORECASE ) # arrêté n°XXXX-XXX text = re.sub( r'(arrêté\s+n°)(\d+-\d+)', lambda m: f"{m.group(1)}{prononcer_numero(m.group(2))}", text, flags=re.IGNORECASE ) # ordonnance n°XXXX-XXX text = re.sub( r'(ordonnance\s+n°)(\d+-\d+)', lambda m: f"{m.group(1)}{prononcer_numero(m.group(2))}", text, flags=re.IGNORECASE ) # Conseil d'Etat text = text.replace("Conseil d'Etat", "Conseil d'État") return text # === SCRIPT DIALOGUÉ === # Variété des questions du présentateur pour éviter la répétition _HOST_QUESTIONS_BREVE = [ lambda t: f"Marc, que faut-il retenir concernant {t} ?", lambda t: f"Marc, pouvez-vous nous en dire plus sur {t} ?", lambda t: f"Marc, qu'est-ce que cet arrêt nous apprend sur {t} ?", lambda t: f"Marc, expliquez-nous les enjeux liés à {t}.", lambda t: f"Marc, quelle est la portée de cet arrêt relatif à {t} ?", lambda t: f"Marc, pourquoi cet arrêt sur {t} est-il significatif ?", ] _HOST_QUESTIONS_FOCUS = [ lambda t: f"Merci Sophie. Et concernant plus précisément {t} ?", lambda t: f"Quelles sont les conséquences concrètes pour les employeurs et les salariés ?", lambda t: f"Sophie, quels conseils pratiques donnez-vous suite à ce point ?", ] def create_radio_script_dialogue( date_range: str, breves_summary: str, focus_title: str, focus_summary: str, actualites_legislatives: str = "", breves_arrets: list = None, focus_arrets: list = None, focus_paragraphs: list = None, ) -> dict: """ Construit le script radio juridique à partir des extraits du PDF. Args: date_range: plage de dates du PDF (ex: "04 au 07 mai 2026") breves_summary: résumé des brèves de la page 1 focus_title: titre du focus (page 2) focus_summary: résumé du focus actualites_legislatives: actualités législatives et réglementaires (optionnel) breves_arrets: liste de dicts {"title": ..., "body": ...} pour page 1 focus_arrets: liste de dicts {"title": ..., "body": ...} pour page 2 focus_paragraphs: liste de paragraphes du focus (optionnel, pour split en segments) Returns: dict compatible avec generate_radio_emission() """ date_str = _date_fr() script = {} # === JINGLE === script["intro_jingle"] = { "voice_role": "jingle", "text": "Radio Juridique", } # === INTRO === intro_text = ( f"Bonjour à toutes et à tous, bienvenue dans votre Radio Juridique. " f"Nous sommes le {date_str}. " f"Aujourd'hui, on fait le point sur la veille juridique de la Confédération F.O., " f"pour la période du {date_range}. " f"Je suis Henri. " f"J'accompagne Marc, juriste en droit du travail, pour décrypter les brèves de la semaine, " f"et Sophie, également juriste, pour notre focus sur un point de droit important. " f"C'est parti." ) script["intro_host"] = { "voice_role": "presentateur", "text": prononcer_reference_juridique(intro_text), } # === PARTIE 1 : ACTUALITÉS LÉGISLATIVES ET RÉGLEMENTAIRES === if actualites_legislatives and actualites_legislatives.strip(): script["partie1_intro"] = { "voice_role": "presentateur", "text": ( f"Première partie : les actualités législatives et réglementaires. " f"Pour la période du {date_range}, voici ce qui a été publié." ), } script["partie1_expert"] = { "voice_role": "expert_breves", "text": prononcer_reference_juridique(actualites_legislatives), } script["partie1_transition"] = { "voice_role": "presentateur", "text": "Merci Marc. Passons maintenant à la jurisprudence.", } else: script["partie1_intro"] = { "voice_role": "presentateur", "text": ( f"Première partie : les actualités législatives et réglementaires. " f"Pour la période du {date_range}, aucune actualité législative et réglementaire " f"importante n'est intervenue cette semaine. Rien de nouveau dans les textes. " f"Passons directement à la jurisprudence." ), } script["partie1_transition"] = { "voice_role": "presentateur", "text": "Passons maintenant à la jurisprudence.", } # === PARTIE 2 : BRÈVES DE JURISPRUDENCE === if breves_arrets and len(breves_arrets) > 0: script["partie2_intro"] = { "voice_role": "presentateur", "text": ( f"Deuxième partie : les brèves de jurisprudence. " f"Le bulletin juridique de F.O. pour la période du {date_range} " f"fait le point sur les arrêts récents de la Cour de cassation, chambre sociale. " f"Marc, qu'est-ce qui se dégage de cette semaine ?" ), } script["partie2_expert_intro"] = { "voice_role": "expert_breves", "text": ( "Alors Henri, cette semaine, la Cour de cassation, chambre sociale, " "a rendu plusieurs arrêts intéressants. " "Je vais vous en parler un par un." ), } for idx, arrest in enumerate(breves_arrets): title = arrest.get("title", "") body = arrest.get("body", "") # Relance variée du présentateur question_fn = _HOST_QUESTIONS_BREVE[idx % len(_HOST_QUESTIONS_BREVE)] script[f"partie2_q{idx}_host"] = { "voice_role": "presentateur", "text": prononcer_reference_juridique(question_fn(title)), } script[f"partie2_q{idx}_expert"] = { "voice_role": "expert_breves", "text": prononcer_reference_juridique(_format_expert_response(title, body)), } script["partie2_transition"] = { "voice_role": "presentateur", "text": ( "Merci Marc pour ce tour d'horizon. Passons maintenant à notre focus de la semaine." ), } else: script["partie2_intro"] = { "voice_role": "presentateur", "text": ( f"Deuxième partie : les brèves de jurisprudence. " f"Pour la période du {date_range}, Marc, qu'est-ce qui se dégage de cette semaine ?" ), } script["partie2_expert"] = { "voice_role": "expert_breves", "text": prononcer_reference_juridique(breves_summary), } script["partie2_transition"] = { "voice_role": "presentateur", "text": "Merci Marc. Passons au focus.", } # === PARTIE 3 : FOCUS === script["partie3_intro"] = { "voice_role": "presentateur", "text": ( f"Troisième partie : notre focus juridique. " f"Cette semaine, le bulletin de F.O. aborde un point important : " f"{focus_title}. " f"Sophie, pouvez-vous nous éclairer sur ce sujet ?" ), } # Sophie répond directement sans répéter le titre script["partie3_expert_intro"] = { "voice_role": "expert_focus", "text": prononcer_reference_juridique(focus_summary), } # === SPLIT FOCUS EN PLUSIEURS SEGMENTS === if focus_arrets and len(focus_arrets) > 0: # Intro focus focus_intro = ( "Alors, pour ce sujet, c'est un point très complet qui couvre plusieurs aspects. " "Je vais vous détailler tout ça." ) script["partie3_q1_expert"] = { "voice_role": "expert_focus", "text": prononcer_reference_juridique(focus_intro), } # Chaque arrêt du focus dans son propre segment for idx, arr in enumerate(focus_arrets): title_arr = arr.get("title", "") body_arr = arr.get("body", "") clean_body = body_arr.replace("\n", " ").replace(" ", " ").strip() sentences = [s.strip() for s in clean_body.split(".") if s.strip()] question_fn = _HOST_QUESTIONS_FOCUS[idx % len(_HOST_QUESTIONS_FOCUS)] script[f"partie3_q{idx + 2}_host"] = { "voice_role": "presentateur", "text": prononcer_reference_juridique(question_fn(title_arr)), } if sentences: script[f"partie3_q{idx + 2}_expert"] = { "voice_role": "expert_focus", "text": prononcer_reference_juridique(" ".join(sentences)), } script["partie3_outro"] = { "voice_role": "expert_focus", "text": ( "Pour conclure, c'est un point très important à surveiller " "pour les employeurs comme pour les salariés." ), } elif focus_paragraphs and len(focus_paragraphs) > 0: # Use paragraphs as segments (for focus content without clear arrêt structure) focus_intro = ( "Alors, pour ce sujet, c'est un point très complet qui couvre plusieurs aspects. " "Je vais vous détailler tout ça." ) script["partie3_q1_expert"] = { "voice_role": "expert_focus", "text": prononcer_reference_juridique(focus_intro), } for idx, para in enumerate(focus_paragraphs): clean_para = para.replace("\n", " ").replace(" ", " ").strip() sentences = [s.strip() for s in clean_para.split(".") if s.strip()] if sentences: script[f"partie3_q{idx + 2}_expert"] = { "voice_role": "expert_focus", "text": prononcer_reference_juridique(" ".join(sentences)), } script["partie3_outro"] = { "voice_role": "expert_focus", "text": ( "Pour conclure, c'est un point très important à surveiller " "pour les employeurs comme pour les salariés." ), } else: script["partie3_q1_expert"] = { "voice_role": "expert_focus", "text": prononcer_reference_juridique(_format_focus_conclusion(focus_title, focus_arrets)), } # === CONCLUSION === script["outro"] = { "voice_role": "presentateur", "text": ( f"Merci à Marc et à Sophie pour ces éclairages juridiques. " f"C'était votre Radio Juridique pour la période du {date_range}. " f"On se retrouve très bientôt pour un nouveau tour d'horizon de la jurisprudence. " f"Bonne journée à toutes et à tous." ), } return script def _format_expert_response(title: str, body: str) -> str: """Formate une réponse d'expert pour la TTS — contenu complet, pas de limite. Le présentateur a déjà annoncé le titre, on va droit au but. Pas de préambule qui répète le titre — uniquement le corps. """ clean_body = body.replace("\n", " ").replace(" ", " ").strip() sentences = [s.strip() for s in clean_body.split(".") if s.strip()] if sentences: return " ".join(sentences) return "" def _format_focus_conclusion(title: str, focus_arrets: list = None) -> str: """Formate la conclusion du focus — contenu complet avec tous les détails. Ne répète PAS le titre déjà annoncé par le présentateur. Va directement au contenu. """ parts = [] if focus_arrets and len(focus_arrets) > 0: for arr in focus_arrets: title_arr = arr.get("title", "") body_arr = arr.get("body", "") clean_body = body_arr.replace("\n", " ").replace(" ", " ").strip() sentences = [s.strip() for s in clean_body.split(".") if s.strip()] if title_arr: parts.append(f"Concernant {title_arr}.") if sentences: parts.append(" ".join(sentences)) return " ".join(parts) if parts else "" # === MAIN (test) === if __name__ == "__main__": engine = _get_engine_name() print(f"Génération Radio Juridique (test — {engine})...") script = create_radio_script_dialogue( date_range="04 au 07 mai 2026", breves_summary="Cette semaine, plusieurs arrêts de la Cour de cassation, chambre sociale, ont été rendus.", focus_title="Attention à l'indemnité de préavis", focus_summary="Les arrêts de la Cour de cassation du 9 avril 2026 rappellent que l'employeur ne peut pas compenser l'indemnité de préavis avec le salaire.", actualites_legislatives="Aucune actualité législative et réglementaire importante n'est intervenue cette semaine.", breves_arrets=[ {"title": "visite de reprise et dispositions conventionnelles", "body": "La durée minimale d'absence pour maladie avant visite de reprise est celle fixée par les conventions collectives, pas le décret réglementaire. Les délais du décret 2022-679 ne sont pas d'ordre public absolu. (Cass. soc., 6-5-26, n°24-13599)."}, {"title": "journaliste et résiliation judiciaire", "body": "La commission arbitrale doit être saisie pour déterminer l'indemnité du journaliste avec plus de 15 ans d'ancienneté. L'article L 7112-4 s'applique même si c'est l'autorité judiciaire qui a jugé le licenciement. (Cass. soc., 6-5-26, n°25-12049)."}, {"title": "forfait-jours et validité", "body": "Toute convention de forfait-jours doit assurer le respect des durées raisonnables de travail et des repos. (Cass. soc., 6-5-26, n°24-10699)."}, ], focus_arrets=[ {"title": "indemnité de préavis et compensation salariale", "body": "L'employeur ne peut pas opérer une retenue de salaire pour compenser le préavis. Article L. 3251-1 du code du travail. (Cass. soc. 9-4-26, n°25-10995)."}, {"title": "prise d'acte et indemnité de préavis", "body": "Quand la prise d'acte produit les effets d'une démission, le salarié est redevable de l'indemnité compensatrice de préavis. (Cass. soc., 9-4-26, n°24-21017)."}, ], ) from validate_juridique import validate_juridique_script, format_report errors, warnings = validate_juridique_script(script) print(format_report(errors, warnings)) print(f"\nTotal segments: {len(script)}") for name, seg in script.items(): print(f" {name}: {seg['voice_role']} ({len(seg['text'])} chars)") output = str(AUDIO_DIR / "radio-juridique-test.mp3") asyncio.run(generate_radio_emission(script, output)) print(f"\nFichier généré : {output}")