From e3610e84ea668a12bc310eb17027dd2183ca3009 Mon Sep 17 00:00:00 2001 From: Cal Corum Date: Tue, 3 Mar 2026 11:36:50 -0600 Subject: [PATCH] fix: implement actual maintenance mode flag in /admin-maintenance (#28) - Add `maintenance_mode: bool = False` flag to `SBABot.__init__` - Register a global `@tree.interaction_check` that blocks non-admin users from all commands when maintenance mode is active - Update `admin_maintenance` command to set `self.bot.maintenance_mode` and log the state change, replacing the no-op placeholder comment Co-Authored-By: Claude Sonnet 4.6 --- bot.py | 17 ++ commands/admin/management.py | 507 ++++++++++++++++++----------------- 2 files changed, 274 insertions(+), 250 deletions(-) diff --git a/bot.py b/bot.py index b0fdef0..04a16e6 100644 --- a/bot.py +++ b/bot.py @@ -80,11 +80,28 @@ class SBABot(commands.Bot): ) self.logger = logging.getLogger("discord_bot_v2") + self.maintenance_mode: bool = False async def setup_hook(self): """Called when the bot is starting up.""" self.logger.info("Setting up bot...") + @self.tree.interaction_check + async def maintenance_check(interaction: discord.Interaction) -> bool: + """Block non-admin users when maintenance mode is enabled.""" + if not self.maintenance_mode: + return True + if ( + isinstance(interaction.user, discord.Member) + and interaction.user.guild_permissions.administrator + ): + return True + await interaction.response.send_message( + "🔧 The bot is currently in maintenance mode. Please try again later.", + ephemeral=True, + ) + return False + # Load command packages await self._load_command_packages() diff --git a/commands/admin/management.py b/commands/admin/management.py index 5377ca7..8950c4c 100644 --- a/commands/admin/management.py +++ b/commands/admin/management.py @@ -3,6 +3,7 @@ Admin Management Commands Administrative commands for league management and bot maintenance. """ + import asyncio from typing import List, Dict, Any @@ -23,145 +24,145 @@ from services.player_service import player_service class AdminCommands(commands.Cog): """Administrative command handlers for league management.""" - + def __init__(self, bot: commands.Bot): self.bot = bot - self.logger = get_contextual_logger(f'{__name__}.AdminCommands') - + self.logger = get_contextual_logger(f"{__name__}.AdminCommands") + async def interaction_check(self, interaction: discord.Interaction) -> bool: """Check if user has admin permissions.""" # Check if interaction is from a guild and user is a Member if not isinstance(interaction.user, discord.Member): await interaction.response.send_message( - "❌ Admin commands can only be used in a server.", - ephemeral=True + "❌ Admin commands can only be used in a server.", ephemeral=True ) return False if not interaction.user.guild_permissions.administrator: await interaction.response.send_message( "❌ You need administrator permissions to use admin commands.", - ephemeral=True + ephemeral=True, ) return False return True - + @app_commands.command( - name="admin-status", - description="Display bot status and system information" + name="admin-status", description="Display bot status and system information" ) @league_admin_only() @logged_command("/admin-status") async def admin_status(self, interaction: discord.Interaction): """Display comprehensive bot status information.""" await interaction.response.defer() - + # Gather system information guilds_count = len(self.bot.guilds) users_count = sum(guild.member_count or 0 for guild in self.bot.guilds) commands_count = len([cmd for cmd in self.bot.tree.walk_commands()]) - + # Bot uptime calculation - uptime = discord.utils.utcnow() - self.bot.user.created_at if self.bot.user else None - uptime_str = f"{uptime.days} days" if uptime else "Unknown" - - embed = EmbedTemplate.create_base_embed( - title="🤖 Bot Status - Admin Panel", - color=EmbedColors.INFO + uptime = ( + discord.utils.utcnow() - self.bot.user.created_at if self.bot.user else None ) - + uptime_str = f"{uptime.days} days" if uptime else "Unknown" + + embed = EmbedTemplate.create_base_embed( + title="🤖 Bot Status - Admin Panel", color=EmbedColors.INFO + ) + # System Stats embed.add_field( name="System Information", value=f"**Guilds:** {guilds_count}\n" - f"**Users:** {users_count:,}\n" - f"**Commands:** {commands_count}\n" - f"**Uptime:** {uptime_str}", - inline=True + f"**Users:** {users_count:,}\n" + f"**Commands:** {commands_count}\n" + f"**Uptime:** {uptime_str}", + inline=True, ) - - # Bot Information + + # Bot Information embed.add_field( name="Bot Information", value=f"**Latency:** {round(self.bot.latency * 1000)}ms\n" - f"**Version:** Discord.py {discord.__version__}\n" - f"**Current Season:** {get_config().sba_season}", - inline=True + f"**Version:** Discord.py {discord.__version__}\n" + f"**Current Season:** {get_config().sba_season}", + inline=True, ) - + # Cog Status loaded_cogs = list(self.bot.cogs.keys()) embed.add_field( name="Loaded Cogs", - value="\n".join([f"✅ {cog}" for cog in loaded_cogs[:10]]) + - (f"\n... and {len(loaded_cogs) - 10} more" if len(loaded_cogs) > 10 else ""), - inline=False + value="\n".join([f"✅ {cog}" for cog in loaded_cogs[:10]]) + + ( + f"\n... and {len(loaded_cogs) - 10} more" + if len(loaded_cogs) > 10 + else "" + ), + inline=False, ) - + embed.set_footer(text="Admin Status • Use /admin-help for more commands") await interaction.followup.send(embed=embed) - + @app_commands.command( name="admin-help", - description="Display available admin commands and their usage" + description="Display available admin commands and their usage", ) @league_admin_only() @logged_command("/admin-help") async def admin_help(self, interaction: discord.Interaction): """Display comprehensive admin help information.""" await interaction.response.defer() - + embed = EmbedTemplate.create_base_embed( title="🛠️ Admin Commands - Help", description="Available administrative commands for league management", - color=EmbedColors.PRIMARY + color=EmbedColors.PRIMARY, ) - + # System Commands embed.add_field( name="System Management", value="**`/admin-status`** - Display bot status and information\n" - "**`/admin-reload `** - Reload a specific cog\n" - "**`/admin-sync`** - Sync application commands\n" - "**`/admin-clear `** - Clear messages from channel\n" - "**`/admin-clear-scorecards`** - Clear live scorebug channel and hide it", - inline=False + "**`/admin-reload `** - Reload a specific cog\n" + "**`/admin-sync`** - Sync application commands\n" + "**`/admin-clear `** - Clear messages from channel\n" + "**`/admin-clear-scorecards`** - Clear live scorebug channel and hide it", + inline=False, ) - + # League Management embed.add_field( name="League Management", value="**`/admin-season `** - Set current season\n" - "**`/admin-announce `** - Send announcement to channel\n" - "**`/admin-maintenance `** - Toggle maintenance mode\n" - "**`/admin-process-transactions [week]`** - Manually process weekly transactions", - inline=False + "**`/admin-announce `** - Send announcement to channel\n" + "**`/admin-maintenance `** - Toggle maintenance mode\n" + "**`/admin-process-transactions [week]`** - Manually process weekly transactions", + inline=False, ) - + # User Management embed.add_field( - name="User Management", + name="User Management", value="**`/admin-timeout `** - Timeout a user\n" - "**`/admin-kick `** - Kick a user\n" - "**`/admin-ban `** - Ban a user", - inline=False + "**`/admin-kick `** - Kick a user\n" + "**`/admin-ban `** - Ban a user", + inline=False, ) - + embed.add_field( name="Usage Notes", value="• All admin commands require Administrator permissions\n" - "• Commands are logged for audit purposes\n" - "• Use with caution - some actions are irreversible", - inline=False + "• Commands are logged for audit purposes\n" + "• Use with caution - some actions are irreversible", + inline=False, ) - + embed.set_footer(text="Administrator Permissions Required") await interaction.followup.send(embed=embed) - - @app_commands.command( - name="admin-reload", - description="Reload a specific bot cog" - ) + + @app_commands.command(name="admin-reload", description="Reload a specific bot cog") @app_commands.describe( cog="Name of the cog to reload (e.g., 'commands.players.info')" ) @@ -170,53 +171,52 @@ class AdminCommands(commands.Cog): async def admin_reload(self, interaction: discord.Interaction, cog: str): """Reload a specific cog for hot-swapping code changes.""" await interaction.response.defer() - + try: # Attempt to reload the cog await self.bot.reload_extension(cog) - + embed = EmbedTemplate.create_base_embed( title="✅ Cog Reloaded Successfully", description=f"Successfully reloaded `{cog}`", - color=EmbedColors.SUCCESS + color=EmbedColors.SUCCESS, ) - + embed.add_field( name="Reload Details", value=f"**Cog:** {cog}\n" - f"**Status:** Successfully reloaded\n" - f"**Time:** {discord.utils.utcnow().strftime('%H:%M:%S UTC')}", - inline=False + f"**Status:** Successfully reloaded\n" + f"**Time:** {discord.utils.utcnow().strftime('%H:%M:%S UTC')}", + inline=False, ) - + except commands.ExtensionNotFound: embed = EmbedTemplate.create_base_embed( title="❌ Cog Not Found", description=f"Could not find cog: `{cog}`", - color=EmbedColors.ERROR + color=EmbedColors.ERROR, ) except commands.ExtensionNotLoaded: embed = EmbedTemplate.create_base_embed( title="❌ Cog Not Loaded", description=f"Cog `{cog}` is not currently loaded", - color=EmbedColors.ERROR + color=EmbedColors.ERROR, ) except Exception as e: embed = EmbedTemplate.create_base_embed( title="❌ Reload Failed", description=f"Failed to reload `{cog}`: {str(e)}", - color=EmbedColors.ERROR + color=EmbedColors.ERROR, ) - + await interaction.followup.send(embed=embed) - + @app_commands.command( - name="admin-sync", - description="Sync application commands with Discord" + name="admin-sync", description="Sync application commands with Discord" ) @app_commands.describe( local="Sync to this guild only (fast, for development)", - clear_local="Clear locally synced commands (does not sync after clearing)" + clear_local="Clear locally synced commands (does not sync after clearing)", ) @league_admin_only() @logged_command("/admin-sync") @@ -224,7 +224,7 @@ class AdminCommands(commands.Cog): self, interaction: discord.Interaction, local: bool = False, - clear_local: bool = False + clear_local: bool = False, ): """Sync slash commands with Discord API.""" await interaction.response.defer() @@ -235,20 +235,24 @@ class AdminCommands(commands.Cog): if not interaction.guild_id: raise ValueError("Cannot clear local commands outside of a guild") - self.logger.info(f"Clearing local commands for guild {interaction.guild_id}") - self.bot.tree.clear_commands(guild=discord.Object(id=interaction.guild_id)) + self.logger.info( + f"Clearing local commands for guild {interaction.guild_id}" + ) + self.bot.tree.clear_commands( + guild=discord.Object(id=interaction.guild_id) + ) embed = EmbedTemplate.create_base_embed( title="✅ Local Commands Cleared", description=f"Cleared all commands synced to this guild", - color=EmbedColors.SUCCESS + color=EmbedColors.SUCCESS, ) embed.add_field( name="Clear Details", value=f"**Guild ID:** {interaction.guild_id}\n" - f"**Time:** {discord.utils.utcnow().strftime('%H:%M:%S UTC')}\n" - f"**Note:** Commands not synced after clearing", - inline=False + f"**Time:** {discord.utils.utcnow().strftime('%H:%M:%S UTC')}\n" + f"**Note:** Commands not synced after clearing", + inline=False, ) await interaction.followup.send(embed=embed) return @@ -270,25 +274,29 @@ class AdminCommands(commands.Cog): embed = EmbedTemplate.create_base_embed( title="✅ Commands Synced Successfully", description=f"Synced {len(synced_commands)} application commands {sync_type}", - color=EmbedColors.SUCCESS + color=EmbedColors.SUCCESS, ) # Show some of the synced commands command_names = [cmd.name for cmd in synced_commands[:10]] embed.add_field( name="Synced Commands", - value="\n".join([f"• /{name}" for name in command_names]) + - (f"\n... and {len(synced_commands) - 10} more" if len(synced_commands) > 10 else ""), - inline=False + value="\n".join([f"• /{name}" for name in command_names]) + + ( + f"\n... and {len(synced_commands) - 10} more" + if len(synced_commands) > 10 + else "" + ), + inline=False, ) embed.add_field( name="Sync Details", value=f"**Total Commands:** {len(synced_commands)}\n" - f"**Sync Type:** {sync_type.title()}\n" - f"**Guild ID:** {interaction.guild_id or 'N/A'}\n" - f"**Time:** {discord.utils.utcnow().strftime('%H:%M:%S UTC')}", - inline=False + f"**Sync Type:** {sync_type.title()}\n" + f"**Guild ID:** {interaction.guild_id or 'N/A'}\n" + f"**Time:** {discord.utils.utcnow().strftime('%H:%M:%S UTC')}", + inline=False, ) except Exception as e: @@ -296,7 +304,7 @@ class AdminCommands(commands.Cog): embed = EmbedTemplate.create_base_embed( title="❌ Sync Failed", description=f"Failed to sync commands: {str(e)}", - color=EmbedColors.ERROR + color=EmbedColors.ERROR, ) await interaction.followup.send(embed=embed) @@ -310,7 +318,9 @@ class AdminCommands(commands.Cog): Use this when slash commands aren't synced yet and you can't access /admin-sync. Syncs to the current guild only (for multi-bot scenarios). """ - self.logger.info(f"Prefix command !admin-sync invoked by {ctx.author} in {ctx.guild}") + self.logger.info( + f"Prefix command !admin-sync invoked by {ctx.author} in {ctx.guild}" + ) try: # Sync to current guild only (not globally) for multi-bot scenarios @@ -319,25 +329,29 @@ class AdminCommands(commands.Cog): embed = EmbedTemplate.create_base_embed( title="✅ Commands Synced Successfully", description=f"Synced {len(synced_commands)} application commands", - color=EmbedColors.SUCCESS + color=EmbedColors.SUCCESS, ) # Show some of the synced commands command_names = [cmd.name for cmd in synced_commands[:10]] embed.add_field( name="Synced Commands", - value="\n".join([f"• /{name}" for name in command_names]) + - (f"\n... and {len(synced_commands) - 10} more" if len(synced_commands) > 10 else ""), - inline=False + value="\n".join([f"• /{name}" for name in command_names]) + + ( + f"\n... and {len(synced_commands) - 10} more" + if len(synced_commands) > 10 + else "" + ), + inline=False, ) embed.add_field( name="Sync Details", value=f"**Total Commands:** {len(synced_commands)}\n" - f"**Sync Type:** Local Guild\n" - f"**Guild ID:** {ctx.guild.id}\n" - f"**Time:** {discord.utils.utcnow().strftime('%H:%M:%S UTC')}", - inline=False + f"**Sync Type:** Local Guild\n" + f"**Guild ID:** {ctx.guild.id}\n" + f"**Time:** {discord.utils.utcnow().strftime('%H:%M:%S UTC')}", + inline=False, ) embed.set_footer(text="💡 Use /admin-sync local:True for guild-only sync") @@ -347,57 +361,60 @@ class AdminCommands(commands.Cog): embed = EmbedTemplate.create_base_embed( title="❌ Sync Failed", description=f"Failed to sync commands: {str(e)}", - color=EmbedColors.ERROR + color=EmbedColors.ERROR, ) await ctx.send(embed=embed) - + @app_commands.command( - name="admin-clear", - description="Clear messages from the current channel" - ) - @app_commands.describe( - count="Number of messages to delete (1-100)" + name="admin-clear", description="Clear messages from the current channel" ) + @app_commands.describe(count="Number of messages to delete (1-100)") @league_admin_only() @logged_command("/admin-clear") async def admin_clear(self, interaction: discord.Interaction, count: int): """Clear a specified number of messages from the channel.""" if count < 1 or count > 100: await interaction.response.send_message( - "❌ Count must be between 1 and 100.", - ephemeral=True + "❌ Count must be between 1 and 100.", ephemeral=True ) return - + await interaction.response.defer() # Verify channel type supports purge - if not isinstance(interaction.channel, (discord.TextChannel, discord.Thread, discord.VoiceChannel, discord.StageChannel)): + if not isinstance( + interaction.channel, + ( + discord.TextChannel, + discord.Thread, + discord.VoiceChannel, + discord.StageChannel, + ), + ): await interaction.followup.send( - "❌ Cannot purge messages in this channel type.", - ephemeral=True + "❌ Cannot purge messages in this channel type.", ephemeral=True ) return try: deleted = await interaction.channel.purge(limit=count) - + embed = EmbedTemplate.create_base_embed( title="🗑️ Messages Cleared", description=f"Successfully deleted {len(deleted)} messages", - color=EmbedColors.SUCCESS + color=EmbedColors.SUCCESS, ) - + embed.add_field( name="Clear Details", value=f"**Messages Deleted:** {len(deleted)}\n" - f"**Channel:** {interaction.channel.mention}\n" - f"**Requested:** {count} messages\n" - f"**Time:** {discord.utils.utcnow().strftime('%H:%M:%S UTC')}", - inline=False + f"**Channel:** {interaction.channel.mention}\n" + f"**Requested:** {count} messages\n" + f"**Time:** {discord.utils.utcnow().strftime('%H:%M:%S UTC')}", + inline=False, ) - + # Send confirmation and auto-delete after 5 seconds message = await interaction.followup.send(embed=embed) await asyncio.sleep(5) @@ -406,46 +423,43 @@ class AdminCommands(commands.Cog): await message.delete() except discord.NotFound: pass # Message already deleted - + except discord.Forbidden: await interaction.followup.send( - "❌ Missing permissions to delete messages.", - ephemeral=True + "❌ Missing permissions to delete messages.", ephemeral=True ) except Exception as e: await interaction.followup.send( - f"❌ Failed to clear messages: {str(e)}", - ephemeral=True + f"❌ Failed to clear messages: {str(e)}", ephemeral=True ) - + @app_commands.command( - name="admin-announce", - description="Send an announcement to the current channel" + name="admin-announce", description="Send an announcement to the current channel" ) @app_commands.describe( message="Announcement message to send", - mention_everyone="Whether to mention @everyone (default: False)" + mention_everyone="Whether to mention @everyone (default: False)", ) @league_admin_only() @logged_command("/admin-announce") async def admin_announce( - self, - interaction: discord.Interaction, + self, + interaction: discord.Interaction, message: str, - mention_everyone: bool = False + mention_everyone: bool = False, ): """Send an official announcement to the channel.""" await interaction.response.defer() - + embed = EmbedTemplate.create_base_embed( title="📢 League Announcement", description=message, - color=EmbedColors.PRIMARY + color=EmbedColors.PRIMARY, ) - + embed.set_footer( text=f"Announcement by {interaction.user.display_name}", - icon_url=interaction.user.display_avatar.url + icon_url=interaction.user.display_avatar.url, ) # Send with or without mention based on flag @@ -453,72 +467,72 @@ class AdminCommands(commands.Cog): await interaction.followup.send(content="@everyone", embed=embed) else: await interaction.followup.send(embed=embed) - + # Log the announcement self.logger.info( f"Announcement sent by {interaction.user} in {interaction.channel}: {message[:100]}..." ) - + @app_commands.command( - name="admin-maintenance", - description="Toggle maintenance mode for the bot" + name="admin-maintenance", description="Toggle maintenance mode for the bot" ) - @app_commands.describe( - mode="Turn maintenance mode on or off" + @app_commands.describe(mode="Turn maintenance mode on or off") + @app_commands.choices( + mode=[ + app_commands.Choice(name="On", value="on"), + app_commands.Choice(name="Off", value="off"), + ] ) - @app_commands.choices(mode=[ - app_commands.Choice(name="On", value="on"), - app_commands.Choice(name="Off", value="off") - ]) @league_admin_only() @logged_command("/admin-maintenance") async def admin_maintenance(self, interaction: discord.Interaction, mode: str): """Toggle maintenance mode to prevent normal command usage.""" await interaction.response.defer() - - # This would typically set a global flag or database value - # For now, we'll just show the interface - + is_enabling = mode.lower() == "on" + self.bot.maintenance_mode = is_enabling + self.logger.info( + f"Maintenance mode {'enabled' if is_enabling else 'disabled'} by {interaction.user} (id={interaction.user.id})" + ) status_text = "enabled" if is_enabling else "disabled" color = EmbedColors.WARNING if is_enabling else EmbedColors.SUCCESS - + embed = EmbedTemplate.create_base_embed( title=f"🔧 Maintenance Mode {status_text.title()}", description=f"Maintenance mode has been **{status_text}**", - color=color + color=color, ) - + if is_enabling: embed.add_field( name="Maintenance Active", value="• Normal commands are disabled\n" - "• Only admin commands are available\n" - "• Users will see maintenance message", - inline=False + "• Only admin commands are available\n" + "• Users will see maintenance message", + inline=False, ) else: embed.add_field( name="Maintenance Ended", value="• All commands are now available\n" - "• Normal bot operation resumed\n" - "• Users can access all features", - inline=False + "• Normal bot operation resumed\n" + "• Users can access all features", + inline=False, ) - + embed.add_field( name="Status Change", value=f"**Changed by:** {interaction.user.mention}\n" - f"**Time:** {discord.utils.utcnow().strftime('%H:%M:%S UTC')}\n" - f"**Mode:** {status_text.title()}", - inline=False + f"**Time:** {discord.utils.utcnow().strftime('%H:%M:%S UTC')}\n" + f"**Mode:** {status_text.title()}", + inline=False, ) - + await interaction.followup.send(embed=embed) @app_commands.command( name="admin-clear-scorecards", - description="Manually clear the live scorebug channel and hide it from members" + description="Manually clear the live scorebug channel and hide it from members", ) @league_admin_only() @logged_command("/admin-clear-scorecards") @@ -539,17 +553,17 @@ class AdminCommands(commands.Cog): if not guild: await interaction.followup.send( - "❌ Could not find guild. Check configuration.", - ephemeral=True + "❌ Could not find guild. Check configuration.", ephemeral=True ) return - live_scores_channel = discord.utils.get(guild.text_channels, name='live-sba-scores') + live_scores_channel = discord.utils.get( + guild.text_channels, name="live-sba-scores" + ) if not live_scores_channel: await interaction.followup.send( - "❌ Could not find #live-sba-scores channel.", - ephemeral=True + "❌ Could not find #live-sba-scores channel.", ephemeral=True ) return @@ -569,7 +583,7 @@ class AdminCommands(commands.Cog): visibility_success = await set_channel_visibility( live_scores_channel, visible=False, - reason="Admin manual clear via /admin-clear-scorecards" + reason="Admin manual clear via /admin-clear-scorecards", ) if visibility_success: @@ -580,25 +594,25 @@ class AdminCommands(commands.Cog): # Create success embed embed = EmbedTemplate.success( title="Live Scorebug Channel Cleared", - description="Successfully cleared the #live-sba-scores channel" + description="Successfully cleared the #live-sba-scores channel", ) embed.add_field( name="Clear Details", value=f"**Channel:** {live_scores_channel.mention}\n" - f"**Messages Deleted:** {deleted_count}\n" - f"**Visibility:** {visibility_status}\n" - f"**Time:** {discord.utils.utcnow().strftime('%H:%M:%S UTC')}", - inline=False + f"**Messages Deleted:** {deleted_count}\n" + f"**Visibility:** {visibility_status}\n" + f"**Time:** {discord.utils.utcnow().strftime('%H:%M:%S UTC')}", + inline=False, ) embed.add_field( name="Next Steps", value="• Channel is now hidden from @everyone\n" - "• Bot retains full access to the channel\n" - "• Channel will auto-show when games are published\n" - "• Live scorebug tracker runs every 3 minutes", - inline=False + "• Bot retains full access to the channel\n" + "• Channel will auto-show when games are published\n" + "• Live scorebug tracker runs every 3 minutes", + inline=False, ) await interaction.followup.send(embed=embed) @@ -606,18 +620,17 @@ class AdminCommands(commands.Cog): except discord.Forbidden: await interaction.followup.send( "❌ Missing permissions to clear messages or modify channel permissions.", - ephemeral=True + ephemeral=True, ) except Exception as e: self.logger.error(f"Error clearing scorecards: {e}", exc_info=True) await interaction.followup.send( - f"❌ Failed to clear channel: {str(e)}", - ephemeral=True + f"❌ Failed to clear channel: {str(e)}", ephemeral=True ) @app_commands.command( name="admin-process-transactions", - description="[ADMIN] Manually process all transactions for the current week (or specified week)" + description="[ADMIN] Manually process all transactions for the current week (or specified week)", ) @app_commands.describe( week="Week number to process (optional, defaults to current week)" @@ -625,9 +638,7 @@ class AdminCommands(commands.Cog): @league_admin_only() @logged_command("/admin-process-transactions") async def admin_process_transactions( - self, - interaction: discord.Interaction, - week: int | None = None + self, interaction: discord.Interaction, week: int | None = None ): """ Manually process all transactions for the current week. @@ -649,7 +660,7 @@ class AdminCommands(commands.Cog): if not current: await interaction.followup.send( "❌ Could not get current league state from the API.", - ephemeral=True + ephemeral=True, ) return @@ -659,31 +670,33 @@ class AdminCommands(commands.Cog): self.logger.info( f"Processing transactions for week {target_week}, season {target_season}", - requested_by=str(interaction.user) + requested_by=str(interaction.user), ) # Get all non-frozen, non-cancelled transactions for the target week using service layer - transactions = await transaction_service.get_all_items(params=[ - ('season', str(target_season)), - ('week_start', str(target_week)), - ('week_end', str(target_week)), - ('frozen', 'false'), - ('cancelled', 'false') - ]) + transactions = await transaction_service.get_all_items( + params=[ + ("season", str(target_season)), + ("week_start", str(target_week)), + ("week_end", str(target_week)), + ("frozen", "false"), + ("cancelled", "false"), + ] + ) if not transactions: embed = EmbedTemplate.info( title="No Transactions to Process", - description=f"No non-frozen, non-cancelled transactions found for Week {target_week}" + description=f"No non-frozen, non-cancelled transactions found for Week {target_week}", ) embed.add_field( name="Search Criteria", value=f"**Season:** {target_season}\n" - f"**Week:** {target_week}\n" - f"**Frozen:** No\n" - f"**Cancelled:** No", - inline=False + f"**Week:** {target_week}\n" + f"**Frozen:** No\n" + f"**Cancelled:** No", + inline=False, ) await interaction.followup.send(embed=embed) @@ -692,7 +705,9 @@ class AdminCommands(commands.Cog): # Count total transactions total_count = len(transactions) - self.logger.info(f"Found {total_count} transactions to process for week {target_week}") + self.logger.info( + f"Found {total_count} transactions to process for week {target_week}" + ) # Process each transaction success_count = 0 @@ -702,12 +717,10 @@ class AdminCommands(commands.Cog): # Create initial status embed processing_embed = EmbedTemplate.loading( title="Processing Transactions", - description=f"Processing {total_count} transactions for Week {target_week}..." + description=f"Processing {total_count} transactions for Week {target_week}...", ) processing_embed.add_field( - name="Progress", - value="Starting...", - inline=False + name="Progress", value="Starting...", inline=False ) status_message = await interaction.followup.send(embed=processing_embed) @@ -718,7 +731,7 @@ class AdminCommands(commands.Cog): await self._execute_player_update( player_id=transaction.player.id, new_team_id=transaction.newteam.id, - player_name=transaction.player.name + player_name=transaction.player.name, ) success_count += 1 @@ -729,11 +742,11 @@ class AdminCommands(commands.Cog): 0, name="Progress", value=f"Processed {idx}/{total_count} transactions\n" - f"✅ Successful: {success_count}\n" - f"❌ Failed: {failure_count}", - inline=False + f"✅ Successful: {success_count}\n" + f"❌ Failed: {failure_count}", + inline=False, ) - await status_message.edit(embed=processing_embed) # type: ignore + await status_message.edit(embed=processing_embed) # type: ignore # Rate limiting: 100ms delay between requests await asyncio.sleep(0.1) @@ -741,45 +754,45 @@ class AdminCommands(commands.Cog): except Exception as e: failure_count += 1 error_info = { - 'player': transaction.player.name, - 'player_id': transaction.player.id, - 'new_team': transaction.newteam.abbrev, - 'error': str(e) + "player": transaction.player.name, + "player_id": transaction.player.id, + "new_team": transaction.newteam.abbrev, + "error": str(e), } errors.append(error_info) self.logger.error( f"Failed to execute transaction for {error_info['player']}", - player_id=error_info['player_id'], - new_team=error_info['new_team'], - error=e + player_id=error_info["player_id"], + new_team=error_info["new_team"], + error=e, ) # Create completion embed if failure_count == 0: completion_embed = EmbedTemplate.success( title="Transactions Processed Successfully", - description=f"All {total_count} transactions for Week {target_week} have been processed." + description=f"All {total_count} transactions for Week {target_week} have been processed.", ) elif success_count == 0: completion_embed = EmbedTemplate.error( title="Transaction Processing Failed", - description=f"Failed to process all {total_count} transactions for Week {target_week}." + description=f"Failed to process all {total_count} transactions for Week {target_week}.", ) else: completion_embed = EmbedTemplate.warning( title="Transactions Partially Processed", - description=f"Some transactions for Week {target_week} failed to process." + description=f"Some transactions for Week {target_week} failed to process.", ) completion_embed.add_field( name="Processing Summary", value=f"**Total Transactions:** {total_count}\n" - f"**✅ Successful:** {success_count}\n" - f"**❌ Failed:** {failure_count}\n" - f"**Week:** {target_week}\n" - f"**Season:** {target_season}", - inline=False + f"**✅ Successful:** {success_count}\n" + f"**❌ Failed:** {failure_count}\n" + f"**Week:** {target_week}\n" + f"**Season:** {target_season}", + inline=False, ) # Add error details if there were failures @@ -792,17 +805,15 @@ class AdminCommands(commands.Cog): error_text += f"\n... and {len(errors) - 5} more errors" completion_embed.add_field( - name="Errors", - value=error_text, - inline=False + name="Errors", value=error_text, inline=False ) completion_embed.add_field( name="Next Steps", value="• Verify transactions in the database\n" - "• Check #transaction-log channel for posted moves\n" - "• Review any errors and retry if necessary", - inline=False + "• Check #transaction-log channel for posted moves\n" + "• Review any errors and retry if necessary", + inline=False, ) completion_embed.set_footer( @@ -810,13 +821,13 @@ class AdminCommands(commands.Cog): ) # Update the status message with final results - await status_message.edit(embed=completion_embed) # type: ignore + await status_message.edit(embed=completion_embed) # type: ignore self.logger.info( f"Transaction processing complete for week {target_week}", success=success_count, failures=failure_count, - total=total_count + total=total_count, ) except Exception as e: @@ -824,16 +835,13 @@ class AdminCommands(commands.Cog): embed = EmbedTemplate.error( title="Transaction Processing Failed", - description=f"An error occurred while processing transactions: {str(e)}" + description=f"An error occurred while processing transactions: {str(e)}", ) await interaction.followup.send(embed=embed, ephemeral=True) async def _execute_player_update( - self, - player_id: int, - new_team_id: int, - player_name: str + self, player_id: int, new_team_id: int, player_name: str ) -> bool: """ Execute a player roster update via service layer. @@ -854,13 +862,12 @@ class AdminCommands(commands.Cog): f"Updating player roster", player_id=player_id, player_name=player_name, - new_team_id=new_team_id + new_team_id=new_team_id, ) # Execute player team update via service layer updated_player = await player_service.update_player_team( - player_id=player_id, - new_team_id=new_team_id + player_id=player_id, new_team_id=new_team_id ) # Verify update was successful @@ -869,7 +876,7 @@ class AdminCommands(commands.Cog): f"Successfully updated player roster", player_id=player_id, player_name=player_name, - new_team_id=new_team_id + new_team_id=new_team_id, ) return True else: @@ -877,7 +884,7 @@ class AdminCommands(commands.Cog): f"Player update returned no response", player_id=player_id, player_name=player_name, - new_team_id=new_team_id + new_team_id=new_team_id, ) return False @@ -888,11 +895,11 @@ class AdminCommands(commands.Cog): player_name=player_name, new_team_id=new_team_id, error=e, - exc_info=True + exc_info=True, ) raise async def setup(bot: commands.Bot): """Load the admin commands cog.""" - await bot.add_cog(AdminCommands(bot)) \ No newline at end of file + await bot.add_cog(AdminCommands(bot))