Update app.py

app.py

@@ -1,19 +1,20 @@
 """
 MediaTranscriberPro - Hugging Face Space
-Standalone audio/video transcription with Gradio UI + API
+Final Fix for DNS/IPv6 Issues
 """
-# ============================================================================
-# 🔧 MAGIC PATCH: FORCE IPv4
-# هذا الكود يجبر التطبيق على استخدام IPv4 فقط لحل مشكلة DNS مع يوتيوب
+# ---------------------------------------------------------
+# LAYER 1: SYSTEM SOCKET PATCH (Must be at the very top)
+# ---------------------------------------------------------
 import socket
-def force_ipv4_getaddrinfo(host, port, family=0, type=0, proto=0, flags=0):
-    # Force usage of AF_INET (IPv4)
-    return socket.getaddrinfo_original(host, port, socket.AF_INET, type, proto, flags)
+import os
 
-if not hasattr(socket, 'getaddrinfo_original'):
-    socket.getaddrinfo_original = socket.getaddrinfo
-    socket.getaddrinfo = force_ipv4_getaddrinfo
-# ============================================================================
+# Force IPv4 for all socket connections
+old_getaddrinfo = socket.getaddrinfo
+def new_getaddrinfo(*args, **kwargs):
+    responses = old_getaddrinfo(*args, **kwargs)
+    return [response for response in responses if response[0] == socket.AF_INET]
+socket.getaddrinfo = new_getaddrinfo
+# ---------------------------------------------------------
 
 import gradio as gr
 import logging
@@ -21,246 +22,143 @@ import tempfile
 import shutil
 import subprocess
 import re
-import yt_dlp  # سنستخدم المكتبة مباشرة الآن
+import yt_dlp
 from pathlib import Path
 from dataclasses import dataclass
 from typing import Optional, Callable
-from enum import Enum
-
-# ======================== Configuration ========================
-SUPPORTED_AUDIO = {".mp3", ".wav", ".m4a", ".aac", ".ogg", ".opus", ".flac"}
-SUPPORTED_VIDEO = {".mp4", ".mkv", ".avi", ".mov", ".webm"}
-SUPPORTED_MEDIA = SUPPORTED_AUDIO | SUPPORTED_VIDEO
-
-URL_PATTERNS = {
-    "youtube": r"(?:https?://)?(?:www\.|m\.)?(?:youtube\.com/(?:watch\?v=|shorts/)|youtu\.be/)[\w-]+",
-    "facebook": r"(?:https?://)?(?:www\.|m\.|web\.)?facebook\.com/(?:watch/?\?v=|videos/|posts/|reel/)",
-    "googledrive": r"(?:https?://)?drive\.google\.com/(?:file/d/|open\?id=|uc\?id=)[\w-]+",
-}
 
+# Logging Setup
 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
 logger = logging.getLogger(__name__)
 
-# ======================== Data Models ========================
-@dataclass
-class DownloadResult:
-    success: bool
-    filepath: Optional[Path] = None
-    error: Optional[str] = None
+# Constants
+SUPPORTED_MEDIA = {".mp3", ".wav", ".m4a", ".aac", ".ogg", ".opus", ".flac", ".mp4", ".mkv", ".avi", ".mov", ".webm"}
 
 @dataclass
-class ConversionResult:
+class Result:
     success: bool
-    wav_path: Optional[Path] = None
+    data: Optional[str] = None
+    file_path: Optional[str] = None
     error: Optional[str] = None
 
-@dataclass
-class TranscriptionResult:
-    success: bool
-    text: str = ""
-    segments: list = None
-    error: Optional[str] = None
-
-# ======================== Core: MediaDownloader ========================
 class MediaDownloader:
-    """Download using yt_dlp Python Library (More Robust)"""
-    
-    def __init__(self, output_dir: Optional[Path] = None):
-        self.output_dir = Path(output_dir) if output_dir else Path(tempfile.mkdtemp())
+    def __init__(self, output_dir):
+        self.output_dir = output_dir
         self.output_dir.mkdir(parents=True, exist_ok=True)
-    
-    def detect_url_type(self, url: str) -> Optional[str]:
-        for platform, pattern in URL_PATTERNS.items():
-            if re.search(pattern, url, re.IGNORECASE):
-                return platform
-        return None
-    
-    def download_url(self, url: str, progress_callback: Optional[Callable] = None) -> DownloadResult:
+
+    def download(self, url, progress=None):
         try:
-            platform = self.detect_url_type(url)
-            if not platform:
-                return DownloadResult(success=False, error="Unsupported URL")
+            if progress: progress(0.1, "Initializing download...")
             
-            if progress_callback:
-                progress_callback(0.1, "Initializing download...")
-            
-            # إعدادات التحميل المباشرة
+            # LAYER 2: YT-DLP SPECIFIC OPTIONS
             ydl_opts = {
                 'format': 'bestaudio/best',
                 'outtmpl': str(self.output_dir / '%(title)s.%(ext)s'),
                 'noplaylist': True,
-                'source_address': '0.0.0.0', # زيادة في التأكيد لإجبار IPv4
+                'force_ipv4': True,      # <--- يجبر المكتبة على استخدام IPv4
+                'nocheckcertificate': True, # <--- يتجاوز أخطاء SSL
+                'socket_timeout': 30,    # <--- يزيد وقت الانتظار
                 'quiet': True,
                 'no_warnings': True,
+                # LAYER 3: USER AGENT SPOOFING
+                'user_agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.212 Safari/537.36',
             }
 
-            logger.info(f"Downloading from {platform}: {url}")
-            
             with yt_dlp.YoutubeDL(ydl_opts) as ydl:
                 info = ydl.extract_info(url, download=True)
                 filename = ydl.prepare_filename(info)
-                downloaded_file = Path(filename)
-            
-            if not downloaded_file.exists():
-                # محاولة البحث اليدوي في حالة اختلاف الاسم
-                files = list(self.output_dir.glob("*"))
-                media_files = [f for f in files if f.suffix.lower() in SUPPORTED_MEDIA]
-                if not media_files:
-                     return DownloadResult(success=False, error="File downloaded but not found!")
-                downloaded_file = max(media_files, key=lambda x: x.stat().st_mtime)
+                file_path = Path(filename)
+                
+            # Fallback check if filename differs
+            if not file_path.exists():
+                potential_files = list(self.output_dir.glob("*"))
+                if not potential_files:
+                    return Result(False, error="Download finished but file not found.")
+                file_path = max(potential_files, key=lambda x: x.stat().st_mtime)
 
-            if progress_callback:
-                progress_callback(1.0, "Download complete")
-            
-            return DownloadResult(success=True, filepath=downloaded_file)
-            
-        except Exception as e:
-            logger.error(f"Download failed: {e}")
-            return DownloadResult(success=False, error=str(e))
+            return Result(True, file_path=str(file_path))
 
-# ======================== Core: AudioConverter ========================
-class AudioConverter:
-    """Convert media to WAV using FFmpeg"""
-    
-    def convert(self, input_path: Path, progress_callback: Optional[Callable] = None) -> ConversionResult:
-        try:
-            if progress_callback:
-                progress_callback(0.1, "Converting to WAV...")
-            
-            output_path = input_path.with_suffix('.wav')
-            
-            cmd = [
-                "ffmpeg", "-i", str(input_path),
-                "-ar", "16000",
-                "-ac", "1",
-                "-y",
-                str(output_path)
-            ]
-            
-            logger.info(f"Converting {input_path.name} to WAV")
-            subprocess.run(cmd, capture_output=True, check=True)
-            
-            return ConversionResult(success=True, wav_path=output_path)
-            
         except Exception as e:
-            logger.error(f"Conversion failed: {e}")
-            return ConversionResult(success=False, error=str(e))
+            logger.error(f"Download Error: {e}")
+            return Result(False, error=str(e))
 
-# ======================== Core: SpeechTranscriber ========================
-class SpeechTranscriber:
-    """Transcribe audio using faster-whisper"""
-    
-    def __init__(self, model_size: str = "medium"):
-        try:
+class Processor:
+    def __init__(self):
+        self.tmp = Path(tempfile.mkdtemp())
+        self.downloader = MediaDownloader(self.tmp / "download")
+        
+        # Lazy load whisper to save startup time
+        self.model = None
+
+    def load_model(self):
+        if not self.model:
             from faster_whisper import WhisperModel
-            logger.info(f"Loading Whisper model: {model_size}")
-            self.model = WhisperModel(model_size, device="cpu", compute_type="int8")
-        except Exception as e:
-            logger.error(f"Error loading model: {e}")
-            self.model = None
-    
-    def transcribe(self, wav_path: Path, language: str = "ar", progress_callback: Optional[Callable] = None) -> TranscriptionResult:
-        try:
-            if not self.model:
-                return TranscriptionResult(success=False, error="Model not loaded")
-            
-            if progress_callback:
-                progress_callback(0.1, "Transcribing...")
-            
-            lang_code = language.split('-')[0] if '-' in language else language
-            
-            segments_iter, info = self.model.transcribe(
-                str(wav_path),
-                language=lang_code,
-                beam_size=5,
-                vad_filter=True
-            )
-            segments = list(segments_iter)
-            text = " ".join([s.text for s in segments])
-            
-            return TranscriptionResult(success=True, text=text.strip(), segments=segments)
-            
-        except Exception as e:
-            logger.error(f"Transcription failed: {e}")
-            return TranscriptionResult(success=False, error=str(e))
+            self.model = WhisperModel("medium", device="cpu", compute_type="int8")
 
-# ======================== Utilities ========================
-def generate_srt(segments, output_path: Path):
-    with open(output_path, 'w', encoding='utf-8') as f:
-        for i, seg in enumerate(segments, 1):
-            start = format_timestamp(seg.start)
-            end = format_timestamp(seg.end)
-            f.write(f"{i}\n{start} --> {end}\n{seg.text.strip()}\n\n")
-
-def format_timestamp(seconds: float) -> str:
-    hours = int(seconds // 3600)
-    minutes = int((seconds % 3600) // 60)
-    secs = int(seconds % 60)
-    millis = int((seconds % 1) * 1000)
-    return f"{hours:02d}:{minutes:02d}:{secs:02d},{millis:03d}"
-
-# ======================== Main Pipeline ========================
-class TranscriptionPipeline:
-    def __init__(self):
-        self.temp_dir = Path(tempfile.mkdtemp())
-        self.downloader = MediaDownloader(output_dir=self.temp_dir / "downloads")
-        self.converter = AudioConverter()
-        self.transcriber = SpeechTranscriber(model_size="medium")
-    
-    def process(self, source_url, upload_file, language, gen_txt, gen_srt, progress=gr.Progress()):
+    def run(self, url, upload, lang, progress=gr.Progress()):
         try:
-            # 1. Source
-            if upload_file:
-                source_path = Path(upload_file)
-            elif source_url:
-                res = self.downloader.download_url(source_url, lambda p, m: progress(p*0.3, desc=m))
-                if not res.success: return f"Download Error: {res.error}", None, None, None
-                source_path = res.filepath
+            # 1. Acquire Media
+            target_file = None
+            if upload:
+                target_file = Path(upload)
+            elif url:
+                res = self.downloader.download(url, progress)
+                if not res.success: return f"❌ Error: {res.error}", None, None
+                target_file = Path(res.file_path)
             else:
-                return "No input provided", None, None, None
+                return "Please provide URL or File", None, None
+
+            # 2. Transcribe
+            progress(0.3, "Loading Model...")
+            self.load_model()
             
-            # 2. Convert
-            res_conv = self.converter.convert(source_path, lambda p, m: progress(0.4, desc=m))
-            if not res_conv.success: return f"Convert Error: {res_conv.error}", None, None, None
+            progress(0.5, "Transcribing...")
+            lang_code = lang.split("-")[0]
+            segments, _ = self.model.transcribe(str(target_file), language=lang_code, beam_size=5)
             
-            # 3. Transcribe
-            res_trans = self.transcriber.transcribe(res_conv.wav_path, language, lambda p, m: progress(0.7, desc=m))
-            if not res_trans.success: return f"Transcribe Error: {res_trans.error}", None, None, None
+            # Collect result
+            full_text = []
+            srt_content = []
+            for i, seg in enumerate(segments, 1):
+                full_text.append(seg.text)
+                # Simple SRT formatting
+                start = f"{int(seg.start//3600):02}:{int((seg.start%3600)//60):02}:{int(seg.start%60):02},000"
+                end = f"{int(seg.end//3600):02}:{int((seg.end%3600)//60):02}:{int(seg.end%60):02},000"
+                srt_content.append(f"{i}\n{start} --> {end}\n{seg.text.strip()}\n")
             
-            # 4. Output
-            out_dir = self.temp_dir / "outputs"
-            out_dir.mkdir(exist_ok=True)
+            text_str = " ".join(full_text)
+            srt_str = "\n".join(srt_content)
             
-            txt_file = str(out_dir / "transcription.txt")
-            with open(txt_file, 'w', encoding='utf-8') as f: f.write(res_trans.text)
+            # Save files
+            out_txt = self.tmp / "transcript.txt"
+            out_srt = self.tmp / "subs.srt"
+            out_txt.write_text(text_str, encoding="utf-8")
+            out_srt.write_text(srt_str, encoding="utf-8")
             
-            srt_file = None
-            if gen_srt:
-                srt_file = str(out_dir / "subtitles.srt")
-                generate_srt(res_trans.segments, Path(srt_file))
-                
-            return f"Done! Length: {len(res_trans.text)} chars", res_trans.text, txt_file, srt_file
+            return f"✅ Done! ({len(text_str)} chars)", str(out_txt), str(out_srt)
 
         except Exception as e:
-            return f"Error: {e}", None, None, None
+            return f"❌ Critical Error: {str(e)}", None, None
 
-# ======================== Interface ========================
-def create_interface():
-    pipeline = TranscriptionPipeline()
-    with gr.Blocks() as demo:
-        gr.Markdown("# MediaTranscriber Pro")
-        with gr.Row():
-            url = gr.Textbox(label="URL")
-            upl = gr.File(label="Upload File")
-        lang = gr.Dropdown(["ar", "en"], value="ar", label="Language")
-        btn = gr.Button("Start")
-        out_txt = gr.Textbox(label="Transcript", interactive=False) # Fixed interactive
-        with gr.Row():
-            dl_txt = gr.File(label="TXT")
-            dl_srt = gr.File(label="SRT")
-            
-        btn.click(pipeline.process, [url, upl, lang, gr.Checkbox(value=True), gr.Checkbox(value=True)], [gr.Markdown(), out_txt, dl_txt, dl_srt])
-    return demo
+# UI Setup
+proc = Processor()
+
+with gr.Blocks(title="Transcriber Pro") as demo:
+    gr.Markdown("## 🎙️ Media Transcriber Pro (IPv4 Fix)")
+    
+    with gr.Row():
+        url_in = gr.Textbox(label="YouTube URL")
+        file_in = gr.File(label="Upload File")
+    
+    lang_in = gr.Dropdown(["ar", "en"], value="ar", label="Language")
+    btn = gr.Button("Transcribe", variant="primary")
+    
+    status = gr.Textbox(label="Status")
+    with gr.Row():
+        f1 = gr.File(label="TXT")
+        f2 = gr.File(label="SRT")
+    
+    btn.click(proc.run, [url_in, file_in, lang_in], [status, f1, f2])
 
 if __name__ == "__main__":
-    create_interface().launch(server_name="0.0.0.0", server_port=7860)
+    demo.launch(server_name="0.0.0.0", server_port=7860)