Spaces:
Sleeping
Sleeping
| import os, uuid, asyncio | |
| from typing import Dict | |
| from pydub import AudioSegment | |
| import edge_tts | |
| from .config import TMP_DIR | |
| EDGE_VOICES: Dict[str, str] = {} | |
| async def fetch_edge_voices_async(): | |
| """Charge dynamiquement toutes les voix FR/NL depuis Edge-TTS.""" | |
| global EDGE_VOICES | |
| try: | |
| voices = await edge_tts.list_voices() | |
| filtered = [v for v in voices if v.get("Locale", "").startswith(("fr", "nl"))] | |
| filtered.sort(key=lambda v: (v.get("Locale"), v.get("Gender"), v.get("ShortName"))) | |
| EDGE_VOICES = { | |
| f"{v['ShortName']} - {v['Locale']} ({v['Gender']})": v["ShortName"] | |
| for v in filtered | |
| } | |
| print(f"[Edge-TTS] {len(EDGE_VOICES)} voix FR/NL chargées.") | |
| except Exception as e: | |
| print(f"[Edge-TTS] Erreur chargement voix : {e}") | |
| EDGE_VOICES.update({ | |
| "fr-FR-DeniseNeural - fr-FR (Female)": "fr-FR-DeniseNeural", | |
| "nl-NL-MaaikeNeural - nl-NL (Female)": "nl-NL-MaaikeNeural", | |
| }) | |
| def init_edge_voices(): | |
| """Démarre le chargement asynchrone sans bloquer Gradio.""" | |
| try: | |
| loop = asyncio.get_event_loop() | |
| if loop.is_running(): | |
| import nest_asyncio | |
| nest_asyncio.apply() | |
| loop.create_task(fetch_edge_voices_async()) | |
| else: | |
| loop.run_until_complete(fetch_edge_voices_async()) | |
| except RuntimeError: | |
| asyncio.run(fetch_edge_voices_async()) | |
| def get_edge_voices(lang="fr"): | |
| """Retourne les voix déjà chargées (selon la langue).""" | |
| global EDGE_VOICES | |
| if not EDGE_VOICES: | |
| init_edge_voices() | |
| if lang == "fr": | |
| return [v for k, v in EDGE_VOICES.items() if k.startswith("fr-")] | |
| elif lang == "nl": | |
| return [v for k, v in EDGE_VOICES.items() if k.startswith("nl-")] | |
| return list(EDGE_VOICES.values()) | |
| async def _edge_tts_async(text, voice, outfile): | |
| communicate = edge_tts.Communicate(text, voice) | |
| await communicate.save(outfile) | |
| return outfile | |
| def tts_edge(text: str, voice: str = "fr-FR-DeniseNeural") -> str: | |
| """Génère un fichier WAV avec Edge-TTS (et fallback gTTS).""" | |
| out_mp3 = os.path.join(TMP_DIR, f"edge_{uuid.uuid4().hex}.mp3") | |
| try: | |
| try: | |
| loop = asyncio.get_event_loop() | |
| if loop.is_running(): | |
| import nest_asyncio | |
| nest_asyncio.apply() | |
| except RuntimeError: | |
| pass | |
| asyncio.run(_edge_tts_async(text, voice, out_mp3)) | |
| out_wav = os.path.join(TMP_DIR, f"edge_{uuid.uuid4().hex}.wav") | |
| AudioSegment.from_file(out_mp3).export(out_wav, format="wav") | |
| os.remove(out_mp3) | |
| return out_wav | |
| except Exception as e: | |
| print(f"[Edge-TTS] Erreur : {e} → fallback gTTS") | |
| return tts_gtts(text, lang="fr" if voice.startswith("fr") else "nl") | |
| def tts_gtts(text: str, lang: str = "fr") -> str: | |
| from gtts import gTTS | |
| out = os.path.join(TMP_DIR, f"gtts_{uuid.uuid4().hex}.mp3") | |
| gTTS(text=text, lang=lang).save(out) | |
| out_wav = os.path.join(TMP_DIR, f"gtts_{uuid.uuid4().hex}.wav") | |
| AudioSegment.from_file(out).export(out_wav, format="wav") | |
| os.remove(out) | |
| return out_wav | |
| def normalize_audio_to_wav(in_path: str) -> str: | |
| """Convertit n'importe quel format en WAV 44.1kHz stéréo.""" | |
| from pydub import AudioSegment | |
| wav_path = os.path.join(TMP_DIR, f"norm_{uuid.uuid4().hex}.wav") | |
| snd = AudioSegment.from_file(in_path) | |
| snd = snd.set_frame_rate(44100).set_channels(2).set_sample_width(2) | |
| snd.export(wav_path, format="wav") | |
| return wav_path | |