Update app.py

app.py

@@ -1,360 +1,146 @@
-# app.py
 import os
-import time
-import math
-import shutil
-import tempfile
-import glob
-import subprocess
-from pathlib import Path
-
 import gradio as gr
+import tempfile
 import yt_dlp
-import requests
+import subprocess
 from pydub import AudioSegment
 from groq import Groq
 
-# -----------------------
-# Config / Env
-# -----------------------
+# ✅ Environment Variable Check
 GROQ_API_KEY = os.getenv("GROQ_API_KEY")
-# TPM limit (tokens per minute) for your Groq org. Default 8000 (from your error).
-GROQ_TPM_LIMIT = int(os.getenv("GROQ_TPM_LIMIT", "8000"))
-# Desired output tokens per chunk when summarizing
-OUT_TOKENS_PER_CHUNK = int(os.getenv("OUT_TOKENS_PER_CHUNK", "120"))
-# Approx characters per token (conservative)
-CHARS_PER_TOKEN = int(os.getenv("CHARS_PER_TOKEN", "4"))
-# Chunk size in characters (input chunk) - conservative to keep tokens low
-CHUNK_CHARS = int(os.getenv("CHUNK_CHARS", "1800"))
-
 if not GROQ_API_KEY:
-    raise EnvironmentError("Please set GROQ_API_KEY in the Space settings.")
+    raise EnvironmentError("Please set GROQ_API_KEY.")
 
 groq_client = Groq(api_key=GROQ_API_KEY)
 
-# -----------------------
-# Utilities
-# -----------------------
-def estimate_tokens(text: str) -> int:
-    """Rough token estimate: characters / CHARS_PER_TOKEN."""
-    if not text:
-        return 1
-    return max(1, math.ceil(len(text) / CHARS_PER_TOKEN))
-
-def rate_limit_sleep(input_tokens: int, output_tokens: int):
-    """
-    Sleep time to respect TPM limit.
-    formula: sleep_seconds = (input+output)/TPM_LIMIT * 60
-    This spaces requests so token-per-minute rate stays under limit.
-    """
-    tokens_needed = input_tokens + output_tokens
-    if GROQ_TPM_LIMIT <= 0:
-        return
-    sleep_seconds = (tokens_needed / GROQ_TPM_LIMIT) * 60.0
-    # enforce a small minimum to avoid too quick back-to-back calls
-    if sleep_seconds < 0.5:
-        sleep_seconds = 0.5
-    time.sleep(sleep_seconds)
-
-# -----------------------
-# Audio / YouTube helpers
-# -----------------------
-def download_youtube_audio(youtube_url: str):
-    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",
-        }],
-    }
+# ✅ Convert audio to 16kHz mono WAV
+def convert_to_wav(input_path):
     try:
-        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
-            info = ydl.extract_info(youtube_url, download=True)
-        files = glob.glob(os.path.join(tmpdir, "*"))
-        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 and files:
-            audio_path = files[0]
-        return audio_path, tmpdir
-    except Exception as e:
-        try:
-            shutil.rmtree(tmpdir)
-        except:
-            pass
-        return f"❌ Error downloading YouTube audio: {e}", None
-
-def convert_to_wav_safe(input_path: str):
-    """
-    Convert input file to 16kHz mono WAV, always produce a new temp file.
-    Uses ffmpeg CLI and falls back to pydub if needed.
-    """
-    p = Path(input_path)
-    tmpf = tempfile.NamedTemporaryFile(suffix=".wav", delete=False, prefix="conv_")
-    tmpf.close()
-    out_wav = tmpf.name
-    try:
-        # call ffmpeg
-        res = subprocess.run(
+        tmp_wav = tempfile.NamedTemporaryFile(suffix=".wav", delete=False, prefix="conv_")
+        out_wav = tmp_wav.name
+        result = subprocess.run(
             ["ffmpeg", "-y", "-i", str(input_path), "-ar", "16000", "-ac", "1", out_wav],
-            stdout=subprocess.PIPE, stderr=subprocess.PIPE
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE
         )
-        if res.returncode != 0 or not Path(out_wav).exists():
-            # fallback to pydub
+        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:
-        # cleanup
-        try:
-            if Path(out_wav).exists():
-                os.remove(out_wav)
-        except:
-            pass
-        raise RuntimeError(f"Error converting to WAV: {e}")
-
-def split_text_to_chunks(text: str, chunk_chars: int = CHUNK_CHARS):
-    """Split long text into character-based chunks (preserving word boundaries)."""
-    words = text.split()
-    chunks = []
-    current = []
-    current_len = 0
-    for w in words:
-        if current_len + len(w) + 1 > chunk_chars and current:
-            chunks.append(" ".join(current))
-            current = [w]
-            current_len = len(w) + 1
-        else:
-            current.append(w)
-            current_len += len(w) + 1
-    if current:
-        chunks.append(" ".join(current))
-    return chunks
+        raise RuntimeError(f"❌ Error converting to WAV: {e}")
 
-# -----------------------
-# Transcription (chunked audio)
-# -----------------------
-def split_wav_to_chunks(wav_path: str, max_ms=5*60*1000):
-    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_audio_with_groq(wav_or_audio_path: str):
-    # convert first
-    wav = convert_to_wav_safe(wav_or_audio_path)
+# ✅ Download YouTube Audio (with clear error messages)
+def download_youtube_audio(youtube_url):
     try:
-        audio_chunks = split_wav_to_chunks(wav)
-        pieces = []
-        for chunk_path in audio_chunks:
-            with open(chunk_path, "rb") as f:
-                resp = groq_client.audio.transcriptions.create(model="whisper-large-v3", file=f)
-            # flexibly extract text
-            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)
-            pieces.append(text_piece.strip())
-            # cleanup chunk right away
-            try:
-                os.remove(chunk_path)
-            except:
-                pass
-        aggregated = "\n".join([p for p in pieces if p])
-        return aggregated
-    finally:
-        # cleanup wav
-        try:
-            if os.path.exists(wav):
-                os.remove(wav)
-        except:
-            pass
+        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:
+                ydl.download([youtube_url])
+            return tmp_file.name
+    except Exception as e:
+        raise RuntimeError(f"❌ Error downloading YouTube audio: {e}")
 
-# -----------------------
-# Summarization (Urdu) with chunking + rate-limiting
-# -----------------------
-def summarise_in_urdu_with_groq(full_text: str):
-    """
-    1) Split long transcript into manageable text chunks (CHUNK_CHARS).
-    2) For each chunk, call Groq LLM to produce a short Urdu summary (OUT_TOKENS_PER_CHUNK).
-    3) Rate-limit between calls based on estimated tokens so we don't exceed TPM.
-    4) Combine chunk summaries and ask Groq for a final condensed Urdu summary.
-    """
-    # 1) split text
-    text_chunks = split_text_to_chunks(full_text, chunk_chars=CHUNK_CHARS)
-    chunk_summaries = []
+# ✅ Transcription using Groq Whisper
+def transcribe_audio(audio_path):
+    try:
+        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
+                )
+            transcript += response.text + "\n"
+        return transcript.strip()
+    except Exception as e:
+        return f"❌ Error during transcription: {e}"
 
-    for idx, chunk in enumerate(text_chunks, start=1):
-        # build prompt in Urdu
+# ✅ Summarization with “Detailed Mode”
+def summarize_text(text, lang):
+    try:
         prompt = (
-            "مندرجہ ذیل عبارت کا مختصر، صاف اور روان اردو خلاصہ (2-3 جملے) لکھیں:\n\n" + chunk
+            f"Create a detailed, structured and comprehensive English summary of the following text. "
+            f"Cover key points, ideas, examples and conclusions clearly:\n\n{text}"
+            if lang == "English"
+            else f"مندرجہ ذیل عبارت کا تفصیلی، منظم اور جامع اردو خلاصہ تحریر کریں۔ خلاصے میں اہم نکات، مثالیں اور نتائج کو واضح طور پر بیان کریں:\n\n{text}"
         )
-        # estimate tokens
-        input_tokens = estimate_tokens(chunk)
-        output_tokens = OUT_TOKENS_PER_CHUNK
-        # Call Groq chat completion
-        try:
-            resp = groq_client.chat.completions.create(
-                model="openai/gpt-oss-120b",
-                messages=[{"role": "user", "content": prompt}],
-                temperature=0.3,
-                max_tokens=output_tokens
-            )
-            # extract content
-            summary_text = ""
-            try:
-                summary_text = resp.choices[0].message.content.strip()
-            except Exception:
-                # fallback to dict parsing
-                if isinstance(resp, dict):
-                    # sometimes resp may contain 'choices' list etc.
-                    ch = resp.get("choices")
-                    if ch and isinstance(ch, list) and len(ch) > 0:
-                        summ = ch[0].get("message", {}).get("content") or ch[0].get("text") or ""
-                        summary_text = summ.strip() if summ else ""
-                if not summary_text and isinstance(resp, str):
-                    summary_text = resp.strip()
-            if not summary_text:
-                summary_text = "[خلاصہ دستیاب نہیں — خالی نتیجہ]"
-            chunk_summaries.append(summary_text)
-        except Exception as e:
-            return f"❌ Summarization failed while processing chunk {idx}: {e}"
-
-        # 3) rate-limit sleep to respect TPM
-        rate_limit_sleep(input_tokens, output_tokens)
-
-    # 4) meta-summary: combine chunk summaries and condense
-    combined = "\n".join(chunk_summaries)
-    meta_prompt = (
-        "مندرجہ ذیل مختصر خلاصوں کو ایک مربوط، جامع اور مختصر اردو خلاصہ میں تبدیل کریں (3-6 جملے):\n\n"
-        + combined
-    )
-    input_tokens = estimate_tokens(combined)
-    output_tokens = int(OUT_TOKENS_PER_CHUNK * 1.5)  # allow slightly larger for meta summary
-    try:
-        resp2 = groq_client.chat.completions.create(
+        response = groq_client.chat.completions.create(
             model="openai/gpt-oss-120b",
-            messages=[{"role": "user", "content": meta_prompt}],
-            temperature=0.3,
-            max_tokens=output_tokens
+            messages=[{"role": "user", "content": prompt}],
+            temperature=0.6,
         )
-        final_summary = ""
-        try:
-            final_summary = resp2.choices[0].message.content.strip()
-        except Exception:
-            if isinstance(resp2, dict):
-                ch = resp2.get("choices")
-                if ch and isinstance(ch, list) and len(ch) > 0:
-                    final_summary = ch[0].get("message", {}).get("content") or ch[0].get("text") or ""
-            if not final_summary and isinstance(resp2, str):
-                final_summary = resp2.strip()
-        if not final_summary:
-            final_summary = "[Meta-summary failed — empty result]"
+        return response.choices[0].message.content.strip()
     except Exception as e:
-        return f"❌ Meta-summarization failed: {e}"
-
-    # final rate-limit sleep
-    rate_limit_sleep(input_tokens, output_tokens)
-    return final_summary
+        return f"❌ Summarization failed: {e}"
 
-# -----------------------
-# Main pipeline: handles YouTube or uploaded file
-# -----------------------
-def process_input(youtube_url, uploaded_audio, summary_lang):
-    tempdir = None
+# ✅ Step 1: Transcription
+def process_transcription(youtube_url, audio_file):
     try:
-        if youtube_url and youtube_url.strip():
-            audio_path, tempdir = download_youtube_audio(youtube_url.strip())
-            if isinstance(audio_path, str) and audio_path.startswith("❌"):
-                return audio_path, "", ""
-        elif uploaded_audio:
-            audio_path = uploaded_audio
+        if youtube_url:
+            try:
+                audio_path = download_youtube_audio(youtube_url)
+            except Exception as e:
+                return f"❌ Error downloading YouTube audio: {e}", ""
+        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
-        transcript = transcribe_audio_with_groq(audio_path)
+        wav_path = convert_to_wav(audio_path)
+        transcript = transcribe_audio(wav_path)
         if transcript.startswith("❌"):
-            return transcript, "", ""
+            return transcript, ""
+        return "✅ Transcription Completed!", transcript
+    except Exception as e:
+        return f"❌ Error: {e}", ""
 
-        # Summarize: English or Urdu
-        if summary_lang == "English":
-            # Keep previous fast English summarization approach using Groq with a reasonable output size
-            en_prompt = "Summarize the following text in clear English (3-6 sentences):\n\n" + transcript
-            input_tokens = estimate_tokens(transcript)
-            output_tokens = 180
-            try:
-                resp = groq_client.chat.completions.create(
-                    model="openai/gpt-oss-120b",
-                    messages=[{"role": "user", "content": en_prompt}],
-                    temperature=0.3,
-                    max_tokens=output_tokens
-                )
-                summary_text = ""
-                try:
-                    summary_text = resp.choices[0].message.content.strip()
-                except Exception:
-                    if isinstance(resp, dict):
-                        ch = resp.get("choices")
-                        if ch and isinstance(ch, list) and len(ch) > 0:
-                            summary_text = ch[0].get("message", {}).get("content") or ch[0].get("text") or ""
-                    if not summary_text and isinstance(resp, str):
-                        summary_text = resp.strip()
-                if not summary_text:
-                    summary_text = "[Empty summary]"
-                # rate-limit sleep
-                rate_limit_sleep(input_tokens, output_tokens)
-                return "✅ Done", transcript, summary_text
-            except Exception as e:
-                return f"❌ English summarization failed: {e}", transcript, ""
-        else:
-            # Urdu summarization via chunked approach
-            ur_summary = summarise_in_urdu_with_groq(transcript)
-            if ur_summary.startswith("❌"):
-                return ur_summary, transcript, ""
-            return "✅ Done", transcript, ur_summary
+# ✅ Step 2: Generate Detailed Summary
+def process_summary(transcript, lang):
+    if not transcript or transcript.startswith("❌"):
+        return "❌ Please transcribe audio first."
+    return summarize_text(transcript, lang)
 
-    finally:
-        if tempdir:
-            try:
-                shutil.rmtree(tempdir)
-            except:
-                pass
+# ✅ Gradio Interface
+with gr.Blocks(title="🎧 Urdu & English Audio Transcriber + Summarizer") as app:
+    gr.Markdown("## 🎧 AI Audio & YouTube Transcriber — English & Urdu")
 
-# -----------------------
-# Gradio UI
-# -----------------------
-with gr.Blocks(title="SmartTranscribe — Urdu/English (rate-limited summaries)") as demo:
-    gr.Markdown("## SmartTranscribe — Upload audio or paste YouTube link. Choose 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)")
+
+    transcribe_btn = gr.Button("📝 Step 1: Transcribe Audio / Video")
+    summarize_btn = gr.Button("🧩 Step 2: Generate Comprehensive Summary")
+
     status = gr.Textbox(label="Status")
-    transcript_box = gr.Textbox(label="Transcription", lines=12)
-    summary_box = gr.Textbox(label="Summary", lines=8)
-    process_btn.click(fn=process_input, inputs=[youtube_input, audio_input, summary_lang], outputs=[status, transcript_box, summary_box])
+    transcript_box = gr.Textbox(label="📝 Transcription", lines=8)
+    summary_box = gr.Textbox(label="📘 Detailed Summary", lines=8)
 
-if __name__ == "__main__":
-    demo.launch()
+    transcribe_btn.click(process_transcription, [youtube_link, audio_input], [status, transcript_box])
+    summarize_btn.click(process_summary, [transcript_box, lang_choice], [summary_box])
 
+app.launch()