ring-scheduler-3cx/src/three_cx_client.py

import logging
import requests

logger = logging.getLogger(__name__)


class ThreeCXClient:
    """Client for 3CX V20 cloud-hosted management API (XAPI)."""

    def __init__(self, domain: str, auth_mode: str = "user", **auth_kwargs):
        """
        Args:
            domain: Your 3CX FQDN (e.g. "yourcompany.3cx.us")
            auth_mode: "user" for extension login, "oauth" for Enterprise API client
            auth_kwargs: credentials — see _authenticate_user / _authenticate_oauth
        """
        self.base_url = f"https://{domain}"
        self.session = requests.Session()
        self.session.headers.update({
            "OData-Version": "4.0",
            "Content-Type": "application/json",
        })

        if auth_mode == "oauth":
            self._authenticate_oauth(auth_kwargs["client_id"], auth_kwargs["client_secret"])
        else:
            self._authenticate_user(auth_kwargs["username"], auth_kwargs["password"])

    def _authenticate_user(self, username: str, password: str):
        """Authenticate via extension/user credentials (any license tier)."""
        resp = self.session.post(
            f"{self.base_url}/webclient/api/Login/GetAccessToken",
            json={"SecurityCode": "", "Username": username, "Password": password},
        )
        resp.raise_for_status()
        data = resp.json()
        token = data.get("Token", {}).get("access_token") or data.get("access_token")
        if not token:
            raise ValueError(f"Failed to get access token. Response: {data}")
        self.session.headers.update({"Authorization": f"Bearer {token}"})
        logger.info("Authenticated to 3CX via user credentials")

    def _authenticate_oauth(self, client_id: str, client_secret: str):
        """Authenticate via OAuth2 client credentials (Enterprise license required).
        API client must be created in 3CX Admin > Integrations > API."""
        resp = self.session.post(
            f"{self.base_url}/connect/token",
            data={
                "client_id": client_id,
                "client_secret": client_secret,
                "grant_type": "client_credentials",
            },
            headers={"Content-Type": "application/x-www-form-urlencoded"},
        )
        resp.raise_for_status()
        token = resp.json()["access_token"]
        self.session.headers.update({"Authorization": f"Bearer {token}"})
        logger.info("Authenticated to 3CX via OAuth2 client credentials")

    def get_ring_group(self, extension_number: str) -> dict:
        """Fetch ring group config by extension number."""
        resp = self.session.get(
            f"{self.base_url}/xapi/v1/RingGroups/Pbx.GetByNumber(number='{extension_number}')",
        )
        resp.raise_for_status()
        return resp.json()

    def update_ring_group_forwarding(
        self,
        ring_group_id: int,
        closed_destination: str,
        holiday_destination: str,
    ):
        """Update the OutOfOfficeRoute and HolidaysRoute on a ring group.

        Args:
            ring_group_id: Numeric ID from the ring group entity
            closed_destination: Extension number for after-hours routing
            holiday_destination: Extension number for holiday routing
        """
        payload = {
            "OutOfOfficeRoute": {
                "IsPromptEnabled": False,
                "Route": {
                    "To": "Extension",
                    "Number": closed_destination,
                    "External": "",
                },
            },
            "HolidaysRoute": {
                "IsPromptEnabled": False,
                "Route": {
                    "To": "Extension",
                    "Number": holiday_destination,
                    "External": "",
                },
            },
        }
        resp = self.session.patch(
            f"{self.base_url}/xapi/v1/RingGroups({ring_group_id})",
            json=payload,
        )
        resp.raise_for_status()
        logger.info(
            "Updated ring group %s: closed->Ext %s, holiday->Ext %s",
            ring_group_id,
            closed_destination,
            holiday_destination,
        )
        return resp.status_code