import asyncio import logging import time from typing import Any, Callable, Awaitable from ari.client import AriClient from ari.debug_log import log_ari_event from ari.events import AriEventsListener from bot_actions import BotActionTrigger from call_metrics import create_metrics_tracker from call_metrics.tracker import MetricsTracker from calls.models import CallState from calls.registry import CallRegistry from calls.state_machine import CallStateMachine, PROCESSED_CHANNELS from config import Settings, get_settings from crm.client import CrmApiClient, CrmApiError from crm.factory import create_crm_client from media.manager import MediaManager from services.agent_session import AgentSessionStore from services.bot_session import BotSessionManager logger = logging.getLogger(__name__) BroadcastFn = Callable[[dict[str, Any]], Awaitable[None]] # Watchdog canal bot: lento si el WS ARI está sano; más frecuente si falla. _BOT_WATCH_INTERVAL_WS_OK = 10.0 _BOT_WATCH_INTERVAL_WS_DOWN = 2.0 class CallService: def __init__( self, settings: Settings | None = None, broadcast: BroadcastFn | None = None, crm: CrmApiClient | None = None, metrics: MetricsTracker | None = None, ) -> None: self.settings = settings or get_settings() self.ari = AriClient(self.settings) self.registry = CallRegistry() self.media_manager = MediaManager(self.settings) self.crm = crm if crm is not None else create_crm_client(self.settings) self._owns_crm = crm is None self.metrics = metrics if metrics is not None else create_metrics_tracker(self.settings) self.bot_manager = BotSessionManager( self.media_manager, self.settings, on_update=self._notify, on_action=self._on_bot_action, metrics=self.metrics, ) self.state_machine = CallStateMachine( self.ari, self.registry, self.settings, media_manager=self.media_manager, call_resolver=self.resolve_call_for_event, get_agent_endpoint=self.get_agent_endpoint, get_registered_extension=self.get_registered_extension, on_bot_ready=self._on_bot_ready, ) self._broadcast = broadcast self.agent_session = AgentSessionStore() self.ari_listener = AriEventsListener(self._handle_ari_event, self.settings) self._pending_outbound: dict[str, str] = {} self._outbound_calls: dict[str, CallState] = {} self._cleaned_calls: set[str] = set() def resolve_call_for_event( self, channel_id: str | None, event: dict[str, Any] ) -> CallState | None: """Busca la llamada por canal, call_id UUID en args o caché de salientes.""" args = event.get("args") or [] call_id = None for raw in args: value = str(raw).strip().strip("'\"") if value and CallStateMachine._looks_like_call_id(value): call_id = value break if channel_id: by_ch = self.registry.get_by_channel(channel_id) if by_ch: return by_ch if call_id: found = self.registry.get(call_id) if found: return found cached = self._outbound_calls.get(call_id) if cached: self.registry.add(cached) if channel_id: self.registry.link_channel(cached, channel_id) logger.info( "Llamada %s re-sincronizada al registry (Stasis)", call_id, ) return cached if channel_id: for cid, ch in self._pending_outbound.items(): if ch == channel_id: cached = self._outbound_calls.get(cid) if cached: self.registry.add(cached) self.registry.link_channel(cached, channel_id) return cached return None def set_broadcast(self, fn: BroadcastFn) -> None: self._broadcast = fn self.bot_manager.set_on_update(self._notify) async def _on_bot_action( self, call: CallState, trigger: BotActionTrigger ) -> None: """Acciones disparadas por frases del bot (TE TRANSFIERO, etc.).""" action = (trigger.action or "").strip().lower() if action == "transfer": await self.transfer_call_to_number(call, trigger.target) return if action == "update_plan": logger.info("Actualizando plan call=%s — cortando llamada", call.call_id) await self.hangup(call.call_id) return logger.warning( "Acción de bot desconocida %r call=%s", trigger.action, call.call_id, ) async def transfer_call_to_number( self, call: CallState, number: str ) -> None: """Detiene el bot, saca externalMedia y origina destino al puente.""" target = "".join(c for c in (number or "") if c.isdigit() or c == "+") if not target: logger.error( "Transferencia sin destino válido call=%s", call.call_id, ) return if call.transfer_in_progress or call.agent_leg_originated: logger.info( "Transferencia ya en curso/hecha call=%s", call.call_id, ) return if call.status in ("ended", "failed"): return if not call.bridge_id: logger.error( "Transferencia sin bridge call=%s", call.call_id, ) return call.transfer_in_progress = True call.transfer_target = target call.agent_state = "transferring" await self._notify(call) endpoint = self.settings.format_endpoint(target) logger.info( "Transferencia call=%s → %s (%s)", call.call_id, target, endpoint, ) # Cortar conversación bot (STT/LLM/TTS) await self.bot_manager.stop_for_call(call.call_id) # Sacar UnicastRTP del puente para que quede cliente ↔ destino media_ch = call.external_media_channel_id if media_ch and call.bridge_id: try: await self.ari.remove_from_bridge(call.bridge_id, media_ch) except Exception as exc: logger.warning( "No se pudo quitar media del puente call=%s: %s", call.call_id, exc, ) try: await self.ari.hangup(media_ch) except Exception as exc: logger.debug( "Hangup media post-transfer call=%s: %s", call.call_id, exc, ) call.external_media_channel_id = None call.external_media_attached = False if media_ch in call.channel_ids: call.channel_ids = [c for c in call.channel_ids if c != media_ch] await self.media_manager.close_session(call.call_id) try: channel = await self.ari.originate_channel( endpoint, caller_id=self.settings.outbound_caller_id, use_stasis=True, app_args=[call.call_id, "agent"], ) call.agent_leg_originated = True self.registry.link_channel(call, channel["id"]) logger.info( "Pata transferencia originada call=%s channel=%s → %s", call.call_id, channel["id"], endpoint, ) except Exception as exc: call.transfer_in_progress = False call.agent_leg_originated = False call.agent_state = "error" logger.error( "Fallo originar transferencia call=%s → %s: %s", call.call_id, endpoint, exc, ) await self._notify(call) return call.agent_state = "transferred" await self._notify(call) async def _on_bot_ready(self, call: CallState) -> None: logger.info( "Iniciando conversación bot call=%s providers stt=%s tts=%s llm=%s", call.call_id, self.settings.stt_provider, self.settings.tts_provider, self.settings.llm_provider, ) # CRM en paralelo con el arranque del bot (RTP wait ~1–8s suele cubrir la latencia). poliza_task = asyncio.create_task( self._load_poliza_for_call(call), name=f"poliza-{call.call_id[:8]}", ) await self.bot_manager.start_for_call(call, poliza_task=poliza_task) # Watchdog: fallback si el WS ARI no entrega StasisEnd. customer_channel = next( ( cid for cid in call.channel_ids if cid != call.external_media_channel_id ), call.channel_ids[0] if call.channel_ids else None, ) if customer_channel: asyncio.create_task( self._watch_bot_channel(call.call_id, customer_channel), name=f"watch-bot-{call.call_id[:8]}", ) await self._notify(call) async def _load_poliza_for_call(self, call: CallState) -> None: """GET /polizas/{document_id} y guarda el resultado en call.poliza_data.""" doc = (call.document_id or "").strip() if not doc: logger.warning( "Bot call=%s sin document_id: no se consulta póliza CRM", call.call_id, ) return if not self.settings.crm_api_enabled or not self.crm.enabled: logger.info("CRM deshabilitado: se omite carga de póliza call=%s", call.call_id) return try: poliza = await self.crm.get_poliza(doc) except CrmApiError as exc: logger.error( "Error CRM polizas/%s call=%s: %s", doc, call.call_id, exc, ) return if not poliza: logger.warning( "Póliza no encontrada CRM polizas/%s call=%s", doc, call.call_id, ) return call.poliza_data = poliza.to_public() logger.info( "Póliza cargada call=%s doc=%s poliza=%s producto=%s cliente=%s", call.call_id, doc, poliza.poliza, poliza.producto, poliza.nombre_completo, ) async def startup(self) -> None: try: await self.ari.cleanup_all() except Exception as exc: logger.warning("ARI cleanup on startup failed: %s", exc) PROCESSED_CHANNELS.clear() await self.metrics.startup() self.ari_listener.start() if not self.ari_listener.task_alive: logger.error("No se pudo iniciar la tarea del WebSocket ARI") elif not self.ari_listener.connected: logger.info( "WebSocket ARI conectando… (HTTP ARI alcanzable=%s)", await self.ari_healthy(), ) # No bloquea el arranque HTTP: carga el modelo Ollama en background asyncio.create_task( self._warmup_llm_background(), name="llm-warmup-startup", ) async def _warmup_llm_background(self) -> None: try: await self.bot_manager.warmup_llm() except Exception: logger.exception("Warmup LLM en startup falló") async def shutdown(self) -> None: await self.bot_manager.close() await self.metrics.shutdown() await self.ari_listener.stop() await self.ari.close() if self._owns_crm: await self.crm.close() async def _handle_ari_event(self, event: dict[str, Any]) -> None: ''' if self.settings.ari_debug: log_ari_event( event, full=self.settings.ari_debug_full_events, ) ''' call = await self.state_machine.handle_event(event) if call: self._touch_metrics(call) if call and call.status in ("ended", "failed"): await self._cleanup_after_remote_end(call) if call: await self._notify(call) if event.get("type") == "StasisStart": channel_id = (event.get("channel") or {}).get("id") call = self.resolve_call_for_event(channel_id, event) if call: self._touch_metrics(call) await self._notify(call) def _touch_metrics(self, call: CallState) -> None: """Asegura CallSession en memoria y marca answerTime si ya contestó.""" self.metrics.start_call(call) if call.status in ("answered", "talking"): self.metrics.mark_answered(call.call_id) async def _notify(self, call: CallState) -> None: if self._broadcast: await self._broadcast( {"type": "call_update", "call": call.to_public()} ) async def _cleanup_after_remote_end(self, call: CallState) -> None: """Detiene bot, cuelga canales restantes y libera media/bridge.""" call_id = call.call_id if call_id in self._cleaned_calls: return self._cleaned_calls.add(call_id) # Cortar bot YA: no seguir mandando audio a STT/LLM tras el hangup. await self.bot_manager.stop_for_call(call_id) # Una sola INSERT MySQL con el resumen de la llamada reason = call.disconnect_reason or "cleanup" await self.metrics.finalize_and_persist(call, reason=reason) channel_ids = set(call.channel_ids) if call.external_media_channel_id: channel_ids.add(call.external_media_channel_id) for channel_id in channel_ids: try: await self.ari.hangup(channel_id) except Exception as exc: logger.debug( "Hangup residual %s call=%s: %s", channel_id, call_id, exc, ) await self.media_manager.close_session(call_id) self._pending_outbound.pop(call_id, None) self._outbound_calls.pop(call_id, None) if call.bridge_id: try: await self.ari.delete_bridge(call.bridge_id) except Exception as exc: logger.debug("Bridge %s ya eliminado: %s", call.bridge_id, exc) call.bridge_id = None call.external_media_channel_id = None call.external_media_attached = False call.bot_media_ready = False call.agent_state = None self.registry.remove(call_id) logger.info("Cleanup completo call=%s (bot detenido, canales colgados)", call_id) async def _remote_hangup_detected(self, call: CallState, reason: str) -> None: if call.status in ("ended", "failed"): return if reason and not call.disconnect_reason: call.disconnect_reason = reason call.finalize("ended") logger.info("Colgado remoto call_id=%s (%s)", call.call_id, reason) await self._cleanup_after_remote_end(call) await self._notify(call) def get_agent_endpoint(self) -> str | None: if self.settings.agent_endpoint: return self.settings.agent_endpoint agent = self.agent_session.current if not agent: return None return self.settings.agent_endpoint_template.format(extension=agent.extension) def get_outbound_caller_id(self) -> str: agent = self.agent_session.current if agent: return f"{agent.username} <{agent.extension}>" return self.settings.outbound_caller_id def get_registered_extension(self) -> str | None: agent = self.agent_session.current return agent.extension if agent else None async def register_agent( self, username: str, extension: str, password: str ) -> dict: username = username.strip() extension = extension.strip() if not username or not extension or not password: raise ValueError("username, extension y password son obligatorios") ari_ok = await self.ari_healthy() if not ari_ok: raise RuntimeError("ARI no disponible") endpoint = self.settings.agent_endpoint_template.format(extension=extension) endpoint_verified = False if endpoint.startswith("PJSIP/"): resource = endpoint.split("/", 1)[1] try: info = await self.ari.get_endpoint("PJSIP", resource) endpoint_verified = info is not None except Exception as exc: logger.warning( "No se pudo verificar endpoint %s: %s", endpoint, exc, ) agent = self.agent_session.register(username, extension, password) logger.info( "Agente registrado ext=%s user=%s endpoint=%s verified=%s", extension, username, endpoint, endpoint_verified, ) return { "connected": True, "agent": agent.to_public(), "endpoint": endpoint, "endpoint_verified": endpoint_verified, "webrtc_enabled": self.settings.webrtc_enabled, "inbound_ready": True, "sip": self.settings.get_sip_config(), } def get_agent_status(self) -> dict: agent = self.agent_session.current endpoint = self.get_agent_endpoint() return { "connected": agent is not None, "agent": agent.to_public() if agent else None, "endpoint": endpoint, "webrtc_enabled": self.settings.webrtc_enabled, "inbound_ready": agent is not None, "sip": self.settings.get_sip_config(), } async def answer_inbound(self, call_id: str) -> CallState: call = self.get_call(call_id) if not call: raise KeyError(call_id) try: await self.state_machine.answer_inbound_call(call) except ValueError as exc: raise ValueError(str(exc)) from exc self._touch_metrics(call) await self._notify(call) return call async def start_outbound(self, number: str) -> CallState: endpoint = self.settings.format_endpoint(number) call = CallState( direction="outbound", status="ringing", number=number, ) self.registry.add(call) self._outbound_calls[call.call_id] = call self._cleaned_calls.discard(call.call_id) self.metrics.start_call(call) logger.info( "Llamada saliente registrada call_id=%s (registry=%d)", call.call_id, len(self.registry.list_all()), ) try: channel = await self.ari.originate_channel( endpoint, caller_id=self.get_outbound_caller_id(), use_stasis=True, app_args=[call.call_id, "customer"], ) channel_id = channel["id"] self.registry.link_channel(call, channel_id) self._pending_outbound[call.call_id] = channel_id asyncio.create_task( self._watch_outbound_channel(call.call_id, channel_id), name=f"watch-outbound-{call.call_id[:8]}", ) except Exception as exc: logger.error("Originate failed: %s", exc) call.status = "failed" call.disconnect_reason = call.disconnect_reason or f"originate failed: {exc}" call.finalize("failed") await self.metrics.finalize_and_persist( call, reason=call.disconnect_reason ) await self._notify(call) return call async def hangup(self, call_id: str) -> bool: call = self.registry.get(call_id) if not call: return False for channel_id in list(call.channel_ids): try: await self.ari.hangup(channel_id) except Exception as exc: logger.warning("Hangup channel %s failed: %s", channel_id, exc) if call.external_media_channel_id and call.external_media_channel_id not in call.channel_ids: try: await self.ari.hangup(call.external_media_channel_id) except Exception as exc: logger.warning( "Hangup externalMedia %s failed: %s", call.external_media_channel_id, exc, ) call.finalize("ended") if not call.disconnect_reason: call.disconnect_reason = "hangup API" await self._cleanup_after_remote_end(call) await self._notify(call) return True def list_calls(self) -> list[CallState]: return self.registry.list_all() def get_call(self, call_id: str) -> CallState | None: return self.registry.get(call_id) or self._outbound_calls.get(call_id) async def sync_outbound_media(self, call_id: str) -> CallState | None: """Si el WS ARI falla, configura puente/media vía HTTP cuando el canal está Up.""" call = self.get_call(call_id) if not call or call.direction != "outbound": return call if call.external_media_attached and call.bridge_id: return call channel_id = ( self._pending_outbound.get(call_id) or (call.channel_ids[0] if call.channel_ids else None) ) if not channel_id: return call try: ch = await self.ari.get_channel(channel_id) except Exception as exc: logger.debug("sync_outbound_media %s: %s", call_id, exc) if call.outbound_stasis_setup or call.status in ("answered", "talking"): await self._remote_hangup_detected(call, "canal no existe en ARI") return call if ch.get("state") != "Up": return call event = { "type": "StasisStart", "args": [call_id, "customer"], "channel": ch, } updated = await self.state_machine.handle_event(event) if updated: await self._notify(updated) return updated return call async def _watch_outbound_channel(self, call_id: str, channel_id: str) -> None: """Polling ARI: setup al contestar y detectar colgado remoto sin WebSocket.""" setup_done = False for _ in range(600): await asyncio.sleep(0.5) call = self.get_call(call_id) if not call or call.status in ("ended", "failed"): return try: ch = await self.ari.get_channel(channel_id) except Exception: if setup_done or call.outbound_stasis_setup: await self._remote_hangup_detected( call, f"canal {channel_id} cerrado en Asterisk" ) return state = ch.get("state") if state == "Up": if not setup_done: logger.info( "Canal %s Up — setup por polling (call_id=%s)", channel_id, call_id, ) await self.sync_outbound_media(call_id) setup_done = True call = self.get_call(call_id) if call and call.external_media_attached and call.bridge_id: setup_done = True elif setup_done and state in ("Down", "Ringing"): await self._remote_hangup_detected( call, f"canal cliente pasó a {state}" ) return async def _watch_bot_channel(self, call_id: str, channel_id: str) -> None: """Fallback de colgado remoto si el WS ARI pierde StasisEnd. Con WS sano: poll cada 10s (StasisEnd es el camino normal). Con WS caído o sin eventos recientes: poll cada 2s. """ while True: interval = ( _BOT_WATCH_INTERVAL_WS_OK if self.ari_ws_healthy else _BOT_WATCH_INTERVAL_WS_DOWN ) await asyncio.sleep(interval) call = self.get_call(call_id) if not call or call.status in ("ended", "failed"): return if not self.bot_manager.is_running(call_id): return try: await self.ari.get_channel(channel_id) except Exception: await self._remote_hangup_detected( call, f"canal bot cliente {channel_id} cerrado en Asterisk (watchdog)", ) return async def ensure_webrtc_ready(self, call_id: str) -> CallState: """Sesión WebRTC lista y reintento de externalMedia si hace falta.""" call = self.registry.get(call_id) or self._outbound_calls.get(call_id) if not call: raise KeyError(call_id) if not self.registry.get(call_id): self.registry.add(call) if not self.settings.webrtc_enabled: return call await self.media_manager.prepare_session(call_id) if not call.external_media_attached: try: await self.media_manager.attach_external_media( call, self.ari, self.registry ) except Exception as exc: logger.warning( "externalMedia pendiente para %s: %s", call_id, exc, ) return call @property def ari_connected(self) -> bool: return self.ari_listener.connected @property def ari_events_recent(self) -> bool: last = self.ari_listener.last_event_at return last is not None and (time.monotonic() - last) < 45.0 @property def ari_ws_healthy(self) -> bool: return self.ari_listener.connected or self.ari_events_recent async def ari_healthy(self) -> bool: return await self.ari.health_check()