215 lines
8.0 KiB
Python
215 lines
8.0 KiB
Python
import asyncio
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
|
|
import httpx
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.api.dependencies import orchestrator
|
|
from app.config import get_settings
|
|
from app.domain.schemas import TranscriptChunk, TranscriptSegment, TranscriptionJobRead
|
|
from app.infrastructure.database import SessionLocal
|
|
|
|
BASE_URL = "https://api.soniox.com/v1"
|
|
|
|
|
|
class SonioxError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def _message_from_response(response: httpx.Response, fallback: str) -> str:
|
|
try:
|
|
body = response.json()
|
|
return str(body.get("message") or body.get("error_message") or fallback)
|
|
except Exception:
|
|
return fallback
|
|
|
|
|
|
def _token_from_json(raw: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"text": raw.get("text") or "",
|
|
"start_ms": int(raw.get("start_ms") or 0),
|
|
"end_ms": int(raw.get("end_ms") or 0),
|
|
"confidence": float(raw.get("confidence") or 0),
|
|
"speaker": str(raw.get("speaker") or "unknown"),
|
|
}
|
|
|
|
|
|
def _ends_sentence(text: str) -> bool:
|
|
return text.rstrip().endswith((".", "?", "!"))
|
|
|
|
|
|
def group_by_speaker(tokens: list[dict[str, Any]]) -> list[TranscriptSegment]:
|
|
segments: list[TranscriptSegment] = []
|
|
active_speaker: str | None = None
|
|
buffer = ""
|
|
start_ms = 0
|
|
end_ms = 0
|
|
|
|
def flush() -> None:
|
|
nonlocal buffer
|
|
text = buffer.strip()
|
|
if active_speaker is None or not text:
|
|
buffer = ""
|
|
return
|
|
segments.append(
|
|
TranscriptSegment(
|
|
id=str(uuid4()),
|
|
speaker=active_speaker,
|
|
original_text=text,
|
|
edited_text=text,
|
|
start_ms=start_ms,
|
|
end_ms=end_ms,
|
|
)
|
|
)
|
|
buffer = ""
|
|
|
|
for token in tokens:
|
|
speaker = token["speaker"] or "unknown"
|
|
if active_speaker is not None and speaker != active_speaker:
|
|
flush()
|
|
if active_speaker != speaker:
|
|
active_speaker = speaker
|
|
start_ms = token["start_ms"]
|
|
buffer += token["text"]
|
|
end_ms = token["end_ms"]
|
|
if _ends_sentence(token["text"]):
|
|
flush()
|
|
active_speaker = None
|
|
|
|
flush()
|
|
return segments
|
|
|
|
|
|
class TranscriptionJobs:
|
|
"""In-memory Soniox job store for the demo runtime."""
|
|
|
|
def __init__(self) -> None:
|
|
self._jobs: dict[str, TranscriptionJobRead] = {}
|
|
|
|
def create(self) -> TranscriptionJobRead:
|
|
job = TranscriptionJobRead(
|
|
job_id=str(uuid4()),
|
|
status="uploading",
|
|
created_at=datetime.now(timezone.utc),
|
|
)
|
|
self._jobs[job.job_id] = job
|
|
return job
|
|
|
|
def get(self, job_id: str) -> TranscriptionJobRead | None:
|
|
return self._jobs.get(job_id)
|
|
|
|
def update(self, job_id: str, **changes: Any) -> TranscriptionJobRead:
|
|
job = self._jobs[job_id].model_copy(update=changes)
|
|
self._jobs[job_id] = job
|
|
return job
|
|
|
|
|
|
jobs = TranscriptionJobs()
|
|
|
|
|
|
class SonioxTranscriptionService:
|
|
def __init__(self) -> None:
|
|
self.settings = get_settings()
|
|
|
|
async def process_upload(self, job_id: str, audio: bytes, filename: str, request_like: Any) -> None:
|
|
try:
|
|
if not self.settings.soniox_api_key:
|
|
raise SonioxError("SONIOX_API_KEY is not configured.")
|
|
file_id = await self._upload_file(audio, filename)
|
|
jobs.update(job_id, status="queued")
|
|
transcription_id = await self._create_transcription(file_id)
|
|
jobs.update(job_id, status="processing", transcription_id=transcription_id)
|
|
await self._wait_until_complete(transcription_id, job_id)
|
|
segments = await self._fetch_segments(transcription_id)
|
|
action_ids = await self._ingest_segments(segments, request_like)
|
|
jobs.update(job_id, status="completed", segments=segments, generated_action_ids=action_ids)
|
|
except Exception as exc:
|
|
jobs.update(job_id, status="error", error=str(exc))
|
|
|
|
async def _upload_file(self, audio: bytes, filename: str) -> str:
|
|
async with httpx.AsyncClient(timeout=60) as client:
|
|
response = await client.post(
|
|
f"{BASE_URL}/files",
|
|
headers={"Authorization": f"Bearer {self.settings.soniox_api_key}"},
|
|
files={"file": (filename, audio, "application/octet-stream")},
|
|
)
|
|
if not response.is_success:
|
|
raise SonioxError(_message_from_response(response, "Soniox file upload failed."))
|
|
data = response.json()
|
|
file_id = str(data.get("id") or data.get("file_id") or "")
|
|
if not file_id:
|
|
raise SonioxError("Soniox did not return a file id.")
|
|
return file_id
|
|
|
|
async def _create_transcription(self, file_id: str) -> str:
|
|
async with httpx.AsyncClient(timeout=30) as client:
|
|
response = await client.post(
|
|
f"{BASE_URL}/transcriptions",
|
|
headers={
|
|
"Authorization": f"Bearer {self.settings.soniox_api_key}",
|
|
"Content-Type": "application/json",
|
|
},
|
|
content=json.dumps(
|
|
{
|
|
"model": self.settings.soniox_model,
|
|
"file_id": file_id,
|
|
"enable_speaker_diarization": True,
|
|
}
|
|
),
|
|
)
|
|
if response.status_code != 201:
|
|
raise SonioxError(_message_from_response(response, "Soniox transcription creation failed."))
|
|
return str(response.json()["id"])
|
|
|
|
async def _wait_until_complete(self, transcription_id: str, job_id: str) -> None:
|
|
async with httpx.AsyncClient(timeout=30) as client:
|
|
for _ in range(self.settings.soniox_poll_attempts):
|
|
response = await client.get(
|
|
f"{BASE_URL}/transcriptions/{transcription_id}",
|
|
headers={"Authorization": f"Bearer {self.settings.soniox_api_key}"},
|
|
)
|
|
if response.status_code != 200:
|
|
raise SonioxError(_message_from_response(response, "Unable to read Soniox status."))
|
|
data = response.json()
|
|
status = data.get("status") or "processing"
|
|
if status == "completed":
|
|
return
|
|
if status == "error":
|
|
raise SonioxError(str(data.get("error_message") or "Soniox transcription failed."))
|
|
jobs.update(job_id, status="processing")
|
|
await asyncio.sleep(self.settings.soniox_poll_interval_seconds)
|
|
raise SonioxError("Timed out waiting for Soniox transcription.")
|
|
|
|
async def _fetch_segments(self, transcription_id: str) -> list[TranscriptSegment]:
|
|
async with httpx.AsyncClient(timeout=30) as client:
|
|
response = await client.get(
|
|
f"{BASE_URL}/transcriptions/{transcription_id}/transcript",
|
|
headers={"Authorization": f"Bearer {self.settings.soniox_api_key}"},
|
|
)
|
|
if response.status_code != 200:
|
|
raise SonioxError(_message_from_response(response, "Unable to download Soniox transcript."))
|
|
tokens = [_token_from_json(item) for item in response.json().get("tokens", []) if isinstance(item, dict)]
|
|
return group_by_speaker(tokens)
|
|
|
|
async def _ingest_segments(self, segments: list[TranscriptSegment], request_like: Any) -> list[int]:
|
|
session: Session = SessionLocal()
|
|
action_ids: list[int] = []
|
|
try:
|
|
for segment in segments:
|
|
action = await orchestrator(request_like).ingest(
|
|
session,
|
|
TranscriptChunk(
|
|
speaker=segment.speaker,
|
|
timestamp=datetime.now(timezone.utc),
|
|
text=segment.edited_text,
|
|
),
|
|
)
|
|
if action:
|
|
action_ids.append(action.id)
|
|
finally:
|
|
session.close()
|
|
return action_ids
|