app.py

# app.py — جسر: خادم موحّد على Hugging Face Space (ZeroGPU)
# يجمع ثلاث خدمات في مكان واحد:
#   /tts        → توليد كلام بصوت مستنسخ (Chatterbox Multilingual)
#   /translate  → ترجمة نصية (API خارجي: MyMemory/Microsoft/Google)
#   /stt        → تفريغ صوتي (Whisper)
#
# ⚠️ يعمل داخل نسخة (Duplicate) من الـ Space الرسمي:
#    ResembleAI/Chatterbox-Multilingual-TTS-V3  (كود Chatterbox في src/)

import os
import sys

sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "src"))

import spaces          # مطلوب لـ ZeroGPU
import torch
import torchaudio
import gradio as gr

# ============================================================
#  اللغات المدعومة
# ============================================================
LANGS = {
    "العربية (ar)": "ar",
    "Türkçe (tr)": "tr",
    "English (en)": "en",
    "हिन्दी (hi)": "hi",
    "Español (es)": "es",
    "Deutsch (de)": "de",
    "Français (fr)": "fr",
}
LANG_CODES = set(LANGS.values())
DEFAULT_VOICE = "default_voice.wav"

# ============================================================
#  1) نموذج الكلام — Chatterbox (يُحمّل مرة واحدة)
# ============================================================
_tts_model = None

def get_tts_model():
    global _tts_model
    if _tts_model is None:
        from chatterbox.mtl_tts import ChatterboxMultilingualTTS
        _tts_model = ChatterboxMultilingualTTS.from_pretrained(device="cuda")
    return _tts_model

# (الترجمة الآن عبر API خارجي - انظر دالة translate)

# ============================================================
#  3) نموذج التفريغ — Whisper
# ============================================================
_asr = None

def get_asr():
    global _asr
    if _asr is None:
        import whisper
        # جرّب أسماء نموذج turbo بالترتيب (تختلف بين إصدارات whisper)
        # نبدأ بالأخف لسرعة التحميل وتجنّب انقطاع ZeroGPU، ثم الأثقل
        candidates = ["small", "medium", "large-v3-turbo", "turbo"]
        last_err = None
        for name in candidates:
            try:
                _asr = whisper.load_model(name, device="cuda")
                print(f"✅ Whisper loaded: {name}")
                break
            except Exception as e:
                last_err = e
                print(f"⚠️ تعذّر تحميل {name}: {e}")
        if _asr is None:
            raise RuntimeError(f"فشل تحميل أي نموذج Whisper: {last_err}")
    return _asr

# ============================================================
#  الدوال المعروضة كـ API
# ============================================================
@spaces.GPU(duration=120)
def tts(text: str, language: str, exaggeration: float = 0.5,
        cfg_weight: float = 0.5, voice_ref=None):
    """توليد كلام بصوت مستنسخ. يعيد مسار WAV."""
    text = (text or "").strip()
    if not text:
        raise gr.Error("أدخل نصاً أولاً")
    if len(text) > 1000:
        raise gr.Error("الحد الأقصى 1000 حرف")
    lang = LANGS.get(language, language)
    if lang not in LANG_CODES:
        raise gr.Error(f"لغة غير مدعومة: {language}")
    ref = voice_ref if voice_ref else DEFAULT_VOICE
    if not os.path.exists(ref):
        raise gr.Error("ملف الصوت المرجعي default_voice.wav غير موجود — ارفعه للـ Space")
    model = get_tts_model()
    wav = model.generate(
        text, language_id=lang, audio_prompt_path=ref,
        exaggeration=float(exaggeration), cfg_weight=float(cfg_weight),
    )
    out_path = "/tmp/speech.wav"
    torchaudio.save(out_path, wav, model.sr)
    return out_path


def translate(text: str, source: str, target: str):
    """ترجمة نص عبر API خارجي (بلا GPU - سريع وموثوق).

    الافتراضي: MyMemory (مجاني، بلا مفتاح - للتجربة).
    للإنتاج: اضبط متغير البيئة على أحد هذين للجودة الأعلى:
      - MS_TRANSLATOR_KEY  (Microsoft Translator - 2 مليون حرف/شهر مجاناً)
      - GOOGLE_TRANSLATE_KEY (Google Cloud Translation)
    """
    import urllib.parse
    import urllib.request
    import json as _json

    text = (text or "").strip()
    if not text:
        return ""
    tgt = LANGS.get(target, target)
    src = LANGS.get(source, source)
    if tgt not in LANG_CODES:
        raise gr.Error(f"لغة هدف غير مدعومة: {target}")
    if src == tgt:
        return text

    # 1) Microsoft Translator (إن وُجد المفتاح) - الأفضل للإنتاج
    ms_key = os.environ.get("MS_TRANSLATOR_KEY", "")
    if ms_key:
        try:
            region = os.environ.get("MS_TRANSLATOR_REGION", "global")
            url = ("https://api.cognitive.microsofttranslator.com/translate"
                   f"?api-version=3.0&from={src}&to={tgt}")
            body = _json.dumps([{"Text": text}]).encode("utf-8")
            req = urllib.request.Request(url, data=body, headers={
                "Ocp-Apim-Subscription-Key": ms_key,
                "Ocp-Apim-Subscription-Region": region,
                "Content-Type": "application/json",
            })
            with urllib.request.urlopen(req, timeout=15) as r:
                data = _json.loads(r.read())
            return data[0]["translations"][0]["text"].strip()
        except Exception as e:
            print(f"MS Translator error: {e}", flush=True)

    # 2) Google Cloud Translation (إن وُجد المفتاح)
    g_key = os.environ.get("GOOGLE_TRANSLATE_KEY", "")
    if g_key:
        try:
            url = f"https://translation.googleapis.com/language/translate/v2?key={g_key}"
            body = urllib.parse.urlencode({
                "q": text, "source": src, "target": tgt, "format": "text",
            }).encode("utf-8")
            with urllib.request.urlopen(url, data=body, timeout=15) as r:
                data = _json.loads(r.read())
            return data["data"]["translations"][0]["translatedText"].strip()
        except Exception as e:
            print(f"Google Translate error: {e}", flush=True)

    # 3) MyMemory (مجاني بلا مفتاح - افتراضي للتجربة)
    try:
        q = urllib.parse.quote(text)
        url = (f"https://api.mymemory.translated.net/get"
               f"?q={q}&langpair={src}|{tgt}")
        with urllib.request.urlopen(url, timeout=15) as r:
            data = _json.loads(r.read())
        result = data.get("responseData", {}).get("translatedText", "")
        if result:
            return result.strip()
        raise RuntimeError("لا توجد نتيجة من MyMemory")
    except Exception as e:
        import traceback
        traceback.print_exc()
        raise gr.Error(f"فشل الترجمة: {type(e).__name__}: {e}")


@spaces.GPU(duration=120)
def stt(audio_path: str, language: str = ""):
    """تفريغ صوتي عبر Whisper. يعيد النص."""
    if not audio_path or not os.path.exists(audio_path):
        return ""
    lang = LANGS.get(language, language) if language else None
    try:
        asr = get_asr()
        kwargs = {}
        if lang and lang in LANG_CODES:
            kwargs["language"] = lang
        result = asr.transcribe(audio_path, **kwargs)
        return (result.get("text") or "").strip()
    except Exception as e:
        # أظهر الخطأ الحقيقي بدل رسالة عامة
        raise gr.Error(f"فشل التفريغ: {type(e).__name__}: {e}")

# ============================================================
#  واجهة Gradio + نقاط API
# ============================================================
with gr.Blocks(title="جسر — خادم موحّد") as demo:
    gr.Markdown("# 🌉 جسر — ترجمة + استنساخ صوتي (7 لغات)")

    with gr.Tab("🔊 توليد كلام"):
        with gr.Row():
            with gr.Column():
                text_in = gr.Textbox(label="النص", lines=4,
                                     placeholder="اكتب النص هنا...")
                lang_in = gr.Dropdown(choices=list(LANGS.keys()),
                                      value="العربية (ar)", label="اللغة")
                with gr.Accordion("إعدادات متقدمة", open=False):
                    exag = gr.Slider(0.25, 1.0, value=0.5, step=0.05,
                                     label="قوة التعبير")
                    cfg = gr.Slider(0.2, 0.8, value=0.5, step=0.05,
                                    label="التطابق مع الصوت المرجعي")
                    voice_in = gr.Audio(label="صوت مرجعي مخصص (اختياري)",
                                        type="filepath")
                btn = gr.Button("🔊 توليد", variant="primary")
            with gr.Column():
                audio_out = gr.Audio(label="الناتج", type="filepath",
                                     autoplay=True)
        btn.click(tts, inputs=[text_in, lang_in, exag, cfg, voice_in],
                  outputs=audio_out, api_name="tts")

    with gr.Tab("🌍 ترجمة"):
        with gr.Row():
            with gr.Column():
                mt_text = gr.Textbox(label="النص", lines=3)
                mt_src = gr.Dropdown(choices=list(LANGS.keys()),
                                     value="العربية (ar)", label="من")
                mt_tgt = gr.Dropdown(choices=list(LANGS.keys()),
                                     value="English (en)", label="إلى")
                mt_btn = gr.Button("ترجم", variant="primary")
            mt_out = gr.Textbox(label="الترجمة", lines=3)
        mt_btn.click(translate, inputs=[mt_text, mt_src, mt_tgt],
                     outputs=mt_out, api_name="translate")

    with gr.Tab("🎤 تفريغ صوتي"):
        with gr.Row():
            with gr.Column():
                stt_audio = gr.Audio(label="ملف صوتي", type="filepath")
                stt_lang = gr.Dropdown(choices=[""] + list(LANGS.keys()),
                                       value="", label="اللغة (اختياري)")
                stt_btn = gr.Button("فرّغ", variant="primary")
            stt_out = gr.Textbox(label="النص", lines=3)
        stt_btn.click(stt, inputs=[stt_audio, stt_lang],
                      outputs=stt_out, api_name="stt")

demo.queue(max_size=20).launch(ssr_mode=False)