app.py

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

# إعداد التسجيل
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('app.log'),
        logging.StreamHandler(sys.stdout)
    ]
)
logger = logging.getLogger(__name__)

# قائمة اللغات المدعومة
SUPPORTED_LANGUAGES = {
    'ar': 'العربية',
    'en': 'English',
    'fr': 'Français',
    'es': 'Español',
    'de': 'Deutsch'
}

# تعيين أصوات لكل لغة
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:
        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)}")
        return None

def text_to_speech(text, lang, progress=gr.Progress()):
    """واجهة لتحويل النص إلى صوت"""
    if not text:
        logger.warning("لم يتم تقديم نص للتحويل إلى صوت")
        return None
        
    try:
        progress(0.2, desc="جاري تجهيز الصوت...")
        logger.info(f"بدء تحويل النص إلى صوت باللغة: {lang}")
        
        # تقسيم النص إلى أجزاء إذا كان طويلاً
        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)
        
        if not audio_files:
            logger.error("لم يتم إنشاء أي ملفات صوتية")
            return None
            
        # إذا كان هناك جزء واحد فقط
        if len(audio_files) == 1:
            return audio_files[0]
            
        # دمج الملفات الصوتية إذا كان هناك أكثر من جزء
        from pydub import AudioSegment
        
        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 final_path
            
    except Exception as e:
        logger.error(f"خطأ في تحويل النص إلى صوت: {str(e)}")
        return None

def create_document(original_text, translated_text, source_lang, target_lang, progress=gr.Progress()):
    """إنشاء ملف Word يحتوي على النص الأصلي والترجمة"""
    try:
        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)
        
        # حفظ الملف
        progress(0.9, desc="جاري حفظ المستند...")
        temp_path = tempfile.mktemp(suffix='.docx')
        doc.save(temp_path)
        
        progress(1.0, desc="تم إنشاء المستند بنجاح!")
        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 format_timestamp(seconds):
    """تحويل الثواني إلى تنسيق MM:SS"""
    minutes = int(seconds // 60)
    seconds = int(seconds % 60)
    return f"{minutes:02d}:{seconds:02d}"        

def process_video(video, source_lang="en", target_lang="ar", progress=gr.Progress()):
    """معالجة الفيديو واستخراج النص وترجمته"""
    if video is None:
        return {
            "error": "الرجاء رفع ملف فيديو",
            "original": "",
            "translated": "",
            "document": None
        }
    
    try:
        # حفظ الفيديو مؤقتاً
        progress(0.1, desc="جاري تحميل الفيديو...")
        temp_path = video.name
        logger.info(f"تم استلام ملف فيديو: {temp_path}")
        
        # تحميل نموذج Whisper
        progress(0.3, desc="جاري تحميل نموذج التعرف على الكلام...")
        model = whisper.load_model("base")
        
        # استخراج النص مع التوقيت
        progress(0.5, desc="جاري استخراج النص من الفيديو...")
        result = model.transcribe(temp_path, language=source_lang)
        
        # تنسيق النص مع التوقيت
        transcribed_text = ""
        segments_for_translation = []
        
        # تجميع النص الأصلي مع التوقيت
        for segment in result["segments"]:
            start_time = format_timestamp(segment["start"])
            text = segment["text"].strip()
            transcribed_text += f"[{start_time}] {text}\n"
            segments_for_translation.append((start_time, text))
        
        logger.info("تم استخراج النص بنجاح")
        
        # ترجمة النص مع الحفاظ على التوقيت
        progress(0.7, desc="جاري ترجمة النص...")
        translated_segments = []
        translator = GoogleTranslator(source=source_lang, target=target_lang)
        
        for start_time, text in segments_for_translation:
            try:
                translated_text = translator.translate(text)
                translated_segments.append(f"[{start_time}] {translated_text}")
                logger.info(f"تمت ترجمة المقطع: {text} -> {translated_text}")
            except Exception as e:
                logger.error(f"خطأ في ترجمة المقطع: {str(e)}")
                translated_segments.append(f"[{start_time}] [خطأ في الترجمة: {text}]")
        
        translated_text = "\n".join(translated_segments)
        
        # إنشاء ملف Word
        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 {
            "error": f"حدث خطأ: {str(e)}",
            "original": "",
            "translated": "",
            "document": None
        }

def create_ui():
    """إنشاء واجهة المستخدم"""
    with gr.Blocks(theme=gr.themes.Soft(
        primary_hue="blue",
        secondary_hue="indigo",
    )) as demo:
        gr.Markdown(
            """
            # 🎥 منصة تحويل الفيديو إلى نص مع الترجمة
            ### قم برفع فيديو للحصول على النص والترجمة مع إمكانية تحويل النص إلى صوت
            """
        )
        
        with gr.Row():
            with gr.Column(scale=2):
                video_input = gr.File(
                    label="📁 رفع فيديو",
                    file_types=["video"],
                    elem_id="video_input"
                )
            with gr.Column(scale=1):
                source_lang = gr.Dropdown(
                    choices=list(SUPPORTED_LANGUAGES.keys()),
                    value="en",
                    label="🗣️ لغة الفيديو الأصلية"
                )
                target_lang = gr.Dropdown(
                    choices=list(SUPPORTED_LANGUAGES.keys()),
                    value="ar",
                    label="🌐 لغة الترجمة"
                )
                process_btn = gr.Button("🎯 معالجة الفيديو", variant="primary")
        
        with gr.Row():
            error_output = gr.Textbox(label="⚠️ الأخطاء", visible=False)
        
        with gr.Tabs():
            with gr.TabItem("📝 النص الأصلي"):
                original_text = gr.Textbox(
                    label="النص المستخرج من الفيديو",
                    lines=10,
                    elem_classes=["ltr"]
                )
                with gr.Row():
                    generate_original_audio = gr.Button("🔊 توليد الصوت", variant="secondary")
                    original_audio = gr.Audio(label="الصوت", visible=True)
            
            with gr.TabItem("🔄 النص المترجم"):
                translated_text = gr.Textbox(
                    label="النص المترجم",
                    lines=10,
                    elem_classes=["rtl"]
                )
                with gr.Row():
                    generate_translated_audio = gr.Button("🔊 توليد الصوت", variant="secondary")
                    translated_audio = gr.Audio(label="الصوت", 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: gr.update(value=result["original"], elem_classes=[original_classes]),
                translated_text: gr.update(value=result["translated"], elem_classes=[translated_classes]),
                download_btn: result["document"]
            }
        
        # ربط الأحداث
        process_btn.click(
            fn=update_ui,
            inputs=[video_input, source_lang, target_lang],
            outputs=[error_output, original_text, translated_text, download_btn]
        )
        
        generate_original_audio.click(
            fn=text_to_speech,
            inputs=[original_text, source_lang],
            outputs=[original_audio]
        )
        
        generate_translated_audio.click(
            fn=text_to_speech,
            inputs=[translated_text, target_lang],
            outputs=[translated_audio]
        )
    
    return demo

if __name__ == "__main__":
    try:
        logger.info("بدء تشغيل التطبيق")
        demo = create_ui()
        demo.launch()
    except Exception as e:
        logger.error(f"خطأ في تشغيل التطبيق: {str(e)}")