151 lines
5.8 KiB
Python
151 lines
5.8 KiB
Python
import aiohttp
|
|
import logging
|
|
import json
|
|
from typing import Optional, Dict, Any
|
|
from config import (
|
|
MIKOPBX_HOST,
|
|
MIKOPBX_API_TOKEN,
|
|
MIKOPBX_ADMIN_LOGIN,
|
|
MIKOPBX_ADMIN_PASSWORD
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
logger.setLevel(logging.DEBUG)
|
|
|
|
class MikoPBXService:
|
|
def __init__(self):
|
|
self.host = MIKOPBX_HOST.rstrip('/')
|
|
self.api_token = MIKOPBX_API_TOKEN
|
|
self.admin_login = MIKOPBX_ADMIN_LOGIN
|
|
self.admin_password = MIKOPBX_ADMIN_PASSWORD
|
|
self.session = None
|
|
|
|
async def _get_session(self):
|
|
if self.session is None:
|
|
self.session = aiohttp.ClientSession()
|
|
return self.session
|
|
|
|
def _get_auth_headers(self) -> Dict[str, str]:
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json"
|
|
}
|
|
if self.api_token:
|
|
headers["Authorization"] = f"Bearer {self.api_token}"
|
|
logger.info(">>> Using Bearer Token authentication")
|
|
elif self.admin_login and self.admin_password:
|
|
import base64
|
|
credentials = f"{self.admin_login}:{self.admin_password}"
|
|
encoded = base64.b64encode(credentials.encode()).decode()
|
|
headers["Authorization"] = f"Basic {encoded}"
|
|
logger.info(">>> Using Basic Auth (fallback)")
|
|
return headers
|
|
|
|
async def _make_request(self, method: str, endpoint: str, json_data: Optional[Dict] = None, params: Optional[Dict] = None):
|
|
session = await self._get_session()
|
|
url = f"{self.host}{endpoint}"
|
|
headers = self._get_auth_headers()
|
|
|
|
logger.info(f"\n{'='*70}")
|
|
logger.info(f">>> REQUEST: {method.upper()} {url}")
|
|
if params:
|
|
logger.info(f">>> PARAMS: {params}")
|
|
if json_data:
|
|
logger.info(f">>> BODY:\n{json.dumps(json_data, indent=2, ensure_ascii=False)}")
|
|
|
|
try:
|
|
if method.upper() == "GET":
|
|
async with session.get(url, params=params, headers=headers) as resp:
|
|
text = await resp.text()
|
|
logger.info(f"<<< RESPONSE [{resp.status}]:\n{text[:3000]}")
|
|
try:
|
|
return await resp.json()
|
|
except:
|
|
return {"result": False, "raw": text}
|
|
elif method.upper() == "DELETE":
|
|
async with session.delete(url, headers=headers) as resp:
|
|
text = await resp.text()
|
|
logger.info(f"<<< RESPONSE [{resp.status}]:\n{text[:3000]}")
|
|
try:
|
|
return await resp.json()
|
|
except:
|
|
return {"result": False, "raw": text}
|
|
else:
|
|
async with session.post(url, json=json_data, headers=headers) as resp:
|
|
text = await resp.text()
|
|
logger.info(f"<<< RESPONSE [{resp.status}]:\n{text[:3000]}")
|
|
try:
|
|
return await resp.json()
|
|
except:
|
|
return {"result": False, "raw": text}
|
|
except Exception as e:
|
|
logger.error(f"!!! REQUEST ERROR: {e}")
|
|
return {"result": False, "messages": [str(e)]}
|
|
|
|
# ==================== EMPLOYEES (correct v3 endpoint) ====================
|
|
|
|
async def create_extension(self, number: str, secret: str, username: str = "", email: str = "") -> Dict[str, Any]:
|
|
"""
|
|
Create new employee (SIP extension) using the correct endpoint
|
|
POST /pbxcore/api/v3/employees
|
|
"""
|
|
logger.info(f"\n>>> Creating employee/extension {number} via /v3/employees ...")
|
|
|
|
payload = {
|
|
"number": number,
|
|
"user_username": username or f"User {number}",
|
|
"sip_secret": secret,
|
|
}
|
|
|
|
if email:
|
|
payload["user_email"] = email
|
|
|
|
# Optional sensible defaults
|
|
payload["sip_enableRecording"] = True
|
|
payload["sip_dtmfmode"] = "auto"
|
|
payload["sip_transport"] = "udp,tcp"
|
|
payload["fwd_ringlength"] = 45
|
|
|
|
logger.info(f">>> Sending create employee request...")
|
|
|
|
response = await self._make_request(
|
|
"POST",
|
|
"/pbxcore/api/v3/employees",
|
|
json_data=payload
|
|
)
|
|
|
|
if response.get("result"):
|
|
logger.info(">>> SUCCESS: Employee created!")
|
|
return {
|
|
"success": True,
|
|
"number": number,
|
|
"data": response.get("data", {})
|
|
}
|
|
else:
|
|
error = response.get("messages") or response.get("error") or ["Unknown error"]
|
|
logger.error(f">>> Creation failed: {error}")
|
|
return {
|
|
"success": False,
|
|
"error": error
|
|
}
|
|
|
|
async def get_extension(self, number: str) -> Optional[Dict]:
|
|
"""Get employee by number"""
|
|
response = await self._make_request("GET", f"/pbxcore/api/v3/employees/{number}")
|
|
return response.get("data") if response.get("result") else None
|
|
|
|
async def set_extension_secret(self, number: str, new_secret: str) -> Dict[str, Any]:
|
|
current = await self.get_extension(number)
|
|
if not current:
|
|
return {"success": False, "error": "Extension not found"}
|
|
current["sip_secret"] = new_secret
|
|
response = await self._make_request("PUT", f"/pbxcore/api/v3/employees/{number}", json_data=current)
|
|
return {"success": response.get("result", False), "messages": response.get("messages", [])}
|
|
|
|
async def delete_extension(self, number: str) -> Dict[str, Any]:
|
|
response = await self._make_request("DELETE", f"/pbxcore/api/v3/employees/{number}")
|
|
return {"success": response.get("result", False), "messages": response.get("messages", [])}
|
|
|
|
async def close(self):
|
|
if self.session:
|
|
await self.session.close() |