aibchecker/whisper.cpp

1import os2import shutil3import tempfile4from typing import Any, Dict, List, Optional5 6from fastapi import (  # type: ignore[import]7    FastAPI,8    File,9    UploadFile,10    HTTPException,11)12from fastapi.responses import JSONResponse  # type: ignore[import]13 14try:15    # Import lazily; model will be instantiated on first use16    from faster_whisper import WhisperModel  # type: ignore17except Exception as import_error:  # pragma: no cover18    WhisperModel = None  # type: ignore19    _import_error: Optional[Exception] = import_error20else:21    _import_error = None22 23 24app = FastAPI(title="faster-whisper REST API")25 26 27def _get_bool_env(name: str, default: bool = False) -> bool:28    val = os.getenv(name)29    if val is None:30        return default31    return val.strip().lower() in {"1", "true", "yes", "y", "on"}32 33 34class _ModelHolder:35    _model: Optional[WhisperModel] = None  # type: ignore36 37    @classmethod38    def get_model(cls) -> "WhisperModel":  # type: ignore39        if _import_error is not None:40            raise RuntimeError(41                f"Failed to import faster-whisper: {_import_error}"42            )43 44        if cls._model is None:45            model_size = os.getenv("WHISPER_MODEL", "small")46            device = os.getenv("WHISPER_DEVICE", "cpu")47            compute_type = os.getenv("WHISPER_COMPUTE_TYPE", "int8")48            # Construct model once and reuse across requests49            cls._model = WhisperModel(50                model_size,51                device=device,52                compute_type=compute_type,53            )  # type: ignore54        return cls._model55 56 57@app.get("/")58def root() -> Dict[str, Any]:59    return {"status": "ok"}60 61 62@app.get("/health")63def health() -> Dict[str, Any]:64    return {"status": "ok"}65 66 67@app.post("/inference")68def transcribe(69    file: UploadFile = File(...),70    language: Optional[str] = None,71    beam_size: Optional[int] = None,72    vad: Optional[bool] = None,73    word_timestamps: Optional[bool] = None,74) -> JSONResponse:75    try:76        model = _ModelHolder.get_model()77    except Exception as e:  # pragma: no cover78        raise HTTPException(status_code=500, detail=str(e))79 80    # Fallback to env-configurable defaults81    effective_beam_size = (82        int(os.getenv("WHISPER_BEAM_SIZE", "5"))83        if beam_size is None84        else beam_size85    )86    effective_vad = (87        _get_bool_env("WHISPER_VAD", False)88        if vad is None89        else vad90    )91    effective_word_ts = (92        _get_bool_env("WHISPER_WORD_TIMESTAMPS", False)93        if word_timestamps is None94        else word_timestamps95    )96 97    # Persist upload to a temporary file so ffmpeg/backends can access it98    try:99        with tempfile.NamedTemporaryFile(100            delete=False,101            suffix=os.path.splitext(file.filename or "audio")[1],102        ) as tmp:103            shutil.copyfileobj(file.file, tmp)104            tmp_path = tmp.name105    except Exception as e:106        raise HTTPException(107            status_code=400,108            detail=f"Failed to read uploaded file: {e}",109        )110    finally:111        try:112            file.file.close()113        except Exception:114            pass115    try:116        segments_iter, info = model.transcribe(117            tmp_path,118            beam_size=effective_beam_size,119            language=language,120            vad_filter=effective_vad,121            word_timestamps=effective_word_ts,122        )123 124        segments = list(segments_iter)125 126        # Build response127        all_text = "".join(seg.text for seg in segments)128 129        segments_json: List[Dict[str, Any]] = []130        for seg in segments:131            seg_obj: Dict[str, Any] = {132                "id": seg.id,133                "start": seg.start,134                "end": seg.end,135                "text": seg.text,136                "avg_log_prob": seg.avg_logprob,137                "compression_ratio": seg.compression_ratio,138                "no_speech_prob": seg.no_speech_prob,139            }140            if effective_word_ts and getattr(seg, "words", None):141                seg_obj["words"] = [142                    {143                        "start": w.start,144                        "end": w.end,145                        "word": w.word,146                        "probability": getattr(w, "probability", None),147                    }148                    for w in seg.words149                ]150            segments_json.append(seg_obj)151 152        result: Dict[str, Any] = {153            "text": all_text.strip(),154            "segments": segments_json,155            "info": {156                "language": getattr(info, "language", None),157                "language_probability": getattr(158                    info, "language_probability", None159                ),160                "duration": getattr(info, "duration", None),161            },162        }163        return JSONResponse(result)164    except Exception as e:165        raise HTTPException(166            status_code=500,167            detail=f"Transcription error: {e}",168        )169    finally:170        try:171            os.unlink(tmp_path)172        except Exception:173            pass174