231 lines
9.5 KiB
Python
231 lines
9.5 KiB
Python
"""
|
|
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
|
|
|
|
# pylint: disable=import-error
|
|
from highscores import Category, CategoryGroup, Guild, Score, session
|
|
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,
|
|
username: 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.
|
|
username (str): User to remove from highscores
|
|
"""
|
|
|
|
stmt = select(distinct(Score.CategoryId)).where(Score.UserName == username)
|
|
category_ids: "list[int]" = list(session.scalars(stmt))
|
|
|
|
stmt = select(Score).where(Score.UserName == username)
|
|
for score in session.scalars(stmt):
|
|
session.delete(score)
|
|
session.commit()
|
|
|
|
group_ids = select(Category.GroupId).where(Category.CategoryId.in_(category_ids)).subquery()
|
|
|
|
stmt = select(CategoryGroup).where(CategoryGroup.GroupId.in_(group_ids))
|
|
|
|
for group in session.scalars(stmt):
|
|
await send_highscore_message(interaction.guild.get_channel(group.ChannelId), group)
|
|
|
|
await interaction.send(f'User "{username}" was removed from all tables.')
|
|
|
|
@user.subcommand(description="Deletes a user's scores from a category.")
|
|
async def from_category(
|
|
self,
|
|
interaction: nextcord.Interaction,
|
|
group_id: int = nextcord.SlashOption(description="Group to delete user from.", required=True),
|
|
category_id: int = nextcord.SlashOption(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_name, and user_name.
|
|
|
|
Args:
|
|
interaction (nextcord.Interaction): Interaction that invoked the command.
|
|
group_id (int): Group to remove user from. Is SlashOption.
|
|
category_id (int): Category to remove user from. Is SlashOption.
|
|
user_name (str): The user to remove. Is SlashOption.
|
|
"""
|
|
|
|
stmt = select(CategoryGroup).where(CategoryGroup.GroupId == group_id)
|
|
group = session.scalar(stmt)
|
|
stmt = select(Score).where(Score.CategoryId == category_id)
|
|
|
|
for score in session.scalars(stmt):
|
|
session.delete(score)
|
|
session.commit()
|
|
|
|
await send_highscore_message(interaction.guild.get_channel(group.ChannelId), group)
|
|
await interaction.response.send_message(f'User "{user_name}" removed from "{category_id}".', ephemeral=True)
|
|
|
|
@user.subcommand(description="Deletes a user's scores from entire group.")
|
|
async def from_group(
|
|
self,
|
|
interaction: nextcord.Interaction,
|
|
group_id: int = nextcord.SlashOption(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 (int): Group to remove user from. Is SlashOption.
|
|
user_name (str): The user to remove. Is SlashOption.
|
|
"""
|
|
|
|
stmt = select(CategoryGroup).where(CategoryGroup.GroupId == group_id)
|
|
group = session.scalar(stmt)
|
|
category_ids = select(Category.Id).where(Category.GroupId == group_id).subquery()
|
|
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(interaction.guild.get_channel(group.ChannelId), group)
|
|
await interaction.response.send_message(f'User "{user_name}" removed from all categories.', ephemeral=True)
|
|
|
|
@from_category.on_autocomplete("table_name")
|
|
@from_group.on_autocomplete("table_name")
|
|
async def autocomplete_table(self, interaction: nextcord.Interaction, table_name: str):
|
|
"""
|
|
Autocomplete function for table names.
|
|
|
|
Args:
|
|
interaction (nextcord.Interaction): Interaction that invoked the command.
|
|
table_name (str): String to be autocompleted with Table name.
|
|
"""
|
|
await interaction.response.send_autocomplete(autofill_groups(interaction.guild_id, table_name))
|
|
|
|
@from_category.on_autocomplete("category_name")
|
|
async def autocomplete_category(self, interaction: nextcord.Interaction, category_name: str, table_name: str):
|
|
"""
|
|
Autocomplete function for category names.
|
|
|
|
Args:
|
|
interaction (nextcord.Interaction): Interaction that invoked the command.
|
|
category_name (str): String to be autocompleted with Category name.
|
|
table_name (str): Table to find Category from.
|
|
"""
|
|
await interaction.response.send_autocomplete(
|
|
autofill_categories(interaction.guild_id, table_name, category_name)
|
|
)
|
|
|
|
@from_category.on_autocomplete("user_name")
|
|
@from_group.on_autocomplete("user_name")
|
|
async def autocomplete_user(
|
|
self, interaction: nextcord.Interaction, user_name: str, table_name: str, category_name: Optional[str]
|
|
):
|
|
"""
|
|
Autocomplete functions for user names.
|
|
|
|
Args:
|
|
interaction (nextcord.Interaction): Interaction that invoked the command.
|
|
user_name (str): String to be autocompleted with User name.
|
|
table_name (str): Table to find category from.
|
|
category_name (Optional[str]): Category to find User from, can be None.
|
|
"""
|
|
await interaction.response.send_autocomplete(
|
|
autofill_users_in_group_category(interaction.guild_id, table_name, category_name, 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.
|
|
"""
|
|
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("username")
|
|
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))
|