Update app.py

app.py

@@ -1,17 +1,360 @@
----
-title: SmartTranscribe — Rate-limited Urdu/English Summaries
-emoji: 🎙️
-colorFrom: indigo
-colorTo: purple
-sdk: gradio
-sdk_version: "4.44.0"
-app_file: app.py
-pinned: true
----
-
-# SmartTranscribe
-
-Notes:
-- This version respects Groq tokens-per-minute (TPM) limits by chunking transcripts and rate-limiting requests.
-- You can tweak TPM via env var `GROQ_TPM_LIMIT` (default 8000).
-- You can adjust chunk size via `CHUNK_CHARS` and output size via `OUT_TOKENS_PER_CHUNK`.
+# 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 yt_dlp
+import requests
+from pydub import AudioSegment
+from groq import Groq
+
+# -----------------------
+# Config / Env
+# -----------------------
+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.")
+
+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",
+        }],
+    }
+    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(
+            ["ffmpeg", "-y", "-i", str(input_path), "-ar", "16000", "-ac", "1", out_wav],
+            stdout=subprocess.PIPE, stderr=subprocess.PIPE
+        )
+        if res.returncode != 0 or not Path(out_wav).exists():
+            # fallback to pydub
+            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
+
+# -----------------------
+# Transcription (chunked audio)
+# -----------------------
+def split_wav_to_chunks(wav_path: str, max_ms=5*60*1000):
+    audio = AudioSegment.from_file(wav_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)
+    return chunks
+
+def transcribe_audio_with_groq(wav_or_audio_path: str):
+    # convert first
+    wav = convert_to_wav_safe(wav_or_audio_path)
+    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
+
+# -----------------------
+# 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 = []
+
+    for idx, chunk in enumerate(text_chunks, start=1):
+        # build prompt in Urdu
+        prompt = (
+            "مندرجہ ذیل عبارت کا مختصر، صاف اور روان اردو خلاصہ (2-3 جملے) لکھیں:\n\n" + chunk
+        )
+        # 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(
+            model="openai/gpt-oss-120b",
+            messages=[{"role": "user", "content": meta_prompt}],
+            temperature=0.3,
+            max_tokens=output_tokens
+        )
+        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]"
+    except Exception as e:
+        return f"❌ Meta-summarization failed: {e}"
+
+    # final rate-limit sleep
+    rate_limit_sleep(input_tokens, output_tokens)
+    return final_summary
+
+# -----------------------
+# Main pipeline: handles YouTube or uploaded file
+# -----------------------
+def process_input(youtube_url, uploaded_audio, summary_lang):
+    tempdir = None
+    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
+        else:
+            return "❌ Please upload an audio file or paste a YouTube link.", "", ""
+
+        # Transcribe
+        transcript = transcribe_audio_with_groq(audio_path)
+        if transcript.startswith("❌"):
+            return transcript, "", ""
+
+        # 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
+
+    finally:
+        if tempdir:
+            try:
+                shutil.rmtree(tempdir)
+            except:
+                pass
+
+# -----------------------
+# 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")
+    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])
+
+if __name__ == "__main__":
+    demo.launch()
+