import asyncio import json import logging import ssl import time from pathlib import Path from typing import Any, Awaitable, Callable import httpx import websockets from .config import Settings log = logging.getLogger(__name__) class ThreeCXClient: def __init__(self, settings: Settings): self.s = settings self._token = None self._expires_at = 0.0 self._token_lock = asyncio.Lock() async def token(self) -> str: if self._token and time.time() < self._expires_at - 60: return self._token async with self._token_lock: if self._token and time.time() < self._expires_at - 60: return self._token secret = Path(self.s.threecx_api_key).read_text().strip() async with httpx.AsyncClient( verify=self.s.threecx_verify_tls, timeout=15, ) as client: response = await client.post( f"{self.s.threecx_base_url}/connect/token", data={ "client_id": self.s.threecx_client_id, "client_secret": secret, "grant_type": "client_credentials", }, ) response.raise_for_status() data = response.json() self._token = data["access_token"] self._expires_at = time.time() + int(data.get("expires_in", 3600)) return self._token async def invalidate_token(self) -> None: async with self._token_lock: self._token = None self._expires_at = 0.0 async def get_dn(self, dn: str) -> dict[str, Any]: for attempt in range(2): token = await self.token() async with httpx.AsyncClient( verify=self.s.threecx_verify_tls, timeout=15, ) as client: response = await client.get( f"{self.s.threecx_base_url}/callcontrol/{dn}", headers={"Authorization": f"Bearer {token}"}, ) if response.status_code == 401 and attempt == 0: log.warning("3CX token rejected for get_dn(%s); refreshing token", dn) await self.invalidate_token() continue response.raise_for_status() return response.json() raise RuntimeError("3CX get_dn failed after token refresh") async def get_participants(self, dn: str) -> list[dict[str, Any]]: for attempt in range(2): token = await self.token() async with httpx.AsyncClient( verify=self.s.threecx_verify_tls, timeout=15, ) as client: response = await client.get( f"{self.s.threecx_base_url}/callcontrol/{dn}/participants", headers={"Authorization": f"Bearer {token}"}, ) if response.status_code == 401 and attempt == 0: log.warning( "3CX token rejected for get_participants(%s); refreshing token", dn, ) await self.invalidate_token() continue response.raise_for_status() data = response.json() if isinstance(data, list): return data return data.get("participants", []) raise RuntimeError("3CX get_participants failed after token refresh") async def make_call( self, dn: str, device_id: str, destination: str, timeout_sec: int = 30, ) -> dict[str, Any]: for attempt in range(2): token = await self.token() async with httpx.AsyncClient( verify=self.s.threecx_verify_tls, timeout=15, ) as client: response = await client.post( f"{self.s.threecx_base_url}/callcontrol/{dn}/devices/{device_id}/makecall", headers={"Authorization": f"Bearer {token}"}, json={ "destination": destination, "timeoutSec": timeout_sec, }, ) if response.status_code == 401 and attempt == 0: log.warning( "3CX token rejected for make_call(%s -> %s); refreshing token", dn, destination, ) await self.invalidate_token() continue response.raise_for_status() return response.json() raise RuntimeError("3CX make_call failed after token refresh") async def get_call_log( self, period_from: str, period_to: str, top: int = 500, skip: int = 0, ) -> dict[str, Any]: """Read current 3CX CDR/CallLog data via ReportCallLogData.""" for attempt in range(2): token = await self.token() params = ( f"periodFrom={period_from}," f"periodTo={period_to}," "sourceType=0," "sourceFilter=''," "destinationType=0," "destinationFilter=''," "callsType=0," "callTimeFilterType=0," "callTimeFilterFrom='0:00:0'," "callTimeFilterTo='0:00:0'," "hidePcalls=true" ) url = ( f"{self.s.threecx_base_url}/xapi/v1/" f"ReportCallLogData/Pbx.GetCallLogData({params})" f"?$top={top}&$skip={skip}&$count=true" ) async with httpx.AsyncClient( verify=self.s.threecx_verify_tls, timeout=120, ) as client: response = await client.get( url, headers={ "Authorization": f"Bearer {token}", "Accept": "application/json", }, ) if response.status_code == 401 and attempt == 0: log.warning("3CX token rejected for get_call_log; refreshing token") await self.invalidate_token() continue response.raise_for_status() return response.json() raise RuntimeError("3CX get_call_log failed after token refresh") async def download_recording(self, rec_id: int) -> tuple[bytes, str]: """Download a historical 3CX recording by recording ID.""" token = await self.token() url = ( f"{self.s.threecx_base_url}/xapi/v1/" f"Recordings/Pbx.DownloadRecording(recId={int(rec_id)})" ) async with httpx.AsyncClient( verify=self.s.threecx_verify_tls, timeout=120, ) as client: response = await client.get( url, headers={ "Authorization": f"Bearer {token}", "Accept": "audio/x-wav,*/*", }, ) if response.status_code == 401: await self.invalidate_token() token = await self.token() async with httpx.AsyncClient( verify=self.s.threecx_verify_tls, timeout=120, ) as client: response = await client.get( url, headers={ "Authorization": f"Bearer {token}", "Accept": "audio/x-wav,*/*", }, ) response.raise_for_status() content_type = response.headers.get( "content-type", "audio/x-wav", ) return response.content, content_type async def websocket(self, on_event: Callable[[dict[str, Any]], Awaitable[None]]): token = await self.token() uri = self.s.threecx_base_url.replace("https://", "wss://").replace("http://", "ws://") uri += "/callcontrol/ws" ssl_context = None if uri.startswith("wss://") and not self.s.threecx_verify_tls: ssl_context = ssl.create_default_context() ssl_context.check_hostname = False ssl_context.verify_mode = ssl.CERT_NONE async with websockets.connect( uri, additional_headers={"Authorization": f"Bearer {token}"}, ssl=ssl_context, ping_interval=20, ping_timeout=20, ) as ws: await ws.send(json.dumps({ "RequestID": "telephony-middleware", "Path": "/callcontrol", })) log.info("3CX WebSocket connected") async for raw in ws: try: message = json.loads(raw) except json.JSONDecodeError: continue await on_event(message) async def run_websocket(self, on_event): delay = 2 while True: try: await self.websocket(on_event) delay = 2 except asyncio.CancelledError: raise except Exception: log.exception("3CX WebSocket failed; reconnect in %ss", delay) await asyncio.sleep(delay) delay = min(delay * 2, 60)