Files
audio-transcription/app/clients/deepgram_client.py
T
2026-07-10 09:54:05 +02:00

57 lines
2.0 KiB
Python

"""Klient pro Deepgram pre-recorded přepis (REST, přes httpx).
Nepoužíváme těžké SDK — voláme REST endpoint přímo, aby nebyl závislý na verzích.
"""
import httpx
from ..config import DEEPGRAM_BASE_URL, UPSTREAM_TIMEOUT_SECONDS
from ..errors import UpstreamError, raise_for_upstream
from ..logging_config import get_logger
log = get_logger("audio-transcription.deepgram")
async def transcribe(
*,
api_key: str,
audio: bytes,
content_type: str,
model: str,
language: str,
diarize: bool,
smart_format: bool,
) -> str:
"""Přepíše audio Deepgramem a vrátí prostý transkript (channels[0].alternatives[0].transcript)."""
params = {
"model": model,
"language": language,
"diarize": "true" if diarize else "false",
"smart_format": "true" if smart_format else "false",
}
headers = {
"Authorization": f"Token {api_key}",
"Content-Type": content_type or "application/octet-stream",
}
url = f"{DEEPGRAM_BASE_URL}/listen"
try:
async with httpx.AsyncClient(timeout=UPSTREAM_TIMEOUT_SECONDS) as client:
resp = await client.post(url, params=params, headers=headers, content=audio)
except httpx.HTTPError as exc:
log.error("Deepgram request failed: %s", exc)
raise UpstreamError(f"Deepgram request se nezdařil: {exc}") from exc
if resp.status_code >= 400:
# Deepgram vrací chybu v JSON; nikdy nelogujeme klíč (ten je jen v hlavičce).
log.warning("Deepgram returned %s: %s", resp.status_code, resp.text[:500])
raise_for_upstream("Deepgram", resp.status_code, resp.text)
data = resp.json()
try:
channels = data["results"]["channels"]
alternative = channels[0]["alternatives"][0]
return alternative.get("transcript", "") or ""
except (KeyError, IndexError, TypeError) as exc:
log.error("Unexpected Deepgram response shape: %s", exc)
raise UpstreamError("Neočekávaná struktura odpovědi z Deepgramu.") from exc