Update app.py

app.py

@@ -1,290 +1,150 @@
-# app.py
 import os
 import gradio as gr
 import tempfile
-import shutil
-import glob
-import subprocess
-import requests
 import yt_dlp
-from pathlib import Path
+import subprocess
 from pydub import AudioSegment
 from groq import Groq
 
-# -----------------------
-# Environment variables
-# -----------------------
+# ✅ Environment Variables
 GROQ_API_KEY = os.getenv("GROQ_API_KEY")
-HUGGINGFACE_API_TOKEN = os.getenv("HUGGINGFACE_API_TOKEN")
+if not GROQ_API_KEY:
+    raise EnvironmentError("Please set GROQ_API_KEY.")
 
-if not GROQ_API_KEY or not HUGGINGFACE_API_TOKEN:
-    raise EnvironmentError("Please set GROQ_API_KEY and HUGGINGFACE_API_TOKEN in Space settings.")
-
-# Initialize Groq client
 groq_client = Groq(api_key=GROQ_API_KEY)
 
-# -----------------------
-# Utilities
-# -----------------------
-
+# ✅ Download YouTube Audio
 def download_youtube_audio(youtube_url):
-    """
-    Download audio into a temporary directory and return the downloaded filepath.
-    """
-    tmpdir = tempfile.mkdtemp(prefix="yt_")
-    outtmpl = os.path.join(tmpdir, "%(id)s.%(ext)s")
-    ydl_opts = {
-        "format": "bestaudio/best",
-        "outtmpl": outtmpl,
-        "quiet": True,
-        "no_warnings": True,
-        "postprocessors": [{
-            "key": "FFmpegExtractAudio",
-            "preferredcodec": "mp3",
-            "preferredquality": "192",
-        }],
-    }
-    try:
+    with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as tmp_file:
+        ydl_opts = {
+            "format": "bestaudio/best",
+            "outtmpl": tmp_file.name,
+            "quiet": True,
+            "postprocessors": [{
+                "key": "FFmpegExtractAudio",
+                "preferredcodec": "mp3",
+                "preferredquality": "192",
+            }],
+        }
         with yt_dlp.YoutubeDL(ydl_opts) as ydl:
-            info = ydl.extract_info(youtube_url, download=True)
-        # find the downloaded file
-        files = glob.glob(os.path.join(tmpdir, "*"))
-        if not files:
-            raise FileNotFoundError("Downloaded file not found.")
-        # pick the first audio file (mp3)
-        audio_path = None
-        for f in files:
-            if f.lower().endswith((".mp3", ".m4a", ".wav", ".webm", ".aac", ".ogg")):
-                audio_path = f
-                break
-        if audio_path is None:
-            audio_path = files[0]
-        return audio_path, tmpdir
-    except Exception as e:
-        # cleanup on failure
-        try:
-            shutil.rmtree(tmpdir)
-        except:
-            pass
-        return f"❌ Error downloading audio: {e}", None
+            ydl.download([youtube_url])
+        return tmp_file.name
 
+# ✅ Convert to WAV safely
 def convert_to_wav(input_path):
-    """
-    Convert any audio/video to 16kHz mono WAV using ffmpeg.
-    Always creates a new temp wav file (never overwrites input).
-    Returns path to wav file.
-    """
-    import tempfile
-    from pydub import AudioSegment
-
     try:
         tmp_wav = tempfile.NamedTemporaryFile(suffix=".wav", delete=False, prefix="conv_")
-        tmp_wav.close()
         out_wav = tmp_wav.name
-
-        # Use ffmpeg CLI
         result = subprocess.run(
             ["ffmpeg", "-y", "-i", str(input_path), "-ar", "16000", "-ac", "1", out_wav],
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE
         )
-
-        if result.returncode != 0 or not os.path.exists(out_wav):
-            print("⚠️ ffmpeg failed, trying pydub fallback...")
-            try:
-                audio = AudioSegment.from_file(input_path)
-                audio = audio.set_frame_rate(16000).set_channels(1)
-                audio.export(out_wav, format="wav")
-            except Exception as e:
-                raise RuntimeError(f"Both ffmpeg and pydub conversion failed: {e}")
-
+        if result.returncode != 0:
+            audio = AudioSegment.from_file(input_path)
+            audio = audio.set_frame_rate(16000).set_channels(1)
+            audio.export(out_wav, format="wav")
         return out_wav
-
     except Exception as e:
-        raise RuntimeError(f"Conversion error: {e}")
+        raise RuntimeError(f"❌ Error converting to WAV: {e}")
 
-def split_audio_to_chunks(wav_path, max_ms=5*60*1000):
-    """
-    Split WAV into chunks of max_ms milliseconds (default 5 minutes).
-    Returns list of chunk file paths.
-    """
-    audio = AudioSegment.from_file(wav_path)
+# ✅ Split long audio into 5-min chunks
+def split_audio(file_path, max_duration_ms=5*60*1000):
+    audio = AudioSegment.from_file(file_path)
     chunks = []
-    for i in range(0, len(audio), max_ms):
-        chunk = audio[i:i+max_ms]
-        tmpf = tempfile.NamedTemporaryFile(suffix=".wav", delete=False, prefix="chunk_")
-        tmpf.close()
-        chunk.export(tmpf.name, format="wav")
-        chunks.append(tmpf.name)
+    for i in range(0, len(audio), max_duration_ms):
+        chunk = audio[i:i + max_duration_ms]
+        temp_chunk = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
+        chunk.export(temp_chunk.name, format="wav")
+        chunks.append(temp_chunk.name)
     return chunks
 
-def transcribe_with_groq_chunks(audio_path):
-    """
-    Convert to wav, split into chunks, send each to Groq Whisper and combine results.
-    Returns aggregated transcript string or error message string starting with ❌
-    """
-    try:
-        wav = convert_to_wav(audio_path)
-    except Exception as e:
-        return f"❌ Error converting to WAV: {e}"
-
+# ✅ Transcription using Groq Whisper
+def transcribe_audio(audio_path):
     try:
-        chunks = split_audio_to_chunks(wav)
-    except Exception as e:
-        return f"❌ Error splitting audio: {e}"
-
-    transcript_pieces = []
-    try:
-        for idx, chunk_path in enumerate(chunks, start=1):
-            with open(chunk_path, "rb") as f:
-                # call Groq transcription API
-                resp = groq_client.audio.transcriptions.create(
+        chunks = split_audio(audio_path)
+        transcript = ""
+        for chunk in chunks:
+            with open(chunk, "rb") as f:
+                response = groq_client.audio.transcriptions.create(
                     model="whisper-large-v3",
                     file=f
                 )
-            # resp may be str, object with .text, or dict
-            text_piece = ""
-            if isinstance(resp, str):
-                text_piece = resp
-            elif hasattr(resp, "text"):
-                text_piece = getattr(resp, "text")
-            elif isinstance(resp, dict):
-                text_piece = resp.get("text") or resp.get("transcription") or ""
-            else:
-                text_piece = str(resp)
-            transcript_pieces.append(text_piece.strip())
+            transcript += response.text + "\n"
+        return transcript.strip()
     except Exception as e:
         return f"❌ Error during transcription: {e}"
-    finally:
-        # cleanup chunk files and wav
-        for c in chunks:
-            try:
-                os.remove(c)
-            except:
-                pass
-        try:
-            if os.path.exists(wav):
-                os.remove(wav)
-        except:
-            pass
-
-    aggregated = "\n".join([p for p in transcript_pieces if p])
-    return aggregated if aggregated else "❌ Empty transcription result."
 
-def summarize_via_hf(text, model, hf_token, params=None):
-    """
-    Call Hugging Face Inference HTTP API for summarization.
-    model: model repo id (e.g., 'facebook/bart-large-cnn')
-    params: optional dict for 'parameters' in request
-    """
-    url = f"https://api-inference.huggingface.co/models/{model}"
-    headers = {"Authorization": f"Bearer {hf_token}"}
-    payload = {"inputs": text}
-    if params:
-        payload["parameters"] = params
+# ✅ Chunk-wise summarization using Groq LLM
+def summarize_text(text, lang):
     try:
-        r = requests.post(url, headers=headers, json=payload, timeout=120)
-    except Exception as e:
-        return f"❌ HTTP error contacting Hugging Face: {e}"
-    if r.status_code != 200:
-        # try to surface error message
-        try:
-            info = r.json()
-            return f"❌ Summarization failed: {info}"
-        except:
-            return f"❌ Summarization failed: HTTP {r.status_code}"
-    try:
-        out = r.json()
-        # typical responses:
-        # - [{'summary_text': '...'}] (BART)
-        # - [{'generated_text': '...'}] (some models)
-        if isinstance(out, list) and len(out) > 0:
-            first = out[0]
-            if isinstance(first, dict) and "summary_text" in first:
-                return first["summary_text"]
-            if isinstance(first, dict) and "generated_text" in first:
-                return first["generated_text"]
-            if isinstance(first, str):
-                return first
-            # fallback: stringify first element
-            return str(first)
-        elif isinstance(out, dict) and "summary_text" in out:
-            return out["summary_text"]
-        elif isinstance(out, str):
-            return out
-        else:
-            return str(out)
+        chunks = [text[i:i+2000] for i in range(0, len(text), 2000)]
+        summaries = []
+
+        for idx, chunk in enumerate(chunks):
+            prompt = (
+                f"Summarize the following text in English:\n\n{chunk}"
+                if lang == "English"
+                else f"مندرجہ ذیل عبارت کا جامع اور رواں اردو خلاصہ تحریر کریں:\n\n{chunk}"
+            )
+            response = groq_client.chat.completions.create(
+                model="openai/gpt-oss-120b",
+                messages=[{"role": "user", "content": prompt}],
+                temperature=0.6,
+            )
+            summaries.append(response.choices[0].message.content.strip())
+
+        # Meta-summary
+        combined = "\n".join(summaries)
+        final_prompt = (
+            f"Combine and condense these summaries into one clear, fluent English summary:\n\n{combined}"
+            if lang == "English"
+            else f"مندرجہ ذیل خلاصوں کو یکجا کر کے ایک مختصر مگر جامع اردو خلاصہ تحریر کریں:\n\n{combined}"
+        )
+        final_response = groq_client.chat.completions.create(
+            model="openai/gpt-oss-120b",
+            messages=[{"role": "user", "content": final_prompt}],
+            temperature=0.6,
+        )
+        return final_response.choices[0].message.content.strip()
     except Exception as e:
-        return f"❌ Error parsing summarization response: {e}"
-
-# -----------------------
-# Main pipeline
-# -----------------------
+        return f"❌ Summarization failed: {e}"
 
-def process_input(youtube_url, uploaded_audio, summary_lang):
-    """
-    Main handler for Gradio:
-    - Accepts optional youtube_url or uploaded_audio (filepath)
-    - summary_lang: 'English' or 'Urdu'
-    Returns: status, transcript, summary
-    """
-    tempdir_to_cleanup = None
+# ✅ Main Function
+def process_input(youtube_url, audio_file, lang):
     try:
-        if youtube_url and youtube_url.strip():
-            audio_path_or_err, tmpdir = download_youtube_audio(youtube_url.strip())
-            if isinstance(audio_path_or_err, str) and audio_path_or_err.startswith("❌"):
-                return audio_path_or_err, "", ""
-            audio_path = audio_path_or_err
-            tempdir_to_cleanup = tmpdir
-        elif uploaded_audio:
-            audio_path = uploaded_audio
+        if youtube_url:
+            audio_path = download_youtube_audio(youtube_url)
+        elif audio_file:
+            audio_path = audio_file
         else:
-            return "❌ Please upload an audio file or paste a YouTube link.", "", ""
+            return "❌ Please upload an audio or paste YouTube link.", "", ""
 
-        # Transcribe (with chunking)
-        transcript = transcribe_with_groq_chunks(audio_path)
+        wav_path = convert_to_wav(audio_path)
+        transcript = transcribe_audio(wav_path)
         if transcript.startswith("❌"):
             return transcript, "", ""
 
-        # Summarize using appropriate model via HF HTTP
-        if summary_lang == "English":
-            model = "facebook/bart-large-cnn"
-            params = {"min_length": 30, "max_length": 250}
-            summary = summarize_via_hf(transcript, model, HUGGINGFACE_API_TOKEN, params=params)
-        else:  # Urdu
-            # Use mBART: we ask the model to produce an Urdu summary.
-            model = "facebook/mbart-large-50-many-to-many-mmt"
-            # Provide a short Urdu instruction + the text (in case transcript is English; model will try to summarize in Urdu)
-            prompt_text = f"مندرجہ ذیل عبارت کا جامع اردو خلاصہ لکھیں:\n\n{transcript}"
-            params = {"min_length": 30, "max_length": 250}
-            summary = summarize_via_hf(prompt_text, model, HUGGINGFACE_API_TOKEN, params=params)
-
-        return "✅ Transcription & Summarization completed.", transcript, summary
+        summary = summarize_text(transcript, lang)
+        return "✅ Transcription Completed!", transcript, summary
+    except Exception as e:
+        return f"❌ Error: {e}", "", ""
 
-    finally:
-        # cleanup downloaded tempdir (if any)
-        if tempdir_to_cleanup:
-            try:
-                shutil.rmtree(tempdir_to_cleanup)
-            except:
-                pass
+# ✅ Gradio Interface
+with gr.Blocks(title="🎧 Urdu & English Audio Summarizer") as app:
+    gr.Markdown("## 🎧 Transcribe & Summarize English or Urdu Audio / YouTube Videos")
 
-# -----------------------
-# Gradio UI
-# -----------------------
-with gr.Blocks(title="SmartTranscribe — YouTube + Upload (Urdu/English)") as demo:
-    gr.Markdown("## SmartTranscribe — Upload audio or paste YouTube link. Select summary language (English/Urdu).")
     with gr.Row():
-        youtube_input = gr.Textbox(label="YouTube Link (optional)", placeholder="https://www.youtube.com/watch?v=...")
-        summary_lang = gr.Dropdown(choices=["English", "Urdu"], value="English", label="Summary language")
-    audio_input = gr.Audio(type="filepath", label="Upload or Record audio (optional)")
-    process_btn = gr.Button("Transcribe & Summarize")
+        youtube_link = gr.Textbox(label="📺 YouTube Link (optional)")
+        lang_choice = gr.Dropdown(["English", "Urdu"], value="English", label="🌐 Summary Language")
+
+    audio_input = gr.Audio(type="filepath", label="🎙️ Upload Audio (optional)")
+    btn = gr.Button("🚀 Transcribe & Summarize")
 
-    status_box = gr.Textbox(label="Status")
-    transcript_box = gr.Textbox(label="Transcription", lines=10)
-    summary_box = gr.Textbox(label="Summary", lines=8)
+    status = gr.Textbox(label="Status")
+    transcript_box = gr.Textbox(label="📝 Transcription", lines=8)
+    summary_box = gr.Textbox(label="🧩 Summary", lines=8)
 
-    process_btn.click(fn=process_input, inputs=[youtube_input, audio_input, summary_lang], outputs=[status_box, transcript_box, summary_box])
+    btn.click(process_input, [youtube_link, audio_input, lang_choice], [status, transcript_box, summary_box])
 
-if __name__ == "__main__":
-    demo.launch()
+app.launch()