app.py

import asyncio
import os
import tempfile
from concurrent.futures import ThreadPoolExecutor

import edge_tts
import gradio as gr

DEFAULT_VOICE = "en-US-JennyNeural"

# Used only if the live voice list can't be fetched at startup.
FALLBACK_VOICES = {
    "Urdu - Uzma (Female)": "ur-PK-UzmaNeural",
    "Urdu - Asad (Male)": "ur-PK-AsadNeural",
    "English (US) - Jenny (Female)": "en-US-JennyNeural",
    "English (US) - Guy (Male)": "en-US-GuyNeural",
    "English (US) - Aria (Female)": "en-US-AriaNeural",
    "English (UK) - Sonia (Female)": "en-GB-SoniaNeural",
    "English (UK) - Ryan (Male)": "en-GB-RyanNeural",
    "English (India) - Neerja (Female)": "en-IN-NeerjaNeural",
    "Hindi - Swara (Female)": "hi-IN-SwaraNeural",
    "Hindi - Madhur (Male)": "hi-IN-MadhurNeural",
    "Arabic (Saudi) - Hamed (Male)": "ar-SA-HamedNeural",
    "Arabic (Saudi) - Zariyah (Female)": "ar-SA-ZariyahNeural",
    "Spanish (Spain) - Elvira (Female)": "es-ES-ElviraNeural",
    "French - Denise (Female)": "fr-FR-DeniseNeural",
    "German - Katja (Female)": "de-DE-KatjaNeural",
    "Japanese - Nanami (Female)": "ja-JP-NanamiNeural",
    "Korean - SunHi (Female)": "ko-KR-SunHiNeural",
    "Pashto - Gul Nawaz (Male)": "ps-AF-GulNawazNeural",
}


def load_voices():
    """Fetch every voice Edge TTS currently offers, Urdu first."""
    try:
        # Run in a worker thread so the main thread's event loop is left
        # untouched (asyncio.run on the main thread breaks Gradio's queue).
        with ThreadPoolExecutor(max_workers=1) as pool:
            voices = pool.submit(lambda: asyncio.run(edge_tts.list_voices())).result(timeout=30)
    except Exception as exc:  # network issue, API change, etc.
        print(f"Could not fetch live voice list ({exc}); using fallback list.")
        return list(FALLBACK_VOICES.items())

    items = []
    for v in voices:
        short = v["ShortName"]
        name = short.split("-", 2)[-1].replace("Neural", "")
        label = f'{v["Locale"]} - {name} ({v["Gender"]})'
        items.append((label, short))

    items.sort(key=lambda x: (not x[1].startswith("ur-"), x[0]))
    return items or list(FALLBACK_VOICES.items())


VOICES = load_voices()
VALID_VOICES = {value for _, value in VOICES}


async def text_to_speech_edge(text, voice, speed):
    text = (text or "").strip()
    if not text:
        raise gr.Error("Please enter some text first.")

    if voice not in VALID_VOICES:
        voice = DEFAULT_VOICE

    rate = f"{int(speed):+d}%"
    communicate = edge_tts.Communicate(text, voice, rate=rate)

    # edge-tts outputs MP3 data, so use the .mp3 extension.
    with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as tmp:
        tmp_path = tmp.name

    try:
        await communicate.save(tmp_path)
    except Exception as exc:
        raise gr.Error(f"Speech synthesis failed: {exc}")

    return f"Speech synthesis completed ({voice}, speed {rate}).", tmp_path


demo = gr.Interface(
    fn=text_to_speech_edge,
    inputs=[
        gr.Textbox(lines=5, label="Input Text", placeholder="Type or paste your text here..."),
        gr.Dropdown(
            choices=VOICES,
            value=DEFAULT_VOICE if DEFAULT_VOICE in VALID_VOICES else VOICES[0][1],
            label="Choose the Voice Model",
        ),
        gr.Slider(-50, 50, value=0, step=5, label="Speed (%)"),
    ],
    outputs=[
        gr.Textbox(label="Status"),
        gr.Audio(type="filepath", label="Exported Audio"),
    ],
    title="Edge TTS",
    description="Microsoft Edge Text-To-Speech (Forked & Fixed Ilaria TTS)",
    allow_flagging="never" if int(gr.__version__.split(".")[0]) < 5 else None,
)

if __name__ == "__main__":
    demo.queue().launch(
        server_name="0.0.0.0",
        server_port=int(os.environ.get("PORT", 7860)),
    )