68 lines
2.1 KiB
Python
68 lines
2.1 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import aiohttp
|
|
|
|
from app.config import ElevenLabsConfig
|
|
|
|
|
|
ELEVENLABS_SPEECH_TO_TEXT_URL = "https://api.elevenlabs.io/v1/speech-to-text"
|
|
|
|
|
|
class TranscriptionError(RuntimeError):
|
|
pass
|
|
|
|
|
|
async def transcribe_file(
|
|
path: Path,
|
|
config: ElevenLabsConfig,
|
|
*,
|
|
content_type: str | None = None,
|
|
) -> str:
|
|
if not config.api_key:
|
|
raise TranscriptionError("ElevenLabs API key is not configured")
|
|
|
|
form = aiohttp.FormData()
|
|
form.add_field("model_id", config.model_id)
|
|
form.add_field("tag_audio_events", "false")
|
|
if config.language_code:
|
|
form.add_field("language_code", config.language_code)
|
|
|
|
timeout = aiohttp.ClientTimeout(total=config.request_timeout_seconds)
|
|
headers = {"xi-api-key": config.api_key}
|
|
|
|
with path.open("rb") as media_file:
|
|
form.add_field(
|
|
"file",
|
|
media_file,
|
|
filename=path.name,
|
|
content_type=content_type or "application/octet-stream",
|
|
)
|
|
try:
|
|
async with aiohttp.ClientSession(timeout=timeout) as session:
|
|
async with session.post(
|
|
ELEVENLABS_SPEECH_TO_TEXT_URL,
|
|
data=form,
|
|
headers=headers,
|
|
) as response:
|
|
body = await response.text()
|
|
if response.status >= 400:
|
|
raise TranscriptionError(
|
|
f"ElevenLabs returned HTTP {response.status}: {body[:500]}"
|
|
)
|
|
except (aiohttp.ClientError, TimeoutError) as exc:
|
|
raise TranscriptionError("ElevenLabs request failed") from exc
|
|
|
|
try:
|
|
payload: Any = json.loads(body)
|
|
except json.JSONDecodeError as exc:
|
|
raise TranscriptionError("ElevenLabs returned invalid JSON") from exc
|
|
|
|
text = payload.get("text") if isinstance(payload, dict) else None
|
|
if not isinstance(text, str) or not text.strip():
|
|
raise TranscriptionError("ElevenLabs returned an empty transcript")
|
|
return text.strip()
|