Spaces:
Sleeping
Sleeping
File size: 4,442 Bytes
05f09ca 4ebd109 4a1191e 84a4e3e 05f09ca a4dc04b d173109 84a4e3e 4a1191e 84a4e3e 4a1191e a4dc04b 4a1191e a4dc04b 84a4e3e a4dc04b 4a1191e a4dc04b 4a1191e 84a4e3e 05f09ca 4ebd109 84a4e3e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | 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
)
|