3cx-telefonie-middleware/app/threecx.py

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)