Compare commits

..
18 Commits
Author SHA1 Message Date
imcompiling c25d354fc0 Correct some parameter usage, handle errors if searches return nothing. 2025-03-14 16:09:52 -04:00
imcompiling e5d90a71d9 Correct parameter usage. 2025-03-14 16:02:18 -04:00
imcompiling 74ce22dd26 Was saving the message id to the wrong spot. 2025-03-13 00:39:47 -04:00
imcompiling 170ce1cbfb Use correct message id reference. 2025-03-12 21:30:56 -04:00
imcompiling e1b761e848 Use correct variable for guild name. 2025-03-12 21:29:01 -04:00
imcompiling 68592a7df8 Typo on col usage. 2025-03-12 21:27:29 -04:00
imcompiling b28938f66a Had col around the wrong variables. 2025-03-12 21:27:12 -04:00
imcompiling 3abbff13d5 Use walrus operator when assigning and checking for None. 2025-03-12 09:50:58 -04:00
imcompiling 0702198f0a Fix lint smells and update model usage in view. 2025-03-04 13:24:08 -05:00
imcompiling 43df7aeda7 Fix lint smells and update model usage in update. 2025-03-04 13:20:14 -05:00
imcompiling c8becb8203 Fix lint smells and update model usage in submission. 2025-03-04 13:15:20 -05:00
imcompiling 44d813504a Fix lint smells and update model usage in refresh and set. 2025-03-04 12:47:44 -05:00
imcompiling 3fc8910c9a Fix lint smell. 2025-03-04 11:19:06 -05:00
imcompiling 563b6ab888 Fix lint smells. Update model usage. 2025-03-04 11:18:40 -05:00
imcompiling d7e44d3a22 Fix lint smells and update model usage. 2025-03-04 11:04:10 -05:00
imcompiling a99b488ae1 Message id isn't known at object creation time. 2025-03-04 11:03:45 -05:00
imcompiling 85d0077107 Fix smells and update model uses for updating username verify. 2025-03-04 09:53:09 -05:00
imcompiling 9510c9eddc Add init for views directory. Fix smells and update model usage in delete confirm. 2025-03-04 09:47:39 -05:00
17 changed files with 220 additions and 191 deletions
+2 -1
View File
@@ -5,11 +5,12 @@
"highscore",
"highscores",
"imcompiling",
"iterrows",
"nextcord",
"ondelete"
],
"python.analysis.diagnosticSeverityOverrides": {
"reportOptionalMemberAccess": false,
"reportOptionalMemberAccess": "warning",
"reportArgumentType": "information",
"reportAttributeAccessIssue": "warning"
}
+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 -1
View File
@@ -11,7 +11,7 @@ class HelpCog(commands.Cog, name="Help commands."):
"""Default constructor for HelpCog class. Does nothing."""
@nextcord.slash_command()
async def help(self, interaction: nextcord.Interaction):
async def help(self, interaction: nextcord.Interaction) -> None:
"""Parent slash command for help. Never called."""
@help.subcommand(description="Describes how to get started with Agon.")
+8 -8
View File
@@ -1,13 +1,13 @@
"""
Commands for refreshing highscore messages.
"""
"""Commands for refreshing highscore messages."""
import nextcord
from nextcord.ext import commands
from sqlalchemy import select
from sqlalchemy.orm import Session
from sqlmodel import col
from database.db import CategoryGroup, engine
from database.db import engine
from database.models import CategoryGroup
from utilities.utils import autofill_groups, send_highscore_message
@@ -43,13 +43,13 @@ class RefreshCog(commands.Cog, name="Refresh Cog"):
with Session(engine) as session:
stmt = (
select(CategoryGroup)
.where(CategoryGroup.GuildId == interaction.guild_id)
.where(CategoryGroup.GroupId == int(group_id))
.where(col(CategoryGroup.guild_id) == interaction.guild_id)
.where(col(CategoryGroup.id) == int(group_id))
)
group = session.scalar(stmt)
await send_highscore_message(session, interaction.guild, group)
await interaction.response.send_message(f"Updated group {group.GroupName}!", ephemeral=True)
await interaction.response.send_message(f"Updated group {group.name}!", ephemeral=True)
@refresh.subcommand()
async def all(self, interaction: nextcord.Interaction) -> None:
@@ -67,7 +67,7 @@ class RefreshCog(commands.Cog, name="Refresh Cog"):
interaction (nextcord.Interaction): Interaction that invoked the command.
"""
with Session(engine) as session:
stmt = select(CategoryGroup).where(CategoryGroup.GuildId == interaction.guild_id)
stmt = select(CategoryGroup).where(col(CategoryGroup.guild_id) == interaction.guild_id)
# awaiting response because posting scores takes a while
await interaction.response.defer(ephemeral=True)
+47 -44
View File
@@ -1,31 +1,25 @@
"""
Set command heirarchy.
"""
"""Set command hierarchy."""
import nextcord
from nextcord.ext import commands
from sqlalchemy import select
from sqlalchemy.orm import Session
from sqlmodel import col
from database.db import CategoryGroup, Guild, engine
from database.db import engine
from database.models import CategoryGroup, Guild
from utilities.utils import MAX_LINE_LENGTH, autofill_groups
class SetCog(commands.Cog, name="Set command."):
"""
Class holds commands for setting variables in guild's database.
"""
"""Class holds commands for setting variables in guild's database."""
def __init__(self) -> None:
"""
Default constructor for SetCog class. Does nothing.
"""
"""Default constructor for SetCog class. Does nothing."""
@nextcord.slash_command(default_member_permissions=nextcord.Permissions(administrator=True))
async def set(self, interaction: nextcord.Interaction):
"""
Parent slash command for setting settings. Never called.
"""
async def set(self, interaction: nextcord.Interaction) -> None:
"""Parent slash command for setting settings. Never called."""
@set.subcommand(description="Registers the channel for a Category Group's messages.")
async def group_output_channel(
@@ -50,17 +44,18 @@ class SetCog(commands.Cog, name="Set command."):
channel (nextcord.abc.GuildChannel, optional): Channel to register
Is a SlashOption. Limited to text channels, Defaults to None.
"""
if not channel:
channel = interaction.channel
channel_id = channel.id
channel_id = interaction.channel.id if not channel else channel.id
with Session(engine) as session:
stmt = (
select(CategoryGroup)
.where(CategoryGroup.GuildId == interaction.guild_id)
.where(CategoryGroup.GroupId == int(group_id))
.where(col(CategoryGroup.guild_id) == interaction.guild_id)
.where(col(CategoryGroup.id) == int(group_id))
)
group = session.scalar(stmt)
group.ChannelId = channel_id
if (group := session.scalar(stmt)) is None:
await interaction.response.send_message("Group or Category was not found.", ephemeral=True)
return
group.channel_id = channel_id
session.commit()
await interaction.send(f"Registered <#{channel_id}>.", ephemeral=True)
@@ -92,14 +87,17 @@ class SetCog(commands.Cog, name="Set command."):
with Session(engine) as session:
stmt = (
select(CategoryGroup)
.where(CategoryGroup.GuildId == interaction.guild_id)
.where(CategoryGroup.GroupId == int(group_id))
.where(col(CategoryGroup.guild_id) == interaction.guild_id)
.where(col(CategoryGroup.id) == int(group_id))
)
group = session.scalar(stmt)
group.PadLength = length
if (group := session.scalar(stmt)) is None:
await interaction.response.send_message("Group or Category was not found.", ephemeral=True)
return
group.pad_length = length
session.commit()
await interaction.response.send_message(
f"Category pad length set to {length} for {group.GroupName}", ephemeral=True
f"Category pad length set to {length} for {group.name}", ephemeral=True
)
@set.subcommand(
@@ -120,17 +118,17 @@ class SetCog(commands.Cog, name="Set command."):
Args:
interaction (nextcord.Interaction): Interaction that invoked the command
channel Optional[nextcord.abc.GuildChannel], optional): Channel to register
channel (nextcord.abc.GuildChannel): Channel to register
Is a SlashOption. Limited to text channels, Defaults to None.
"""
if not channel:
channel = interaction.channel
channel_id = interaction.channel.id if not channel else channel.id
with Session(engine) as session:
channel_id = channel.id
stmt = select(Guild).where(Guild.GuildId == interaction.guild_id)
guild = session.scalar(stmt)
guild.SubmissionChannel = channel_id
stmt = select(Guild).where(col(Guild.id) == interaction.guild_id)
if (guild := session.scalar(stmt)) is None:
await interaction.response.send_message("Guild was not found.", ephemeral=True)
return
guild.submission_channel = channel_id
session.commit()
await interaction.send(f"Registered <#{channel_id}>.", ephemeral=True)
@@ -150,16 +148,20 @@ class SetCog(commands.Cog, name="Set command."):
Args:
interaction (nextcord.Interaction): Interaction that invoked the command
group_id (str): String representation of Group id to set username length in. Is SlashOption.
length (int): Number of characters. Is SlashOption. Defaults to 12.
"""
with Session(engine) as session:
stmt = (
select(CategoryGroup)
.where(CategoryGroup.GuildId == interaction.guild_id)
.where(CategoryGroup.GroupId == int(group_id))
.where(col(CategoryGroup.guild_id) == interaction.guild_id)
.where(col(CategoryGroup.id) == int(group_id))
)
group = session.scalar(stmt)
group.UsernameLength = length
if (group := session.scalar(stmt)) is None:
await interaction.response.send_message("Group or Category was not found.", ephemeral=True)
return
group.username_length = length
session.commit()
await interaction.response.send_message(f"Username length set to {length}.", ephemeral=True)
@@ -182,20 +184,21 @@ class SetCog(commands.Cog, name="Set command."):
channel (Optional[nextcord.abc.GuildChannel]): Channel to register
Is a SlashOption. Limited to text channels, Defaults to None.
"""
if not channel:
channel = interaction.channel
channel_id = channel.id
channel_id = interaction.channel.id if not channel else channel.id
with Session(engine) as session:
stmt = select(Guild).where(Guild.GuildId == interaction.guild_id)
guild = session.scalar(stmt)
guild.VerificationChannel = channel_id
stmt = select(Guild).where(col(Guild.id) == interaction.guild_id)
if (guild := session.scalar(stmt)) is None:
await interaction.response.send_message("Guild was not found.", ephemeral=True)
return
guild.verification_channel = channel_id
session.commit()
await interaction.send(f"Registered <#{channel_id}>.", ephemeral=True)
@group_output_channel.on_autocomplete("group_id")
@pad_length.on_autocomplete("group_id")
@username_length.on_autocomplete("group_id")
async def group_autocomplete(self, interaction: nextcord.Interaction, group_id: str):
async def group_autocomplete(self, interaction: nextcord.Interaction, group_id: str) -> None:
"""
Sends results from autofill_groups back to group_output_channel for autocompletion of group_id.
+38 -28
View File
@@ -6,9 +6,11 @@ import nextcord
from nextcord.ext import commands
from sqlalchemy import select
from sqlalchemy.orm import Session
from sqlmodel import col
from cogs.submission.verification_view import VerificationView
from database.db import Category, CategoryGroup, Guild, Submission, engine
from database.db import engine
from database.models import Category, CategoryGroup, Guild, Submission
from main import logger
from utilities.utils import autofill_categories, autofill_groups, submit_score
@@ -19,8 +21,10 @@ class SubmissionCog(commands.Cog):
def __init__(self) -> None:
return
# TODO(imcompiling): fix too many return statements when reworking command.
# https://gitea.cheezers.org/imcompiling/agon/issues/76
@nextcord.slash_command()
async def submit( # noqa: PLR0913
async def submit( # noqa: PLR0911, PLR0913
self,
interaction: nextcord.Interaction,
group_id: str = nextcord.SlashOption(name="group", required=True, description="Group to submit score to."),
@@ -56,40 +60,46 @@ class SubmissionCog(commands.Cog):
with Session(engine) as session:
stmt = (
select(Category).where(Category.GroupId == int(group_id)).where(Category.CategoryId == int(category_id))
select(Category)
.where(col(Category.group_id) == int(group_id))
.where(col(Category.id) == int(category_id))
)
category: Category = session.scalar(stmt)
if (category := session.scalar(stmt)) is None:
await interaction.response.send_message("Category not found.", ephemeral=True)
return
# check if verification is required. if no picture, deny
if category.IsValidated and proof is None:
if category.is_validated and proof is None:
await interaction.response.send_message(
"This category required verification. Please attach a picture and re-submit.", ephemeral=True
)
return
# check with regex if score is in correct format
verified_score = await self.verify_score(interaction, score, is_time_record=category.IsTimeRecord)
# check if verify_score was successful. It replies with corrections required itself.
if verified_score is None:
if (
verified_score := await self.verify_score(interaction, score, is_time_record=category.is_time_record)
) is None:
# This response message will be filled out better once command is split up to
# know which type it expects...
await interaction.response.send_message("Format was incorrect.", ephemeral=True)
return
# Create submission verification request or submit score if verification not needed.
if not category.IsValidated:
if not category.is_validated:
await submit_score(interaction, int(group_id), int(category_id), username, verified_score)
await interaction.response.send_message("Score has been submitted!", ephemeral=True)
return
stmt = select(Guild).where(Guild.GuildId == interaction.guild_id)
verification_channel_id: int = session.scalar(stmt).VerificationChannel
stmt = select(col(Guild.verification_channel)).where(col(Guild.id) == interaction.guild_id)
channel = interaction.guild.get_channel(verification_channel_id)
if (verification_channel_id := session.scalar(stmt)) is None:
error_response = "No verification channel was registered. Contact server admins."
await interaction.response.send_message(error_response, ephemeral=True)
return
if channel is None:
error_response = (
"Registered Verification channel does not exist or was never registered. Contact server admins."
)
if (channel := interaction.guild.get_channel(verification_channel_id)) is None:
error_response = "Registered Verification channel does not exist. Contact server admins."
await interaction.response.send_message(error_response, ephemeral=True)
return
@@ -111,13 +121,13 @@ class SubmissionCog(commands.Cog):
# Save the submission data for later use.
submission: Submission = Submission(
CategoryId=category_id,
GroupId=group_id,
GuildId=interaction.guild_id,
MessageId=verification_message.id,
Score=verified_score,
UserId=interaction.user.id,
UserName=username,
category_id=int(category_id),
group_id=int(group_id),
guild_id=interaction.guild_id,
message_id=verification_message.id,
score=verified_score,
user_id=interaction.user.id,
username=username,
)
session.add(submission)
@@ -151,15 +161,15 @@ class SubmissionCog(commands.Cog):
# Get group name from database
stmt = (
select(CategoryGroup.GroupName)
.where(CategoryGroup.GuildId == interaction.guild_id)
.where(CategoryGroup.GroupId == group_id)
select(col(CategoryGroup.name))
.where(col(CategoryGroup.guild_id) == interaction.guild_id)
.where(col(CategoryGroup.id) == group_id)
)
embed.add_field(name="Group", value=session.scalar(stmt), inline=False)
# Get category name from database
stmt = (
select(Category.CategoryName).where(Category.GroupId == group_id).where(Category.CategoryId == category_id)
select(col(Category.name)).where(col(Category.group_id) == group_id).where(col(Category.id) == category_id)
)
embed.add_field(name="Category", value=session.scalar(stmt), inline=False)
embed.add_field(name="Score", value=score, inline=False)
+20 -21
View File
@@ -1,22 +1,20 @@
"""
Commands for updating information/variables held in database.
"""
"""Commands for updating information/variables held in database."""
import nextcord
from nextcord.ext import commands
from sqlalchemy import select
from sqlalchemy.orm import Session
from sqlmodel import col
from database.db import Guild, engine
from database.db import engine
from database.models import Guild
from utilities.utils import autofill_users_in_guild
from .views.update_username_verify import UpdateUsernameVerifyView
class UpdateCog(commands.Cog, name="Update commands."):
"""
Class holds commands and views needed for updating things in bot.
"""
"""Class holds commands and views needed for updating things in bot."""
def __init__(self, bot: commands.Bot):
"""
@@ -28,17 +26,13 @@ class UpdateCog(commands.Cog, name="Update commands."):
self._bot: commands.Bot = bot
@commands.Cog.listener()
async def on_ready(self):
"""
Called when cog is loaded. Adds views required for DataManagement commands.
"""
async def on_ready(self) -> None:
"""Called when cog is loaded. Adds views required for DataManagement commands."""
self._bot.add_view(UpdateUsernameVerifyView())
@nextcord.slash_command()
async def update(self, interaction: nextcord.Interaction):
"""
Parent slash command for updating things. Never called.
"""
async def update(self, interaction: nextcord.Interaction) -> None:
"""Parent slash command for updating things. Never called."""
@update.subcommand(description="Request a username update. Admins must approve.")
async def username(
@@ -46,10 +40,11 @@ class UpdateCog(commands.Cog, name="Update commands."):
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),
):
) -> None:
"""
Creates a request to send to admins for changing a username. Uses
UpdateUsernameVerifyView to do the work.
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
@@ -57,9 +52,13 @@ class UpdateCog(commands.Cog, name="Update commands."):
new_username (str): Username to replace old with.
"""
with Session(engine) as session:
stmt = select(Guild.VerificationChannel).where(Guild.GuildId == interaction.guild_id)
stmt = select(col(Guild.verification_channel)).where(col(Guild.id) == interaction.guild_id)
verify_channel = interaction.guild.get_channel(session.scalar(stmt))
if not isinstance(verify_channel, nextcord.TextChannel):
await interaction.send("Verification channel not set. Cannot process request.", ephemeral=True)
return
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)
@@ -71,12 +70,12 @@ class UpdateCog(commands.Cog, name="Update commands."):
await interaction.send("Request sent to admins for verification.", ephemeral=True)
@username.on_autocomplete("old_username")
async def all_users_autocomplete(self, interaction: nextcord.Interaction, username: str):
async def all_users_autocomplete(self, interaction: nextcord.Interaction, username: str) -> None:
"""
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.
username (str): String to run autocompletion against.
"""
await interaction.response.send_autocomplete(autofill_users_in_guild(interaction.guild_id, username))
+9 -7
View File
@@ -5,8 +5,10 @@ import pandas as pd
from nextcord.ext import commands
from sqlalchemy import 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 MAX_LINE_LENGTH, autofill_categories, autofill_groups, format_score_string
@@ -46,17 +48,17 @@ class ViewCog(commands.Cog, name="View Command"):
# get group, category, and all scores sorted correctly.
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).where(Category.CategoryId == int(category_id))
stmt = select(Category).where(col(Category.id) == int(category_id))
category = session.scalar(stmt)
d_f = pd.read_sql_query(
sql=session.query(Score).where(Score.CategoryId == category_id).statement, con=engine
sql=session.query(Score).where(col(Score.category_id) == category_id).statement, con=engine
)
scores = d_f.sort_values("Score", ascending=category.IsAscending).reset_index()
scores = d_f.sort_values("Score", ascending=category.is_ascending).reset_index()
# get the longest string
longest_str: int = scores.UserName.map(len).max()
@@ -67,13 +69,13 @@ class ViewCog(commands.Cog, name="View Command"):
row["UserName"],
row["Score"],
longest_str,
category.IsTimeRecord,
category.is_time_record,
i + 1, # type: ignore[reportOperatorIssue]
MAX_LINE_LENGTH,
)
)
message_str: str = f"**{group.GroupName} {category.CategoryName} Leaderboard!**\n```"
message_str: str = f"**{group.name} {category.name} Leaderboard!**\n```"
if len(str_list) == 0:
message_str += "No Submissions!"
else:
+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 -2
View File
@@ -26,10 +26,9 @@ class GuildListener(commands.Cog, name="Guild Listener"):
"""
with Session(engine) as session:
stmt = select(Guild).where(col(Guild.id == guild.id))
guild_data = session.scalar(stmt)
# just in case check if the guild already exists
if guild_data is None:
if (guild_data := session.scalar(stmt)) is None:
new_guild = Guild(id=guild.id, name=guild.name)
session.add(new_guild)
else:
+2 -2
View File
@@ -31,7 +31,7 @@ class GuildData(commands.Cog, name="Guild Data"):
with Session(engine) as session:
for guild in self.bot.guilds:
# update given guild name for guild id
stmt = select(Guild).where(col(Guild.id == guild.id))
stmt = select(Guild).where(col(Guild.id) == guild.id)
guild_data = session.scalar(stmt)
# just in case checking if the guild doesn't exist.
@@ -39,6 +39,6 @@ class GuildData(commands.Cog, name="Guild Data"):
new_guild = Guild(id=guild.id, name=guild.name)
session.add(new_guild)
else:
guild_data.GuildName = guild.name
guild_data.name = guild.name
session.commit()
+1 -3
View File
@@ -39,9 +39,7 @@ class VerificationView(nextcord.ui.View):
.where(col(Submission.message_id) == interaction.message.id)
)
submission = session.scalar(stmt)
if submission is None:
if (submission := session.scalar(stmt)) is None:
await interaction.response.send_message("Could not find submission data in database.", ephemeral=True)
await interaction.message.delete()
return
+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)))
+3 -3
View File
@@ -64,7 +64,7 @@ def autofill_groups(guild_id: int, group_begin: str) -> "dict[str, str]":
"""
with Session(engine) as session:
# Select all groups in guild.
stmt = select(CategoryGroup).wherecol((CategoryGroup.guild_id) == guild_id)
stmt = select(CategoryGroup).where(col(CategoryGroup.guild_id) == guild_id)
# dict has to have key values of same type for discord.
group_dict: dict[str, str] = {group.name: str(group.id) for group in session.scalars(stmt)}
@@ -347,7 +347,7 @@ async def send_highscore_message(session: Session, guild: nextcord.Guild, group:
if group.message_id is None:
# message doesn't exist.
message: nextcord.Message = await channel.send(highscore_string)
group.MessageId = message.id
group.message_id = message.id
session.commit()
else:
message: nextcord.Message = await channel.fetch_message(group.message_id)
@@ -390,7 +390,7 @@ async def submit_score(
existing_score = session.scalar(stmt)
if existing_score:
existing_score.Score = score_f
existing_score.score = score_f
else:
score_obj = Score(username=username, score=score_f, category_id=category_id)
session.add(score_obj)