""" Pterodactyl/Pelican Panel API Client Module Provides a clean interface for interacting with Pterodactyl and Pelican Panel APIs. Supports both client API (server operations) and application API (admin operations). """ import asyncio import logging from typing import Any, Dict, List, Optional import aiohttp from config import Config from exceptions import APIError, ServerNotFoundError logger = logging.getLogger("pterodisbot") class PanelAPI: """ Unified API client for Pterodactyl and Pelican Panel. Handles all HTTP requests to the panel API with proper authentication, error handling, and rate limiting protection. Features: - Automatic session management - Dual API key support (client + application) - Concurrent request protection via asyncio.Lock - Comprehensive error handling and logging - Response parsing with structured error information Args: config: Validated Config instance containing panel credentials """ def __init__(self, config: Config) -> None: """ Initialize the Panel API client. Args: config: Validated Config instance with panel URL and API keys """ self.config = config self.panel_url = config.pterodactyl_url.rstrip("/") self.client_api_key = config.client_api_key self.application_api_key = config.application_api_key self.session: Optional[aiohttp.ClientSession] = None self._lock = asyncio.Lock() self._is_initialized = False logger.info(f"Initialized {config.get_panel_display_name()} API client") async def initialize(self) -> None: """Initialize the aiohttp client session.""" if self.session is None or self.session.closed: self.session = aiohttp.ClientSession() logger.debug("Created new aiohttp ClientSession") self._is_initialized = True async def close(self) -> None: """Cleanly close the aiohttp session.""" if self.session and not self.session.closed: await self.session.close() logger.debug("Closed aiohttp ClientSession") self._is_initialized = False async def _request( self, method: str, endpoint: str, data: Optional[Dict[str, Any]] = None, use_application_key: bool = False, ) -> Dict[str, Any]: """ Make an authenticated request to the panel API. Args: method: HTTP method (GET, POST, PUT, DELETE) endpoint: API endpoint (e.g., 'application/servers') data: Optional JSON payload for POST/PUT requests use_application_key: Whether to use application API key (admin endpoints) Returns: Dictionary containing API response data Raises: APIError: For network errors, HTTP errors, or parsing failures """ if not self._is_initialized or self.session is None or self.session.closed: await self.initialize() url = f"{self.panel_url}/api/{endpoint}" api_key_type = "Application" if use_application_key else "Client" logger.debug(f"Preparing {method} request to {endpoint} using {api_key_type} API key") api_key = self.application_api_key if use_application_key else self.client_api_key headers = { "Authorization": f"Bearer {api_key}", "Accept": "application/json", "Content-Type": "application/json", } try: async with self._lock: logger.debug(f"Acquired lock for API request to {endpoint}") async with self.session.request( method, url, headers=headers, json=data ) as response: if response.status == 204: logger.debug(f"Received 204 No Content from {endpoint}") return {"status": "success"} response_data = await response.json() logger.debug( f"Received response from {endpoint} with status {response.status}" ) if response.status >= 400: error_msg = self._extract_error_message(response_data) logger.error( f"API request to {endpoint} failed with status " f"{response.status}: {error_msg}" ) raise APIError( message=error_msg, endpoint=endpoint, status_code=response.status, ) return response_data except aiohttp.ClientError as e: logger.error(f"Network error during API request to {endpoint}: {str(e)}") raise APIError( message=f"Network error: {str(e)}", endpoint=endpoint, ) except Exception as e: logger.error(f"Unexpected error during API request to {endpoint}: {str(e)}") raise APIError( message=f"Unexpected error: {str(e)}", endpoint=endpoint, ) def _extract_error_message(self, response_data: Dict[str, Any]) -> str: """ Extract human-readable error message from API response. Args: response_data: Parsed JSON response from API Returns: Formatted error message string """ if "errors" in response_data and isinstance(response_data["errors"], list): errors = response_data["errors"] if errors and isinstance(errors[0], dict): return errors[0].get("detail", "Unknown API error") return str(errors[0]) if errors else "Unknown API error" return response_data.get("message", "Unknown error") # ============================================================ # Server Management Methods (Client API) # ============================================================ async def get_server_resources(self, server_id: str) -> Dict[str, Any]: """ Get resource usage for a specific server. Args: server_id: The panel server identifier Returns: Dictionary containing server resource usage and state Raises: ServerNotFoundError: If server doesn't exist APIError: For other API errors """ logger.debug(f"Fetching resource usage for server {server_id}") try: response = await self._request( "GET", f"client/servers/{server_id}/resources" ) return response except APIError as e: if e.status_code == 404: raise ServerNotFoundError(f"Server {server_id} not found") raise async def send_power_action(self, server_id: str, action: str) -> Dict[str, Any]: """ Send a power action to a server. Args: server_id: The panel server identifier action: Power action (start, stop, restart) Returns: Dictionary containing API response status Raises: ValueError: If action is invalid APIError: For API errors """ valid_actions = ["start", "stop", "restart"] if action not in valid_actions: raise ValueError( f"Invalid power action: {action}. Must be one of: {valid_actions}" ) logger.info(f"Sending {action} command to server {server_id}") result = await self._request( "POST", f"client/servers/{server_id}/power", {"signal": action} ) if result.get("status") == "success": logger.info(f"Successfully executed {action} on server {server_id}") else: logger.error(f"Failed to execute {action} on server {server_id}") return result # ============================================================ # Server Information Methods (Application API) # ============================================================ async def get_servers(self) -> List[Dict[str, Any]]: """ Get a list of all servers from the panel. Uses application API key as this is an admin endpoint. Returns: List of server dictionaries with full attributes """ logger.info("Fetching list of all servers from panel") response = await self._request("GET", "application/servers", use_application_key=True) servers = response.get("data", []) logger.info(f"Retrieved {len(servers)} servers from panel") return servers async def get_server_details(self, server_id: str) -> Dict[str, Any]: """ Get detailed server information including allocations. Uses application API key as this is an admin endpoint. Args: server_id: The panel server identifier Returns: Dictionary containing detailed server information """ logger.debug(f"Fetching detailed information for server {server_id}") return await self._request( "GET", f"application/servers/{server_id}", use_application_key=True ) async def get_server_allocations(self, server_id: str) -> Dict[str, Any]: """ Get allocation information for a server (IP addresses and ports). Uses application API key as this is an admin endpoint. Args: server_id: The panel server identifier Returns: Dictionary containing server allocation information """ logger.debug(f"Fetching allocation information for server {server_id}") return await self._request( "GET", f"application/servers/{server_id}/allocations", use_application_key=True, ) async def get_client_server_details(self, server_id: str) -> Dict[str, Any]: """ Get server details using client API (for connection info). Args: server_id: The panel server identifier Returns: Dictionary containing server details with allocations """ logger.debug(f"Fetching client server details for {server_id}") return await self._request("GET", f"client/servers/{server_id}") # ============================================================ # Convenience Methods # ============================================================ async def get_server_default_allocation( self, server_id: str ) -> tuple[str, str]: """ Get the default allocation IP and port for a server. Args: server_id: The panel server identifier Returns: Tuple of (ip_alias, port) Raises: ServerNotFoundError: If server or allocations not found APIError: For API errors """ details = await self.get_client_server_details(server_id) if details.get("status") == "error": raise ServerNotFoundError( details.get("message", "Failed to get server details") ) attributes = details.get("attributes", {}) relationships = attributes.get("relationships", {}) allocations = relationships.get("allocations", {}).get("data", []) if not allocations: raise ServerNotFoundError("No allocations found for this server") # Find default allocation or fallback to first default_allocation = next( ( alloc for alloc in allocations if alloc.get("attributes", {}).get("is_default", False) ), allocations[0], ) allocation_attrs = default_allocation.get("attributes", {}) ip_alias = allocation_attrs.get("ip_alias", "Unknown") port = str(allocation_attrs.get("port", "Unknown")) return ip_alias, port