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
from pydub import AudioSegment

# إعداد التسجيل
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    handlers=[
        logging.FileHandler("app.log"),
        logging.StreamHandler(),
    ],
)
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"]

# وظيفة تنظيف النص
def clean_text(text):
    return (
        text.replace("’", "'")
        .replace("«", "")
        .replace("»", "")
        .replace("\n", " ")
        .strip()
    )

# وظيفة توليد الصوت
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):
    if not text:
        logger.warning("لم يتم تقديم نص للتحويل إلى صوت")
        return None

    try:
        text = clean_text(text)
        max_length = 1000
        text_parts = [text[i : i + max_length] for i in range(0, len(text), max_length)]

        audio_files = []
        for part in text_parts:
            audio_path = asyncio.run(generate_speech(part, lang))
            if audio_path:
                audio_files.append(audio_path)

        if len(audio_files) == 1:
            return audio_files[0]

        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:
            os.remove(file)

        return final_path

    except Exception as e:
        logger.error(f"خطأ في تحويل النص إلى صوت: {str(e)}")
        return None

# الترجمة
def translate_text(text, source_lang, target_lang):
    if source_lang == target_lang:
        return text

    try:
        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 = [translator.translate(part) for part in text_parts]
        return " ".join(translated_parts)

    except Exception as e:
        logger.error(f"خطأ في الترجمة: {str(e)}")
        return f"خطأ في الترجمة: {str(e)}"

# معالجة الفيديو
def process_video(video, source_lang="en", target_lang="ar"):
    if video is None:
        return {
            "error": "الرجاء رفع ملف فيديو",
            "original": "",
            "translated": "",
        }

    try:
        temp_path = video.name
        model = whisper.load_model("base")
        result = model.transcribe(temp_path, language=source_lang)
        transcribed_text = result["text"]
        translated_text = translate_text(transcribed_text, source_lang, target_lang)
        return {
            "error": None,
            "original": transcribed_text,
            "translated": translated_text,
        }

    except Exception as e:
        logger.error(f"خطأ في معالجة الفيديو: {str(e)}")
        return {
            "error": f"حدث خطأ: {str(e)}",
            "original": "",
            "translated": "",
        }

# إنشاء الواجهة
def create_ui():
    with gr.Blocks() as demo:
        gr.Markdown("# 🎥 تحويل الفيديو إلى نصوص وصوت")
        with gr.Row():
            video_input = gr.File(label="📁 رفع فيديو", file_types=["video"])
            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("🎯 معالجة الفيديو")

        with gr.Tabs():
            with gr.TabItem("النص الأصلي"):
                original_text = gr.Textbox(label="النص المستخرج", lines=10)
                original_audio = gr.Audio(label="الصوت")
            with gr.TabItem("النص المترجم"):
                translated_text = gr.Textbox(label="النص المترجم", lines=10)
                translated_audio = gr.Audio(label="الصوت")

        def update_ui(video, src_lang, tgt_lang):
            result = process_video(video, src_lang, tgt_lang)
            return {
                original_text: result["original"],
                translated_text: result["translated"],
            }

        process_btn.click(
            fn=update_ui,
            inputs=[video_input, source_lang, target_lang],
            outputs=[original_text, translated_text],
        )

    return demo


if __name__ == "__main__":
    demo = create_ui()
    demo.launch()