121 lines
3.9 KiB
Python
121 lines
3.9 KiB
Python
"""
|
|
WhisperService — распознавание речи через OpenAI-совместимый API.
|
|
|
|
Принимает аудио-байты и MIME-тип, возвращает распознанный текст.
|
|
"""
|
|
|
|
import logging
|
|
import tempfile
|
|
from typing import Optional
|
|
|
|
import requests
|
|
|
|
from config import WHISPER_URL, WHISPER_KEY, WHISPER_MODEL
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
class WhisperService:
|
|
"""
|
|
Сервис транскрибации аудио в текст.
|
|
|
|
Использует OpenAI-совместимый эндпоинт /v1/audio/transcriptions.
|
|
Поддерживает форматы: webm, wav, mp3, ogg, flac.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
api_url: str = WHISPER_URL,
|
|
api_key: str = WHISPER_KEY,
|
|
model: str = WHISPER_MODEL,
|
|
):
|
|
"""
|
|
Args:
|
|
api_url: URL эндпоинта транскрибации.
|
|
api_key: API-ключ (Bearer-токен).
|
|
model: Название модели Whisper.
|
|
"""
|
|
self._url = api_url
|
|
self._key = api_key
|
|
self._model = model
|
|
|
|
def transcribe(
|
|
self,
|
|
audio_bytes: bytes,
|
|
mime_type: str = "audio/webm",
|
|
language: str = "ru",
|
|
) -> str:
|
|
"""
|
|
Отправляет аудио на распознавание и возвращает текст.
|
|
|
|
Args:
|
|
audio_bytes: Сырые байты аудиофайла.
|
|
mime_type: MIME-тип аудио (audio/webm, audio/wav, ...).
|
|
language: Язык распознавания (ISO 639-1).
|
|
|
|
Returns:
|
|
Распознанный текст (может быть пустой строкой).
|
|
|
|
Raises:
|
|
RuntimeError: При ошибке API или отсутствии ключа.
|
|
"""
|
|
if not self._key:
|
|
raise RuntimeError("API-ключ для Whisper не задан (WHISPER_KEY)")
|
|
|
|
# Определяем расширение файла по MIME-типу
|
|
suffix = self._mime_to_suffix(mime_type)
|
|
|
|
# Пишем аудио во временный файл (requests требует file-like object)
|
|
with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp:
|
|
tmp.write(audio_bytes)
|
|
tmp.flush()
|
|
tmp.seek(0)
|
|
|
|
log.info(
|
|
"Whisper: отправка %d байт (тип=%s, модель=%s)",
|
|
len(audio_bytes),
|
|
mime_type,
|
|
self._model,
|
|
)
|
|
|
|
resp = requests.post(
|
|
self._url,
|
|
headers={"Authorization": f"Bearer {self._key}"},
|
|
files={"file": (f"audio{suffix}", tmp, mime_type)},
|
|
data={
|
|
"model": self._model,
|
|
"language": language,
|
|
},
|
|
timeout=30,
|
|
)
|
|
|
|
if resp.status_code != 200:
|
|
log.error("Whisper API error %d: %s", resp.status_code, resp.text)
|
|
raise RuntimeError(
|
|
f"Ошибка распознавания ({resp.status_code}): {resp.text[:200]}"
|
|
)
|
|
|
|
result = resp.json()
|
|
text = result.get("text", "").strip()
|
|
|
|
log.info("Whisper: распознано %d символов", len(text))
|
|
return text
|
|
|
|
@staticmethod
|
|
def _mime_to_suffix(mime_type: str) -> str:
|
|
"""Сопоставляет MIME-тип с расширением файла."""
|
|
mapping = {
|
|
"audio/webm": ".webm",
|
|
"audio/wav": ".wav",
|
|
"audio/wave": ".wav",
|
|
"audio/mp3": ".mp3",
|
|
"audio/mpeg": ".mp3",
|
|
"audio/ogg": ".ogg",
|
|
"audio/flac": ".flac",
|
|
}
|
|
return mapping.get(mime_type, ".webm")
|
|
|
|
|
|
# Глобальный экземпляр сервиса (переиспользуется между запросами)
|
|
whisper_service = WhisperService()
|