Update transcriber.py

transcriber.py

@@ -1,223 +1,23 @@
-"""
-Department 2 — Transcriber
-UPGRADE APPLIED:
-  1. Military-Grade VAD: Silero VAD filters audio prior to Whisper processing.
-     This drops the hallucination rate to zero by physically isolating human speech.
-"""
-
 import os
-import time
 import logging
-import subprocess
-import tempfile
-import shutil
+from groq import Groq
 
 logger = logging.getLogger(__name__)
 
-LANG_TO_WHISPER = {"auto": None, "en": "en", "te": "te", "hi": "hi", "ta": "ta", "kn": "kn"}
-GROQ_SUPPORTED_LANGS = {"en", "hi", "ta", "es", "fr", "de", "ja", "zh", "ar", "pt", "ru", "it"}
-FORCE_LANGUAGE_HINT = {"te", "kn", "hi", "ta"}
-CHUNK_SEC   = 60
-MAX_RETRIES = 3
-
 class Transcriber:
     def __init__(self):
-        self.groq_key       = os.environ.get("GROQ_API_KEY", "")
-        self._groq_client   = None
-        self._local_model   = None
-        self._last_segments = []
-
-        if self.groq_key:
-            print("[Transcriber] Groq API key found — primary = Groq Whisper large-v3")
-            self._init_groq()
-        else:
-            print("[Transcriber] No GROQ_API_KEY — local Whisper loads on first use")
-
-    def transcribe(self, audio_path: str, language: str = "auto"):
-        lang_hint = LANG_TO_WHISPER.get(language, None)
-        duration  = self._get_duration(audio_path)
-        self._last_segments = []
-
-        if duration <= CHUNK_SEC:
-            return self._transcribe_single(audio_path, lang_hint)
-        return self._transcribe_chunked(audio_path, lang_hint, duration)
-
-    def _transcribe_chunked(self, audio_path, language, duration):
-        tmp_dir = tempfile.mkdtemp()
-        chunks, start, idx = [], 0, 0
-
-        while start < duration:
-            cp = os.path.join(tmp_dir, f"chunk_{idx:03d}.wav")
-            subprocess.run([
-                "ffmpeg", "-y", "-i", audio_path, "-ss", str(start), "-t", str(CHUNK_SEC),
-                "-acodec", "pcm_s16le", "-ar", "16000", "-ac", "1", cp
-            ], capture_output=True)
-            if os.path.exists(cp): chunks.append((cp, start))
-            start += CHUNK_SEC
-            idx   += 1
-
-        all_texts, all_segments = [], []
-        detected, method = language or "en", "unknown"
-
-        for i, (chunk_path, offset) in enumerate(chunks):
-            try:
-                text, lang, m = self._transcribe_single(chunk_path, language, skip_preprocess=True)
-                all_texts.append(text.strip())
-                detected, method = lang, m
-                for seg in self._last_segments:
-                    all_segments.append({
-                        'word': seg['word'],
-                        'start': round(seg['start'] + offset, 3),
-                        'end': round(seg['end'] + offset, 3),
-                    })
-            except Exception as e:
-                logger.warning(f"Chunk {i+1} failed: {e}")
-
-        shutil.rmtree(tmp_dir, ignore_errors=True)
-        self._last_segments = all_segments
-        return " ".join(t for t in all_texts if t), detected, f"{method} (chunked)"
-
-    def _transcribe_single(self, audio_path, language, skip_preprocess=False):
-        preprocessed, temp_created = audio_path, False
-        vad_path, vad_created = audio_path, False
-
-        try:
-            # 1. Standardize format
-            if not skip_preprocess:
-                preprocessed, temp_created = self._preprocess_for_whisper(audio_path)
-                
-            # 2. UPGRADE: Apply Silero VAD to kill hallucinations
-            vad_path, vad_created = self._apply_silero_vad(preprocessed)
-
-            # 3. Transcribe clean speech
-            if self._groq_client is not None:
-                try: return self._transcribe_groq(vad_path, language)
-                except Exception as e:
-                    logger.warning(f"Groq failed ({e}), falling back to local")
-                    if self._local_model is None: self._init_local()
-            return self._transcribe_local(vad_path, language)
-            
-        finally:
-            for path, created in [(preprocessed, temp_created), (vad_path, vad_created)]:
-                if created and path != audio_path:
-                    try: os.unlink(path)
-                    except: pass
-
-    # ══════════════════════════════════════════════════════════════════
-    # UPGRADE 2: MILITARY-GRADE VAD (SILERO)
-    # ══════════════════════════════════════════════════════════════════
-    def _apply_silero_vad(self, audio_path):
-        """Isolates pure speech blocks so Whisper never hears static/silence."""
-        try:
-            import torch
-            import torchaudio
-            
-            # Load model directly from Torch Hub cache
-            model, utils = torch.hub.load(repo_or_dir='snakers4/silero-vad',
-                                          model='silero_vad', force_reload=False, trust_repo=True)
-            
-            get_speech_timestamps, save_audio, read_audio, _, _ = utils
-            wav = read_audio(audio_path, sampling_rate=16000)
-            
-            # Extract timestamps of undeniable human speech
-            speech_timestamps = get_speech_timestamps(wav, model, sampling_rate=16000, min_silence_duration_ms=250)
-            
-            if not speech_timestamps:
-                return audio_path, False 
-                
-            # Stitch chunks together with a tiny 100ms pad to prevent words from blending
-            chunks = []
-            pad = torch.zeros(int(16000 * 0.1))
-            for ts in speech_timestamps:
-                chunks.append(wav[ts['start']:ts['end']])
-                chunks.append(pad)
-                
-            final_wav = torch.cat(chunks).unsqueeze(0)
-            
-            fd, out_path = tempfile.mkstemp(suffix="_silero_cleaned.wav")
-            os.close(fd)
-            save_audio(out_path, final_wav, sampling_rate=16000)
-            
-            print("[Transcriber] ✅ Silero VAD applied. Hallucination risk neutralized.")
-            return out_path, True
-            
-        except Exception as e:
-            logger.warning(f"[Transcriber] Silero VAD failed: {e}. Passing raw audio to Whisper.")
-            return audio_path, False
-
-    def _preprocess_for_whisper(self, audio_path: str):
-        try:
-            fd, out_path = tempfile.mkstemp(suffix="_16k.wav")
-            os.close(fd)
-            if subprocess.run([
-                "ffmpeg", "-y", "-i", audio_path, "-ar", "16000", "-ac", "1", "-acodec", "pcm_s16le", out_path
-            ], capture_output=True).returncode == 0:
-                return out_path, True
-            return audio_path, False
-        except: return audio_path, False
-
-    def _init_groq(self):
-        try:
-            from groq import Groq
-            self._groq_client = Groq(api_key=self.groq_key)
-        except: self._groq_client = None
-
-    def _transcribe_groq(self, audio_path, language=None):
-        if language and language not in GROQ_SUPPORTED_LANGS:
-            if language not in FORCE_LANGUAGE_HINT: language = None
-        t0, resp, last_exc = time.time(), None, None
-
-        for attempt in range(1, MAX_RETRIES + 1):
-            try:
-                with open(audio_path, "rb") as f:
-                    kwargs = dict(file=f, model="whisper-large-v3", response_format="verbose_json", timestamp_granularities=["word"], temperature=0.0)
-                    if language: kwargs["language"] = language
-                    resp = self._groq_client.audio.transcriptions.create(**kwargs)
-                break
-            except Exception as e:
-                last_exc = e
-                if "429" in str(e).lower() or "rate" in str(e).lower(): time.sleep(2 ** attempt)
-                else: raise  
-
-        if resp is None: raise RuntimeError(f"Groq failed: {last_exc}")
-
-        transcript    = resp.text.strip()
-        detected_lang = self._norm(getattr(resp, "language", language or "en") or "en")
-        
-        self._last_segments = [{'word': getattr(w, 'word', str(w)).strip(), 'start': float(getattr(w, 'start', 0.0)), 'end': float(getattr(w, 'end', 0.0))} for w in (getattr(resp, "words", []) or [])]
-        print(f"[Transcriber] ✅ Groq done in {time.time()-t0:.2f}s")
-        return transcript, detected_lang, "Groq Whisper large-v3"
-
-    def _init_local(self):
-        try:
-            from faster_whisper import WhisperModel
-            self._local_model = WhisperModel("large-v3", device="cpu", compute_type="int8")
-        except: self._local_model = None
-
-    def _transcribe_local(self, audio_path, language=None):
-        t0 = time.time()
-        if self._local_model is None: self._init_local()
-        
-        segments, info = self._local_model.transcribe(
-            audio_path, language=language, beam_size=5, word_timestamps=True,
-            condition_on_previous_text=False, temperature=[0.0, 0.2, 0.4, 0.6]
-        )
-
-        all_words, text_parts = [], []
-        for seg in segments:
-            text_parts.append(seg.text.strip())
-            if seg.words:
-                all_words.extend([{'word': w.word.strip(), 'start': round(w.start, 3), 'end': round(w.end, 3)} for w in seg.words])
-
-        self._last_segments = all_words
-        print(f"[Transcriber] ✅ Local Whisper done in {time.time()-t0:.2f}s")
-        return " ".join(text_parts).strip(), info.language or language or "en", "faster-whisper large-v3 int8"
-
-    def _get_duration(self, audio_path):
-        try: return float(subprocess.run(["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", audio_path], capture_output=True, text=True).stdout.strip())
-        except: return 0.0
-
-    @staticmethod
-    def _norm(raw):
-        m = {"english":"en","telugu":"te","hindi":"hi","tamil":"ta","kannada":"kn"}
-        return m.get(raw.lower(), raw[:2].lower() if len(raw) >= 2 else raw)
+        self.api_key = os.environ.get("GROQ_API_KEY", "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")
+        self.client = Groq(api_key=self.api_key)
+        print("[Transcriber] Groq API Loaded — Whisper-v3")
+
+    def transcribe(self, audio_path: str) -> dict:
+        with open(audio_path, "rb") as file:
+            transcription = self.client.audio.transcriptions.create(
+                file=(audio_path, file.read()),
+                model="whisper-large-v3",
+                response_format="verbose_json",
+            )
+        return {
+            "text": transcription.text,
+            "segments": transcription.segments
+        }