Update app.py

app.py

@@ -1,17 +1,15 @@
 import gradio as gr
 import whisper
 import os
+import asyncio
+import edge_tts
+from transformers import pipeline
+from deep_translator import GoogleTranslator
+from docx import Document
 import tempfile
 from datetime import datetime
 import logging
 import sys
-from transformers import pipeline, SpeechT5Processor, SpeechT5ForTextToSpeech, SpeechT5HifiGan
-from deep_translator import GoogleTranslator
-from docx import Document
-import time
-import torch
-import soundfile as sf
-import numpy as np
 
 # إعداد التسجيل
 logging.basicConfig(
@@ -24,10 +22,6 @@ logging.basicConfig(
 )
 logger = logging.getLogger(__name__)
 
-# إنشاء مجلد للملفات المؤقتة
-TEMP_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'temp_files')
-os.makedirs(TEMP_DIR, exist_ok=True)
-
 # قائمة اللغات المدعومة
 SUPPORTED_LANGUAGES = {
     'ar': 'العربية',
@@ -37,113 +31,158 @@ SUPPORTED_LANGUAGES = {
     'de': 'Deutsch'
 }
 
-def cleanup_temp_files():
-    """تنظيف الملفات المؤقتة"""
+# تعيين أصوات لكل لغة
+VOICE_MAPPINGS = {
+    'ar': 'ar-EG-ShakirNeural',
+    'en': 'en-US-EricNeural',
+    'fr': 'fr-FR-HenriNeural',
+    'es': 'es-ES-AlvaroNeural',
+    'de': 'de-DE-ConradNeural'
+}
+
+# تحديد اللغات RTL
+RTL_LANGUAGES = ['ar']
+
+async def generate_speech(text, lang):
+    """توليد الصوت باستخدام edge-tts"""
     try:
-        current_time = time.time()
-        for filename in os.listdir(TEMP_DIR):
-            file_path = os.path.join(TEMP_DIR, filename)
-            if os.path.getmtime(file_path) < current_time - 300:  # 5 minutes
-                try:
-                    os.remove(file_path)
-                except Exception as e:
-                    logger.error(f"خطأ في حذف الملف المؤقت {file_path}: {str(e)}")
+        voice = VOICE_MAPPINGS.get(lang, 'en-US-EricNeural')
+        communicate = edge_tts.Communicate(text, voice)
+        
+        audio_path = tempfile.mktemp(suffix='.mp3')
+        await communicate.save(audio_path)
+        
+        # التحقق من وجود الملف وحجمه
+        if os.path.exists(audio_path) and os.path.getsize(audio_path) > 0:
+            logger.info(f"تم إنشاء ملف صوتي: {audio_path}")
+            return audio_path
+        else:
+            logger.error("فشل إنشاء ملف صوتي صالح")
+            return None
+            
     except Exception as e:
-        logger.error(f"خطأ في تنظيف الملفات المؤقتة: {str(e)}")
+        logger.error(f"خطأ في توليد الصوت: {str(e)}")
+        return None
 
 def text_to_speech(text, lang, progress=gr.Progress()):
-    """تحويل النص إلى صوت"""
+    """واجهة لتحويل النص إلى صوت"""
     if not text:
+        logger.warning("لم يتم تقديم نص للتحويل إلى صوت")
         return None
-
-    try:
-        progress(0.2, desc="جاري تحميل النماذج...")
         
-        # تحميل النماذج
-        processor = SpeechT5Processor.from_pretrained("microsoft/speecht5_tts")
-        model = SpeechT5ForTextToSpeech.from_pretrained("microsoft/speecht5_tts")
-        vocoder = SpeechT5HifiGan.from_pretrained("microsoft/speecht5_hifigan")
+    try:
+        progress(0.2, desc="جاري تجهيز الصوت...")
+        logger.info(f"بدء تحويل النص إلى صوت باللغة: {lang}")
         
         # تقسيم النص إلى أجزاء إذا كان طويلاً
-        max_length = 500
-        text_chunks = [text[i:i+max_length] for i in range(0, len(text), max_length)]
+        max_length = 1000
+        text_parts = [text[i:i+max_length] for i in range(0, len(text), max_length)]
+        
+        # إنشاء ملف صوتي لكل جزء
+        audio_files = []
+        for i, part in enumerate(text_parts):
+            progress((i + 1) / len(text_parts), desc=f"معالجة الجزء {i+1} من {len(text_parts)}...")
+            audio_path = asyncio.run(generate_speech(part, lang))
+            if audio_path:
+                audio_files.append(audio_path)
         
-        all_speech = []
-        for i, chunk in enumerate(text_chunks):
-            progress((i + 1) / len(text_chunks) * 0.8, desc=f"معالجة الجزء {i+1} من {len(text_chunks)}...")
+        if not audio_files:
+            logger.error("لم يتم إنشاء أي ملفات صوتية")
+            return None
             
-            # تجهيز النص
-            inputs = processor(text=chunk, return_tensors="pt")
+        # إذا كان هناك جزء واحد فقط
+        if len(audio_files) == 1:
+            return audio_files[0]
             
-            # توليد الصوت
-            speech = model.generate_speech(inputs["input_ids"], vocoder)
-            all_speech.append(speech.numpy())
-
-        # دمج كل الأجزاء
-        final_speech = np.concatenate(all_speech)
+        # دمج الملفات الصوتية إذا كان هناك أكثر من جزء
+        from pydub import AudioSegment
         
-        # حفظ الملف
-        output_file = os.path.join(TEMP_DIR, f'speech_{time.time()}.wav')
-        sf.write(output_file, final_speech, 16000)
+        final_audio = AudioSegment.from_mp3(audio_files[0])
+        for audio_file in audio_files[1:]:
+            final_audio += AudioSegment.from_mp3(audio_file)
+        
+        final_path = tempfile.mktemp(suffix='.mp3')
+        final_audio.export(final_path, format="mp3")
+        
+        # تنظيف الملفات المؤقتة
+        for file in audio_files:
+            try:
+                os.remove(file)
+            except:
+                pass
         
         progress(1.0, desc="تم إنشاء الصوت بنجاح!")
-        return output_file
-
+        return final_path
+            
     except Exception as e:
         logger.error(f"خطأ في تحويل النص إلى صوت: {str(e)}")
         return None
 
-def translate_text(text, source_lang, target_lang, progress=gr.Progress()):
-    """ترجمة النص"""
-    if source_lang == target_lang:
-        return text
-    
-    try:
-        progress(0.3, desc="جاري الترجمة...")
-        translator = GoogleTranslator(source=source_lang, target=target_lang)
-        
-        # تقسيم النص إلى أجزاء صغيرة
-        max_chunk_size = 1000
-        chunks = [text[i:i+max_chunk_size] for i in range(0, len(text), max_chunk_size)]
-        translated_chunks = []
-        
-        for i, chunk in enumerate(chunks):
-            progress((i + 1) / len(chunks), desc=f"ترجمة الجزء {i+1} من {len(chunks)}...")
-            translated_chunk = translator.translate(chunk)
-            translated_chunks.append(translated_chunk)
-            time.sleep(0.5)  # تأخير صغير لتجنب التحميل الزائد
-        
-        return ' '.join(translated_chunks)
-    except Exception as e:
-        logger.error(f"خطأ في الترجمة: {str(e)}")
-        return f"خطأ في الترجمة: {str(e)}"
-
 def create_document(original_text, translated_text, source_lang, target_lang, progress=gr.Progress()):
-    """إنشاء ملف Word"""
+    """إنشاء ملف Word يحتوي على النص الأصلي والترجمة"""
     try:
-        progress(0.3, desc="جاري إنشاء المستند...")
+        progress(0, desc="جاري إنشاء المستند...")
         doc = Document()
         doc.add_heading('النص الأصلي والترجمة', 0)
         
+        progress(0.3, desc="جاري إضافة المحتوى...")
+        # إضافة التاريخ والوقت
         doc.add_paragraph(f'تم الإنشاء في: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}')
         
+        # إضافة النص الأصلي
         doc.add_heading(f'النص الأصلي ({SUPPORTED_LANGUAGES[source_lang]})', level=1)
         doc.add_paragraph(original_text)
         
+        progress(0.6, desc="جاري إضافة الترجمة...")
+        # إضافة الترجمة
         doc.add_heading(f'الترجمة ({SUPPORTED_LANGUAGES[target_lang]})', level=1)
         doc.add_paragraph(translated_text)
         
-        doc_path = os.path.join(TEMP_DIR, f'document_{time.time()}.docx')
-        doc.save(doc_path)
+        # حفظ الملف
+        progress(0.9, desc="جاري حفظ المستند...")
+        temp_path = tempfile.mktemp(suffix='.docx')
+        doc.save(temp_path)
         
         progress(1.0, desc="تم إنشاء المستند بنجاح!")
-        return doc_path
+        logger.info(f"تم إنشاء مستند Word: {temp_path}")
+        return temp_path
+        
     except Exception as e:
         logger.error(f"خطأ في إنشاء المستند: {str(e)}")
         return None
 
+def translate_text(text, source_lang, target_lang, progress=gr.Progress()):
+    """ترجمة النص باستخدام deep-translator"""
+    if source_lang == target_lang:
+        return text
+    
+    try:
+        progress(0.3, desc="جاري الترجمة...")
+        logger.info(f"بدء الترجمة من {source_lang} إلى {target_lang}")
+        
+        translator = GoogleTranslator(source=source_lang, target=target_lang)
+        
+        # تقسيم النص إلى أجزاء إذا كان طويلاً
+        max_length = 5000
+        text_parts = [text[i:i+max_length] for i in range(0, len(text), max_length)]
+        
+        translated_parts = []
+        for i, part in enumerate(text_parts):
+            progress((i + 1) / len(text_parts), desc=f"ترجمة الجزء {i+1} من {len(text_parts)}...")
+            translated_part = translator.translate(part)
+            translated_parts.append(translated_part)
+        
+        translated_text = ' '.join(translated_parts)
+        progress(1.0, desc="تمت الترجمة بنجاح!")
+        
+        return translated_text
+        
+    except Exception as e:
+        logger.error(f"خطأ في الترجمة: {str(e)}")
+        return f"خطأ في الترجمة: {str(e)}"
+
 def process_video(video, source_lang="en", target_lang="ar", progress=gr.Progress()):
-    """معالجة الفيديو"""
+    """معالجة الفيديو واستخراج النص وترجمته"""
     if video is None:
         return {
             "error": "الرجاء رفع ملف فيديو",
@@ -153,8 +192,10 @@ def process_video(video, source_lang="en", target_lang="ar", progress=gr.Progres
         }
     
     try:
-        cleanup_temp_files()
+        # حفظ الفيديو مؤقتاً
         progress(0.1, desc="جاري تحميل الفيديو...")
+        temp_path = video.name
+        logger.info(f"تم استلام ملف فيديو: {temp_path}")
         
         # تحميل نموذج Whisper
         progress(0.3, desc="جاري تحميل نموذج التعرف على الكلام...")
@@ -162,8 +203,9 @@ def process_video(video, source_lang="en", target_lang="ar", progress=gr.Progres
         
         # استخراج النص
         progress(0.5, desc="جاري استخراج النص من الفيديو...")
-        result = model.transcribe(video.name, language=source_lang)
+        result = model.transcribe(temp_path, language=source_lang)
         transcribed_text = result["text"]
+        logger.info("تم استخراج النص بنجاح")
         
         # ترجمة النص
         progress(0.7, desc="جاري ترجمة النص...")
@@ -173,12 +215,14 @@ def process_video(video, source_lang="en", target_lang="ar", progress=gr.Progres
         progress(0.9, desc="جاري إنشاء المستند...")
         doc_path = create_document(transcribed_text, translated_text, source_lang, target_lang)
         
+        progress(1.0, desc="تمت المعالجة بنجاح!")
         return {
             "error": None,
             "original": transcribed_text,
             "translated": translated_text,
             "document": doc_path
         }
+        
     except Exception as e:
         logger.error(f"خطأ في معالجة الفيديو: {str(e)}")
         return {
@@ -190,7 +234,10 @@ def process_video(video, source_lang="en", target_lang="ar", progress=gr.Progres
 
 def create_ui():
     """إنشاء واجهة المستخدم"""
-    with gr.Blocks(theme=gr.themes.Soft(primary_hue="blue")) as demo:
+    with gr.Blocks(theme=gr.themes.Soft(
+        primary_hue="blue",
+        secondary_hue="indigo",
+    )) as demo:
         gr.Markdown(
             """
             # 🎥 منصة تحويل الفيديو إلى نص مع الترجمة
@@ -202,7 +249,8 @@ def create_ui():
             with gr.Column(scale=2):
                 video_input = gr.File(
                     label="📁 رفع فيديو",
-                    file_types=["video"]
+                    file_types=["video"],
+                    elem_id="video_input"
                 )
             with gr.Column(scale=1):
                 source_lang = gr.Dropdown(
@@ -217,13 +265,15 @@ def create_ui():
                 )
                 process_btn = gr.Button("🎯 معالجة الفيديو", variant="primary")
         
-        error_output = gr.Textbox(label="⚠️ الأخطاء", visible=False)
+        with gr.Row():
+            error_output = gr.Textbox(label="⚠️ الأخطاء", visible=False)
         
         with gr.Tabs():
             with gr.TabItem("📝 النص الأصلي"):
                 original_text = gr.Textbox(
                     label="النص المستخرج من الفيديو",
-                    lines=10
+                    lines=10,
+                    elem_classes=["ltr"]
                 )
                 with gr.Row():
                     generate_original_audio = gr.Button("🔊 توليد الصوت", variant="secondary")
@@ -232,20 +282,30 @@ def create_ui():
             with gr.TabItem("🔄 النص المترجم"):
                 translated_text = gr.Textbox(
                     label="النص المترجم",
-                    lines=10
+                    lines=10,
+                    elem_classes=["rtl"]
                 )
                 with gr.Row():
                     generate_translated_audio = gr.Button("🔊 توليد الصوت", variant="secondary")
                     translated_audio = gr.Audio(label="الصوت", visible=True)
         
-        download_btn = gr.File(label="📥 تحميل المستند (Word)", visible=True)
+        with gr.Row():
+            download_btn = gr.File(
+                label="📥 تحميل المستند (Word)",
+                interactive=False
+            )
         
         def update_ui(video, src_lang, tgt_lang):
             result = process_video(video, src_lang, tgt_lang)
+            
+            # تحديث اتجاه النص
+            original_classes = "rtl" if src_lang in RTL_LANGUAGES else "ltr"
+            translated_classes = "rtl" if tgt_lang in RTL_LANGUAGES else "ltr"
+            
             return {
                 error_output: gr.update(value=result["error"], visible=bool(result["error"])),
-                original_text: result["original"],
-                translated_text: result["translated"],
+                original_text: gr.update(value=result["original"], elem_classes=[original_classes]),
+                translated_text: gr.update(value=result["translated"], elem_classes=[translated_classes]),
                 download_btn: result["document"]
             }