Pterodisbot/api.py

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