diff --git a/.env.example b/.env.example index ff73675..86e4afe 100644 --- a/.env.example +++ b/.env.example @@ -80,14 +80,3 @@ SEMANTIC_ENDPOINTING_HOLD_MS=600 REALTIME_VOICE_FILLER_DELAY_MS=300 REALTIME_VOICE_TTS_CHUNK_SOFT_MIN_CHARS=10 REALTIME_VOICE_TTS_CHUNK_SOFT_MIN_WORDS=2 - -CRM_INTERACTION_ENABLED=false -CRM_INTERACTION_SERVICE_URL=http://interaction-service:8000 -CRM_APP_TOKEN_SECRET= -CRM_QUEUE_ID= - -SALES_SYNC_ENABLED=true -SALES_SERVICE_URL=http://sales-service:8000 -SALES_SYNC_TIMEOUT_SECONDS=8 -REALTIME_VOICE_QUEUE_ID= -REALTIME_VOICE_QUEUE_CODE= diff --git a/.idea/.gitignore b/.idea/.gitignore deleted file mode 100644 index 30cf57e..0000000 --- a/.idea/.gitignore +++ /dev/null @@ -1,10 +0,0 @@ -# Default ignored files -/shelf/ -/workspace.xml -# Editor-based HTTP Client requests -/httpRequests/ -# Ignored default folder with query files -/queries/ -# Datasource local storage ignored files -/dataSources/ -/dataSources.local.xml diff --git a/.idea/checkstyle-idea.xml b/.idea/checkstyle-idea.xml deleted file mode 100644 index ae8e08c..0000000 --- a/.idea/checkstyle-idea.xml +++ /dev/null @@ -1,16 +0,0 @@ - - - - 13.4.0 - JavaOnly - true - - - \ No newline at end of file diff --git a/.idea/git_toolbox_prj.xml b/.idea/git_toolbox_prj.xml deleted file mode 100644 index 02b915b..0000000 --- a/.idea/git_toolbox_prj.xml +++ /dev/null @@ -1,15 +0,0 @@ - - - - - - - \ No newline at end of file diff --git a/.idea/material_theme_project_new.xml b/.idea/material_theme_project_new.xml deleted file mode 100644 index bfc5c67..0000000 --- a/.idea/material_theme_project_new.xml +++ /dev/null @@ -1,12 +0,0 @@ - - - - - - \ No newline at end of file diff --git a/.idea/misc.xml b/.idea/misc.xml deleted file mode 100644 index 31e1ebc..0000000 --- a/.idea/misc.xml +++ /dev/null @@ -1,6 +0,0 @@ - - - - - - \ No newline at end of file diff --git a/.idea/modules.xml b/.idea/modules.xml deleted file mode 100644 index 2f9d213..0000000 --- a/.idea/modules.xml +++ /dev/null @@ -1,8 +0,0 @@ - - - - - - - - \ No newline at end of file diff --git a/.idea/realtime_voice_service.iml b/.idea/realtime_voice_service.iml deleted file mode 100644 index d6ebd48..0000000 --- a/.idea/realtime_voice_service.iml +++ /dev/null @@ -1,9 +0,0 @@ - - - - - - - - - \ No newline at end of file diff --git a/.idea/vcs.xml b/.idea/vcs.xml deleted file mode 100644 index 35eb1dd..0000000 --- a/.idea/vcs.xml +++ /dev/null @@ -1,6 +0,0 @@ - - - - - - \ No newline at end of file diff --git a/audioop_compat.py b/audioop_compat.py deleted file mode 100644 index ef99ee9..0000000 --- a/audioop_compat.py +++ /dev/null @@ -1,110 +0,0 @@ -from __future__ import annotations - -import math -import struct -from array import array - -try: - import audioop as _audioop -except ModuleNotFoundError: - try: - import audioop_lts as _audioop # type: ignore[import-not-found] - except ModuleNotFoundError: - _audioop = None - - -def _native_is_little_endian() -> bool: - return struct.pack("=h", 1) == struct.pack(" array: - if width != 2: - raise NotImplementedError("audioop_compat fallback currently supports only 16-bit PCM") - samples = array("h") - samples.frombytes(fragment) - if not _native_is_little_endian(): - samples.byteswap() - return samples - - -def _write_int16_samples(samples: array) -> bytes: - output = array("h", samples) - if not _native_is_little_endian(): - output.byteswap() - return output.tobytes() - - -def _clip_int16(value: float) -> int: - return max(-32768, min(32767, int(round(value)))) - - -class _AudioopFallback: - @staticmethod - def rms(fragment: bytes, width: int) -> int: - samples = _read_int16_samples(fragment, width) - if not samples: - return 0 - mean_square = sum(sample * sample for sample in samples) / len(samples) - return int(math.sqrt(mean_square)) - - @staticmethod - def max(fragment: bytes, width: int) -> int: - samples = _read_int16_samples(fragment, width) - if not samples: - return 0 - return max(abs(sample) for sample in samples) - - @staticmethod - def tomono(fragment: bytes, width: int, lfactor: float, rfactor: float) -> bytes: - samples = _read_int16_samples(fragment, width) - if len(samples) % 2 != 0: - raise ValueError("Stereo PCM must contain an even number of samples") - mono = array("h") - for index in range(0, len(samples), 2): - left = samples[index] - right = samples[index + 1] - mono.append(_clip_int16((left * lfactor) + (right * rfactor))) - return _write_int16_samples(mono) - - @staticmethod - def ratecv( - fragment: bytes, - width: int, - nchannels: int, - inrate: int, - outrate: int, - state, - weightA: int = 1, - weightB: int = 0, - ) -> tuple[bytes, None]: - del state, weightA, weightB - if nchannels <= 0: - raise ValueError("nchannels must be positive") - if inrate <= 0 or outrate <= 0: - raise ValueError("Sample rates must be positive") - samples = _read_int16_samples(fragment, width) - if not samples or inrate == outrate: - return fragment, None - if len(samples) % nchannels != 0: - raise ValueError("PCM fragment size does not match channel count") - - frame_count = len(samples) // nchannels - output_frame_count = max(1, int(round(frame_count * outrate / inrate))) - output = array("h") - - for out_index in range(output_frame_count): - position = out_index * inrate / outrate - left_index = min(int(position), frame_count - 1) - right_index = min(left_index + 1, frame_count - 1) - fraction = max(0.0, min(1.0, position - left_index)) - for channel_index in range(nchannels): - left_sample = samples[(left_index * nchannels) + channel_index] - right_sample = samples[(right_index * nchannels) + channel_index] - interpolated = left_sample + ((right_sample - left_sample) * fraction) - output.append(_clip_int16(interpolated)) - return _write_int16_samples(output), None - - -audioop = _audioop or _AudioopFallback() - -__all__ = ["audioop"] diff --git a/core/filler_audio.py b/core/filler_audio.py index bc287dd..ff91e16 100644 --- a/core/filler_audio.py +++ b/core/filler_audio.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import audioop import io import logging import os @@ -9,7 +10,6 @@ import wave from collections.abc import AsyncIterable from collections.abc import Iterable -from realtime_voice_service.audioop_compat import audioop from realtime_voice_service.providers.base import BaseTTS diff --git a/core/session.py b/core/session.py index dc809b9..f3452e4 100644 --- a/core/session.py +++ b/core/session.py @@ -1214,14 +1214,6 @@ class CallSession: def conversation(self) -> tuple[tuple[str, str], ...]: return tuple(self._conversation) - @property - def customer_name(self) -> str | None: - return self._customer_name - - @property - def session_language(self) -> str | None: - return self._session_language - async def run(self) -> None: if self._assistant_task is not None: raise RuntimeError("CallSession.run() can only be called once per session") diff --git a/main.py b/main.py index 6328309..dcc08bc 100644 --- a/main.py +++ b/main.py @@ -6,12 +6,10 @@ import logging import os import uuid from contextlib import asynccontextmanager -from datetime import datetime, timezone from fastapi import FastAPI, WebSocket from realtime_voice_service import crm_client -from realtime_voice_service import sales_sync_client from realtime_voice_service.core.filler_audio import FillerAudioLibrary from realtime_voice_service.core.session import CallSession from realtime_voice_service.core.session import _normalize_voice_pronunciation @@ -150,52 +148,6 @@ def _llm_model_name() -> str: return str(os.getenv("OPENAI_LLM_MODEL", "gpt-4o-mini")).strip() or "gpt-4o-mini" -def _utc_now_iso() -> str: - return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") - - -def _conversation_transcript(session: CallSession) -> str | None: - lines = [f"{speaker}: {text.strip()}" for speaker, text in session.conversation if str(text).strip()] - if not lines: - return None - return "\n".join(lines) - - -def _conversation_summary(session: CallSession) -> str | None: - user_turns = [text.strip() for speaker, text in session.conversation if speaker == "user" and str(text).strip()] - assistant_turns = [text.strip() for speaker, text in session.conversation if speaker == "assistant" and str(text).strip()] - if not user_turns and not assistant_turns: - return None - parts: list[str] = [] - if user_turns: - parts.append(f"Customer asked: {user_turns[0][:220]}") - if assistant_turns: - parts.append(f"Assistant response: {assistant_turns[-1][:220]}") - if session.customer_name: - parts.append(f"Customer name: {session.customer_name}") - return " | ".join(parts)[:700] - - -def _session_sync_metadata( - session: CallSession, - transport: BaseMediaTransport, - *, - failure_reason: str | None = None, -) -> dict[str, object]: - return { - "protocol": transport.protocol, - "sample_rate_hz": transport.sample_rate_hz, - "frame_duration_ms": transport.frame_duration_ms, - "frame_bytes": transport.frame_bytes, - "session_language": session.session_language, - "customer_name": session.customer_name, - "interruptions": list(session.interruptions), - "latency_ms": dict(session.last_latency_ms), - "conversation_entries": len(session.conversation), - "failure_reason": failure_reason, - } - - class RealtimeVoiceService: def __init__(self) -> None: @@ -324,53 +276,19 @@ class RealtimeVoiceService: async def _run_transport_session(self, transport: BaseMediaTransport) -> None: session = self._build_session(transport) - started_at = _utc_now_iso() interaction_id = await crm_client.create_interaction(call_id=session.session_id) - await sales_sync_client.sync_voice_session( - call_id=session.session_id, - interaction_id=interaction_id, - voice_session_id=session.session_id, - ai_state=str(session.state.value).lower(), - telephony_status="connected", - call_status="started", - started_at=started_at, - summary="Realtime voice session started", - metadata=_session_sync_metadata(session, transport), - ) - failure_reason: str | None = None - call_status = "completed" async with self._track_session(session): - try: - LOGGER.info( - "starting realtime session %s via %s sample_rate=%s frame_ms=%s frame_bytes=%s", - session.session_id, - transport.protocol, - transport.sample_rate_hz, - transport.frame_duration_ms, - transport.frame_bytes, - ) - await session.run() - except Exception as exc: - call_status = "failed" - failure_reason = str(exc)[:300] - raise - finally: - await sales_sync_client.sync_voice_session( - call_id=session.session_id, - interaction_id=interaction_id, - caller_name=session.customer_name, - voice_session_id=session.session_id, - ai_state=str(session.state.value).lower(), - telephony_status="ended", - call_status=call_status, - started_at=started_at, - ended_at=_utc_now_iso(), - summary=_conversation_summary(session), - transcript_text=_conversation_transcript(session), - metadata=_session_sync_metadata(session, transport, failure_reason=failure_reason), - ) - if interaction_id: - await crm_client.close_interaction(interaction_id) + LOGGER.info( + "starting realtime session %s via %s sample_rate=%s frame_ms=%s frame_bytes=%s", + session.session_id, + transport.protocol, + transport.sample_rate_hz, + transport.frame_duration_ms, + transport.frame_bytes, + ) + await session.run() + if interaction_id: + await crm_client.close_interaction(interaction_id) def _build_session(self, transport: BaseMediaTransport) -> CallSession: return CallSession( @@ -446,8 +364,6 @@ async def health() -> dict[str, object]: "sample_rate_hz": service.sample_rate_hz, "active_sessions": service.active_session_count, "audiosocket_port": _audiosocket_port(), - "sales_sync_enabled": os.getenv("SALES_SYNC_ENABLED", "0"), - "crm_interaction_enabled": os.getenv("CRM_INTERACTION_ENABLED", "0"), } diff --git a/providers/stt.py b/providers/stt.py index 754b17f..8e455c0 100644 --- a/providers/stt.py +++ b/providers/stt.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import audioop import base64 import contextlib import io @@ -12,7 +13,6 @@ import wave from collections.abc import Sequence from urllib.parse import urlencode -from realtime_voice_service.audioop_compat import audioop from realtime_voice_service.providers.base import BaseSTT from realtime_voice_service.providers.base import BaseSTTStream from realtime_voice_service.providers.base import PartialTranscriptCallback diff --git a/providers/tts.py b/providers/tts.py index f5311d1..001e283 100644 --- a/providers/tts.py +++ b/providers/tts.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import audioop import base64 import json import logging @@ -11,7 +12,6 @@ from collections.abc import AsyncIterable from urllib.parse import quote from urllib.parse import urlencode -from realtime_voice_service.audioop_compat import audioop from realtime_voice_service.providers.base import BaseTTS diff --git a/sales_sync_client.py b/sales_sync_client.py deleted file mode 100644 index aeb7fcf..0000000 --- a/sales_sync_client.py +++ /dev/null @@ -1,126 +0,0 @@ -from __future__ import annotations - -import logging -import os -from typing import Any - -import httpx - -from realtime_voice_service.crm_client import _issue_service_token - - -LOGGER = logging.getLogger("uvicorn.error") - - -def _enabled() -> bool: - return os.getenv("SALES_SYNC_ENABLED", "0").strip().lower() in {"1", "true", "yes", "on"} - - -def _base_url() -> str: - return str(os.getenv("SALES_SERVICE_URL", "http://sales-service:8000")).rstrip("/") - - -def _timeout_seconds() -> float: - raw = str(os.getenv("SALES_SYNC_TIMEOUT_SECONDS", "8")).strip() - try: - return max(float(raw), 1.0) - except ValueError: - return 8.0 - - -def _secret() -> str: - return str(os.getenv("CRM_APP_TOKEN_SECRET", "")).strip() - - -def _service_headers() -> dict[str, str] | None: - secret = _secret() - if not secret: - LOGGER.warning("sales sync: CRM_APP_TOKEN_SECRET not configured") - return None - token = _issue_service_token(secret) - return {"Authorization": f"Bearer {token}"} - - -def _queue_id() -> str | None: - raw = str( - os.getenv("REALTIME_VOICE_QUEUE_ID") - or os.getenv("CRM_QUEUE_ID") - or "" - ).strip() - return raw or None - - -def _queue_code() -> str | None: - raw = str(os.getenv("REALTIME_VOICE_QUEUE_CODE", "")).strip() - return raw or None - - -async def sync_voice_session( - *, - call_id: str, - interaction_id: str | None = None, - caller_number: str | None = None, - caller_name: str | None = None, - voice_session_id: str | None = None, - ai_session_id: str | None = None, - ai_state: str | None = None, - handoff_reason: str | None = None, - telephony_status: str | None = None, - call_status: str | None = None, - started_at: str | None = None, - ended_at: str | None = None, - summary: str | None = None, - transcript_text: str | None = None, - metadata: dict[str, Any] | None = None, -) -> dict[str, Any] | None: - if not _enabled(): - return None - headers = _service_headers() - if headers is None: - return None - - payload: dict[str, Any] = { - "call_id": call_id, - "interaction_id": interaction_id, - "queue_id": _queue_id(), - "queue_code": _queue_code(), - "caller_number": caller_number, - "caller_name": caller_name, - "voice_session_id": voice_session_id or call_id, - "ai_session_id": ai_session_id, - "ai_state": ai_state, - "handoff_reason": handoff_reason, - "telephony_status": telephony_status, - "call_status": call_status, - "started_at": started_at, - "ended_at": ended_at, - "summary": summary, - "transcript_text": transcript_text, - "metadata": { - "source_service": "realtime_voice_service", - **(metadata or {}), - }, - } - try: - async with httpx.AsyncClient(timeout=_timeout_seconds()) as client: - response = await client.post( - f"{_base_url()}/internal/sales-sync/voice", - json=payload, - headers=headers, - ) - response.raise_for_status() - data = response.json() - LOGGER.info( - "sales sync: voice session synced call_id=%s voice_session_id=%s status=%s", - call_id, - voice_session_id or call_id, - call_status, - ) - return data if isinstance(data, dict) else None - except Exception: - LOGGER.exception( - "sales sync: failed to sync voice session call_id=%s voice_session_id=%s", - call_id, - voice_session_id or call_id, - ) - return None diff --git a/transports/audiosocket.py b/transports/audiosocket.py index 183e96a..1f0f7c2 100644 --- a/transports/audiosocket.py +++ b/transports/audiosocket.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import audioop import logging import os import struct @@ -8,7 +9,6 @@ import time import uuid from typing import Awaitable, Callable -from realtime_voice_service.audioop_compat import audioop from realtime_voice_service.transports.base import BaseMediaTransport