""" Commands for managing the tables/categories/scores for a Guild. """ from typing import Optional import nextcord from nextcord.ext import commands from sqlalchemy import distinct, select from sqlalchemy.orm import Session # pylint: disable=import-error from utilities.sql_data_classes import Category, CategoryGroup, Guild, Score, engine from utilities.utils import ( autofill_categories, autofill_groups, autofill_users_in_group_category, autofill_users_in_guild, send_highscore_message, ) from .views.update_username_verify import UpdateUsernameVerifyView class DataManagementCog(commands.Cog, name="Table Management"): """ Class holds commands for managing the data storage of the highscores. """ def __init__(self, bot: commands.Bot): """ Default constructor for DataManagement class. Sets internal bot variable. """ self._bot: commands.Bot = bot @commands.Cog.listener() async def on_ready(self): """ Called when cog is loaded. Adds views required for DataManagement commands. """ self._bot.add_view(UpdateUsernameVerifyView()) @nextcord.slash_command(default_member_permissions=nextcord.Permissions(administrator=True)) async def delete(self, interaction: nextcord.Interaction): """ Parent slash command for deleting things. Never called. """ @delete.subcommand() async def user(self, interaction: nextcord.Interaction): """ Subcommand parent for deleting user information. Never called. """ @user.subcommand( description="Removes all of a user's scores. Warning: data is destroyed, use with case.", ) async def from_all( self, interaction: nextcord.Interaction, user_name: str = nextcord.SlashOption( description="User to remove scores for. Note: This is not always a user's discord name.", required=True ), ): """ Searches all of guild's scores data for any score from the user. Removes. Updates table message is user was removed from that table. Args: interaction (nextcord.Interaction): Interaction that invoked the command. user_name (str): User to remove from highscores """ with Session(engine) as session: stmt = select(distinct(Score.CategoryId)).where(Score.UserName == user_name) category_ids: "list[int]" = list(session.scalars(stmt)) stmt = select(Score).where(Score.UserName == user_name) for score in session.scalars(stmt): session.delete(score) session.commit() group_ids = select(Category.GroupId).where(Category.CategoryId.in_(category_ids)) stmt = select(CategoryGroup).where(CategoryGroup.GroupId.in_(group_ids)) for group in session.scalars(stmt): await send_highscore_message(session, interaction.guild.get_channel(group.ChannelId), group) await interaction.send(f'User "{user_name}" was removed from all tables.', ephemeral=True) @user.subcommand(description="Deletes a user's scores from a category.") async def from_category( self, interaction: nextcord.Interaction, group_id: str = nextcord.SlashOption(name="group", description="Group to delete user from.", required=True), category_id: str = nextcord.SlashOption( name="category", description="Category to delete user from.", required=True ), user_name: str = nextcord.SlashOption(description="User to delete.", required=True), ): """ Removes a user from a category. Uses autofill functions to fill out group_id, category_id, and user_name. Args: interaction (nextcord.Interaction): Interaction that invoked the command. group_id (str): String representation of Group id to remove user from. Is SlashOption. category_id (str): String representation of Category id to remove user from. Is SlashOption. user_name (str): The user to remove. Is SlashOption. """ with Session(engine) as session: stmt = select(CategoryGroup).where(CategoryGroup.GroupId == int(group_id)) group = session.scalar(stmt) stmt = select(Category.CategoryName).where(Category.CategoryId == int(category_id)) category_name = session.scalar(stmt) stmt = select(Score).where(Score.CategoryId == int(category_id)).where(Score.UserName == user_name) # there should only be one because category id and username are a composite primary key in score session.delete(session.scalar(stmt)) session.commit() await send_highscore_message(session, interaction.guild.get_channel(group.ChannelId), group) await interaction.response.send_message( f'User "{user_name}" removed from "{category_name}".', ephemeral=True ) @user.subcommand(description="Deletes a user's scores from entire group.") async def from_group( self, interaction: nextcord.Interaction, group_id: str = nextcord.SlashOption(name="group", description="Group to delete user from.", required=True), user_name: str = nextcord.SlashOption(description="User to delete.", required=True), ): """ Removes user from a goup. Uses autofill functions to fill out group_id and user_name. Args: interaction (nextcord.Interaction): Interaction that invoked the command. group_id (str): String representation of Group id to remove user from. Is SlashOption. user_name (str): The user to remove. Is SlashOption. """ with Session(engine) as session: stmt = select(CategoryGroup).where(CategoryGroup.GroupId == int(group_id)) group = session.scalar(stmt) category_ids = select(Category.CategoryId).where(Category.GroupId == int(group_id)) stmt = select(Score).where(Score.CategoryId.in_(category_ids)).where(Score.UserName == user_name) for score in session.scalars(stmt): session.delete(score) session.commit() await send_highscore_message(session, interaction.guild.get_channel(group.ChannelId), group) await interaction.response.send_message( f'User "{user_name}" removed from all categories in {group.GroupName}.', ephemeral=True ) @from_category.on_autocomplete("group_id") @from_group.on_autocomplete("group_id") async def autocomplete_group(self, interaction: nextcord.Interaction, group_id: str): """ Autocomplete function for table names. Args: interaction (nextcord.Interaction): Interaction that invoked the command. group_id (str): String to be autocompleted with Group name. """ await interaction.response.send_autocomplete(autofill_groups(interaction.guild_id, group_id)) @from_category.on_autocomplete("category_id") async def autocomplete_category(self, interaction: nextcord.Interaction, category_id: str, group_id: str): """ Autocomplete function for category names. Args: interaction (nextcord.Interaction): Interaction that invoked the command. category_id (str): String to be autocompleted with Category name. group_id (str): Group to find Category from. """ await interaction.response.send_autocomplete( autofill_categories(interaction.guild_id, int(group_id), category_id) ) @from_category.on_autocomplete("user_name") @from_group.on_autocomplete("user_name") async def autocomplete_user( self, interaction: nextcord.Interaction, user_name: str, group_id: str, category_id: Optional[str] = None ): """ Autocomplete functions for user names. Args: interaction (nextcord.Interaction): Interaction that invoked the command. user_name (str): String to be autocompleted with User name. group_id (str): String representation of Group id to find category from. category_id (Optional[str]): String representation of Category id to find User from, can be None. """ await interaction.response.send_autocomplete( autofill_users_in_group_category( interaction.guild_id, int(group_id), category_id if category_id else 0, user_name ) ) @nextcord.slash_command(description="Request a username update. Admins must approve.") async def update_username( self, interaction: nextcord.Interaction, old_username: str = nextcord.SlashOption(description="Username currently stored in highscores.", required=True), new_username: str = nextcord.SlashOption(description="Username to update to.", required=True), ): """ Creates a request to send to admins for changing a username. Uses UpdateUsernameVerifyView to do the work. Args: interaction (nextcord.Interaction): Interaction that invoked the command old_username (str): Username to replace. new_username (str): Username to replace old with. """ with Session(engine) as session: stmt = select(Guild.VerificationChannel).where(Guild.GuildId == interaction.guild_id) verify_channel = interaction.guild.get_channel(session.scalar(stmt)) embed: nextcord.Embed = nextcord.Embed(title="Username change request.") embed.add_field(name="Old Username", value=old_username) embed.add_field(name="New Username", value=new_username) await verify_channel.send( content=f"Username change request from <@{interaction.user.id}>.", embed=embed, view=UpdateUsernameVerifyView(), ) await interaction.send("Request sent to admins for verification.", ephemeral=True) @update_username.on_autocomplete("old_username") @from_all.on_autocomplete("user_name") async def username_autofill(self, interaction: nextcord.Interaction, username: str): """ Sends response from autofill_users_in_guild back to update_username for autocompletion of old_username. Args: interaction (nextcord.Interaction): Interaction with command. old_username (str): String to run autocompletion against. """ await interaction.response.send_autocomplete(autofill_users_in_guild(interaction.guild_id, username))