k.eaven/Pterodisbot
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 5 Wiki Activity

feat: Multi-guild support implementation

Browse Source 
- Added multi-guild support with comma-separated AllowedGuildIDs config
- Per-guild embed tracking with isolated state
- Per-server action locking to prevent race conditions across guilds
- Cross-guild embed updates after power actions
- Debug logging for embed update triggers
- Fixed resource usage field formatting (brackets now properly closed)
- Updated config.ini.sample, generate_config.py for backward compatibility
- Updated README.md with multi-guild documentation
- Updated tests for multi-guild support
This commit is contained in:
k.eaven committed 2026-10-30 10:37:13 +00:00
1 parent e68b248033
commit bec06bd909
10 files changed
+802 -408
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

bot.py
+386 -216
View File
@@ -3,6 +3,7 @@ Main Bot Module
Core bot class and command definitions.
Manages the bot lifecycle, background tasks, and Discord interactions.
Supports multi-guild deployments with independent server tracking per guild.
"""
import asyncio
@@ -10,8 +11,9 @@ import logging
import signal
import sys
import types
from collections import defaultdict
from datetime import datetime
from typing import TYPE_CHECKING, Optional, Tuple
from typing import TYPE_CHECKING, Dict, List, Optional, Tuple
import discord
from discord.ext import commands, tasks
@@ -43,14 +45,15 @@ class PterodactylBot(commands.Bot):
Main bot class for Pterodactyl/Pelican Panel server management.
Handles Discord interactions, embed management, and background tasks.
Manages server status embeds and user commands.
Manages server status embeds and user commands across multiple guilds.
Features:
- Background status update task with smart update detection
- Persistent embed tracking with JSON storage
- Background status update task with smart update detection per guild
- Persistent embed tracking with JSON storage per guild
- Real-time server monitoring with metrics graphs
- Role-based access control
- Single-guild restriction
- Role-based access control per guild
- Multi-guild support with independent server tracking
- Per-server action locking to prevent race conditions
- Graceful shutdown handling
"""
@@ -73,23 +76,121 @@ class PterodactylBot(commands.Bot):
self.metrics_manager = ServerMetricsManager()
self.embed_manager = EmbedManager(config, self.api, self.metrics_manager)
# Server cache from panel
self.server_cache: dict = {}
# Server cache from panel (shared across guilds, read-mostly)
self.server_cache: Dict[str, dict] = {}
self._server_cache_lock = asyncio.Lock()
# Concurrency control
# Concurrency control for background update task
self.update_lock = asyncio.Lock()
# State tracking for smart updates
# Format: {server_id: (state, cpu_usage, last_force_update)}
self.previous_states: dict = {}
# Per-server action locks to serialize concurrent power actions across guilds
# Key: server_id, Value: asyncio.Lock
self._server_action_locks: Dict[str, asyncio.Lock] = defaultdict(asyncio.Lock)
logger.info("Initialized PterodactylBot instance with state tracking")
# State tracking for smart updates per guild
# Format: {guild_id: {server_id: (state, cpu_usage, last_force_update)}}
self.previous_states: Dict[int, Dict[str, Tuple[str, float, Optional[float]]]] = defaultdict(dict)
logger.info(f"Initialized PterodactylBot instance with multi-guild support for {len(config.allowed_guild_ids)} guild(s)")
@classmethod
def get_instance(cls) -> Optional["PterodactylBot"]:
"""Get the singleton bot instance."""
return _bot_instance
def get_server_action_lock(self, server_id: str) -> asyncio.Lock:
"""
Get or create a per-server lock for serializing power actions.
This prevents race conditions when users from multiple guilds
press buttons for the same server simultaneously.
Args:
server_id: Panel server identifier
Returns:
asyncio.Lock for the specified server
"""
return self._server_action_locks[server_id]
async def trigger_embed_update_for_server(self, server_id: str) -> None:
"""
Trigger an immediate embed update for a server across all guilds tracking it.
Called after a successful power action to ensure all guilds see the state change promptly.
Bypasses smart update detection since we know the state changed.
Args:
server_id: Panel server identifier
"""
if server_id not in self.server_cache:
logger.warning(f"Cannot trigger update for unknown server {server_id}")
return
server_data = self.server_cache[server_id]
server_name = server_data["attributes"]["name"]
try:
# Fetch fresh resources
resources = await self.api.get_server_resources(server_id)
current_state = resources["attributes"]["current_state"]
cpu_usage = round(
resources["attributes"]["resources"].get("cpu_absolute", 0), 2
)
# Collect metrics for running servers
if current_state == "running":
memory_usage = round(
resources["attributes"]["resources"].get("memory_bytes", 0)
/ (1024**2),
2,
)
self.metrics_manager.add_server_data(
server_id, server_name, cpu_usage, memory_usage
)
# Convert to models for embed creation
server_obj = ServerData.model_validate(server_data)
resources_obj = ServerResources.model_validate(resources)
current_time = datetime.now().timestamp()
# Update in all guilds that track this server
for guild_id in self.config.allowed_guild_ids:
guild_locations = self.embed_manager.get_guild_locations(guild_id)
if server_id not in guild_locations:
continue
location = guild_locations[server_id]
# Force update (bypass smart detection)
success = await self.embed_manager.update_single_embed(
guild_id=guild_id,
server_id=server_id,
location=location,
server_data=server_obj,
resources=resources_obj,
current_state=current_state,
cpu_usage=cpu_usage,
)
if success:
logger.debug(
f"Cross-guild embed update triggered for {server_name} ({server_id}) in guild {guild_id}: power action"
)
# Update state tracking for this guild
guild_states = self.previous_states[guild_id]
guild_states[server_id] = (current_state, cpu_usage, current_time)
logger.debug(f"Force-updated embed for {server_name} in guild {guild_id}")
else:
logger.warning(f"Failed to force-update embed for {server_name} in guild {guild_id}")
# Small delay between guilds to respect rate limits
await asyncio.sleep(0.3)
except Exception as e:
logger.error(f"Failed to trigger cross-guild update for server {server_id}: {str(e)}")
async def setup_hook(self) -> None:
"""
Bot setup routine called when the bot is starting.
@@ -112,184 +213,241 @@ class PterodactylBot(commands.Bot):
@tasks.loop(seconds=10) # UPDATE_INTERVAL = 10
async def update_status(self) -> None:
"""
Background task to update server status embeds when:
Background task to update server status embeds across all guilds.
For each guild, updates embeds when:
1. Server power state changes (started/stopped/restarted)
2. Significant CPU usage change (>50% difference)
3. First time seeing the server
3. First time seeing the server in that guild
4. Server has been running for 10 minutes (force update for uptime)
This minimizes API calls to Discord and updates while maintaining
This minimizes API calls to Discord while maintaining
real-time awareness of important server changes.
"""
logger.info("Starting optimized server status update cycle")
logger.info("Starting multi-guild server status update cycle")
async with self.update_lock:
try:
# Fetch current server list from panel
# Fetch current server list from panel (once per cycle, shared across guilds)
servers = await self.api.get_servers()
if not servers:
logger.warning("No servers found in panel during update")
return
# Update our local cache with fresh server data
self.server_cache = {s["attributes"]["identifier"]: s for s in servers}
async with self._server_cache_lock:
self.server_cache = {s["attributes"]["identifier"]: s for s in servers}
logger.debug(f"Updated server cache with {len(servers)} servers")
# Clean up metrics for servers that no longer exist
active_server_ids = list(self.server_cache.keys())
self.metrics_manager.cleanup_old_servers(active_server_ids)
# Update statistics
update_count = 0
error_count = 0
missing_count = 0
skipped_count = 0
current_time = datetime.now().timestamp()
# Process each server we're tracking embeds for
for server_id, location in list(self.embed_manager.embed_locations.items()):
# Skip if server no longer exists in panel
if server_id not in self.server_cache:
logger.warning(f"Server {server_id} not found in cache, skipping update")
continue
# Process each guild's embeds
total_updated = 0
total_skipped = 0
total_missing = 0
total_errors = 0
server_data = self.server_cache[server_id]
server_name = server_data["attributes"]["name"]
try:
logger.debug(f"Checking status for server {server_name} ({server_id})")
# Get current server resource usage
resources = await self.api.get_server_resources(server_id)
current_state = resources["attributes"]["current_state"]
cpu_usage = round(
resources["attributes"]["resources"].get("cpu_absolute", 0), 2
)
# Collect metrics data for running servers
if current_state == "running":
memory_usage = round(
resources["attributes"]["resources"].get("memory_bytes", 0)
/ (1024**2),
2,
)
self.metrics_manager.add_server_data(
server_id, server_name, cpu_usage, memory_usage
)
logger.debug(
f"Added metrics data for {server_name}: "
f"CPU={cpu_usage}%, Memory={memory_usage}MB"
)
# Retrieve previous recorded state
prev_state, prev_cpu, last_force_update = self.previous_states.get(
server_id, (None, 0, None)
)
# DECISION LOGIC: Should we update the embed?
needs_update = False
# 1. Check if power state changed (most important)
if current_state != prev_state:
logger.debug(
f"Power state changed for {server_name}: "
f"{prev_state} -> {current_state}"
)
needs_update = True
# 2. Check for significant CPU change (only if server is running)
elif current_state == "running" and abs(cpu_usage - prev_cpu) > 50:
logger.debug(
f"Significant CPU change for {server_name}: "
f"{prev_cpu}% -> {cpu_usage}%"
)
needs_update = True
# 3. First time we're seeing this server (initial update)
elif prev_state is None:
logger.debug(f"First check for {server_name}, performing initial update")
needs_update = True
# 4. Force update every 10 minutes for running servers (for uptime counter)
elif current_state == "running" and (
last_force_update is None
or current_time - last_force_update >= 600
): # 10 minutes = 600 seconds
logger.debug(f"Executing 10-minute force update for running server {server_name}")
needs_update = True
last_force_update = current_time
# PERFORM UPDATE IF NEEDED
if needs_update:
# Convert to models for embed creation
server_obj = ServerData.model_validate(server_data)
resources_obj = ServerResources.model_validate(resources)
# Update the embed
success = await self.embed_manager.update_single_embed(
server_id=server_id,
location=location,
server_data=server_obj,
resources=resources_obj,
current_state=current_state,
cpu_usage=cpu_usage,
)
if success:
update_count += 1
# Update state tracking
new_last_force_update = (
last_force_update
if needs_update
and current_state == "running"
and current_time - (last_force_update or 0) >= 600
else last_force_update
)
self.previous_states[server_id] = (
current_state,
cpu_usage,
new_last_force_update,
)
else:
# No significant changes, but update tracking with current state
self.previous_states[server_id] = (
current_state,
cpu_usage,
last_force_update,
)
skipped_count += 1
logger.debug(f"No changes detected for {server_name}, skipping update")
except discord.NotFound:
# Embed message was deleted - clean up tracking
logger.warning(f"Embed for server {server_id} not found, removing from tracking")
await self.embed_manager.untrack_embed(server_id)
missing_count += 1
except Exception as e:
logger.error(f"Failed to update status for server {server_id}: {str(e)}")
error_count += 1
# Small delay between servers to avoid rate limits
await asyncio.sleep(0.5)
for guild_id in self.config.allowed_guild_ids:
u, s, m, e = await self._update_guild_embeds(guild_id, current_time)
total_updated += u
total_skipped += s
total_missing += m
total_errors += e
# Log summary of this update cycle
logger.info(
f"Update cycle complete: "
f"{update_count} updated, "
f"{skipped_count} skipped, "
f"{missing_count} missing, "
f"{error_count} errors"
f"Multi-guild update cycle complete: "
f"{total_updated} updated, "
f"{total_skipped} skipped, "
f"{total_missing} missing, "
f"{total_errors} errors across {len(self.config.allowed_guild_ids)} guild(s)"
)
except Exception as e:
logger.error(f"Error in update_status task: {str(e)}")
await asyncio.sleep(5)
async def _update_guild_embeds(
self, guild_id: int, current_time: float
) -> Tuple[int, int, int, int]:
"""
Update embeds for a specific guild.
Args:
guild_id: Discord guild ID
current_time: Current timestamp for force-update calculations
Returns:
Tuple of (updated, skipped, missing, errors) counts
"""
guild_locations = self.embed_manager.get_guild_locations(guild_id)
if not guild_locations:
return 0, 0, 0, 0
guild_prev_states = self.previous_states[guild_id]
update_count = 0
skipped_count = 0
missing_count = 0
error_count = 0
for server_id, location in list(guild_locations.items()):
# Skip if server no longer exists in panel
if server_id not in self.server_cache:
logger.warning(f"Server {server_id} not found in cache, skipping update for guild {guild_id}")
continue
server_data = self.server_cache[server_id]
server_name = server_data["attributes"]["name"]
try:
logger.debug(f"Checking status for server {server_name} ({server_id}) in guild {guild_id}")
# Get current server resource usage
resources = await self.api.get_server_resources(server_id)
current_state = resources["attributes"]["current_state"]
cpu_usage = round(
resources["attributes"]["resources"].get("cpu_absolute", 0), 2
)
# Collect metrics data for running servers (shared across guilds)
if current_state == "running":
memory_usage = round(
resources["attributes"]["resources"].get("memory_bytes", 0)
/ (1024**2),
2,
)
self.metrics_manager.add_server_data(
server_id, server_name, cpu_usage, memory_usage
)
logger.debug(
f"Added metrics data for {server_name}: "
f"CPU={cpu_usage}%, Memory={memory_usage}MB"
)
# Retrieve previous recorded state for THIS guild
prev_state, prev_cpu, last_force_update = guild_prev_states.get(
server_id, (None, 0, None)
)
# DECISION LOGIC: Should we update the embed?
needs_update, trigger_reason = self._should_update_embed(
current_state, prev_state, cpu_usage, prev_cpu,
last_force_update, current_time
)
# PERFORM UPDATE IF NEEDED
if needs_update:
logger.debug(
f"Embed update triggered for {server_name} ({server_id}) in guild {guild_id}: {trigger_reason}"
)
# Convert to models for embed creation
server_obj = ServerData.model_validate(server_data)
resources_obj = ServerResources.model_validate(resources)
# Update the embed
success = await self.embed_manager.update_single_embed(
guild_id=guild_id,
server_id=server_id,
location=location,
server_data=server_obj,
resources=resources_obj,
current_state=current_state,
cpu_usage=cpu_usage,
)
if success:
update_count += 1
# Update state tracking for this guild
new_last_force_update = current_time if (
current_state == "running" and
(last_force_update is None or current_time - last_force_update >= 600)
) else last_force_update
guild_prev_states[server_id] = (
current_state,
cpu_usage,
new_last_force_update,
)
else:
# No significant changes, but update tracking with current state
guild_prev_states[server_id] = (
current_state,
cpu_usage,
last_force_update,
)
skipped_count += 1
logger.debug(f"No changes detected for {server_name} in guild {guild_id}, skipping update")
except discord.NotFound:
# Embed message was deleted - clean up tracking for this guild
logger.warning(f"Embed for server {server_id} in guild {guild_id} not found, removing from tracking")
await self.embed_manager.untrack_embed(guild_id, server_id)
missing_count += 1
except Exception as e:
logger.error(f"Failed to update status for server {server_id} in guild {guild_id}: {str(e)}")
error_count += 1
# Small delay between servers to avoid rate limits
await asyncio.sleep(0.5)
return update_count, skipped_count, missing_count, error_count
def _should_update_embed(
self,
current_state: str,
prev_state: Optional[str],
cpu_usage: float,
prev_cpu: float,
last_force_update: Optional[float],
current_time: float,
) -> Tuple[bool, str]:
"""
Determine if an embed needs updating based on smart detection criteria.
This logic is shared between the background task and cross-guild triggers.
Args:
current_state: Current server power state
prev_state: Previously recorded state (or None)
cpu_usage: Current CPU usage percentage
prev_cpu: Previously recorded CPU usage
last_force_update: Timestamp of last forced update (or None)
current_time: Current timestamp
Returns:
Tuple of (should_update: bool, reason: str)
"""
# 1. Check if power state changed (most important)
if current_state != prev_state:
return True, f"power state changed: {prev_state} -> {current_state}"
# 2. Check for significant CPU change (only if server is running)
if current_state == "running" and abs(cpu_usage - prev_cpu) > 50:
return True, f"significant CPU change: {prev_cpu}% -> {cpu_usage}%"
# 3. First time we're seeing this server (initial update)
if prev_state is None:
return True, "first detection (initial update)"
# 4. Force update every 10 minutes for running servers (for uptime counter)
if current_state == "running" and (
last_force_update is None
or current_time - last_force_update >= 600
): # 10 minutes = 600 seconds
return True, f"10-minute force update (last: {last_force_update})"
return False, "no significant changes"
@update_status.before_loop
async def before_update_status(self) -> None:
"""Wait for bot to be ready before starting update task."""
"""Wait for bot to be ready before starting update task, then do initial refresh."""
logger.debug("Waiting for bot readiness before starting update task")
await self.wait_until_ready()
await self.embed_manager.refresh_all_embeds()
# Initial refresh for all guilds
for guild_id in self.config.allowed_guild_ids:
await self.embed_manager.refresh_all_embeds(guild_id)
@update_status.after_loop
async def after_update_status(self) -> None:
@@ -309,7 +467,7 @@ class PterodactylBot(commands.Bot):
async def check_allowed_guild(interaction: discord.Interaction, config: Config) -> bool:
"""
Verify that an interaction is coming from the allowed guild.
Verify that an interaction is coming from an allowed guild.
Args:
interaction: Discord interaction object
@@ -318,13 +476,13 @@ async def check_allowed_guild(interaction: discord.Interaction, config: Config)
Returns:
bool: True if interaction is allowed, False otherwise
"""
if interaction.guild_id != config.allowed_guild_id:
if interaction.guild_id not in config.allowed_guild_ids:
logger.warning(
f"Command attempted from unauthorized guild {interaction.guild_id} "
f"by {interaction.user.name}"
)
await interaction.response.send_message(
"This bot is only available in a specific server.", ephemeral=True
"This bot is not available in this server.", ephemeral=True
)
return False
return True
@@ -346,10 +504,10 @@ async def setup_commands(bot: PterodactylBot) -> None:
@bot.event
async def on_interaction(interaction: discord.Interaction) -> None:
"""Global interaction handler to check guild before processing any interaction."""
if interaction.guild_id != config.allowed_guild_id:
if interaction.guild_id not in config.allowed_guild_ids:
logger.debug(f"Ignoring interaction from unauthorized guild {interaction.guild_id}")
await interaction.response.send_message(
"This bot is only available in a specific server.", ephemeral=True
"This bot is not available in this server.", ephemeral=True
)
return
@@ -359,14 +517,15 @@ async def setup_commands(bot: PterodactylBot) -> None:
logger.info(f"Bot connected as {bot.user.name} (ID: {bot.user.id})")
try:
# Sync commands only to the allowed guild
guild = discord.Object(id=bot.config.allowed_guild_id)
bot.tree.copy_global_to(guild=guild)
synced = await bot.tree.sync(guild=guild)
logger.info(
f"Successfully synced {len(synced)} command(s) to guild {bot.config.allowed_guild_id}: "
f"{[cmd.name for cmd in synced]}"
)
# Sync commands to ALL allowed guilds
for guild_id in bot.config.allowed_guild_ids:
guild = discord.Object(id=guild_id)
bot.tree.copy_global_to(guild=guild)
synced = await bot.tree.sync(guild=guild)
logger.info(
f"Successfully synced {len(synced)} command(s) to guild {guild_id}: "
f"{[cmd.name for cmd in synced]}"
)
except Exception as e:
logger.error(f"Command sync failed: {str(e)}")
@@ -389,12 +548,13 @@ async def setup_commands(bot: PterodactylBot) -> None:
5. Generates dropdown menu with all servers (name + description)
6. Handles user selection via ephemeral dropdown interaction
7. Creates permanent status embed in channel upon selection
8. Manages embed tracking and cleanup of previous embeds
8. Manages embed tracking and cleanup of previous embeds (per guild)
"""
if not await check_allowed_guild(interaction, config):
return
logger.info(f"Server status command invoked by {interaction.user.name}")
guild_id = interaction.guild_id
logger.info(f"Server status command invoked by {interaction.user.name} in guild {guild_id}")
await interaction.response.defer(ephemeral=True)
try:
@@ -407,7 +567,8 @@ async def setup_commands(bot: PterodactylBot) -> None:
)
return
bot.server_cache = {s["attributes"]["identifier"]: s for s in servers}
async with bot._server_cache_lock:
bot.server_cache = {s["attributes"]["identifier"]: s for s in servers}
logger.debug(f"Refreshed server cache with {len(servers)} servers")
# Count online/offline servers
@@ -472,7 +633,7 @@ async def setup_commands(bot: PterodactylBot) -> None:
view=ServerDropdownView(server_options),
ephemeral=True,
)
logger.info(f"Sent server status dashboard to {interaction.user.name} with {len(server_options)} servers")
logger.info(f"Sent server status dashboard to {interaction.user.name} in guild {guild_id} with {len(server_options)} servers")
except Exception as e:
logger.error(f"Server status command failed: {str(e)}")
@@ -481,47 +642,48 @@ async def setup_commands(bot: PterodactylBot) -> None:
)
@bot.tree.command(
name="refresh_embeds", description="Refresh all server status embeds (admin only)"
name="refresh_embeds", description="Refresh all server status embeds in this guild (admin only)"
)
async def refresh_embeds(interaction: discord.Interaction) -> None:
"""Slash command to refresh all server embeds."""
"""Slash command to refresh all server embeds in the current guild."""
if not await check_allowed_guild(interaction, config):
return
logger.info(f"Refresh embeds command invoked by {interaction.user.name}")
guild_id = interaction.guild_id
logger.info(f"Refresh embeds command invoked by {interaction.user.name} in guild {guild_id}")
await interaction.response.defer(ephemeral=True)
# Require administrator permissions
if not interaction.user.guild_permissions.administrator:
logger.warning(f"Unauthorized refresh attempt by {interaction.user.name}")
logger.warning(f"Unauthorized refresh attempt by {interaction.user.name} in guild {guild_id}")
await interaction.followup.send(
"You need administrator permissions to refresh all embeds.", ephemeral=True
)
return
try:
logger.info("Starting full embed refresh per admin request")
deleted, created = await bot.embed_manager.refresh_all_embeds()
logger.info(f"Starting full embed refresh per admin request for guild {guild_id}")
deleted, created = await bot.embed_manager.refresh_all_embeds(guild_id)
await interaction.followup.send(
f"Refreshed all embeds. Deleted {deleted} old embeds, created {created} new ones.",
f"Refreshed all embeds in this guild. Deleted {deleted} old embeds, created {created} new ones.",
ephemeral=True,
)
logger.info(f"Embed refresh completed: {deleted} deleted, {created} created")
logger.info(f"Embed refresh completed for guild {guild_id}: {deleted} deleted, {created} created")
except Exception as e:
logger.error(f"Embed refresh failed: {str(e)}")
logger.error(f"Embed refresh failed for guild {guild_id}: {str(e)}")
await interaction.followup.send(f"Failed to refresh embeds: {str(e)}", ephemeral=True)
@bot.tree.command(
name="purge_embeds",
description="Permanently delete all server status embeds (admin only)",
description="Permanently delete all server status embeds in this guild (admin only)",
)
async def purge_embeds(interaction: discord.Interaction) -> None:
"""
Slash command to permanently purge all server status embeds from Discord channels.
Slash command to permanently purge all server status embeds from the current guild.
Workflow:
1. Validates administrator permissions
2. Checks if any embeds are currently tracked
2. Checks if any embeds are currently tracked in this guild
3. Sends initial progress embed
4. Processes each embed sequentially with error handling
5. Updates progress embed in real-time
@@ -531,35 +693,38 @@ async def setup_commands(bot: PterodactylBot) -> None:
if not await check_allowed_guild(interaction, config):
return
logger.info(f"Purge embeds command invoked by {interaction.user.name}")
guild_id = interaction.guild_id
logger.info(f"Purge embeds command invoked by {interaction.user.name} in guild {guild_id}")
await interaction.response.defer(ephemeral=True)
# Require administrator permissions
if not interaction.user.guild_permissions.administrator:
logger.warning(f"Unauthorized purge attempt by {interaction.user.name}")
logger.warning(f"Unauthorized purge attempt by {interaction.user.name} in guild {guild_id}")
await interaction.followup.send(
"You need administrator permissions to purge all embeds.", ephemeral=True
)
return
try:
logger.info("Starting embed purge per admin request")
logger.info(f"Starting embed purge per admin request for guild {guild_id}")
guild_locations = bot.embed_manager.get_guild_locations(guild_id)
total_embeds = len(guild_locations)
if total_embeds == 0:
await interaction.followup.send(
"No embeds are currently being tracked in this guild. Nothing to purge.", ephemeral=True
)
return
deleted_count = 0
not_found_count = 0
error_count = 0
total_embeds = len(bot.embed_manager.embed_locations)
if total_embeds == 0:
await interaction.followup.send(
"No embeds are currently being tracked. Nothing to purge.", ephemeral=True
)
return
# Create progress embed
progress_embed = discord.Embed(
title="🔄 Purging Server Embeds",
description=f"Processing {total_embeds} embeds...",
description=f"Processing {total_embeds} embeds in this guild...",
color=discord.Color.orange(),
)
progress_embed.add_field(name="Deleted", value="0", inline=True)
@@ -569,31 +734,36 @@ async def setup_commands(bot: PterodactylBot) -> None:
progress_message = await interaction.followup.send(embed=progress_embed, ephemeral=True)
# Process each tracked embed
for server_id, location in list(bot.embed_manager.embed_locations.items()):
# Process each tracked embed in this guild
for server_id, location in list(guild_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"Successfully purged embed for server {server_id}")
except discord.NotFound:
# Verify channel belongs to the correct guild
if channel.guild.id != guild_id:
logger.warning(f"Channel {location['channel_id']} not in guild {guild_id}, skipping")
not_found_count += 1
logger.debug(f"Embed for server {server_id} already deleted")
except discord.Forbidden:
error_count += 1
logger.error(f"Permission denied when deleting embed for server {server_id}")
except Exception as e:
error_count += 1
logger.error(f"Error deleting embed for server {server_id}: {str(e)}")
else:
try:
message = await channel.fetch_message(int(location["message_id"]))
await message.delete()
deleted_count += 1
logger.debug(f"Successfully purged embed for server {server_id} in guild {guild_id}")
except discord.NotFound:
not_found_count += 1
logger.debug(f"Embed for server {server_id} in guild {guild_id} already deleted")
except discord.Forbidden:
error_count += 1
logger.error(f"Permission denied when deleting embed for server {server_id} in guild {guild_id}")
except Exception as e:
error_count += 1
logger.error(f"Error deleting embed for server {server_id} in guild {guild_id}: {str(e)}")
else:
not_found_count += 1
logger.warning(f"Channel not found for server {server_id}")
logger.warning(f"Channel not found for server {server_id} in guild {guild_id}")
# Remove from tracking immediately
await bot.embed_manager.untrack_embed(server_id)
await bot.embed_manager.untrack_embed(guild_id, server_id)
# Update progress every 5 embeds or for the last one
processed = deleted_count + not_found_count + error_count
@@ -610,7 +780,7 @@ async def setup_commands(bot: PterodactylBot) -> None:
except Exception as e:
error_count += 1
logger.error(f"Unexpected error processing server {server_id}: {str(e)}")
logger.error(f"Unexpected error processing server {server_id} in guild {guild_id}: {str(e)}")
# Create results embed
result_embed = discord.Embed(
@@ -627,13 +797,13 @@ async def setup_commands(bot: PterodactylBot) -> None:
value=f"{((deleted_count + not_found_count) / total_embeds * 100):.1f}%",
inline=True,
)
result_embed.set_footer(text="Embed tracking file has been cleared")
result_embed.set_footer(text="Embed tracking for this guild has been cleared")
await progress_message.edit(embed=result_embed)
logger.info(f"Embed purge completed: {deleted_count} deleted, {not_found_count} not found, {error_count} errors")
logger.info(f"Embed purge completed for guild {guild_id}: {deleted_count} deleted, {not_found_count} not found, {error_count} errors")
except Exception as e:
logger.error(f"Embed purge failed: {str(e)}")
logger.error(f"Embed purge failed for guild {guild_id}: {str(e)}")
await interaction.followup.send(f"❌ Failed to purge embeds: {str(e)}", ephemeral=True)