import gradio as gr
import edge_tts
import tempfile
import asyncio
import hashlib
import os
from pathlib import Path
from functools import lru_cache

# Voice Mapping
VOICE_MAP = {
    "رجل (مصري)": "ar-EG-ShakirNeural",
    "سيدة (مصرية)": "ar-EG-SalmaNeural",
    "رجل (سعودي)": "ar-SA-HamedNeural",
    "سيدة (سعودية)": "ar-SA-ZariyahNeural",
    "English (US) M": "en-US-EricNeural",
    "English (US) F": "en-US-AriaNeural"
}

# معالجة 2 طلبات فقط في نفس الوقت (لـ 2 vCPUs)
TTS_SEMAPHORE = asyncio.Semaphore(2)

# Cache directory
CACHE_DIR = Path("./tts_cache")
CACHE_DIR.mkdir(exist_ok=True)

# تنظيف الكاش القديم عند البداية
def cleanup_old_cache(max_files=100):
    """حذف أقدم ملفات الكاش إذا تجاوز العدد الحد المسموح"""
    cache_files = sorted(CACHE_DIR.glob("*.mp3"), key=os.path.getmtime)
    if len(cache_files) > max_files:
        for f in cache_files[:len(cache_files) - max_files]:
            try:
                f.unlink()
            except:
                pass

cleanup_old_cache(max_files=100)

def generate_cache_key(text, voice, rate, pitch):
    """توليد مفتاح فريد للكاش"""
    content = f"{text[:500]}{voice}{rate}{pitch}"  # أول 500 حرف فقط
    return hashlib.md5(content.encode('utf-8')).hexdigest()

async def generate_speech(text, voice, emotion, is_symbol, rate, pitch):
    if not text or not text.strip(): 
        return None
    
    # تقليم النص الطويل جداً (حماية من الإساءة)
    text = text[:5000]
    
    # Defaults
    final_rate = rate if rate and isinstance(rate, str) and len(rate.strip()) > 0 else "+0%"
    final_pitch = pitch if pitch and isinstance(pitch, str) and len(pitch.strip()) > 0 else "+0Hz"
    
    # Voice Selection
    selected_voice = "ar-SA-HamedNeural"
    if voice in VOICE_MAP: 
        selected_voice = VOICE_MAP[voice]
    elif voice in VOICE_MAP.values(): 
        selected_voice = voice
    
    # Check cache
    cache_key = generate_cache_key(text, selected_voice, final_rate, final_pitch)
    cache_path = CACHE_DIR / f"{cache_key}.mp3"
    
    if cache_path.exists():
        print(f"✓ Cache hit: {len(text)} chars")
        return str(cache_path)
    
    print(f"⚙ Generating: {len(text)} chars | {selected_voice}")

    # Semaphore للتحكم في التزامن
    async with TTS_SEMAPHORE:
        try:
            # استخدام cache path مباشرة بدلاً من tempfile
            communicate = edge_tts.Communicate(
                text, 
                selected_voice, 
                rate=final_rate, 
                pitch=final_pitch
            )
            
            # حفظ مباشرة في الكاش
            await asyncio.wait_for(
                communicate.save(str(cache_path)),
                timeout=45.0  # timeout 45 ثانية
            )
            
            return str(cache_path)
            
        except asyncio.TimeoutError:
            print(f"✗ Timeout for {len(text)} chars")
            if cache_path.exists():
                cache_path.unlink()
            raise gr.Error("Request timeout - try shorter text")
        except Exception as e:
            print(f"✗ ERROR: {str(e)}")
            if cache_path.exists():
                cache_path.unlink()
            raise gr.Error(f"TTS Error: {str(e)[:100]}")

# UI Definition
with gr.Blocks(title="Natiq Pro API", css=".gradio-container {max-width: 100%}") as demo:
    with gr.Row(visible=False):
        t = gr.Textbox(label="Text")
        v = gr.Textbox(label="Voice")
        e = gr.Textbox(label="Emotion", value="neutral")
        s = gr.Checkbox(label="Is Symbol", value=True)
        r = gr.Textbox(label="Rate", value="+0%")
        p = gr.Textbox(label="Pitch", value="+0Hz")
    
    o = gr.Audio(label="Output", type="filepath")
    b = gr.Button("Generate", visible=False)
    
    b.click(
        generate_speech, 
        inputs=[t, v, e, s, r, p], 
        outputs=[o], 
        api_name="text_to_speech_edge",
        concurrency_limit=3  # 3 طلبات متزامنة كحد أقصى
    )

if __name__ == "__main__":
    demo.queue(
        default_concurrency_limit=3,
        max_size=30  # قائمة انتظار 30 طلب فقط
    ).launch(
        max_threads=6,  # threads محدودة
        show_error=True
    )