Compare commits

...
5 Commits
6 changed files with 88 additions and 71 deletions
+17 -15
View File
@@ -4,8 +4,10 @@ import nextcord
from nextcord.ext import commands
from sqlalchemy import exc, select
from sqlalchemy.orm import Session
from sqlmodel import col
from database.db import Category, CategoryGroup, engine
from database.db import engine
from database.models import Category, CategoryGroup
from utilities.utils import MAX_LINE_LENGTH, autofill_groups, send_highscore_message
@@ -52,30 +54,30 @@ class CreateCog(commands.Cog, name="Create Cog"):
size (int): How many sores to show in the category message. Is SlashOption.
"""
new_category = Category(
CategoryName=category_name,
GroupId=int(group_id),
Size=size,
IsAscending=is_ascending,
IsTimeRecord=is_time_record,
IsValidated=is_validated,
name=category_name,
group_id=int(group_id),
size=size,
is_ascending=is_ascending,
is_time_record=is_time_record,
is_validated=is_validated,
)
# Open connection to engine and add new category.
with Session(engine) as session:
stmt = select(CategoryGroup).where(CategoryGroup.GroupId == int(group_id))
stmt = select(CategoryGroup).where(col(CategoryGroup.id) == int(group_id))
group = session.scalar(stmt)
try:
session.add(new_category)
session.commit()
await send_highscore_message(session, interaction.guild, group)
await interaction.response.send_message(
f'"{category_name}" has been added to "{group.GroupName}"!', ephemeral=True
f'"{category_name}" has been added to "{group.name}"!', ephemeral=True
)
except exc.IntegrityError:
# Category name already exists in group
session.rollback()
await interaction.response.send_message(
f'"{category_name}" already exists in "{group.GroupName}". Please rename.', ephemeral=True
f'"{category_name}" already exists in "{group.name}". Please rename.', ephemeral=True
)
@create.subcommand(description="Create a new Category Group. Must not exist already for Guild.")
@@ -123,11 +125,11 @@ class CreateCog(commands.Cog, name="Create Cog"):
channel_id: int = channel.id
new_group = CategoryGroup(
GroupName=group_name,
ChannelId=channel_id,
GuildId=interaction.guild_id,
PadLength=pad_length,
UsernameLength=username_length,
name=group_name,
channel_id=channel_id,
guild_id=interaction.guild_id,
pad_length=pad_length,
username_length=username_length,
)
with Session(engine) as session:
+34 -25
View File
@@ -4,8 +4,10 @@ import nextcord
from nextcord.ext import commands
from sqlalchemy import distinct, select
from sqlalchemy.orm import Session
from sqlmodel import col
from database.db import Category, CategoryGroup, Score, engine
from database.db import engine
from database.models import Category, CategoryGroup, Score
from utilities.utils import (
autofill_categories,
autofill_groups,
@@ -35,10 +37,9 @@ class DeleteCog(commands.Cog, name="Delete Cog"):
async def channel_messages(
self,
interaction: nextcord.Interaction,
channel: nextcord.abc.GuildChannel | None = nextcord.SlashOption(
channel: nextcord.TextChannel | None = nextcord.SlashOption(
description="Channel to clear of messages.",
required=False,
channel_types=[nextcord.ChannelType.text],
default=None,
),
) -> None:
@@ -53,13 +54,18 @@ class DeleteCog(commands.Cog, name="Delete Cog"):
Is a SlashOption. Limited to text channels. Defaults to None.
"""
if not channel:
channel = interaction.channel
if isinstance(interaction.channel, nextcord.TextChannel):
channel = interaction.channel
else:
# This should never happen.
await interaction.response.send_message("Channel is not a text channel somehow.", ephemeral=True)
return
with Session(engine) as session:
stmt = (
select(CategoryGroup)
.where(CategoryGroup.GuildId == interaction.guild_id)
.where(CategoryGroup.ChannelId == channel.id)
.where(col(CategoryGroup.guild_id) == interaction.guild_id)
.where(col(CategoryGroup.channel_id) == channel.id)
)
# TODO test when nothing in channel
groups = session.scalars(stmt)
@@ -67,12 +73,13 @@ class DeleteCog(commands.Cog, name="Delete Cog"):
await interaction.response.defer(ephemeral=True)
for group in groups:
try:
message: nextcord.Message = await channel.fetch_message(group.MessageId)
await message.delete()
except nextcord.NotFound:
# silently ignore that messages are not found. They got deleted already.
pass
if group.message_id is not None:
try:
message: nextcord.Message = await channel.fetch_message(group.message_id)
await message.delete()
except nextcord.NotFound:
# silently ignore that messages are not found. They got deleted already.
pass
await interaction.send("Highscore messages deleted.", ephemeral=True)
@@ -96,24 +103,24 @@ class DeleteCog(commands.Cog, name="Delete Cog"):
user_name (str): User to remove from highscores
"""
with Session(engine) as session:
stmt = select(distinct(Score.CategoryId)).where(Score.UserName == user_name)
stmt = select(distinct(col(Score.category_id))).where(col(Score.username) == user_name)
category_ids: list[int] = list(session.scalars(stmt))
stmt = select(Score).where(Score.UserName == user_name)
stmt = select(Score).where(col(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))
group_ids = select(col(Category.group_id)).where(col(Category.id).in_(category_ids))
stmt = select(CategoryGroup).where(CategoryGroup.GroupId.in_(group_ids))
stmt = select(CategoryGroup).where(col(CategoryGroup.id).in_(group_ids))
for group in session.scalars(stmt):
await send_highscore_message(session, interaction.guild, group)
await interaction.send(f'User "{user_name}" was removed from all Category Groups.', ephemeral=True)
@delete.subcommand(description="Delete a category from a group. Deletion is perminant.")
@delete.subcommand(description="Delete a category from a group. Deletion is permanent.")
async def category(
self,
interaction: nextcord.Interaction,
@@ -132,7 +139,7 @@ class DeleteCog(commands.Cog, name="Delete Cog"):
DeleteConfirmView(custom_id=create_delete_id(int(group_id), int(category_id)))
)
@delete.subcommand(description="Delete a group. Deletion is perminant.")
@delete.subcommand(description="Delete a group. Deletion is permanent.")
async def group(
self,
interaction: nextcord.Interaction,
@@ -171,13 +178,15 @@ class DeleteCog(commands.Cog, name="Delete Cog"):
user_name (str): The user to remove. Is SlashOption.
"""
with Session(engine) as session:
stmt = select(CategoryGroup).where(CategoryGroup.GroupId == int(group_id))
stmt = select(CategoryGroup).where(col(CategoryGroup.id) == int(group_id))
group = session.scalar(stmt)
stmt = select(Category.CategoryName).where(Category.CategoryId == int(category_id))
stmt = select(col(Category.name)).where(col(Category.id) == int(category_id))
category_name = session.scalar(stmt)
stmt = select(Score).where(Score.CategoryId == int(category_id)).where(Score.UserName == user_name)
stmt = (
select(Score).where(col(Score.category_id) == int(category_id)).where(col(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))
@@ -204,10 +213,10 @@ class DeleteCog(commands.Cog, name="Delete Cog"):
user_name (str): The user to remove. Is SlashOption.
"""
with Session(engine) as session:
stmt = select(CategoryGroup).where(CategoryGroup.GroupId == int(group_id))
stmt = select(CategoryGroup).where(col(CategoryGroup.id) == 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)
category_ids = select(col(Category.id)).where(col(Category.group_id) == int(group_id))
stmt = select(Score).where(col(Score.category_id).in_(category_ids)).where(col(Score.username) == user_name)
for score in session.scalars(stmt):
session.delete(score)
@@ -215,7 +224,7 @@ class DeleteCog(commands.Cog, name="Delete Cog"):
await send_highscore_message(session, interaction.guild, group)
await interaction.response.send_message(
f'User "{user_name}" removed from all categories in {group.GroupName}.', ephemeral=True
f'User "{user_name}" removed from all categories in {group.name}.', ephemeral=True
)
@from_category.on_autocomplete("group_id")
+1
View File
@@ -0,0 +1 @@
"""Package for views used in commands."""
+19 -15
View File
@@ -1,22 +1,19 @@
"""
Contains view and helper functions for confirming and processing deletes.
"""
"""Contains view and helper functions for confirming and processing deletes."""
import nextcord
from sqlalchemy import delete, select
from sqlalchemy.orm import Session
from database.db import Category, CategoryGroup, engine
from database.db import engine
from database.models import Category, CategoryGroup
from utilities.utils import send_highscore_message
class DeleteConfirmView(nextcord.ui.Modal):
"""
View modal popup for confirming deletions.
"""
"""View modal popup for confirming deletions."""
def __init__(self, custom_id: str):
super().__init__(title="Warning, deletion is perminant.", custom_id=custom_id)
super().__init__(title="Warning, deletion is permanent.", custom_id=custom_id)
self.checker = nextcord.ui.TextInput(
label='Type "delete" to confirm deletion.',
@@ -26,7 +23,12 @@ class DeleteConfirmView(nextcord.ui.Modal):
)
self.add_item(self.checker)
async def callback(self, interaction: nextcord.Interaction):
async def callback(self, interaction: nextcord.Interaction) -> None:
"""Action for confirming deletion.
Args:
interaction (nextcord.Interaction): User interaction object with the modal.
"""
if self.checker.value.lower() != "delete":
await interaction.response.send_message('Response was not "delete". Canceling delete.')
return
@@ -34,21 +36,23 @@ class DeleteConfirmView(nextcord.ui.Modal):
del_vals = parse_delete_id(self.custom_id)
with Session(engine) as session:
stmt = select(CategoryGroup).where(CategoryGroup.GroupId == del_vals["group_id"])
stmt = select(CategoryGroup).where(CategoryGroup.id == del_vals["group_id"])
group = session.scalar(stmt)
if del_vals["category_id"] is None:
# delete entire group
session.delete(group)
# delete the message related to group
channel: nextcord.TextChannel = interaction.guild.get_channel(group.ChannelId)
message: nextcord.Message = await channel.fetch_message(group.MessageId)
channel = interaction.guild.get_channel(group.channel_id)
if not isinstance(channel, nextcord.TextChannel):
return
message: nextcord.Message = await channel.fetch_message(group.message_id)
await message.delete()
else:
stmt = (
delete(Category)
.where(Category.GroupId == del_vals["group_id"])
.where(Category.CategoryId == del_vals["category_id"])
.where(Category.group_id == del_vals["group_id"])
.where(Category.id == del_vals["category_id"])
)
session.execute(stmt)
# update the message
@@ -77,7 +81,7 @@ def parse_delete_id(delete_id: str) -> dict:
Returns a dict containing the values of a delete id.
Args:
id (str): the delete id
delete_id (str): the delete id
Returns:
dict: dict containing values from a delete id.
+16 -15
View File
@@ -3,22 +3,22 @@
import nextcord
from sqlalchemy import distinct, exc, select
from sqlalchemy.orm import Session
from sqlmodel import col
from database.db import Category, CategoryGroup, Score, engine
from database.db import engine
from database.models import Category, CategoryGroup, Score
from utilities.utils import send_highscore_message
class UpdateUsernameVerifyView(nextcord.ui.View):
"""Defines view for allowing a username update."""
def __init__(self):
"""
Default constructor for UpdateUsernameVerifyView. Calls super constructor with no timeout.
"""
def __init__(self) -> None:
"""Default constructor for UpdateUsernameVerifyView. Calls super constructor with no timeout."""
super().__init__(timeout=None)
@nextcord.ui.button(label="Deny", style=nextcord.ButtonStyle.red, custom_id="deny-username-update")
async def deny_button(self, button: nextcord.ui.Button, interaction: nextcord.Interaction):
async def deny_button(self, button: nextcord.ui.Button, interaction: nextcord.Interaction) -> None: # noqa: ARG002
"""
Button for denying username update. Deletes message.
@@ -29,10 +29,11 @@ class UpdateUsernameVerifyView(nextcord.ui.View):
await interaction.message.delete()
@nextcord.ui.button(label="Approve", style=nextcord.ButtonStyle.green, custom_id="approve-username-update")
async def approve_button(self, button: nextcord.ui.Button, interaction: nextcord.Interaction):
async def approve_button(self, button: nextcord.ui.Button, interaction: nextcord.Interaction) -> None: # noqa: ARG002
"""
Button for approving username update. Tells guild data to update username, then updates
message to show it is complete.
Button for approving username update.
Tells guild data to update username, then updates message to show it is complete.
Args:
button (nextcord.ui.Button): unused
@@ -41,20 +42,20 @@ class UpdateUsernameVerifyView(nextcord.ui.View):
with Session(engine) as session:
try:
embed_content = interaction.message.embeds[0].to_dict()
old_username: str = embed_content["fields"][0]["value"]
new_username: str = embed_content["fields"][1]["value"]
old_username: str = embed_content["fields"][0]["value"] # type: ignore reportTypedDictNotRequiredAccess
new_username: str = embed_content["fields"][1]["value"] # type: ignore reportTypedDictNotRequiredAccess
stmt = select(distinct(Score.CategoryId)).where(Score.UserName == old_username)
stmt = select(distinct(col(Score.category_id))).where(col(Score.username) == old_username)
category_ids: list[int] = list(session.scalars(stmt))
stmt = select(Score).where(Score.UserName == old_username)
stmt = select(Score).where(col(Score.username) == old_username)
for score in session.scalars(stmt):
score.UserName = new_username
session.commit()
group_ids = select(Category.GroupId).where(Category.CategoryId.in_(category_ids))
group_ids = select(col(Category.group_id)).where(col(Category.id).in_(category_ids))
stmt = select(CategoryGroup).where(CategoryGroup.GroupId.in_(group_ids))
stmt = select(CategoryGroup).where(col(CategoryGroup.id).in_(group_ids))
for group in session.scalars(stmt):
await send_highscore_message(session, interaction.guild, group)
+1 -1
View File
@@ -64,7 +64,7 @@ class CategoryGroup(SQLModel, table=True):
nullable=False,
)
)
message_id: int = Field(sa_column=Column("MessageId", BIGINT(unsigned=True)))
message_id: int | None = Field(sa_column=Column("MessageId", BIGINT(unsigned=True)), default=None)
pad_length: int = Field(default=12, sa_column=Column("PadLength", SMALLINT(unsigned=True)))
username_length: int = Field(default=12, sa_column=Column("UsernameLength", SMALLINT(unsigned=True)))