"""Tracker en memoria de CallSession + flush a MySQL al colgar.""" from __future__ import annotations import logging from datetime import datetime from typing import Protocol from calls.models import CallState from call_metrics.cost import CostRates from call_metrics.persist import CallMetricsMySqlStore from call_metrics.session import CallSession, utc_now from config import Settings logger = logging.getLogger(__name__) class MetricsTracker(Protocol): """Contrato usado por CallService / BotConversationSession.""" enabled: bool def start_call(self, call: CallState) -> CallSession | None: ... def get(self, call_id: str) -> CallSession | None: ... def mark_answered(self, call_id: str, when: datetime | None = None) -> None: ... def set_providers( self, call_id: str, *, stt: str | None = None, tts: str | None = None, llm: str | None = None, llm_model: str | None = None, ) -> None: ... async def finalize_and_persist( self, call: CallState, *, reason: str, ) -> None: ... async def startup(self) -> None: ... async def shutdown(self) -> None: ... class NullCallMetricsTracker: """No-op cuando ``CALL_METRICS_ENABLED=false``.""" enabled = False def start_call(self, call: CallState) -> CallSession | None: return None def get(self, call_id: str) -> CallSession | None: return None def mark_answered(self, call_id: str, when: datetime | None = None) -> None: return None def set_providers( self, call_id: str, *, stt: str | None = None, tts: str | None = None, llm: str | None = None, llm_model: str | None = None, ) -> None: return None async def finalize_and_persist( self, call: CallState, *, reason: str, ) -> None: return None async def startup(self) -> None: return None async def shutdown(self) -> None: return None class CallMetricsTracker: """Registry in-memory de CallSession + una INSERT MySQL al fin.""" enabled = True def __init__( self, settings: Settings, *, store: CallMetricsMySqlStore | None = None, rates: CostRates | None = None, ) -> None: self.settings = settings self._sessions: dict[str, CallSession] = {} # call_ids ya insertados: evita recrear sesión vacía post-cleanup self._persisted_ids: set[str] = set() self._store = store or CallMetricsMySqlStore( host=settings.call_metrics_mysql_host, port=settings.call_metrics_mysql_port, user=settings.call_metrics_mysql_user, password=settings.call_metrics_mysql_password, database=settings.call_metrics_mysql_database, ) self._rates = rates or CostRates( llm_input_per_1m=settings.call_metrics_price_llm_input_per_1m, llm_output_per_1m=settings.call_metrics_price_llm_output_per_1m, stt_per_minute=settings.call_metrics_price_stt_per_minute, tts_per_1k_chars=settings.call_metrics_price_tts_per_1k_chars, ) async def startup(self) -> None: try: await self._store.connect() except Exception: logger.exception( "Call metrics: no se pudo conectar a MySQL — " "se seguirá acumulando en memoria pero el INSERT fallará" ) async def shutdown(self) -> None: # Flush residual (shutdown del proceso) for call_id, session in list(self._sessions.items()): if session._persisted: continue try: session.finalize("shutdown") cost = self._rates.estimate(session) if self._store.ready: await self._store.persist(session, cost) session._persisted = True self._persisted_ids.add(call_id) except Exception: logger.exception( "Call metrics: fallo flush en shutdown call=%s", call_id ) self._sessions.clear() await self._store.close() def start_call(self, call: CallState) -> CallSession | None: if call.call_id in self._persisted_ids: return None existing = self._sessions.get(call.call_id) if existing and not existing._persisted: # Actualizar metadata si llega más tarde (doc, número) existing.document_id = call.document_id or existing.document_id existing.number = call.number or existing.number existing.direction = call.direction or existing.direction return existing session = CallSession( call_id=call.call_id, start_time=call.started_at or utc_now(), document_id=call.document_id, number=call.number, direction=call.direction, ) if call.status in ("answered", "talking"): session.mark_answered() self._sessions[call.call_id] = session logger.debug("Call metrics: sesión iniciada call=%s", call.call_id) return session def get(self, call_id: str) -> CallSession | None: return self._sessions.get(call_id) def mark_answered(self, call_id: str, when: datetime | None = None) -> None: session = self._sessions.get(call_id) if session: session.mark_answered(when) def set_providers( self, call_id: str, *, stt: str | None = None, tts: str | None = None, llm: str | None = None, llm_model: str | None = None, ) -> None: session = self._sessions.get(call_id) if not session: return if stt: session.stt_provider = stt if tts: session.tts_provider = tts if llm: session.llm_provider = llm if llm_model: session.llm_model = llm_model async def finalize_and_persist( self, call: CallState, *, reason: str, ) -> None: session = self._sessions.get(call.call_id) if session is None: # No inventar una sesión vacía: un segundo evento (StasisEnd + # ChannelDestroyed) pisaría/duplicaría con costo ~0. logger.debug( "Call metrics: finalize sin sesión en memoria call=%s reason=%s", call.call_id, reason, ) return if session._persisted: self._sessions.pop(call.call_id, None) return session.document_id = call.document_id or session.document_id session.number = call.number or session.number session.direction = call.direction or session.direction disconnect = ( reason or getattr(call, "disconnect_reason", None) or "unknown" ) session.finalize(disconnect) cost = self._rates.estimate(session) try: if not self._store.ready: await self._store.connect() await self._store.persist(session, cost) session._persisted = True self._persisted_ids.add(call.call_id) except Exception: logger.exception( "Call metrics: fallo INSERT call=%s (queda en memoria hasta retry/shutdown)", call.call_id, ) return self._sessions.pop(call.call_id, None) logger.info( "Call metrics finalizado call=%s reason=%s duration=%.1fs " "ttfa=%sms cost≈$%.6f silence=%.1f%% avg_resp=%sms", call.call_id, disconnect, session.duration_seconds or 0.0, session.time_to_first_audio_ms, cost, session.silence_percent or 0.0, ( f"{session.avg_response_latency_ms:.0f}" if session.avg_response_latency_ms is not None else "-" ), ) def create_metrics_tracker(settings: Settings) -> MetricsTracker: """Factory: tracker real o no-op según ``CALL_METRICS_ENABLED``.""" if not settings.call_metrics_enabled: logger.info("Call metrics DESHABILITADO (CALL_METRICS_ENABLED=false)") return NullCallMetricsTracker() logger.info( "Call metrics HABILITADO → MySQL %s:%s/%s", settings.call_metrics_mysql_host, settings.call_metrics_mysql_port, settings.call_metrics_mysql_database, ) return CallMetricsTracker(settings)