python_backend/app/services/whisper_local.py

from __future__ import annotations

import asyncio
import tempfile
from pathlib import Path

from faster_whisper import WhisperModel

from ..config import settings

_model: WhisperModel | None = None


def _get_model() -> WhisperModel:
    global _model
    if _model is None:
        _model = WhisperModel(
            settings.whisper_model,
            device=settings.whisper_device,
            compute_type=settings.whisper_compute_type,
        )
    return _model


def _transcribe_file(file_path: str) -> str:
    model = _get_model()
    segments, _info = model.transcribe(file_path, beam_size=1, vad_filter=True)
    return " ".join(segment.text.strip() for segment in segments).strip()


async def transcribe_audio_bytes(audio_bytes: bytes, suffix: str = ".ogg") -> str:
    with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as temp_file:
        temp_file.write(audio_bytes)
        temp_path = temp_file.name

    try:
        return await asyncio.to_thread(_transcribe_file, temp_path)
    finally:
        Path(temp_path).unlink(missing_ok=True)