PTSRaiAgent/app/llm/stt.py

"""Speech-to-text for the patient's voice input (#116, slide 11: record → check the text → send).

The service only transcribes: the text goes back to the patient to check and edit, and reaches the dialog as an
ordinary text message — safety, `scrub()` and the LLM gateway apply there. Audio is never stored or logged.

* `OpenRouterSTT` — an audio-capable model via OpenRouter `/chat/completions` (`input_audio`). TEMPORARY, synthetic
  data only (ADR-014): voice is personal data that `scrub()` cannot depersonalise, and OpenRouter is a foreign
  service (152-FZ). Real patients need an RF provider (SaluteSpeech, #121) behind the same `SpeechToText` contract.
* `SaluteSTT` — Sber SaluteSpeech, synchronous REST recognition of the whole file (RF contour, #121, ADR-016).
  The pilot provider: the only one allowed for real patients.
* `MockSTT` — offline, returns a fixed synthetic phrase (ADR-004: the demo works without network).
"""

import asyncio
import base64
import logging
import ssl
import struct
import time
import uuid
from functools import lru_cache
from pathlib import Path
from typing import Protocol

import httpx
import yaml
from pydantic import BaseModel

from app.config import Settings

log = logging.getLogger(__name__)

_RETRY_STATUSES = {429, 500, 502, 503, 504}
_ATTEMPTS = 3


class SttTexts(BaseModel):
    system: str
    user: str
    silence_marker: str
    mock_transcript: str


@lru_cache
def load_texts(path: Path | None = None) -> SttTexts:
    from app.config import get_settings

    path = path or get_settings().config_dir / "texts" / "stt.yaml"
    return SttTexts.model_validate(yaml.safe_load(path.read_text(encoding="utf-8")))


class SttError(RuntimeError):
    """The provider failed (network, HTTP error, bad response)."""


class SttNoSpeech(SttError):
    """The recording holds no recognisable speech."""


class InvalidAudio(ValueError):
    pass


class WavInfo(BaseModel):
    audio_format: int  # 1 = PCM
    channels: int
    sample_rate: int
    byte_rate: int
    bits: int
    data_offset: int
    data_size: int


def wav_info(audio: bytes) -> WavInfo:
    """Header of a WAV file; raises `InvalidAudio` for anything else."""
    if len(audio) < 44 or audio[:4] != b"RIFF" or audio[8:12] != b"WAVE":
        raise InvalidAudio("not a WAV file")
    pos, fmt = 12, None
    while pos + 8 <= len(audio):
        chunk_id, size = audio[pos : pos + 4], struct.unpack_from("<I", audio, pos + 4)[0]
        if chunk_id == b"fmt " and size >= 16:
            fmt = struct.unpack_from("<HHIIHH", audio, pos + 8)
        elif chunk_id == b"data":
            if not fmt or not fmt[3]:
                break
            audio_format, channels, rate, byte_rate, _, bits = fmt
            return WavInfo(
                audio_format=audio_format,
                channels=channels,
                sample_rate=rate,
                byte_rate=byte_rate,
                bits=bits,
                data_offset=pos + 8,
                data_size=min(size, len(audio) - pos - 8),
            )
        pos += 8 + size + (size & 1)
    raise InvalidAudio("malformed WAV")


def wav_seconds(audio: bytes) -> float:
    """Duration of a PCM WAV file; raises `InvalidAudio` for anything else."""
    info = wav_info(audio)
    return info.data_size / info.byte_rate


class SpeechToText(Protocol):
    name: str

    async def transcribe(self, audio: bytes, fmt: str = "wav") -> str: ...


class MockSTT:
    name = "mock"

    async def transcribe(self, audio: bytes, fmt: str = "wav") -> str:
        return load_texts().mock_transcript


class OpenRouterSTT:
    name = "openrouter"

    def __init__(self, settings: Settings, client: httpx.AsyncClient | None = None) -> None:
        self._key = settings.openrouter_api_key
        self._model = settings.stt_model
        self._url = settings.openrouter_base_url.rstrip("/") + "/chat/completions"
        self._timeout = settings.stt_timeout_s
        self._proxy = settings.openrouter_proxy_url or None
        self._client = client

    async def transcribe(self, audio: bytes, fmt: str = "wav") -> str:
        texts = load_texts()
        content = [
            {"type": "text", "text": texts.user},
            {"type": "input_audio", "input_audio": {"data": base64.b64encode(audio).decode(), "format": fmt}},
        ]
        body = {
            "model": self._model,
            "messages": [{"role": "system", "content": texts.system}, {"role": "user", "content": content}],
            "temperature": 0,
            "max_tokens": 800,
            # Transcription needs no thinking: with reasoning on, gemini-2.5-flash takes ~60 s instead of ~3 s.
            "reasoning": {"enabled": False},
        }
        data = await self._post(body)
        try:
            text = (data["choices"][0]["message"]["content"] or "").strip()
        except (KeyError, IndexError, TypeError) as exc:
            raise SttError("bad response") from exc
        text = text.strip("«»\"' \n")
        if not text or text.lower() == texts.silence_marker.lower():
            raise SttNoSpeech("no speech")
        return text

    async def _post(self, body: dict) -> dict:
        headers = {"Authorization": f"Bearer {self._key}", "X-Title": "PTSRaiAgent prototype (synthetic data)"}
        for attempt in range(_ATTEMPTS):
            try:
                if self._client is not None:
                    resp = await self._client.post(self._url, json=body, headers=headers, timeout=self._timeout)
                else:
                    async with httpx.AsyncClient(timeout=self._timeout, proxy=self._proxy) as client:
                        resp = await client.post(self._url, json=body, headers=headers)
            except httpx.HTTPError as exc:
                if attempt + 1 == _ATTEMPTS:
                    raise SttError(type(exc).__name__) from exc
                await asyncio.sleep(0.5 * 2**attempt)
                continue
            if resp.status_code in _RETRY_STATUSES and attempt + 1 < _ATTEMPTS:
                log.warning("stt openrouter http %s, retry %d", resp.status_code, attempt + 1)
                await asyncio.sleep(0.5 * 2**attempt)
                continue
            if resp.status_code != 200:
                # Provider error bodies carry no patient data — log the message to make failures diagnosable.
                log.error("stt openrouter http %s: %s", resp.status_code, resp.text[:200])
                raise SttError(f"http {resp.status_code}")
            try:
                return resp.json()
            except ValueError as exc:
                raise SttError("bad json") from exc
        raise SttError("retries exhausted")


SALUTE_OAUTH_URL = "https://ngw.devices.sberbank.ru:9443/api/v2/oauth"
SALUTE_RECOGNIZE_URL = "https://smartspeech.sber.ru/rest/v1/speech:recognize"


class _SaluteToken:
    """OAuth access token, cached until a minute before expiry (Sber tokens live 30 min) and shared per process:
    a new token per request would burn the OAuth rate limit."""

    def __init__(self) -> None:
        self.value = ""
        self.expires = 0.0  # time.monotonic() deadline
        self.lock = asyncio.Lock()


@lru_cache(maxsize=4)
def _salute_token(credentials: str, scope: str) -> _SaluteToken:
    return _SaluteToken()


class SaluteSTT:
    name = "salute"

    def __init__(self, settings: Settings, client: httpx.AsyncClient | None = None) -> None:
        self._credentials = settings.salute_speech_credentials.get_secret_value()
        self._scope = settings.salute_speech_scope
        self._timeout = settings.stt_timeout_s
        self._ca = settings.salute_speech_ca
        self._client = client
        self._token = _salute_token(self._credentials, self._scope)

    async def transcribe(self, audio: bytes, fmt: str = "wav") -> str:
        info = wav_info(audio)
        if info.audio_format != 1 or info.bits != 16 or info.channels != 1:
            raise InvalidAudio("SaluteSpeech needs 16-bit mono PCM")
        pcm = audio[info.data_offset : info.data_offset + info.data_size]
        headers = {"Content-Type": f"audio/x-pcm;bit=16;rate={info.sample_rate}"}
        data = await self._post(pcm, headers)
        if data.get("status", 200) != 200 or not isinstance(data.get("result"), list):
            raise SttError("bad response")
        text = " ".join(str(part).strip() for part in data["result"] if str(part).strip())
        if not text:
            raise SttNoSpeech("no speech")
        return text

    async def _get_token(self, client: httpx.AsyncClient, *, force: bool = False) -> str:
        tok = self._token
        if not force and tok.value and time.monotonic() < tok.expires:
            return tok.value
        async with tok.lock:
            if not force and tok.value and time.monotonic() < tok.expires:
                return tok.value
            headers = {
                "Authorization": f"Basic {self._credentials}",
                "RqUID": str(uuid.uuid4()),
                "Content-Type": "application/x-www-form-urlencoded",
                "Accept": "application/json",
            }
            try:
                resp = await client.post(SALUTE_OAUTH_URL, headers=headers, data={"scope": self._scope})
            except httpx.HTTPError as exc:
                raise SttError(f"oauth {type(exc).__name__}") from exc
            if resp.status_code != 200:
                log.error("stt salute oauth http %s", resp.status_code)
                raise SttError(f"oauth http {resp.status_code}")
            try:
                body = resp.json()
                value, expires_at_ms = body["access_token"], float(body["expires_at"])
            except (ValueError, KeyError, TypeError) as exc:
                raise SttError("oauth bad response") from exc
            # expires_at is wall-clock epoch ms from Sber; keep a monotonic deadline with a minute of margin.
            ttl = max(0.0, expires_at_ms / 1000 - time.time() - 60)
            tok.value, tok.expires = value, time.monotonic() + ttl
            log.info("stt salute oauth token refreshed")
            return value

    def _verify(self) -> ssl.SSLContext:
        context = ssl.create_default_context()
        if self._ca is not None:
            context.load_verify_locations(cafile=str(self._ca))
        return context

    async def _post(self, pcm: bytes, headers: dict[str, str]) -> dict:
        if self._client is not None:
            return await self._recognize(self._client, pcm, headers)
        async with httpx.AsyncClient(timeout=self._timeout, verify=self._verify()) as client:
            return await self._recognize(client, pcm, headers)

    async def _recognize(self, client: httpx.AsyncClient, pcm: bytes, headers: dict[str, str]) -> dict:
        refreshed = False
        for attempt in range(_ATTEMPTS):
            token = await self._get_token(client)
            try:
                resp = await client.post(
                    SALUTE_RECOGNIZE_URL,
                    content=pcm,
                    params={"language": "ru-RU"},
                    headers={**headers, "Authorization": f"Bearer {token}"},
                    timeout=self._timeout,
                )
            except httpx.HTTPError as exc:
                if attempt + 1 == _ATTEMPTS:
                    raise SttError(type(exc).__name__) from exc
                await asyncio.sleep(0.5 * 2**attempt)
                continue
            if resp.status_code == 401 and not refreshed:
                # The cached token was revoked or expired early: fetch a new one once.
                refreshed = True
                await self._get_token(client, force=True)
                continue
            if resp.status_code in _RETRY_STATUSES and attempt + 1 < _ATTEMPTS:
                log.warning("stt salute http %s, retry %d", resp.status_code, attempt + 1)
                await asyncio.sleep(0.5 * 2**attempt)
                continue
            if resp.status_code != 200:
                log.error("stt salute http %s: %s", resp.status_code, resp.text[:200])
                raise SttError(f"http {resp.status_code}")
            try:
                return resp.json()
            except ValueError as exc:
                raise SttError("bad json") from exc
        raise SttError("retries exhausted")


def get_stt(settings: Settings) -> SpeechToText | None:
    """The configured provider; `None` = voice input is off.

    `auto` = SaluteSpeech if credentials are set. Otherwise only in the demo (synthetic data): OpenRouter if a key is
    set, else mock. Outside the demo `auto` never picks a foreign service or a fake transcript — voice stays off.
    """
    mode = settings.stt_provider
    has_salute = bool(settings.salute_speech_credentials.get_secret_value())
    if mode == "auto":
        if has_salute:
            mode = "salute"
        elif not settings.demo_clock:
            return None
        else:
            mode = "openrouter" if settings.openrouter_api_key else "mock"
    if mode == "salute":
        return SaluteSTT(settings) if has_salute else None
    if mode == "openrouter":
        return OpenRouterSTT(settings) if settings.openrouter_api_key else None
    if mode == "mock":
        return MockSTT()
    return None