53 lines
1.9 KiB
Python
53 lines
1.9 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
|
|
from backend.app.config import settings
|
|
|
|
|
|
class DeepgramClient:
|
|
endpoint = "https://api.deepgram.com/v1/listen?model=nova-3&smart_format=true&language=es"
|
|
|
|
def transcribe_file(self, path: Path) -> dict:
|
|
if not settings.deepgram_api_key:
|
|
raise RuntimeError("DEEPGRAM_API_KEY no esta configurada")
|
|
|
|
headers = {
|
|
"Authorization": f"Token {settings.deepgram_api_key}",
|
|
"Content-Type": self._content_type(path),
|
|
}
|
|
with path.open("rb") as audio_file:
|
|
response = requests.post(self.endpoint, headers=headers, data=audio_file, timeout=600)
|
|
|
|
if response.status_code == 401:
|
|
raise RuntimeError("API Key de Deepgram invalida")
|
|
if not response.ok:
|
|
try:
|
|
error = response.json()
|
|
except ValueError:
|
|
error = response.text
|
|
raise RuntimeError(f"Deepgram error {response.status_code}: {error}")
|
|
return response.json()
|
|
|
|
def transcript_from_response(self, data: dict) -> tuple[str, float | None]:
|
|
alternative = data.get("results", {}).get("channels", [{}])[0].get("alternatives", [{}])[0]
|
|
transcript = alternative.get("paragraphs", {}).get("transcript") or alternative.get("transcript") or ""
|
|
confidence = alternative.get("confidence")
|
|
if not transcript.strip():
|
|
raise RuntimeError("Deepgram no devolvio transcripcion")
|
|
return transcript.strip(), confidence
|
|
|
|
def _content_type(self, path: Path) -> str:
|
|
ext = path.suffix.lower()
|
|
return {
|
|
".mp3": "audio/mpeg",
|
|
".wav": "audio/wav",
|
|
".m4a": "audio/mp4",
|
|
".mp4": "audio/mp4",
|
|
".webm": "audio/webm",
|
|
".ogg": "audio/ogg",
|
|
".flac": "audio/flac",
|
|
}.get(ext, "application/octet-stream")
|