"""Gestión de sesiones externalMedia + WebRTC / bot por llamada.""" from __future__ import annotations import logging from typing import Any from ari.client import AriClient from calls.models import CallState from calls.registry import CallRegistry from config import Settings, get_settings from media.rtp_session import RtpSession logger = logging.getLogger(__name__) class MediaManager: def __init__(self, settings: Settings | None = None) -> None: self.settings = settings or get_settings() self._sessions: dict[str, object] = {} self._rtp_sessions: dict[str, RtpSession] = {} @property def enabled(self) -> bool: return self.settings.webrtc_enabled def has_session(self, call_id: str) -> bool: return call_id in self._sessions def get_rtp_session(self, call_id: str) -> RtpSession | None: return self._rtp_sessions.get(call_id) def ice_servers(self) -> list[dict[str, Any]]: servers: list[dict[str, Any]] = [] if self.settings.webrtc_stun_url: servers.append({"urls": self.settings.webrtc_stun_url}) return servers async def prepare_rtp_session(self, call_id: str) -> RtpSession: """Socket RTP únicamente (bot o base para WebRTC).""" existing = self._rtp_sessions.get(call_id) if existing and existing.local_port and not existing.is_closed: logger.info( "Reutilizando RTP call_id=%s (puerto %s)", call_id, existing.local_port, ) return existing old_rtp = self._rtp_sessions.pop(call_id, None) if old_rtp: old_rtp.close() rtp = RtpSession() port = await rtp.start( bind_host=self.settings.external_media_bind_host, port=0, ) self._rtp_sessions[call_id] = rtp logger.info("Sesión RTP preparada call_id=%s (puerto %s)", call_id, port) return rtp async def prepare_session(self, call_id: str) -> None: """Un socket RTP + WebRTC por llamada (mismo puerto que externalMedia).""" if call_id in self._sessions and call_id in self._rtp_sessions: rtp = self._rtp_sessions[call_id] if not rtp.is_closed and rtp.local_port: logger.info( "Reutilizando sesión WebRTC call_id=%s (RTP puerto %s)", call_id, rtp.local_port, ) return old = self._sessions.pop(call_id, None) if old: await old.close() # type: ignore[union-attr] # WebRTC nuevo: RTP limpio (el close de WebRTC puede haber cerrado el socket) old_rtp = self._rtp_sessions.pop(call_id, None) if old_rtp: old_rtp.close() rtp = await self.prepare_rtp_session(call_id) try: from media.webrtc_session import WebRtcBridgeSession except ImportError as exc: rtp.close() self._rtp_sessions.pop(call_id, None) raise RuntimeError( "WebRTC requiere aiortc. Ejecutá: pip install aiortc av" ) from exc self._sessions[call_id] = WebRtcBridgeSession( call_id, rtp, ice_servers=self.ice_servers(), ) logger.info( "Sesión WebRTC preparada call_id=%s (RTP puerto %s)", call_id, rtp.local_port, ) def rtp_port(self, call_id: str) -> int | None: rtp = self._rtp_sessions.get(call_id) return rtp.local_port if rtp else None async def _create_external_media_channel( self, call: CallState, ari: AriClient, registry: CallRegistry, *, require_bridge: bool = False, ) -> None: if call.external_media_channel_id: logger.info( "externalMedia ya existe para %s (%s)", call.call_id, call.external_media_channel_id, ) return rtp = self._rtp_sessions.get(call.call_id) if rtp is None or rtp.local_port is None: raise RuntimeError("Puerto RTP no disponible") advertise = self.settings.external_media_advertise_host external_host = f"{advertise}:{rtp.local_port}" logger.info( "Creando externalMedia call=%s Asterisk→RTP %s (bind=%s)", call.call_id, external_host, self.settings.external_media_bind_host, ) try: channel = await ari.create_external_media( external_host=external_host, app_args=[call.call_id, "media"], fmt=self.settings.external_media_format, ) channel_id = channel["id"] registry.link_channel(call, channel_id) call.external_media_channel_id = channel_id await self._prime_rtp_remote_from_asterisk(ari, channel_id, rtp) if call.bridge_id: await ari.add_to_bridge(call.bridge_id, channel_id) call.external_media_attached = True logger.info( "externalMedia %s → %s en puente %s (llamada %s)", channel_id, external_host, call.bridge_id, call.call_id, ) else: if require_bridge: raise RuntimeError( f"Bot sin puente mixing para llamada {call.call_id}" ) logger.info( "externalMedia %s → %s para llamada %s (sin puente aún)", channel_id, external_host, call.call_id, ) except Exception as exc: logger.error( "No se pudo crear externalMedia para %s: %s", call.call_id, exc, ) raise async def _prime_rtp_remote_from_asterisk( self, ari: AriClient, channel_id: str, rtp: RtpSession, ) -> None: """Usa UNICASTRTP_LOCAL_* para poder enviar TTS sin esperar RTP del caller. WebRTC suele mandar silencio continuo; muchos softphones UDP no envían nada hasta que hay audio (VAD). Sin destino prearmado el bot hace timeout. """ host = await ari.get_channel_variable(channel_id, "UNICASTRTP_LOCAL_ADDRESS") port_raw = await ari.get_channel_variable(channel_id, "UNICASTRTP_LOCAL_PORT") if not host or not port_raw: logger.warning( "externalMedia %s sin UNICASTRTP_LOCAL_* (host=%s port=%s)", channel_id, host, port_raw, ) return try: port = int(port_raw) except ValueError: logger.warning( "UNICASTRTP_LOCAL_PORT inválido en %s: %s", channel_id, port_raw, ) return rtp.set_remote(host, port) async def attach_external_media( self, call: CallState, ari: AriClient, registry: CallRegistry, ) -> None: """WebRTC operador: RTP + sesión browser.""" if not self.enabled: return await self.prepare_session(call.call_id) await self._create_external_media_channel(call, ari, registry) async def attach_bot_external_media( self, call: CallState, ari: AriClient, registry: CallRegistry, ) -> None: """Bot IA: solo RTP ↔ Asterisk (sin WebRTC).""" await self.prepare_rtp_session(call.call_id) await self._create_external_media_channel( call, ari, registry, require_bridge=True ) async def apply_offer( self, call_id: str, sdp: str, offer_type: str ) -> dict[str, str]: await self.prepare_session(call_id) session = self._sessions[call_id] return await session.apply_offer(sdp, offer_type) # type: ignore[union-attr] async def add_ice_candidate( self, call_id: str, candidate: str | None, sdp_mid: str | None, sdp_mline_index: int | None, ) -> None: if call_id not in self._sessions: return session = self._sessions[call_id] await session.add_ice_candidate( # type: ignore[union-attr] candidate, sdp_mid, sdp_mline_index ) async def close_session(self, call_id: str) -> None: session = self._sessions.pop(call_id, None) rtp = self._rtp_sessions.pop(call_id, None) if session: await session.close() # type: ignore[union-attr] logger.info("Sesión WebRTC cerrada call_id=%s", call_id) if rtp: rtp.close() if not session: logger.info("Sesión RTP (bot) cerrada call_id=%s", call_id)