Update: Major code refactor
CI/CD Pipeline / Unit Tests (Python 3.12) (push) Successful in 30s
CI/CD Pipeline / Unit Tests (Python 3.13) (push) Successful in 35s
CI/CD Pipeline / Code Quality & Linting (push) Successful in 27s
CI/CD Pipeline / Security Scanning (push) Successful in 2m23s
CI/CD Pipeline / Unit Tests (Python 3.11) (push) Successful in 24s
CI/CD Pipeline / Integration Tests (push) Successful in 26s
CI/CD Pipeline / Generate Test Report (push) Successful in 9s
CI/CD Pipeline / Build Docker Image (push) Successful in 41s
CI/CD Pipeline / CI/CD Pipeline Status (push) Successful in 2s

This commit is contained in:
k.eaven committed 2026-10-24 10:53:09 +00:00
1 parent d891b21ffd
commit 3e03939fd0
13 files changed
+3266 -443

No files matched your search

+315
View File
@@ -0,0 +1,315 @@
"""
Discord Views Module
Contains all Discord UI views and interactive components.
Handles server control buttons, dropdowns, and permission checks.
"""
import logging
from typing import TYPE_CHECKING
import discord
from discord.ext import commands
from config import Config
from exceptions import PermissionError
if TYPE_CHECKING:
from api import PanelAPI
logger = logging.getLogger("pterodisbot")
class ServerStatusView(discord.ui.View):
"""
Interactive view containing server control buttons.
Provides persistent controls for server management with role-based access.
Buttons: Start (green), Stop (red), Restart (blurple), Show Address (grey).
Features:
- Guild restriction (only works in configured server)
- Role-based access control (requires "Game Server User" role)
- Persistent views (timeout=None for infinite persistence)
- Comprehensive error handling with user feedback
"""
def __init__(
self,
server_id: str,
server_name: str,
api: "PanelAPI",
server_data: dict,
) -> None:
"""
Initialize the server status view with control buttons.
Args:
server_id: The server's panel identifier
server_name: Human-readable server name
api: Panel API client instance
server_data: Full server data from panel
"""
super().__init__(timeout=None) # Persistent view
self.server_id = server_id
self.server_name = server_name
self.api = api
self.server_data = server_data
logger.debug(f"Created ServerStatusView for {server_name} ({server_id})")
async def interaction_check(self, interaction: discord.Interaction) -> bool:
"""
Verify the interacting user has the required role and is in the allowed guild.
Args:
interaction: Discord interaction object
Returns:
bool: True if authorized, False otherwise
"""
config = Config.get_instance()
# First check if interaction is from the allowed guild
if interaction.guild_id != config.allowed_guild_id:
logger.warning(f"Unauthorized interaction attempt from guild {interaction.guild_id}")
await interaction.response.send_message(
"This bot is only available in a specific server.", ephemeral=True
)
return False
# Then check for required role
logger.debug(f"Checking permissions for {interaction.user.name} on server {self.server_name}")
has_role = any(role.name == config.required_role for role in interaction.user.roles)
if not has_role:
logger.warning(
f"Permission denied for {interaction.user.name} - missing '{config.required_role}' role"
)
await interaction.response.send_message(
f"You don't have permission to control servers. You need the '{config.required_role}' role.",
ephemeral=True,
)
return False
logger.debug(f"Permission granted for {interaction.user.name}")
return True
async def on_error(
self, interaction: discord.Interaction, error: Exception, item: discord.ui.Item
) -> None:
"""
Handle errors in button interactions.
Args:
interaction: Discord interaction object
error: Exception that occurred
item: The UI item that triggered the error
"""
logger.error(f"View error in {self.server_name} by {interaction.user.name}: {str(error)}")
if not interaction.response.is_done():
await interaction.response.send_message(
"An error occurred while processing your request.", ephemeral=True
)
else:
await interaction.followup.send(
"An error occurred while processing your request.", ephemeral=True
)
@discord.ui.button(
label="Start", style=discord.ButtonStyle.green, custom_id="start_button"
)
async def start_button(
self, interaction: discord.Interaction, button: discord.ui.Button
) -> None:
"""Send a start command to the server."""
logger.info(f"Start button pressed for {self.server_name} by {interaction.user.name}")
await interaction.response.defer(ephemeral=True)
result = await self.api.send_power_action(self.server_id, "start")
if result.get("status") == "success":
message = f"Server '{self.server_name}' is starting..."
logger.info(f"Successfully started server {self.server_name}")
else:
message = f"Failed to start server: {result.get('message', 'Unknown error')}"
logger.error(f"Failed to start server {self.server_name}: {message}")
await interaction.followup.send(message, ephemeral=True)
@discord.ui.button(
label="Stop", style=discord.ButtonStyle.red, custom_id="stop_button"
)
async def stop_button(
self, interaction: discord.Interaction, button: discord.ui.Button
) -> None:
"""Send a stop command to the server."""
logger.info(f"Stop button pressed for {self.server_name} by {interaction.user.name}")
await interaction.response.defer(ephemeral=True)
result = await self.api.send_power_action(self.server_id, "stop")
if result.get("status") == "success":
message = f"Server '{self.server_name}' is stopping..."
logger.info(f"Successfully stopped server {self.server_name}")
else:
message = f"Failed to stop server: {result.get('message', 'Unknown error')}"
logger.error(f"Failed to stop server {self.server_name}: {message}")
await interaction.followup.send(message, ephemeral=True)
@discord.ui.button(
label="Restart", style=discord.ButtonStyle.blurple, custom_id="restart_button"
)
async def restart_button(
self, interaction: discord.Interaction, button: discord.ui.Button
) -> None:
"""Send a restart command to the server."""
logger.info(f"Restart button pressed for {self.server_name} by {interaction.user.name}")
await interaction.response.defer(ephemeral=True)
result = await self.api.send_power_action(self.server_id, "restart")
if result.get("status") == "success":
message = f"Server '{self.server_name}' is restarting..."
logger.info(f"Successfully restarted server {self.server_name}")
else:
message = f"Failed to restart server: {result.get('message', 'Unknown error')}"
logger.error(f"Failed to restart server {self.server_name}: {message}")
await interaction.followup.send(message, ephemeral=True)
@discord.ui.button(
label="Show Address",
style=discord.ButtonStyle.grey,
custom_id="show_address_button",
)
async def show_address_button(
self, interaction: discord.Interaction, button: discord.ui.Button
) -> None:
"""Show server's default allocation IP and port."""
logger.info(f"Show Address button pressed for {self.server_name} by {interaction.user.name}")
try:
await interaction.response.defer(ephemeral=True)
logger.debug(f"Fetching server details for {self.server_id}")
ip_alias, port = await self.api.get_server_default_allocation(self.server_id)
# Create and send embed
embed = discord.Embed(
title=f"{self.server_name} Connection Info",
color=discord.Color.blue(),
description=f"Server ID: `{self.server_id}`",
)
embed.add_field(name="Address", value=f"`{ip_alias}`", inline=True)
embed.add_field(name="Port", value=f"`{port}`", inline=True)
await interaction.followup.send(embed=embed, ephemeral=True)
logger.info(f"Displayed connection info for {self.server_name}")
except Exception as e:
logger.error(f"Failed to show address for {self.server_name}: {str(e)}")
await interaction.followup.send(
"⚠️ Failed to get connection info. The server may not have any ports allocated.",
ephemeral=True,
)
class ServerDropdown(discord.ui.Select):
"""Dropdown menu for server selection in status dashboard."""
def __init__(self, server_options: list) -> None:
"""
Initialize the server dropdown.
Args:
server_options: List of discord.SelectOption objects
"""
super().__init__(
placeholder="Select a server to display...",
options=server_options,
min_values=1,
max_values=1,
)
async def callback(self, interaction: discord.Interaction) -> None:
"""
Handle server selection from dropdown menu.
Creates permanent status embed in the channel for the selected server.
"""
from bot import PterodactylBot
await interaction.response.defer(ephemeral=True)
bot = PterodactylBot.get_instance()
selected_server_id = self.values[0]
server_data = bot.server_cache.get(selected_server_id)
if not server_data:
await interaction.followup.send(
"❌ Selected server no longer available. Please try again.", ephemeral=True
)
return
server_name = server_data["attributes"]["name"]
logger.info(f"User {interaction.user.name} selected server: {server_name}")
try:
# Get current server status for embed creation
resources = await bot.api.get_server_resources(selected_server_id)
# Delete old embed if it exists to prevent duplication
if selected_server_id in bot.embed_manager.embed_locations:
logger.debug(f"Found existing embed for {selected_server_id}, attempting to delete")
try:
old_location = bot.embed_manager.embed_locations[selected_server_id]
old_channel = bot.get_channel(int(old_location["channel_id"]))
if old_channel:
try:
old_message = await old_channel.fetch_message(
int(old_location["message_id"])
)
await old_message.delete()
logger.debug(f"Deleted old embed for {selected_server_id}")
except discord.NotFound:
logger.debug(f"Old embed for {selected_server_id} already deleted")
except Exception as e:
logger.error(f"Failed to delete old embed: {str(e)}")
# Create and send new permanent status embed in channel
from models import ServerData, ServerResources
server_obj = ServerData.model_validate(server_data)
resources_obj = ServerResources.model_validate(resources)
embed, view = await bot.embed_manager.create_server_embed(server_obj, resources_obj)
message = await interaction.channel.send(embed=embed, view=view)
await bot.embed_manager.track_embed(selected_server_id, message)
await interaction.followup.send(
f"✅ **{server_name}** status has been posted in {interaction.channel.mention}",
ephemeral=True,
)
logger.info(f"Successfully posted status for {server_name}")
except Exception as e:
logger.error(f"Failed to create status embed for {server_name}: {str(e)}")
await interaction.followup.send(
f"❌ Failed to create status embed for **{server_name}**: {str(e)}", ephemeral=True
)
class ServerDropdownView(discord.ui.View):
"""View containing the server dropdown with timeout for auto-cleanup."""
def __init__(self, server_options: list, timeout: int = 180) -> None:
"""
Initialize dropdown view with timeout.
Args:
server_options: List of discord.SelectOption objects
timeout: Timeout in seconds (default 3 minutes for ephemeral cleanup)
"""
super().__init__(timeout=timeout)
self.server_options = server_options
self.add_item(ServerDropdown(server_options))
async def on_timeout(self) -> None:
"""Clean up when dropdown times out (ephemeral auto-removal)."""
logger.debug("Server dropdown timed out and was automatically cleaned up")