Update: Major code refactor for modular extensibility

embeds.py

@@ -0,0 +1,424 @@
+"""
+Embed Management Module
+
+Handles creation, tracking, and persistence of Discord server status embeds.
+Manages embed locations storage and provides utilities for embed lifecycle.
+"""
+
+import json
+import logging
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Tuple
+
+import discord
+
+from api import PanelAPI
+from config import Config
+from exceptions import EmbedNotFoundError
+from metrics import ServerMetricsManager
+from models import ServerData, ServerResources
+
+logger = logging.getLogger("pterodisbot")
+
+
+class EmbedManager:
+    """
+    Manages server status embeds lifecycle.
+
+    Handles:
+    - Embed creation with real-time server data
+    - Persistent storage of embed locations (channel_id, message_id)
+    - Embed refresh and recreation
+    - Cleanup of deleted/missing embeds
+    - Rate limit protection during bulk operations
+
+    Attributes:
+        config: Application configuration
+        api: Panel API client
+        metrics_manager: Metrics manager for graphs
+        storage_path: Path to embed locations JSON file
+    """
+
+    def __init__(
+        self,
+        config: Config,
+        api: PanelAPI,
+        metrics_manager: ServerMetricsManager,
+    ) -> None:
+        """
+        Initialize the embed manager.
+
+        Args:
+            config: Validated application configuration
+            api: Initialized PanelAPI instance
+            metrics_manager: ServerMetricsManager instance
+        """
+        self.config = config
+        self.api = api
+        self.metrics_manager = metrics_manager
+        self.storage_path = config.embed_storage_path
+        self.embed_locations: Dict[str, Dict[str, int]] = {}
+        self.previous_states: Dict[str, Tuple[str, float, Optional[float]]] = {}
+
+        # Ensure storage directory exists
+        self.storage_path.parent.mkdir(parents=True, exist_ok=True)
+        logger.info("Initialized EmbedManager")
+
+    async def load_locations(self) -> None:
+        """Load saved embed locations from JSON storage file."""
+        logger.debug("Loading embed locations from storage")
+        if not self.storage_path.exists():
+            logger.info("No existing embed locations file found")
+            return
+
+        try:
+            with open(self.storage_path, "r") as f:
+                self.embed_locations = json.load(f)
+            logger.info(f"Loaded {len(self.embed_locations)} embed locations from storage")
+        except Exception as e:
+            logger.error(f"Failed to load embed locations: {str(e)}")
+
+    async def save_locations(self) -> None:
+        """Save current embed locations to JSON storage file."""
+        logger.debug("Saving embed locations to storage")
+        try:
+            with open(self.storage_path, "w") as f:
+                json.dump(self.embed_locations, f, indent=2)
+            logger.debug("Successfully saved embed locations")
+        except Exception as e:
+            logger.error(f"Failed to save embed locations: {str(e)}")
+
+    async def track_embed(self, server_id: str, message: discord.Message) -> None:
+        """
+        Track a newly created embed in storage.
+
+        Args:
+            server_id: The server's panel identifier
+            message: Discord message containing the embed
+        """
+        logger.debug(f"Tracking new embed for server {server_id} in channel {message.channel.id}")
+        self.embed_locations[server_id] = {
+            "channel_id": str(message.channel.id),
+            "message_id": str(message.id),
+        }
+        await self.save_locations()
+
+    async def untrack_embed(self, server_id: str) -> None:
+        """
+        Remove a server's embed from tracking.
+
+        Args:
+            server_id: The server's panel identifier
+        """
+        self.embed_locations.pop(server_id, None)
+        self.previous_states.pop(server_id, None)
+        await self.save_locations()
+
+    async def create_server_embed(
+        self, server_data: ServerData, resources: ServerResources
+    ) -> Tuple[discord.Embed, discord.ui.View]:
+        """
+        Create a status embed and interactive view for a server.
+
+        Args:
+            server_data: Server information from panel
+            resources: Current resource usage data
+
+        Returns:
+            Tuple of (embed, view) ready for display
+        """
+        attributes = server_data.attributes
+        identifier = attributes.identifier
+        name = attributes.name
+        description = attributes.description or "No description available"
+
+        logger.debug(f"Building status embed for server {name} ({identifier})")
+
+        # Parse resource data
+        current_state = resources.attributes.current_state.title()
+        is_suspended = attributes.suspended
+
+        # Create embed with appropriate color based on status
+        embed = discord.Embed(
+            title=f"{name} - {current_state}",
+            description=description,
+            color=(
+                discord.Color.blue()
+                if current_state.lower() == "running"
+                else discord.Color.red()
+            ),
+            timestamp=datetime.now(),
+        )
+
+        embed.add_field(name="🆔 Server ID", value=f"`{identifier}`", inline=True)
+
+        if is_suspended:
+            embed.add_field(name="ℹ️ Status", value="⛔ `Suspended`", inline=True)
+        else:
+            embed.add_field(name="ℹ️ Status", value="✅ `Active`", inline=True)
+
+        # Add resource usage if server is running
+        if current_state.lower() != "offline":
+            embed = self._add_resource_fields(embed, resources, attributes)
+            embed = self._add_graph_fields(embed, identifier)
+
+        embed.set_footer(text="Last updated")
+
+        # Create interactive view with control buttons
+        from views import ServerStatusView
+
+        view = ServerStatusView(
+            server_id=identifier,
+            server_name=name,
+            api=self.api,
+            server_data=server_data.model_dump(),
+        )
+
+        logger.debug(f"Successfully built status components for {name}")
+        return embed, view
+
+    def _add_resource_fields(
+        self, embed: discord.Embed, resources: ServerResources, attributes: Any
+    ) -> discord.Embed:
+        """Add resource usage fields to embed."""
+        resource_attrs = resources.attributes.resources
+        limits = attributes.limits
+
+        # Current usage
+        cpu_usage = round(resource_attrs.cpu_absolute, 2)
+        memory_usage = round(resource_attrs.memory_bytes / (1024**2), 2)
+        disk_usage = round(resource_attrs.disk_bytes / (1024**2), 2)
+        network_rx = round(resource_attrs.network_rx_bytes / (1024**2), 2)
+        network_tx = round(resource_attrs.network_tx_bytes / (1024**2), 2)
+
+        # Maximum allocated resources
+        cpu_limit = limits.cpu
+        memory_limit = limits.memory
+        disk_limit = limits.disk
+
+        # Format limit values - display ∞ for unlimited (0 limit)
+        def format_limit(value: int, unit: str = "") -> str:
+            if value == 0:
+                return f"{'∞':>8} {unit}"
+            return f"{value:>8} {unit}"
+
+        # Format uptime
+        uptime_ms = resource_attrs.uptime
+        if uptime_ms > 0:
+            uptime_seconds = uptime_ms // 1000
+            if uptime_seconds < 60:
+                uptime_text = f"`{uptime_seconds}s`"
+            elif uptime_seconds < 3600:
+                uptime_text = f"`{uptime_seconds // 60}m {uptime_seconds % 60}s`"
+            elif uptime_seconds < 86400:
+                hours = uptime_seconds // 3600
+                minutes = (uptime_seconds % 3600) // 60
+                uptime_text = f"`{hours}h {minutes}m`"
+            else:
+                days = uptime_seconds // 86400
+                hours = (uptime_seconds % 86400) // 3600
+                uptime_text = f"`{days}d {hours}h`"
+        else:
+            uptime_text = "`Just started`"
+
+        embed.add_field(name="⏱️ Uptime", value=uptime_text, inline=True)
+
+        # Resource usage text box
+        usage_text = (
+            f"```properties\n"
+            f"CPU    : [{cpu_usage:>8} / {format_limit(cpu_limit, '%')}\n"
+            f"Memory : [{memory_usage:>8} / {format_limit(memory_limit, 'MiB')}\n"
+            f"Disk   : [{disk_usage:>8} / {format_limit(disk_limit, 'MiB')}\n"
+            f"```"
+        )
+
+        embed.add_field(name="📊 Resource Usage", value=usage_text, inline=False)
+        embed.add_field(name="Network In", value=f"📥 `{network_rx} MiB`", inline=True)
+        embed.add_field(name="Network Out", value=f"📤 `{network_tx} MiB`", inline=True)
+
+        return embed
+
+    def _add_graph_fields(
+        self, embed: discord.Embed, server_id: str
+    ) -> discord.Embed:
+        """Add metrics graph fields to embed if data available."""
+        server_graphs = self.metrics_manager.get_server_graphs(server_id)
+        if server_graphs and server_graphs.has_sufficient_data:
+            summary = server_graphs.get_data_summary()
+
+            graph_description = (
+                f">>> `Data points: {summary['point_count']}/6`\n"
+                f"`CPU trend: {summary['cpu_trend']} • Memory trend: {summary['memory_trend']}`"
+            )
+
+            embed.add_field(
+                name="📈 Usage Trends (Last Minute)",
+                value=graph_description,
+                inline=False,
+            )
+            embed.set_image(url=f"attachment://metrics_graph_{server_id}.png")
+
+        return embed
+
+    async def refresh_all_embeds(self) -> Tuple[int, int]:
+        """
+        Perform a complete refresh of all server status embeds.
+
+        Creates new embeds and deletes old ones to prevent duplication.
+
+        Returns:
+            Tuple of (deleted_count, created_count)
+        """
+        logger.info("Starting full refresh of all server embeds")
+        from bot import PterodactylBot  # Avoid circular import
+        bot = PterodactylBot.get_instance()
+
+        deleted_count = 0
+        created_count = 0
+        skipped_count = 0
+
+        try:
+            # Get current server list
+            servers = await self.api.get_servers()
+            if not servers:
+                logger.warning("No servers found in panel during refresh")
+                return 0, 0
+
+            server_cache = {s["attributes"]["identifier"]: s for s in servers}
+
+            # Create new embeds
+            new_embeds = {}
+
+            for server_id, server_data in server_cache.items():
+                if server_id not in self.embed_locations:
+                    skipped_count += 1
+                    continue
+
+                channel_id = self.embed_locations[server_id]["channel_id"]
+                channel = bot.get_channel(int(channel_id))
+                if not channel:
+                    logger.warning(f"Channel {channel_id} not found for server {server_id}")
+                    continue
+
+                try:
+                    resources = await self.api.get_server_resources(server_id)
+                    server_obj = ServerData.model_validate(server_data)
+                    resources_obj = ServerResources.model_validate(resources)
+
+                    embed, view = await self.create_server_embed(server_obj, resources_obj)
+                    message = await channel.send(embed=embed, view=view)
+
+                    new_embeds[server_id] = {
+                        "channel_id": str(channel.id),
+                        "message_id": str(message.id),
+                    }
+                    created_count += 1
+                    logger.info(f"Created new embed for server {server_data['attributes']['name']}")
+
+                    await asyncio.sleep(1)  # Rate limit protection
+                except Exception as e:
+                    logger.error(f"Failed to create new embed for server {server_id}: {str(e)}")
+
+            if not new_embeds:
+                logger.warning("No new embeds created during refresh")
+                return 0, 0
+
+            # Delete old embeds
+            for server_id, location in list(self.embed_locations.items()):
+                try:
+                    channel = bot.get_channel(int(location["channel_id"]))
+                    if channel:
+                        try:
+                            message = await channel.fetch_message(int(location["message_id"]))
+                            await message.delete()
+                            deleted_count += 1
+                            logger.debug(f"Deleted old embed for server {server_id}")
+                            await asyncio.sleep(0.5)
+                        except discord.NotFound:
+                            logger.debug(f"Old embed for server {server_id} already deleted")
+                        except Exception as e:
+                            logger.error(f"Failed to delete old embed for server {server_id}: {str(e)}")
+                except Exception as e:
+                    logger.error(f"Error processing old embed for server {server_id}: {str(e)}")
+
+            logger.info(f"Deleted {deleted_count} old embeds, created {created_count} new embeds")
+
+            # Update storage
+            self.embed_locations = new_embeds
+            await self.save_locations()
+
+            return deleted_count, created_count
+
+        except Exception as e:
+            logger.error(f"Critical error during embed refresh: {str(e)}")
+            raise
+
+    async def update_single_embed(
+        self,
+        server_id: str,
+        location: Dict[str, str],
+        server_data: ServerData,
+        resources: ServerResources,
+        current_state: str,
+        cpu_usage: float,
+    ) -> bool:
+        """
+        Update a single server's status embed.
+
+        Args:
+            server_id: Server identifier
+            location: Embed location dict with channel_id and message_id
+            server_data: Current server data
+            resources: Current resource data
+            current_state: Current power state
+            cpu_usage: Current CPU usage
+
+        Returns:
+            True if embed was updated, False otherwise
+        """
+        from bot import PterodactylBot
+        bot = PterodactylBot.get_instance()
+
+        try:
+            embed, view = await self.create_server_embed(server_data, resources)
+
+            channel = bot.get_channel(int(location["channel_id"]))
+            if not channel:
+                logger.warning(f"Channel {location['channel_id']} not found for server {server_id}")
+                return False
+
+            message = await channel.fetch_message(int(location["message_id"]))
+
+            # Handle graph attachments
+            files = []
+            server_graphs = self.metrics_manager.get_server_graphs(server_id)
+
+            if current_state == "running" and server_graphs and server_graphs.has_sufficient_data:
+                combined_graph = server_graphs.generate_combined_graph()
+                if combined_graph:
+                    files.append(
+                        discord.File(
+                            combined_graph,
+                            filename=f"metrics_graph_{server_id}.png",
+                        )
+                    )
+            # Empty files list removes existing attachments
+
+            await message.edit(embed=embed, view=view, attachments=files)
+            logger.debug(f"Updated status for {server_data.attributes.name}")
+            return True
+
+        except discord.NotFound:
+            logger.warning(f"Embed for server {server_id} not found, removing from tracking")
+            self.embed_locations.pop(server_id, None)
+            self.previous_states.pop(server_id, None)
+            await self.save_locations()
+            return False
+        except Exception as e:
+            logger.error(f"Failed to update status for server {server_id}: {str(e)}")
+            return False
+
+
+# Import at bottom to avoid circular imports
+import asyncio