mirror of
https://github.com/Sea-Haven-Industries/ring-scheduler-3cx.git
synced 2026-05-18 20:20:12 +00:00
114 lines
4.1 KiB
Python
114 lines
4.1 KiB
Python
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_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.
|
|
|
|
Args:
|
|
queue_id: Numeric ID from the queue 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/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
|