Xubin RenandGitHub
9c81280300
* feat(webui): add voice transcription input * feat(webui): render ANSI output in code blocks * refactor(webui): isolate voice recorder logic * refactor(transcription): keep websocket ingress thin * refactor(transcription): resolve channel audio settings on demand * style(webui): neutralize voice waveform color * feat(webui): add voice input tooltip * feat(webui): add voice input keyboard shortcut * fix(webui): distinguish voice shortcut platforms * fix(webui): place voice button after model selector * refactor(webui): share voice hold recording helpers * fix(desktop): allow microphone voice input * fix(webui): stabilize token usage month labels * feat(webui): show voice input on settings overview * fix(webui): label voice capability as recognition * fix(webui): align capability overview status * refactor(webui): isolate transcription socket handling * fix(webui): soften silent voice waveform * refactor(audio): clarify transcription service location * docs(transcription): clarify audio and provider boundaries * fix(exec): reduce session output polling flake
47 lines
1.5 KiB
Python
47 lines
1.5 KiB
Python
"""WebUI transcription envelope handling.
|
|
|
|
The WebSocket channel owns transport and subscription fan-out. This module owns
|
|
the WebUI-specific audio transcription action carried over that socket.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from nanobot.audio.transcription import (
|
|
TranscriptionIngressError,
|
|
resolve_transcription_config,
|
|
transcribe_audio_data_url,
|
|
)
|
|
from nanobot.config.loader import load_config
|
|
|
|
_MAX_REQUEST_ID_LENGTH = 80
|
|
|
|
|
|
async def webui_transcription_event(envelope: dict[str, Any]) -> tuple[str, dict[str, Any]]:
|
|
"""Return the WS event name and payload for one WebUI transcription request."""
|
|
request_id = envelope.get("request_id")
|
|
valid_request_id = (
|
|
isinstance(request_id, str)
|
|
and 0 < len(request_id) <= _MAX_REQUEST_ID_LENGTH
|
|
)
|
|
|
|
def error(detail: str, **extra: Any) -> tuple[str, dict[str, Any]]:
|
|
payload: dict[str, Any] = {"detail": detail, **extra}
|
|
if valid_request_id:
|
|
payload["request_id"] = request_id
|
|
return "transcription_error", payload
|
|
|
|
if not valid_request_id:
|
|
return error("invalid_request")
|
|
|
|
try:
|
|
text = await transcribe_audio_data_url(
|
|
envelope.get("data_url"),
|
|
resolve_transcription_config(load_config()),
|
|
duration_ms=envelope.get("duration_ms"),
|
|
)
|
|
except TranscriptionIngressError as exc:
|
|
return error(exc.detail, **exc.extra)
|
|
return "transcription_result", {"request_id": request_id, "text": text}
|