| """Engine: аудио → текст на базе Whisper base (CTranslate2). |
| |
| Модель не загружает torch — инференс через ctranslate2 + faster-whisper. |
| """ |
|
|
| from __future__ import annotations |
|
|
| from pathlib import Path |
| from typing import Any, Dict, List, Optional, Union |
|
|
| from faster_whisper import WhisperModel |
|
|
| DEFAULT_MODEL_DIR = Path(__file__).resolve().parent.parent / "model" |
|
|
|
|
| class AuditEngine: |
| """Расшифровщик речи в текст (ASR, 99 языков).""" |
|
|
| def __init__( |
| self, |
| model_dir: Optional[Union[str, Path]] = None, |
| device: str = "cpu", |
| compute_type: str = "int8", |
| cpu_threads: int = 0, |
| ) -> None: |
| model_dir = Path(model_dir or DEFAULT_MODEL_DIR) |
| if not (model_dir / "model.bin").exists(): |
| raise FileNotFoundError( |
| f"Веса модели не найдены: {model_dir}/model.bin\n" |
| "Запустите сначала: python scripts/build_model.py" |
| ) |
| self.model_dir = model_dir |
| self.model = WhisperModel( |
| str(model_dir), |
| device=device, |
| compute_type=compute_type, |
| cpu_threads=cpu_threads, |
| ) |
|
|
| def transcribe( |
| self, |
| audio: Union[str, Path], |
| language: Optional[str] = None, |
| beam_size: int = 5, |
| vad_filter: bool = False, |
| ) -> Dict[str, Any]: |
| """Транскрибировать аудио. Возвращает dict с текстом и метаданными.""" |
| segments, info = self.model.transcribe( |
| str(audio), |
| language=language, |
| beam_size=beam_size, |
| vad_filter=vad_filter, |
| ) |
| parts: List[Dict[str, Any]] = [] |
| for seg in segments: |
| parts.append( |
| { |
| "start": round(float(seg.start), 2), |
| "end": round(float(seg.end), 2), |
| "text": seg.text.strip(), |
| } |
| ) |
| return { |
| "text": " ".join(p["text"] for p in parts).strip(), |
| "language": info.language, |
| "language_probability": round(float(info.language_probability), 4), |
| "duration": round(float(info.duration), 2), |
| "segments": parts, |
| } |
|
|
