File size: 4,441 Bytes
05f09ca 4ebd109 4a1191e 8854e7d 05f09ca a4dc04b d173109 8854e7d 4a1191e 8854e7d 4a1191e a4dc04b 4a1191e a4dc04b 8854e7d a4dc04b 4a1191e a4dc04b 4a1191e 8854e7d 05f09ca 4ebd109 8854e7d | 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 | 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
) |