JiriUhlir
91 lines
3.2 KiB
Python
91 lines
3.2 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.
|
|
|
|
Když je zapnutá diarizace, vrátí přepis se štítky mluvčích (z `utterances`),
|
|
např. `Mluvčí 0: ...`. Bez diarizace vrátí plochý transkript.
|
|
"""
|
|
params = {
|
|
"model": model,
|
|
"language": language,
|
|
"diarize": "true" if diarize else "false",
|
|
"smart_format": "true" if smart_format else "false",
|
|
# utterances rozseká přepis na promluvy s přiřazeným mluvčím — nutné, aby
|
|
# v textu bylo vidět rozlišení mluvčích (samotný `transcript` je plochý).
|
|
"utterances": "true" if diarize 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:
|
|
results = data["results"]
|
|
# 1) Preferuj diarizovaný výstup z utterances (se štítky mluvčích).
|
|
diarized = _format_utterances(results.get("utterances"))
|
|
if diarized:
|
|
return diarized
|
|
# 2) Fallback na plochý transkript.
|
|
alternative = results["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
|
|
|
|
|
|
def _format_utterances(utterances) -> str:
|
|
"""Sloučí Deepgram utterances do přepisu se štítky mluvčích (`Mluvčí N: ...`).
|
|
|
|
Po sobě jdoucí promluvy stejného mluvčího spojí do jednoho řádku.
|
|
"""
|
|
if not utterances:
|
|
return ""
|
|
lines: list[str] = []
|
|
prev_speaker = None
|
|
for utt in utterances:
|
|
text = (utt.get("transcript") or "").strip()
|
|
if not text:
|
|
continue
|
|
speaker = utt.get("speaker")
|
|
if speaker != prev_speaker:
|
|
lines.append(f"Mluvčí {speaker}: {text}")
|
|
prev_speaker = speaker
|
|
else:
|
|
lines[-1] += " " + text
|
|
return "\n".join(lines)
|