Update app.py

app.py

@@ -1,122 +1,273 @@
+# 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
 from pydub import AudioSegment
 from groq import Groq
-from huggingface_hub import InferenceClient
 
-# ✅ 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 or not HUGGINGFACE_API_TOKEN:
-    raise EnvironmentError("Please set GROQ_API_KEY and 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)
-hf_client = InferenceClient(token=HUGGINGFACE_API_TOKEN)
 
-# ✅ Download YouTube Audio
+# -----------------------
+# Utilities
+# -----------------------
+
 def download_youtube_audio(youtube_url):
-    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",
-            }],
-        }
+    """
+    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 yt_dlp.YoutubeDL(ydl_opts) as ydl:
-            ydl.download([youtube_url])
-        return tmp_file.name
+            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
+
+def convert_to_wav(input_path):
+    """
+    Convert any audio/video to 16kHz mono WAV using ffmpeg CLI for reliability.
+    Returns path to wav file.
+    """
+    p = Path(input_path)
+    out_wav = str(p.with_suffix(".wav"))
+    try:
+        subprocess.run(
+            ["ffmpeg", "-y", "-i", str(input_path), "-ar", "16000", "-ac", "1", out_wav],
+            check=True,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        )
+        return out_wav
+    except Exception as e:
+        raise RuntimeError(f"ffmpeg conversion failed: {e}")
 
-# ✅ Split long audio into 5-minute chunks
-def split_audio(file_path, max_duration_ms=5*60*1000):
-    audio = AudioSegment.from_file(file_path)
+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)
     chunks = []
-    for i in range(0, len(audio), max_duration_ms):
-        chunk = audio[i:i + max_duration_ms]
-        temp_chunk = tempfile.NamedTemporaryFile(suffix=".mp3", delete=False)
-        chunk.export(temp_chunk.name, format="mp3")
-        chunks.append(temp_chunk.name)
+    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)
     return chunks
 
-# ✅ Transcribe with Groq Whisper
-def transcribe_audio(audio_path):
+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:
-        chunks = split_audio(audio_path)
-        transcript = ""
-        for i, chunk in enumerate(chunks):
-            with open(chunk, "rb") as f:
-                response = groq_client.audio.transcriptions.create(
+        wav = convert_to_wav(audio_path)
+    except Exception as e:
+        return f"❌ Error converting to WAV: {e}"
+
+    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(
                     model="whisper-large-v3",
                     file=f
                 )
-            transcript += response.text + "\n"
-        return transcript.strip()
+            # 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())
     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."
 
-# ✅ Summarize with correct API
-def summarize_text(text, lang):
+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
     try:
-        if lang == "English":
-            model = "facebook/bart-large-cnn"
-            inputs = text
-        else:
-            model = "facebook/mbart-large-50-many-to-many-mmt"
-            inputs = f"مندرجہ ذیل عبارت کا جامع اردو خلاصہ لکھیں:\n\n{text}"
-
-        result = hf_client.summarization(
-            model=model,
-            inputs=inputs,
-            max_length=250,
-            min_length=80,
-            temperature=0.7,
-        )
-        if isinstance(result, list):
-            return result[0]["summary_text"]
-        elif isinstance(result, dict) and "summary_text" in result:
-            return result["summary_text"]
+        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(result)
+            return str(out)
     except Exception as e:
-        return f"❌ Error during summarization: {e}"
+        return f"❌ Error parsing summarization response: {e}"
 
-# ✅ Main Pipeline
-def process_input(youtube_url, audio_file, lang):
-    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.", "", ""
+# -----------------------
+# Main pipeline
+# -----------------------
 
-    transcript = transcribe_audio(audio_path)
-    if transcript.startswith("❌"):
-        return transcript, "", ""
+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
+    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
+        else:
+            return "❌ Please upload an audio file or paste a YouTube link.", "", ""
+
+        # Transcribe (with chunking)
+        transcript = transcribe_with_groq_chunks(audio_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)
 
-    summary = summarize_text(transcript, lang)
-    return "✅ Transcription Completed!", transcript, summary
+        return "✅ Transcription & Summarization completed.", transcript, summary
 
-# ✅ Gradio Interface
-with gr.Blocks(title="🎧 Urdu/English Audio Summarizer") as app:
-    gr.Markdown("## 🎧 Urdu & English Audio Summarizer\nUpload an audio file or paste a YouTube link below:")
+    finally:
+        # cleanup downloaded tempdir (if any)
+        if tempdir_to_cleanup:
+            try:
+                shutil.rmtree(tempdir_to_cleanup)
+            except:
+                pass
 
+# -----------------------
+# 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_link = gr.Textbox(label="📺 YouTube Link (optional)")
-        language_choice = gr.Dropdown(["English", "Urdu"], value="English", label="🌐 Summary Language")
-
-    audio_input = gr.Audio(type="filepath", label="🎙️ Upload Audio File (optional)")
-    btn = gr.Button("🚀 Transcribe & Summarize")
+        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")
 
-    status = gr.Textbox(label="Status")
-    transcript_box = gr.Textbox(label="📝 Transcription", lines=8)
-    summary_box = gr.Textbox(label="🧩 Summary", lines=8)
+    status_box = gr.Textbox(label="Status")
+    transcript_box = gr.Textbox(label="Transcription", lines=10)
+    summary_box = gr.Textbox(label="Summary", lines=8)
 
-    btn.click(process_input, [youtube_link, audio_input, language_choice], [status, transcript_box, summary_box])
+    process_btn.click(fn=process_input, inputs=[youtube_input, audio_input, summary_lang], outputs=[status_box, transcript_box, summary_box])
 
-app.launch()
+if __name__ == "__main__":
+    demo.launch()