From efd96e7dd0bfbfdf674675c19467e6eddef2aae6 Mon Sep 17 00:00:00 2001 From: King-of-the-all-Cookies Date: Wed, 24 Jun 2026 17:23:37 +0300 Subject: [PATCH] 5 --- bot/services/mikopbx_service.py | 167 ++++++++++++++------------------ 1 file changed, 74 insertions(+), 93 deletions(-) diff --git a/bot/services/mikopbx_service.py b/bot/services/mikopbx_service.py index efad673..cf56b10 100644 --- a/bot/services/mikopbx_service.py +++ b/bot/services/mikopbx_service.py @@ -1,5 +1,6 @@ import aiohttp import logging +import json from typing import Optional, Dict, Any from config import ( MIKOPBX_HOST, @@ -9,6 +10,7 @@ from config import ( ) logger = logging.getLogger(__name__) +logger.setLevel(logging.DEBUG) class MikoPBXService: def __init__(self): @@ -24,72 +26,112 @@ class MikoPBXService: return self.session def _get_auth_headers(self) -> Dict[str, str]: - """Возвращает заголовки авторизации (Bearer Token в приоритете)""" 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): - """Выполняет запрос к MikoPBX REST API v3""" 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)}") + logger.info(f">>> HEADERS: { {k: v for k, v in headers.items() if k != 'Authorization'} }") + try: if method.upper() == "GET": async with session.get(url, params=params, headers=headers) as resp: - return await resp.json() + 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: - return await resp.json() + 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: - return await resp.json() - + 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"API request error: {e}") + logger.error(f"!!! REQUEST ERROR: {e}") return {"result": False, "messages": [str(e)]} # ==================== EXTENSIONS ==================== async def get_extension_template(self) -> Optional[Dict]: - """Получить шаблон нового сотрудника (API v3)""" - response = await self._make_request("GET", "/pbxcore/api/v3/extensions") - if response.get("result"): - return response.get("data") - logger.error(f"Failed to get template: {response.get('messages')}") - return None + """Получить шаблон нового сотрудника""" + logger.info(">>> Trying to get extension template...") + + # Пробуем v3 + response = await self._make_request( + "GET", + "/pbxcore/api/v3/extensions", + params={"id": ""} + ) + + data = response.get("data") + + if isinstance(data, dict): + logger.info(">>> Template received as dict") + return data + elif isinstance(data, list) and len(data) > 0: + logger.info(">>> Template received as list, using first item as template") + return data[0].copy() if isinstance(data[0], dict) else None + else: + logger.warning(">>> Could not get template from API. Will use minimal structure.") + return None async def create_extension(self, number: str, secret: str = None, username: str = "", email: str = "") -> Dict[str, Any]: - """Создать нового SIP сотрудника (API v3)""" - template = await self.get_extension_template() - if not template: - return {"success": False, "error": "Failed to get template"} + logger.info(f"\n>>> Creating extension {number}...") - # Заполняем шаблон + template = await self.get_extension_template() + + if not template: + logger.warning(">>> Using minimal fallback template") + template = { + "type": "SIP", + "show_in_phonebook": "1", + "is_general_user_number": "1", + } + + # Заполняем данными template["number"] = number if username: template["user_username"] = username if email: template["user_email"] = email - template["type"] = "SIP" - template["show_in_phonebook"] = "1" - template["is_general_user_number"] = "1" - if secret: template["sip_secret"] = secret + logger.info(f">>> Final creation payload prepared") + response = await self._make_request( "POST", "/pbxcore/api/v3/extensions", @@ -103,88 +145,27 @@ class MikoPBXService: "data": response.get("data", {}) } else: + error_msg = response.get("messages") or response.get("error") or ["Unknown error"] return { "success": False, - "error": response.get("messages", ["Unknown error"]) + "error": error_msg } async def get_extension(self, number: str) -> Optional[Dict]: - """Получить сотрудника по номеру""" - response = await self._make_request( - "GET", - f"/pbxcore/api/v3/extensions/{number}" - ) - if response.get("result"): - return response.get("data") - return None + response = await self._make_request("GET", f"/pbxcore/api/v3/extensions/{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/extensions/{number}", - json_data=current - ) - - return { - "success": response.get("result", False), - "messages": response.get("messages", []) - } - - async def change_extension_number(self, old_number: str, new_number: str) -> Dict[str, Any]: - """Сменить номер (удалить + создать заново)""" - current = await self.get_extension(old_number) - if not current: - return {"success": False, "error": "Extension not found"} - - old_secret = current.get("sip_secret", "") - - # Удаляем старый - delete_resp = await self._make_request( - "DELETE", - f"/pbxcore/api/v3/extensions/{old_number}" - ) - - if not delete_resp.get("result"): - return {"success": False, "error": "Failed to delete old extension"} - - # Создаём новый - current["number"] = new_number - current.pop("id", None) - current.pop("sip_uniqid", None) - - create_resp = await self._make_request( - "POST", - "/pbxcore/api/v3/extensions", - json_data=current - ) - - if create_resp.get("result"): - return { - "success": True, - "old_number": old_number, - "new_number": new_number, - "secret": old_secret - } - - return {"success": False, "error": "Failed to create new extension"} + response = await self._make_request("PUT", f"/pbxcore/api/v3/extensions/{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/extensions/{number}" - ) - return { - "success": response.get("result", False), - "messages": response.get("messages", []) - } + response = await self._make_request("DELETE", f"/pbxcore/api/v3/extensions/{number}") + return {"success": response.get("result", False), "messages": response.get("messages", [])} async def close(self): if self.session: