"""Module de conversion texte vers audio via ElevenLabs.""" import time import requests from pathlib import Path from typing import Optional from elevenlabs import ElevenLabs, VoiceSettings import config class TextToSpeech: """Gère la conversion texte vers audio via ElevenLabs API.""" def __init__(self): if not config.ELEVENLABS_API_KEY: raise ValueError( "ELEVENLABS_API_KEY non définie. Configurez-la dans .env" ) self.client = ElevenLabs(api_key=config.ELEVENLABS_API_KEY) self.voice_id = config.ELEVENLABS_VOICE_ID self.model = config.ELEVENLABS_MODEL self.api_key = config.ELEVENLABS_API_KEY self.max_retries = 3 self.retry_delay = 5 # secondes def generate_audio( self, text: str, output_filename: str, voice_id: Optional[str] = None, ) -> Path: """Génère un fichier audio à partir du texte.""" if not text.strip(): raise ValueError("Le texte ne peut pas être vide") voice_id = voice_id or self.voice_id output_path = config.OUTPUT_AUDIO_DIR / f"{output_filename}.{config.AUDIO_FORMAT}" # Vérifier si le texte doit être découpé en chunks if len(text) > config.MAX_TEXT_CHUNK_SIZE: return self._generate_chunked_audio(text, output_path, voice_id) # Essayer les voix par ordre de priorité: Yann > Sebastien > Matilda voices_to_try = [(voice_id, "custom")] if voice_id else [] for vid, vname in config.ELEVENLABS_VOICES: if vid != voice_id: voices_to_try.append((vid, vname)) for i, (vid, vname) in enumerate(voices_to_try): # Essayer avec retries for retry in range(self.max_retries): try: # Utiliser l'API REST directement (plus stable que le SDK) success = self._generate_audio_rest(text, output_path, vid) if success and output_path.stat().st_size > 0: print(f" Voix utilisée: {vname}") return output_path else: raise Exception("Fichier audio vide généré") except Exception as e: error_msg = str(e) if retry < self.max_retries - 1: print(f" ⚠ Retry {retry + 1}/{self.max_retries} pour {vname}: {error_msg[:50]}...") time.sleep(self.retry_delay) else: if i < len(voices_to_try) - 1: next_voice = voices_to_try[i + 1][1] print(f"⚠ Voix {vname} indisponible après {self.max_retries} essais, essai de {next_voice}...") break # Sortir de la boucle retry pour essayer la voix suivante else: raise e return output_path def _generate_audio_rest(self, text: str, output_path: Path, voice_id: str) -> bool: """Génère l'audio via l'API REST directement (plus stable).""" url = f"https://api.elevenlabs.io/v1/text-to-speech/{voice_id}" headers = { "Accept": "audio/mpeg", "Content-Type": "application/json", "xi-api-key": self.api_key } data = { "text": text, "model_id": self.model, "voice_settings": { "stability": 0.5, "similarity_boost": 0.75, "style": 0.0, "use_speaker_boost": True } } response = requests.post(url, json=data, headers=headers, timeout=120) if response.status_code == 200: with open(output_path, "wb") as f: f.write(response.content) return True else: raise Exception(f"API Error {response.status_code}: {response.text[:200]}") def _generate_chunked_audio( self, text: str, output_path: Path, voice_id: str, ) -> Path: """Génère l'audio en plusieurs chunks et les combine.""" chunks = self._split_text(text, config.MAX_TEXT_CHUNK_SIZE) temp_files = [] try: for i, chunk in enumerate(chunks): temp_path = output_path.parent / f"_temp_chunk_{i}.mp3" temp_files.append(temp_path) # Générer le chunk audio audio_generator = self.client.text_to_speech.convert( voice_id=voice_id, text=chunk, model_id=self.model, voice_settings=VoiceSettings( stability=0.5, similarity_boost=0.75, style=0.0, use_speaker_boost=True, ), ) with open(temp_path, "wb") as f: for audio_chunk in audio_generator: f.write(audio_chunk) # Pause pour respecter le rate limiting if i < len(chunks) - 1: time.sleep(0.5) # Combiner les fichiers audio self._combine_audio_files(temp_files, output_path) finally: # Nettoyer les fichiers temporaires for temp_file in temp_files: if temp_file.exists(): temp_file.unlink() return output_path def _split_text(self, text: str, max_size: int) -> list[str]: """Découpe le texte en chunks respectant les limites de phrases.""" chunks = [] current_chunk = "" # Découper par paragraphes d'abord paragraphs = text.split("\n\n") for para in paragraphs: if len(current_chunk) + len(para) + 2 <= max_size: current_chunk += para + "\n\n" else: if current_chunk: chunks.append(current_chunk.strip()) # Si le paragraphe est trop long, le découper par phrases if len(para) > max_size: sentences = self._split_into_sentences(para) current_chunk = "" for sentence in sentences: if len(current_chunk) + len(sentence) + 1 <= max_size: current_chunk += sentence + " " else: if current_chunk: chunks.append(current_chunk.strip()) current_chunk = sentence + " " else: current_chunk = para + "\n\n" if current_chunk.strip(): chunks.append(current_chunk.strip()) return chunks def _split_into_sentences(self, text: str) -> list[str]: """Découpe un texte en phrases.""" import re # Pattern pour découper en phrases sentences = re.split(r"(?<=[.!?])\s+", text) return [s.strip() for s in sentences if s.strip()] def _combine_audio_files(self, input_files: list[Path], output_path: Path) -> None: """Combine plusieurs fichiers MP3 en un seul.""" # Méthode simple: concaténation binaire (fonctionne pour MP3) with open(output_path, "wb") as outfile: for input_file in input_files: with open(input_file, "rb") as infile: outfile.write(infile.read()) def list_voices(self) -> list[dict]: """Liste les voix disponibles.""" response = self.client.voices.get_all() voices = [] for voice in response.voices: voices.append( { "voice_id": voice.voice_id, "name": voice.name, "category": voice.category, "labels": voice.labels, } ) return voices def get_voice_info(self, voice_id: str) -> dict: """Récupère les informations d'une voix.""" voice = self.client.voices.get(voice_id) return { "voice_id": voice.voice_id, "name": voice.name, "category": voice.category, "labels": voice.labels, }