afterhours-shift-manager/src/shared/python/shared/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_group_members(self, group_name: str = "DEFAULT") -> list[dict]:
        """Fetch members of a 3CX group (department).

        Returns members with Number, MemberName, Type, etc.
        """
        # First, find the group ID by name
        resp = self.session.get(
            f"{self.base_url}/xapi/v1/Groups",
            params={"$filter": f"Name eq '{group_name}'"},
        )
        resp.raise_for_status()
        groups = resp.json().get("value", [])
        if not groups:
            logger.warning("Group '%s' not found in 3CX", group_name)
            return []

        group_id = groups[0]["Id"]

        # Fetch members of that group
        resp = self.session.get(f"{self.base_url}/xapi/v1/Groups({group_id})/Members")
        resp.raise_for_status()
        return resp.json().get("value", [])

    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."""
        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

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

    def update_queue_forwarding(
        self,
        queue_id: int,
        closed_destination: str,
        holiday_destination: str,
    ):
        """Update the OutOfOfficeRoute and HolidaysRoute on a queue."""
        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/Queues({queue_id})",
            json=payload,
        )
        resp.raise_for_status()
        logger.info(
            "Updated queue %s: closed->Ext %s, holiday->Ext %s",
            queue_id,
            closed_destination,
            holiday_destination,
        )
        return resp.status_code