"""Voice routes for the OpenAI STT/TTS cascade over WebSocket. Client sends JSON controls plus binary audio chunks: audio_start -> binary audio chunks -> audio_end Server emits: ready -> state(listening) -> state(thinking) -> transcript -> reply -> state(speaking) -> binary audio chunks -> tts_end -> state(idle) When voice is not configured, the route reports degraded state and closes cleanly instead of crashing. """ from __future__ import annotations import json import hashlib import time from typing import Optional from fastapi import APIRouter, WebSocket, WebSocketDisconnect from fastapi.responses import JSONResponse from starlette.websockets import WebSocketState from .. import session_persistence, turn_runtime from ..auth_sessions import get_session, user_has_consent, user_onboarding_complete from ..config import settings from ..deps import Principal, Role from ..engine_client import EngineError, engine_client from ..persona_repository import ( PersonaVoiceMap, get_catalog_persona, get_persona_voice_map, get_session_voice_map, ) from ..runtime_policy import require_runtime_fallback_allowed from ..services import evaluator, orchestrator, state_machine from ..services import voice as voice_svc from ..services.voice import VoicePreset, VoiceUnavailable, resolve_voice, voice_service from ..store import InProcSession, TurnRecord, store router = APIRouter(prefix="/voice", tags=["voice"]) # WebSocket close codes. WS_CLOSE_DEGRADED = 1011 WS_CLOSE_BAD_REQUEST = 1008 WS_CLOSE_UNAUTHORIZED = 1008 # Per-utterance audio cap to avoid unbounded memory growth. _MAX_AUDIO_BYTES = 10 * 1024 * 1024 _PROVIDER_EVENT_MAX_ITEMS = 12 _PROVIDER_EVENT_MAX_STRING = 80 _PROVIDER_EVENT_ALLOWED_KEYS = { "category", "event_type", "type", "kind", "label", "source", "provider", "start_ms", "end_ms", "duration_ms", "confidence", "score", "is_final", } _PROVIDER_EVENT_TAXONOMY = { "barge_in": ("barge_in", "turn_taking"), "interrupt": ("barge_in", "turn_taking"), "interruption": ("barge_in", "turn_taking"), "overlap": ("barge_in", "turn_taking"), "sigh": ("sigh", "paralinguistic"), "sighing": ("sigh", "paralinguistic"), "sob": ("cry", "paralinguistic"), "cry": ("cry", "paralinguistic"), "crying": ("cry", "paralinguistic"), "weep": ("cry", "paralinguistic"), "laugh": ("laugh", "paralinguistic"), "laughter": ("laugh", "paralinguistic"), "breath": ("breath", "paralinguistic"), "breathing": ("breath", "paralinguistic"), "voice_activity": ("voice_activity", "speech_activity"), "vad": ("voice_activity", "speech_activity"), "speech_start": ("speech_start", "speech_activity"), "speech_end": ("speech_end", "speech_activity"), "speech_final": ("speech_final", "speech_activity"), "silence": ("silence", "timing"), "pause": ("silence", "timing"), "long_pause": ("silence", "timing"), "speech_rate": ("speech_rate", "prosody"), "fast_speech": ("speech_rate", "prosody"), "slow_speech": ("speech_rate", "prosody"), "pitch": ("pitch", "prosody"), "intonation": ("intonation", "prosody"), "prosody": ("prosody", "prosody"), "noise": ("background_noise", "audio_quality"), "background_noise": ("background_noise", "audio_quality"), } _PROVIDER_EVENT_TYPE_FIELDS = ("event_type", "type", "kind", "label") @router.get("/health") async def voice_health() -> JSONResponse: """Return voice service readiness.""" available = voice_service.is_available() tts_provider = voice_service.tts_provider() body = { "status": "ok" if available else "degraded", "available": available, "stt_model": voice_svc.STT_MODEL, "tts_model": voice_svc.TTS_MODEL, "tts_provider": tts_provider, "reason": None if available else "OPENAI_API_KEY is not configured", } return JSONResponse(body, status_code=200 if available else 503) @router.websocket("/ws") async def voice_ws(websocket: WebSocket) -> None: """Run one authenticated learner voice cascade.""" await websocket.accept() # Authenticate the same server-side browser session used by REST routes. principal = await _principal_from_websocket(websocket) if principal is None: await _safe_send_json(websocket, {"type": "error", "detail": "not authenticated"}) await _safe_close(websocket, WS_CLOSE_UNAUTHORIZED) return if principal.role != Role.LEARNER and principal.can_access_role(Role.LEARNER): principal = principal.with_role(Role.LEARNER) if principal.role != Role.LEARNER: await _safe_send_json(websocket, {"type": "error", "detail": "only learners can use voice"}) await _safe_close(websocket, WS_CLOSE_UNAUTHORIZED) return # Bind to an existing session first. persona_code creation is dev-only. session_id, voice_preset, err, bind_meta = await _bind_session(websocket, principal) if err is not None: await _safe_send_json(websocket, {"type": "error", "detail": err}) await _safe_close(websocket, WS_CLOSE_BAD_REQUEST) return assert session_id is not None and voice_preset is not None if bind_meta.get("degraded"): await _safe_send_json( websocket, { "type": "degraded", "reason": bind_meta.get("degraded_reason", "voice session binding degraded"), **bind_meta, }, ) # Voice misconfiguration is reported explicitly and then closed cleanly. if not voice_service.is_available(): await _safe_send_json( websocket, {"type": "degraded", "reason": "OPENAI_API_KEY is not configured"}, ) await _safe_close(websocket, WS_CLOSE_DEGRADED) return await _safe_send_json( websocket, { "type": "ready", "session_id": session_id, "voice": voice_preset.openai_voice, "preset": voice_preset.preset, "tts_provider": voice_service.tts_provider(), "state": "idle", **bind_meta, }, ) audio_buf = bytearray() receiving = False audio_started_at: float | None = None last_audio_end_at: float | None = None audio_format: str | None = None audio_sample_rate: int | None = None audio_channels: int | None = None audio_sample_width: int | None = None try: while True: msg = await websocket.receive() mtype = msg.get("type") if mtype == "websocket.disconnect": break # Binary frames are audio chunks. if msg.get("bytes") is not None: if not receiving: # Be tolerant when audio arrives before audio_start. receiving = True audio_started_at = time.monotonic() audio_buf.clear() await _safe_send_json(websocket, {"type": "state", "state": "listening"}) audio_buf.extend(msg["bytes"]) if len(audio_buf) > _MAX_AUDIO_BYTES: await _safe_send_json( websocket, {"type": "error", "detail": "audio too large; please send a shorter utterance"}, ) audio_buf.clear() receiving = False continue # Text frames are JSON controls. text = msg.get("text") if text is None: continue try: ctrl = json.loads(text) except (json.JSONDecodeError, TypeError): await _safe_send_json(websocket, {"type": "error", "detail": "invalid control json"}) continue ctype = ctrl.get("type") if ctype == "audio_start": receiving = True audio_started_at = time.monotonic() audio_format = _safe_str(ctrl.get("format")) audio_sample_rate = _safe_int(ctrl.get("sample_rate")) audio_channels = _safe_int(ctrl.get("channels")) audio_sample_width = _safe_int(ctrl.get("sample_width")) audio_buf.clear() await _safe_send_json(websocket, {"type": "state", "state": "listening"}) elif ctype == "audio_end": receiving = False audio_ended_at = time.monotonic() silence_ms = _safe_int(ctrl.get("silence_ms")) if silence_ms is None and last_audio_end_at is not None and audio_started_at is not None: silence_ms = max(0, int((audio_started_at - last_audio_end_at) * 1000)) end_format = _safe_str(ctrl.get("format")) or audio_format await _handle_utterance( websocket, session_id=session_id, principal=principal, voice_preset=voice_preset, audio=bytes(audio_buf), fmt=end_format, sample_rate=_safe_int(ctrl.get("sample_rate")) or audio_sample_rate, channels=_safe_int(ctrl.get("channels")) or audio_channels, sample_width=_safe_int(ctrl.get("sample_width")) or audio_sample_width, audio_started_at=audio_started_at, audio_ended_at=audio_ended_at, silence_ms=silence_ms, barge_in=_safe_bool(ctrl.get("barge_in")), provider_events=_safe_provider_events(ctrl.get("provider_events")), ) last_audio_end_at = audio_ended_at audio_started_at = None audio_format = None audio_sample_rate = None audio_channels = None audio_sample_width = None audio_buf.clear() elif ctype == "text_turn": # Text-only path for accessibility and deterministic tests. receiving = False audio_buf.clear() learner_text = (ctrl.get("text") or "").strip() if learner_text: await _run_turn_and_speak( websocket, session_id=session_id, principal=principal, voice_preset=voice_preset, learner_text=learner_text, ) elif ctype == "stt_result": receiving = False audio_buf.clear() stt_received_at = time.monotonic() await _handle_stt_result_control( websocket, session_id=session_id, principal=principal, voice_preset=voice_preset, ctrl=ctrl, audio_started_at=audio_started_at, audio_ended_at=stt_received_at, last_audio_end_at=last_audio_end_at, ) last_audio_end_at = stt_received_at audio_started_at = None audio_format = None audio_sample_rate = None audio_channels = None audio_sample_width = None elif ctype == "ping": await _safe_send_json(websocket, {"type": "pong"}) elif ctype == "close": break except WebSocketDisconnect: pass except Exception as e: await _safe_send_json(websocket, {"type": "error", "detail": f"voice ws error: {e}"}) finally: await _safe_close(websocket) async def _handle_stt_result_control( websocket: WebSocket, *, session_id: str, principal: Principal, voice_preset: VoicePreset, ctrl: dict[str, object], audio_started_at: float | None = None, audio_ended_at: float | None = None, last_audio_end_at: float | None = None, ) -> None: learner_text = str(ctrl.get("text") or "").strip() transcript_final = _safe_bool(ctrl.get("final")) silence_ms = _safe_int(ctrl.get("silence_ms")) if silence_ms is None and last_audio_end_at is not None and audio_started_at is not None: silence_ms = max(0, int((audio_started_at - last_audio_end_at) * 1000)) provider_events = _safe_provider_events(ctrl.get("provider_events")) decision = voice_svc.assess_end_of_turn( transcript_text=learner_text, transcript_final=bool(transcript_final), silence_ms=silence_ms, ) await _safe_send_json( websocket, { "type": "eot", "ready": decision.ready, "reason": decision.reason, "silence_ms": decision.silence_ms, "threshold_ms": decision.threshold_ms, }, ) if not decision.ready: await _safe_send_json(websocket, {"type": "state", "state": "listening"}) return await _safe_send_json(websocket, {"type": "state", "state": "thinking"}) await _safe_send_json( websocket, {"type": "transcript", "text": learner_text, "final": True, "speaker": "counselor"}, ) await _run_turn_and_speak( websocket, session_id=session_id, principal=principal, voice_preset=voice_preset, learner_text=learner_text, duration_s=_elapsed_seconds(audio_started_at, audio_ended_at), silence_ms=decision.silence_ms, barge_in=_safe_bool(ctrl.get("barge_in")), provider_events=provider_events, ) async def _handle_utterance( websocket: WebSocket, *, session_id: str, principal: Principal, voice_preset: VoicePreset, audio: bytes, fmt: Optional[str], sample_rate: int | None = None, channels: int | None = None, sample_width: int | None = None, audio_started_at: float | None = None, audio_ended_at: float | None = None, silence_ms: int | None = None, barge_in: bool | None = None, provider_events: list[dict[str, object]] | None = None, ) -> None: """Transcribe one utterance, generate the client reply, then synthesize TTS.""" if not audio: await _safe_send_json(websocket, {"type": "transcript", "text": "", "final": True}) await _safe_send_json(websocket, {"type": "state", "state": "idle"}) return # STT begins after the learner stops speaking. await _safe_send_json(websocket, {"type": "state", "state": "thinking"}) upload_audio, upload_fmt = _normalize_audio_upload( audio, fmt=fmt, sample_rate=sample_rate, channels=channels, sample_width=sample_width, ) filename, content_type = _audio_meta(upload_fmt) try: stt = await voice_service.transcribe( upload_audio, filename=filename, content_type=content_type ) except VoiceUnavailable as e: await _safe_send_json(websocket, {"type": "degraded", "reason": str(e)}) await _safe_send_json(websocket, {"type": "state", "state": "idle"}) return except Exception as e: await _safe_send_json(websocket, {"type": "error", "detail": f"STT failed: {e}"}) await _safe_send_json(websocket, {"type": "state", "state": "idle"}) return learner_text = stt.text audio_ref = _voice_audio_ref(upload_audio, upload_fmt) duration_s = stt.duration or _elapsed_seconds(audio_started_at, audio_ended_at) speech_rate = _estimate_speech_rate(learner_text, duration_s) provider_events = _merge_provider_events(provider_events, getattr(stt, "provider_events", [])) await _safe_send_json( websocket, {"type": "transcript", "text": learner_text, "final": True, "speaker": "counselor"}, ) if not learner_text: await _safe_send_json(websocket, {"type": "state", "state": "idle"}) return await _run_turn_and_speak( websocket, session_id=session_id, principal=principal, voice_preset=voice_preset, learner_text=learner_text, audio_ref=audio_ref, silence_ms=silence_ms, speech_rate=speech_rate, barge_in=barge_in, provider_events=provider_events, ) async def _run_turn_and_speak( websocket: WebSocket, *, session_id: str, principal: Principal, voice_preset: VoicePreset, learner_text: str, audio_ref: str | None = None, duration_s: float | None = None, silence_ms: int | None = None, speech_rate: float | None = None, barge_in: bool | None = None, provider_events: list[dict[str, object]] | None = None, ) -> None: """Run one counseling turn and stream synthesized client speech.""" if speech_rate is None: speech_rate = _estimate_speech_rate(learner_text, duration_s) sess, err = await _load_voice_session(session_id, principal) if sess is None: await _safe_send_json(websocket, {"type": "error", "detail": err or "session not found or ended"}) await _safe_send_json(websocket, {"type": "state", "state": "idle"}) return from . import sessions as session_routes recall = await session_routes.ensure_recall_context(sess) kb_cues = session_routes._KB_CUES_CACHE.get(session_id) or [] ctx = orchestrator.prepare_turn( session_id=session_id, case_id=sess.case_id, card=sess.persona, state=sess.state, learner_text=learner_text, memory=orchestrator.TurnMemory( recall_summary=recall.recall_summary, pinned_facts=recall.pinned_facts, recent_turns=sess.recent_turns(visible_to="client"), kb_behavior_cues=kb_cues, ), theory_mode=sess.theory_mode, ) assert ctx.state_after is not None # Voice needs the full client reply before TTS starts. try: result = await orchestrator.run_turn_generate( ctx, engine_client, eval_hook=evaluator.make_eval_hook( engine_client, audit_hook=session_persistence.record_llm_call_audit, ), audit_hook=session_persistence.record_llm_call_audit, ) except EngineError as e: await _safe_send_json(websocket, {"type": "error", "detail": f"engine unavailable: {e}"}) await _safe_send_json(websocket, {"type": "state", "state": "idle"}) return reply = result.client_reply or "" # Persist only after the client reply has been generated. A failed AI turn # must not leave a learner-only transcript in review or history. await turn_runtime.finalize_completed_turn( sess, ctx, result, context_prefix="voice session", counselor_turn=TurnRecord( turn_seq=ctx.state_after.turn_seq, speaker="counselor", stage=turn_runtime.stage_label(ctx.state_after.stage), text=learner_text, text_masked=ctx.learner_text_masked, audio_ref=audio_ref, silence_ms=silence_ms, speech_rate=speech_rate, barge_in=barge_in, provider_events=provider_events or [], evaluation=result.evaluation, ), ) # Send the final client text before audio playback. await _safe_send_json( websocket, { "type": "reply", "text": reply, "speaker": "client", "stage": result.stage, "effective_openness": round(result.effective_openness, 4), "turn_seq": result.turn_seq, "safety_flagged": result.safety_flagged, "crisis_kind": result.crisis_kind, "crisis_resource": result.crisis_resource, "conversation_stopped": result.conversation_stopped, }, ) if not reply: await _safe_send_json(websocket, {"type": "state", "state": "idle"}) return # TTS speaking state comes before chunk metadata and binary audio. await _safe_send_json( websocket, { "type": "state", "state": "speaking", "voice": voice_preset.openai_voice, "tts_provider": voice_service.tts_provider(), }, ) try: n = 0 async for ck in voice_service.synthesize_stream(reply, voice_preset): # 바이너리 오디오 청크만 송신(프론트가 Web Audio AnalyserNode로 립싱크 자체 산출). await _safe_send_bytes(websocket, ck.audio) n += 1 await _safe_send_json(websocket, {"type": "tts_end", "chunks": n}) except VoiceUnavailable as e: await _safe_send_json(websocket, {"type": "degraded", "reason": str(e)}) except Exception as e: await _safe_send_json(websocket, {"type": "error", "detail": f"TTS failed: {e}"}) await _safe_send_json(websocket, {"type": "state", "state": "idle"}) async def _load_voice_session( session_id: str, principal: Principal, ) -> tuple[InProcSession | None, str | None]: sess, err = await turn_runtime.load_owned_session(session_id, principal) if err == turn_runtime.SessionAccessError.NOT_FOUND: return None, f"unknown session {session_id}" if err == turn_runtime.SessionAccessError.FORBIDDEN: return None, "session does not belong to user" if err == turn_runtime.SessionAccessError.ENDED: return None, "session already ended" assert sess is not None return sess, None async def _principal_from_websocket(websocket: WebSocket) -> Principal | None: """Restore the same server-side browser session used by REST routes.""" raw_cookie = websocket.cookies.get(settings.cookie_name) if raw_cookie is None and settings.environment == "dev": raw_cookie = websocket.cookies.get("vignette_sid") session = await get_session(raw_cookie) if session is None: return None if session.account_status != "approved": return None try: role = Role(session.role) except ValueError: return None return Principal( user_id=session.user_id, role=role, admin_access=session.admin_access, super_admin=session.super_admin, account_status=session.account_status, cohort_ids=session.cohort_ids, email=session.email, display_name=session.display_name, consent_at=session.consent_at, profile_completed_at=session.profile_completed_at, ) async def _bind_session( websocket: WebSocket, principal: Principal, ) -> tuple[Optional[str], Optional[VoicePreset], Optional[str], dict[str, object]]: """Bind an existing session or create a dev-only voice session.""" qp = websocket.query_params explicit_preset = qp.get("preset") session_id = qp.get("session_id") if session_id: sess, err = await _load_voice_session(session_id, principal) if sess is None: return None, None, err or f"unknown session {session_id}", {} vp = await _resolve_session_voice( session_id=session_id, persona_code=sess.persona.code, explicit_preset=explicit_preset, ) return session_id, vp, None, {"degraded": False, "persona_catalog_source": "session"} # persona_code session creation is local-dev only. Production uses REST start. if settings.environment != "dev": return None, None, "session_id required", {} persona_code = qp.get("persona_code") if not persona_code: return None, None, "session_id or persona_code query required", {} if ( principal.profile_completed_at is None and not await user_onboarding_complete(principal.user_id) ): return None, None, "onboarding_required", {} if principal.consent_at is None and not await user_has_consent(principal.user_id): return None, None, "consent_required", {} try: catalog_persona = await get_catalog_persona(persona_code) except Exception: return None, None, "persona catalog database unavailable", {} if catalog_persona is None: return None, None, f"unknown persona {persona_code}", {} card = catalog_persona.card st = state_machine.init_state( params=card.openness_params(), ) sess = await session_persistence.create_session( learner_id=principal.user_id, card=card, theory_mode="humanistic", state=st, session_no=1, carry_rapport=st.rapport_credit, persona_id=catalog_persona.persona_id, persona_version=catalog_persona.version, ) session_source = "database" if sess is None: require_runtime_fallback_allowed("voice session creation") sess = store.create( learner_id=principal.user_id, persona=card, theory_mode="humanistic", state=st, session_no=1, carry_rapport=st.rapport_credit, ) session_source = "runtime" else: store.put(sess) vp = await _resolve_catalog_voice( persona_id=catalog_persona.persona_id, version=catalog_persona.version, persona_code=card.code, explicit_preset=explicit_preset, ) degraded_reasons: list[str] = [] if catalog_persona.degraded: degraded_reasons.append("카탈로그 원본을 확인하지 못해 음성 회기를 시작하지 않습니다") if session_source == "runtime": degraded_reasons.append("세션 저장소 연결 전까지 비영구 개발 런타임 기록을 사용합니다") bind_meta = { "degraded": bool(degraded_reasons), "degraded_reason": "; ".join(degraded_reasons) if degraded_reasons else None, "persona_catalog_source": catalog_persona.source, "session_source": session_source, } return sess.session_id, vp, None, bind_meta async def _resolve_session_voice( *, session_id: str, persona_code: str, explicit_preset: str | None, ) -> VoicePreset: fallback = resolve_voice(persona_code=persona_code, preset=explicit_preset) if explicit_preset: return fallback try: voice_map = await get_session_voice_map(session_id) except Exception: return fallback return _voice_from_map(voice_map, persona_code=persona_code) or fallback async def _resolve_catalog_voice( *, persona_id: str | None, version: int | None, persona_code: str, explicit_preset: str | None, ) -> VoicePreset: fallback = resolve_voice(persona_code=persona_code, preset=explicit_preset) if explicit_preset: return fallback try: voice_map = await get_persona_voice_map(persona_id=persona_id, version=version) except Exception: return fallback return _voice_from_map(voice_map, persona_code=persona_code) or fallback def _voice_from_map( voice_map: PersonaVoiceMap | None, *, persona_code: str, ) -> VoicePreset | None: if voice_map is None: return None return voice_svc.resolve_voice_from_map( provider=voice_map.provider, voice_id=voice_map.voice_id, base_params=voice_map.base_params, persona_code=persona_code, ) def _audio_meta(fmt: Optional[str]) -> tuple[str, str]: """Map the browser audio format to upload metadata.""" f = (fmt or "webm").lower().lstrip(".") table = { "webm": ("audio.webm", "audio/webm"), "ogg": ("audio.ogg", "audio/ogg"), "opus": ("audio.ogg", "audio/ogg"), "wav": ("audio.wav", "audio/wav"), "mp3": ("audio.mp3", "audio/mpeg"), "mp4": ("audio.mp4", "audio/mp4"), "m4a": ("audio.m4a", "audio/mp4"), "pcm": ("audio.wav", "audio/wav"), } return table.get(f, ("audio.webm", "audio/webm")) def _normalize_audio_upload( audio: bytes, *, fmt: Optional[str], sample_rate: int | None = None, channels: int | None = None, sample_width: int | None = None, ) -> tuple[bytes, str]: f = (fmt or "webm").lower().lstrip(".") or "webm" if f != "pcm": return audio, f if sample_width not in (None, 2): raise ValueError("pcm sample_width must be 2 bytes") return _wav_from_pcm16( audio, sample_rate=_bounded_int(sample_rate, default=48000, minimum=8000, maximum=96000), channels=_bounded_int(channels, default=1, minimum=1, maximum=2), ), "wav" def _bounded_int(value: int | None, *, default: int, minimum: int, maximum: int) -> int: if value is None: return default return min(maximum, max(minimum, value)) def _wav_from_pcm16(pcm: bytes, *, sample_rate: int, channels: int) -> bytes: byte_rate = sample_rate * channels * 2 block_align = channels * 2 data_size = len(pcm) header = b"".join( [ b"RIFF", (36 + data_size).to_bytes(4, "little"), b"WAVE", b"fmt ", (16).to_bytes(4, "little"), (1).to_bytes(2, "little"), channels.to_bytes(2, "little"), sample_rate.to_bytes(4, "little"), byte_rate.to_bytes(4, "little"), block_align.to_bytes(2, "little"), (16).to_bytes(2, "little"), b"data", data_size.to_bytes(4, "little"), ] ) return header + pcm def _voice_audio_ref(audio: bytes, fmt: Optional[str]) -> str | None: if not audio: return None f = (fmt or "webm").lower().lstrip(".") or "webm" digest = hashlib.sha256(audio).hexdigest()[:24] return f"voice:{f}:sha256:{digest}" def _elapsed_seconds(started_at: float | None, ended_at: float | None) -> float | None: if started_at is None or ended_at is None: return None return max(0.001, ended_at - started_at) def _estimate_speech_rate(text: str, duration_s: float | None) -> float | None: if not text or not duration_s or duration_s <= 0: return None units = sum(1 for ch in text if not ch.isspace()) if units <= 0: return None return round((units / duration_s) * 60.0, 2) def _safe_int(value: object) -> int | None: if value is None: return None try: return int(value) except (TypeError, ValueError): return None def _safe_str(value: object) -> str | None: if isinstance(value, str): text = value.strip() return text or None return None def _safe_bool(value: object) -> bool | None: if value is None: return None if isinstance(value, bool): return value if isinstance(value, str): normalized = value.strip().lower() if normalized in {"1", "true", "yes", "y"}: return True if normalized in {"0", "false", "no", "n"}: return False return bool(value) def _provider_event_slug(value: object) -> str: text = str(value or "").strip().lower() if not text: return "" chars = [] previous_underscore = False for char in text: if char.isalnum(): chars.append(char) previous_underscore = False elif not previous_underscore: chars.append("_") previous_underscore = True return "".join(chars).strip("_")[:_PROVIDER_EVENT_MAX_STRING] def _provider_event_taxonomy(event: dict[str, object]) -> tuple[str, str]: for field in _PROVIDER_EVENT_TYPE_FIELDS: slug = _provider_event_slug(event.get(field)) if not slug: continue canonical = _PROVIDER_EVENT_TAXONOMY.get(slug) if canonical: return canonical return slug, "unknown" return "", "" def _safe_provider_events(value: object) -> list[dict[str, object]]: if not isinstance(value, list): return [] events: list[dict[str, object]] = [] for item in value[:_PROVIDER_EVENT_MAX_ITEMS]: if not isinstance(item, dict): continue safe: dict[str, object] = {} for key in _PROVIDER_EVENT_ALLOWED_KEYS: raw = item.get(key) if isinstance(raw, bool): safe[key] = raw elif isinstance(raw, (int, float)): safe[key] = raw elif isinstance(raw, str): text = raw.strip() if text: safe[key] = text[:_PROVIDER_EVENT_MAX_STRING] if safe: event_type, category = _provider_event_taxonomy(safe) if event_type: safe["event_type"] = event_type if category: safe["category"] = category events.append(safe) return events def _merge_provider_events(*values: object) -> list[dict[str, object]]: merged: list[dict[str, object]] = [] for value in values: merged.extend(_safe_provider_events(value)) if len(merged) >= _PROVIDER_EVENT_MAX_ITEMS: return merged[:_PROVIDER_EVENT_MAX_ITEMS] return merged async def _safe_send_json(websocket: WebSocket, payload: dict) -> None: if websocket.client_state != WebSocketState.CONNECTED: return try: await websocket.send_text(json.dumps(payload, ensure_ascii=False)) except Exception: pass async def _safe_send_bytes(websocket: WebSocket, data: bytes) -> None: if websocket.client_state != WebSocketState.CONNECTED: return try: await websocket.send_bytes(data) except Exception: pass async def _safe_close(websocket: WebSocket, code: int = 1000) -> None: if websocket.client_state == WebSocketState.DISCONNECTED: return try: await websocket.close(code=code) except Exception: pass __all__ = ["router"]