ai-podcast/backend/services/transcription.py

"""Whisper transcription service"""

import tempfile
import numpy as np
from faster_whisper import WhisperModel
import librosa

WHISPER_MODEL = "distil-large-v3"

# Global model instance (loaded once)
_whisper_model = None


def get_whisper_model() -> WhisperModel:
    """Get or create Whisper model instance"""
    global _whisper_model
    if _whisper_model is None:
        print(f"Loading Whisper {WHISPER_MODEL} model...")
        _whisper_model = WhisperModel(WHISPER_MODEL, device="cpu", compute_type="int8")
        print("Whisper model loaded")
    return _whisper_model


def decode_audio(audio_data: bytes, source_sample_rate: int = None) -> tuple[np.ndarray, int]:
    """
    Decode audio from various formats to numpy array.

    Args:
        audio_data: Raw audio bytes
        source_sample_rate: If provided, treat as raw PCM at this sample rate

    Returns:
        Tuple of (audio array as float32, sample rate)
    """
    # If sample rate is provided, assume raw PCM (from server-side recording)
    if source_sample_rate is not None:
        print(f"Decoding raw PCM at {source_sample_rate}Hz, {len(audio_data)} bytes")
        if len(audio_data) % 2 != 0:
            audio_data = audio_data + b'\x00'
        audio = np.frombuffer(audio_data, dtype=np.int16).astype(np.float32) / 32768.0
        return audio, source_sample_rate

    print(f"First 20 bytes: {audio_data[:20].hex()}")

    # Try to decode with librosa first (handles webm, ogg, wav, mp3, etc via ffmpeg)
    try:
        with tempfile.NamedTemporaryFile(suffix='.webm', delete=False) as f:
            f.write(audio_data)
            temp_path = f.name

        audio, sample_rate = librosa.load(temp_path, sr=None, mono=True)
        print(f"Decoded with librosa: {len(audio)} samples at {sample_rate}Hz")

        import os
        os.unlink(temp_path)

        return audio.astype(np.float32), sample_rate

    except Exception as e:
        print(f"librosa decode failed: {e}, trying raw PCM at 16kHz...")

        # Fall back to raw PCM (16-bit signed int, 16kHz mono - Whisper's rate)
        if len(audio_data) % 2 != 0:
            audio_data = audio_data + b'\x00'

        audio = np.frombuffer(audio_data, dtype=np.int16).astype(np.float32) / 32768.0
        return audio, 16000


async def transcribe_audio(audio_data: bytes, source_sample_rate: int = None,
                           context_hint: str = "") -> str:
    """
    Transcribe audio data to text using Whisper.

    Args:
        audio_data: Audio bytes (webm, ogg, wav, or raw PCM)
        source_sample_rate: If provided, treat audio_data as raw PCM at this rate
        context_hint: Optional extra context for the initial prompt (e.g. caller name/topic)

    Returns:
        Transcribed text
    """
    model = get_whisper_model()

    print(f"Transcribing audio: {len(audio_data)} bytes")

    # Decode audio from whatever format
    audio, detected_sample_rate = decode_audio(audio_data, source_sample_rate)

    print(f"Audio samples: {len(audio)}, duration: {len(audio)/detected_sample_rate:.2f}s")
    print(f"Audio range: min={audio.min():.4f}, max={audio.max():.4f}")

    # Check if audio is too quiet
    if np.abs(audio).max() < 0.01:
        print("Warning: Audio appears to be silent or very quiet")
        return ""

    # Resample to 16kHz for Whisper
    if detected_sample_rate != 16000:
        audio_16k = librosa.resample(audio, orig_sr=detected_sample_rate, target_sr=16000)
        print(f"Resampled to {len(audio_16k)} samples at 16kHz")
    else:
        audio_16k = audio

    # Build initial prompt — context helps Whisper with names and topic-specific words
    initial_prompt = "Luke at the Roost, a late-night radio talk show in Alpine, Texas, in the Big Bend region. The host Luke talks to callers about life, relationships, sports, politics, and pop culture, with his intern Devon. Callers reference Alpine, Marfa, Marathon, Terlingua, Fort Stockton, and Big Bend."
    if context_hint:
        initial_prompt += f" {context_hint}"

    # Transcribe
    segments, info = model.transcribe(
        audio_16k,
        beam_size=5,
        language="en",
        vad_filter=True,
        initial_prompt=initial_prompt,
    )
    segments_list = list(segments)
    text = " ".join([s.text for s in segments_list]).strip()

    print(f"Transcription result: '{text}' (language: {info.language}, prob: {info.language_probability:.2f})")

    return text