Compare commits

...
3 Commits
Author SHA1 Message Date
imcompiling f5e54778d6 Update dependencies in pyproject to have specific version. 2024-10-05 12:33:45 -04:00
imcompiling 19f83ec2a5 run ruff format 2024-10-05 12:28:12 -04:00
imcompiling 7919a63e04 run ruff check --fix 2024-10-05 12:27:14 -04:00
17 changed files with 54 additions and 142 deletions
+1 -6
View File
@@ -2,8 +2,6 @@
Commands for creating items in bot. Commands for creating items in bot.
""" """
from typing import Optional
import nextcord import nextcord
from nextcord.ext import commands from nextcord.ext import commands
from sqlalchemy import exc, select from sqlalchemy import exc, select
@@ -64,7 +62,6 @@ class CreateCog(commands.Cog, name="Create Cog"):
is_validated (bool): If the scores should be validated or not. Is SlashOption. is_validated (bool): If the scores should be validated or not. Is SlashOption.
size (int): How many sores to show in the category message. Is SlashOption. size (int): How many sores to show in the category message. Is SlashOption.
""" """
new_category = Category( new_category = Category(
CategoryName=category_name, CategoryName=category_name,
GroupId=int(group_id), GroupId=int(group_id),
@@ -113,7 +110,7 @@ class CreateCog(commands.Cog, name="Create Cog"):
min_value=1, min_value=1,
max_value=30, max_value=30,
), ),
channel: Optional[nextcord.abc.GuildChannel] = nextcord.SlashOption( channel: nextcord.abc.GuildChannel | None = nextcord.SlashOption(
description="Channel to post Group in. If left empty, defaults to current channel.", description="Channel to post Group in. If left empty, defaults to current channel.",
required=False, required=False,
channel_types=[nextcord.ChannelType.text], channel_types=[nextcord.ChannelType.text],
@@ -132,7 +129,6 @@ class CreateCog(commands.Cog, name="Create Cog"):
channel (Optional[nextcord.abc.GuildChannel]): Channel to post group in. If None, defaults to current channel (Optional[nextcord.abc.GuildChannel]): Channel to post group in. If None, defaults to current
channel. Is SlashOption. channel. Is SlashOption.
""" """
# if no channel provided, use current channel. # if no channel provided, use current channel.
if channel is None: if channel is None:
channel = interaction.channel channel = interaction.channel
@@ -168,5 +164,4 @@ class CreateCog(commands.Cog, name="Create Cog"):
interaction (nextcord.Interaction): Interaction with command. interaction (nextcord.Interaction): Interaction with command.
group_id (str): String to run autocomplete compared to. group_id (str): String to run autocomplete compared to.
""" """
await interaction.response.send_autocomplete(autofill_groups(interaction.guild_id, group_id)) await interaction.response.send_autocomplete(autofill_groups(interaction.guild_id, group_id))
+3 -10
View File
@@ -2,8 +2,6 @@
Commands for deleting groups/categories/scores in database and messages in discord.. Commands for deleting groups/categories/scores in database and messages in discord..
""" """
from typing import Optional
import nextcord import nextcord
from nextcord.ext import commands from nextcord.ext import commands
from sqlalchemy import distinct, select from sqlalchemy import distinct, select
@@ -48,7 +46,7 @@ class DeleteCog(commands.Cog, name="Table Management"):
async def channel_messages( async def channel_messages(
self, self,
interaction: nextcord.Interaction, interaction: nextcord.Interaction,
channel: Optional[nextcord.abc.GuildChannel] = nextcord.SlashOption( channel: nextcord.abc.GuildChannel | None = nextcord.SlashOption(
description="Channel to clear of messages.", description="Channel to clear of messages.",
required=False, required=False,
channel_types=[nextcord.ChannelType.text], channel_types=[nextcord.ChannelType.text],
@@ -64,7 +62,6 @@ class DeleteCog(commands.Cog, name="Table Management"):
channel (Optional[nextcord.abc.GuildChannel]): Channel to clear of messages. channel (Optional[nextcord.abc.GuildChannel]): Channel to clear of messages.
Is a SlashOption. Limited to text channels. Defaults to None. Is a SlashOption. Limited to text channels. Defaults to None.
""" """
if not channel: if not channel:
channel = interaction.channel channel = interaction.channel
@@ -107,10 +104,9 @@ class DeleteCog(commands.Cog, name="Table Management"):
interaction (nextcord.Interaction): Interaction that invoked the command. interaction (nextcord.Interaction): Interaction that invoked the command.
user_name (str): User to remove from highscores user_name (str): User to remove from highscores
""" """
with Session(engine) as session: with Session(engine) as session:
stmt = select(distinct(Score.CategoryId)).where(Score.UserName == user_name) stmt = select(distinct(Score.CategoryId)).where(Score.UserName == user_name)
category_ids: "list[int]" = list(session.scalars(stmt)) category_ids: list[int] = list(session.scalars(stmt))
stmt = select(Score).where(Score.UserName == user_name) stmt = select(Score).where(Score.UserName == user_name)
for score in session.scalars(stmt): for score in session.scalars(stmt):
@@ -185,7 +181,6 @@ class DeleteCog(commands.Cog, name="Table Management"):
category_id (str): String representation of Category 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. user_name (str): The user to remove. Is SlashOption.
""" """
with Session(engine) as session: with Session(engine) as session:
stmt = select(CategoryGroup).where(CategoryGroup.GroupId == int(group_id)) stmt = select(CategoryGroup).where(CategoryGroup.GroupId == int(group_id))
group = session.scalar(stmt) group = session.scalar(stmt)
@@ -219,7 +214,6 @@ class DeleteCog(commands.Cog, name="Table Management"):
group_id (str): String representation of Group id to remove user from. Is SlashOption. group_id (str): String representation of Group id to remove user from. Is SlashOption.
user_name (str): The user to remove. Is SlashOption. user_name (str): The user to remove. Is SlashOption.
""" """
with Session(engine) as session: with Session(engine) as session:
stmt = select(CategoryGroup).where(CategoryGroup.GroupId == int(group_id)) stmt = select(CategoryGroup).where(CategoryGroup.GroupId == int(group_id))
group = session.scalar(stmt) group = session.scalar(stmt)
@@ -267,7 +261,7 @@ class DeleteCog(commands.Cog, name="Table Management"):
@from_category.on_autocomplete("user_name") @from_category.on_autocomplete("user_name")
@from_group.on_autocomplete("user_name") @from_group.on_autocomplete("user_name")
async def user_autocomplete( async def user_autocomplete(
self, interaction: nextcord.Interaction, user_name: str, group_id: str, category_id: Optional[str] = None self, interaction: nextcord.Interaction, user_name: str, group_id: str, category_id: str | None = None
): ):
""" """
Autocomplete functions for user names. Autocomplete functions for user names.
@@ -278,7 +272,6 @@ class DeleteCog(commands.Cog, name="Table Management"):
group_id (str): String representation of Group id to find category from. 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. category_id (Optional[str]): String representation of Category id to find User from, can be None.
""" """
await interaction.response.send_autocomplete( await interaction.response.send_autocomplete(
autofill_users_in_group_category( autofill_users_in_group_category(
interaction.guild_id, int(group_id), category_id if category_id else 0, user_name interaction.guild_id, int(group_id), category_id if category_id else 0, user_name
-1
View File
@@ -30,7 +30,6 @@ class HelpCog(commands.Cog, name="Help commands."):
Args: Args:
interaction (nextcord.Interaction): Interaction that invoked the command. interaction (nextcord.Interaction): Interaction that invoked the command.
""" """
# I hate this but don't want indents in the string. # I hate this but don't want indents in the string.
# pylint: disable=line-too-long # pylint: disable=line-too-long
help_str: str = """Welcome to Agon! Three simple steps are needed to get started. help_str: str = """Welcome to Agon! Three simple steps are needed to get started.
-3
View File
@@ -46,7 +46,6 @@ class RefreshCog(commands.Cog, name="Refresh Cog"):
interaction (nextcord.Interaction): Interaction that invoked the command. interaction (nextcord.Interaction): Interaction that invoked the command.
group_id (str): Str representation of Group id to refresh. Is SlashOption. group_id (str): Str representation of Group id to refresh. Is SlashOption.
""" """
with Session(engine) as session: with Session(engine) as session:
stmt = ( stmt = (
select(CategoryGroup) select(CategoryGroup)
@@ -78,7 +77,6 @@ class RefreshCog(commands.Cog, name="Refresh Cog"):
Args: Args:
interaction (nextcord.Interaction): Interaction that invoked the command. interaction (nextcord.Interaction): Interaction that invoked the command.
""" """
with Session(engine) as session: with Session(engine) as session:
stmt = select(CategoryGroup).where(CategoryGroup.GuildId == interaction.guild_id) stmt = select(CategoryGroup).where(CategoryGroup.GuildId == interaction.guild_id)
@@ -100,5 +98,4 @@ class RefreshCog(commands.Cog, name="Refresh Cog"):
interaction (nextcord.Interaction): Interaction with command. interaction (nextcord.Interaction): Interaction with command.
group_id (str): String to run autocomplete compared to. group_id (str): String to run autocomplete compared to.
""" """
await interaction.response.send_autocomplete(autofill_groups(interaction.guild_id, group_id)) await interaction.response.send_autocomplete(autofill_groups(interaction.guild_id, group_id))
+3 -10
View File
@@ -2,8 +2,6 @@
Set command heirarchy. Set command heirarchy.
""" """
from typing import Optional
import nextcord import nextcord
from nextcord.ext import commands from nextcord.ext import commands
from sqlalchemy import select from sqlalchemy import select
@@ -37,7 +35,7 @@ class SetCog(commands.Cog, name="Set command."):
group_id: str = nextcord.SlashOption( group_id: str = nextcord.SlashOption(
name="group", description="Category Group to assign to channel.", required=True name="group", description="Category Group to assign to channel.", required=True
), ),
channel: Optional[nextcord.abc.GuildChannel] = nextcord.SlashOption( channel: nextcord.abc.GuildChannel | None = nextcord.SlashOption(
description="Channel to register", description="Channel to register",
required=False, required=False,
channel_types=[nextcord.ChannelType.text], channel_types=[nextcord.ChannelType.text],
@@ -53,7 +51,6 @@ class SetCog(commands.Cog, name="Set command."):
channel (nextcord.abc.GuildChannel, optional): Channel to register channel (nextcord.abc.GuildChannel, optional): Channel to register
Is a SlashOption. Limited to text channels, Defaults to None. Is a SlashOption. Limited to text channels, Defaults to None.
""" """
if not channel: if not channel:
channel = interaction.channel channel = interaction.channel
channel_id = channel.id channel_id = channel.id
@@ -93,7 +90,6 @@ class SetCog(commands.Cog, name="Set command."):
group_id (str): String representation of Group id to set pad length in. Is SlashOption. group_id (str): String representation of Group id to set pad length in. Is SlashOption.
length (int): Number of characters. Is SlashOption. Defaults to 26. length (int): Number of characters. Is SlashOption. Defaults to 26.
""" """
with Session(engine) as session: with Session(engine) as session:
stmt = ( stmt = (
select(CategoryGroup) select(CategoryGroup)
@@ -111,7 +107,7 @@ class SetCog(commands.Cog, name="Set command."):
async def submission_channel( async def submission_channel(
self, self,
interaction: nextcord.Interaction, interaction: nextcord.Interaction,
channel: Optional[nextcord.abc.GuildChannel] = nextcord.SlashOption( channel: nextcord.abc.GuildChannel | None = nextcord.SlashOption(
description="Channel to register", description="Channel to register",
required=False, required=False,
channel_types=[nextcord.ChannelType.text], channel_types=[nextcord.ChannelType.text],
@@ -126,7 +122,6 @@ class SetCog(commands.Cog, name="Set command."):
channel Optional[nextcord.abc.GuildChannel], optional): Channel to register channel Optional[nextcord.abc.GuildChannel], optional): Channel to register
Is a SlashOption. Limited to text channels, Defaults to None. Is a SlashOption. Limited to text channels, Defaults to None.
""" """
if not channel: if not channel:
channel = interaction.channel channel = interaction.channel
@@ -156,7 +151,6 @@ class SetCog(commands.Cog, name="Set command."):
interaction (nextcord.Interaction): Interaction that invoked the command interaction (nextcord.Interaction): Interaction that invoked the command
length (int): Number of characters. Is SlashOption. Defaults to 12. length (int): Number of characters. Is SlashOption. Defaults to 12.
""" """
with Session(engine) as session: with Session(engine) as session:
stmt = ( stmt = (
select(CategoryGroup) select(CategoryGroup)
@@ -172,7 +166,7 @@ class SetCog(commands.Cog, name="Set command."):
async def verify_channel( async def verify_channel(
self, self,
interaction: nextcord.Interaction, interaction: nextcord.Interaction,
channel: Optional[nextcord.abc.GuildChannel] = nextcord.SlashOption( channel: nextcord.abc.GuildChannel | None = nextcord.SlashOption(
description="Channel to register", description="Channel to register",
required=False, required=False,
channel_types=[nextcord.ChannelType.text], channel_types=[nextcord.ChannelType.text],
@@ -187,7 +181,6 @@ class SetCog(commands.Cog, name="Set command."):
channel (Optional[nextcord.abc.GuildChannel]): Channel to register channel (Optional[nextcord.abc.GuildChannel]): Channel to register
Is a SlashOption. Limited to text channels, Defaults to None. Is a SlashOption. Limited to text channels, Defaults to None.
""" """
if not channel: if not channel:
channel = interaction.channel channel = interaction.channel
channel_id = channel.id channel_id = channel.id
-1
View File
@@ -57,7 +57,6 @@ class UpdateCog(commands.Cog, name="Update commands."):
old_username (str): Username to replace. old_username (str): Username to replace.
new_username (str): Username to replace old with. new_username (str): Username to replace old with.
""" """
with Session(engine) as session: with Session(engine) as session:
stmt = select(Guild.VerificationChannel).where(Guild.GuildId == interaction.guild_id) stmt = select(Guild.VerificationChannel).where(Guild.GuildId == interaction.guild_id)
verify_channel = interaction.guild.get_channel(session.scalar(stmt)) verify_channel = interaction.guild.get_channel(session.scalar(stmt))
+1 -1
View File
@@ -49,7 +49,7 @@ class ViewCog(commands.Cog, name="View Command"):
group_id (str): String representation of Group id to search from. Is SlashOption. group_id (str): String representation of Group id to search from. Is SlashOption.
category_id (str): String representation of Category id to show leaderboard. Is SlashOption. category_id (str): String representation of Category id to show leaderboard. Is SlashOption.
""" """
str_list: "list[str]" = [] str_list: list[str] = []
# get group, category, and all scores sorted correctly. # get group, category, and all scores sorted correctly.
with Session(engine) as session: with Session(engine) as session:
+1 -3
View File
@@ -2,8 +2,6 @@
Contains view and helper functions for confirming and processing deletes. Contains view and helper functions for confirming and processing deletes.
""" """
from typing import Optional
import nextcord import nextcord
from sqlalchemy import delete, select from sqlalchemy import delete, select
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
@@ -61,7 +59,7 @@ class DeleteConfirmView(nextcord.ui.Modal):
await interaction.response.send_message("Deletion complete.", ephemeral=True) await interaction.response.send_message("Deletion complete.", ephemeral=True)
def create_delete_id(group_id: int, category_id: Optional[int] = None) -> str: def create_delete_id(group_id: int, category_id: int | None = None) -> str:
""" """
Creates a delete id string to tell the view what to delete. Creates a delete id string to tell the view what to delete.
@@ -19,9 +19,7 @@ class UpdateUsernameVerifyView(nextcord.ui.View):
super().__init__(timeout=None) super().__init__(timeout=None)
@nextcord.ui.button(label="Deny", style=nextcord.ButtonStyle.red, custom_id="deny-username-update") @nextcord.ui.button(label="Deny", style=nextcord.ButtonStyle.red, custom_id="deny-username-update")
async def deny_button( async def deny_button(self, button: nextcord.ui.Button, interaction: nextcord.Interaction): # pylint: disable=unused-argument
self, button: nextcord.ui.Button, interaction: nextcord.Interaction
): # pylint: disable=unused-argument
""" """
Button for denying username update. Deletes message. Button for denying username update. Deletes message.
@@ -32,9 +30,7 @@ class UpdateUsernameVerifyView(nextcord.ui.View):
await interaction.message.delete() await interaction.message.delete()
@nextcord.ui.button(label="Approve", style=nextcord.ButtonStyle.green, custom_id="approve-username-update") @nextcord.ui.button(label="Approve", style=nextcord.ButtonStyle.green, custom_id="approve-username-update")
async def approve_button( async def approve_button(self, button: nextcord.ui.Button, interaction: nextcord.Interaction): # pylint: disable=unused-argument
self, button: nextcord.ui.Button, interaction: nextcord.Interaction
): # pylint: disable=unused-argument
""" """
Button for approving username update. Tells guild data to update username, then updates Button for approving username update. Tells guild data to update username, then updates
message to show it is complete. message to show it is complete.
@@ -43,7 +39,6 @@ class UpdateUsernameVerifyView(nextcord.ui.View):
button (nextcord.ui.Button): unused button (nextcord.ui.Button): unused
interaction (nextcord.Interaction): Interaction with the button interaction (nextcord.Interaction): Interaction with the button
""" """
with Session(engine) as session: with Session(engine) as session:
try: try:
embed_content = interaction.message.embeds[0].to_dict() embed_content = interaction.message.embeds[0].to_dict()
@@ -51,7 +46,7 @@ class UpdateUsernameVerifyView(nextcord.ui.View):
new_username: str = embed_content["fields"][1]["value"] new_username: str = embed_content["fields"][1]["value"]
stmt = select(distinct(Score.CategoryId)).where(Score.UserName == old_username) stmt = select(distinct(Score.CategoryId)).where(Score.UserName == old_username)
category_ids: "list[int]" = list(session.scalars(stmt)) category_ids: list[int] = list(session.scalars(stmt))
stmt = select(Score).where(Score.UserName == old_username) stmt = select(Score).where(Score.UserName == old_username)
for score in session.scalars(stmt): for score in session.scalars(stmt):
-2
View File
@@ -25,7 +25,6 @@ class GuildListener(commands.Cog, name="Guild Listener"):
Args: Args:
guild (nextcord.Guild): Guild that was joined. guild (nextcord.Guild): Guild that was joined.
""" """
with Session(engine) as session: with Session(engine) as session:
stmt = select(Guild).where(Guild.GuildId == guild.id) stmt = select(Guild).where(Guild.GuildId == guild.id)
guild_data = session.scalar(stmt) guild_data = session.scalar(stmt)
@@ -53,7 +52,6 @@ class GuildListener(commands.Cog, name="Guild Listener"):
Args: Args:
guild (nextcord.Guild): Guild that was left. guild (nextcord.Guild): Guild that was left.
""" """
with Session(engine) as session: with Session(engine) as session:
stmt = select(Guild).where(Guild.GuildId == guild.id) stmt = select(Guild).where(Guild.GuildId == guild.id)
guild_data = session.scalar(stmt) guild_data = session.scalar(stmt)
-1
View File
@@ -113,7 +113,6 @@ async def birthday_action(message: nextcord.Message) -> None:
Args: Args:
message (nextcord.Message): The message to check for birthday thing. message (nextcord.Message): The message to check for birthday thing.
""" """
today = datetime.today() today = datetime.today()
birthday = datetime(today.year, 7, 31) birthday = datetime(today.year, 7, 31)
chris = 642330132958281750 chris = 642330132958281750
+2 -5
View File
@@ -1,7 +1,5 @@
"""Views used for submission process.""" """Views used for submission process."""
from typing import Optional
import nextcord import nextcord
from nextcord.ext import commands from nextcord.ext import commands
from sqlalchemy import select from sqlalchemy import select
@@ -17,7 +15,7 @@ from .verification_view import VerificationView
def category_select_options(group_id: int) -> "list[nextcord.SelectOption]": def category_select_options(group_id: int) -> "list[nextcord.SelectOption]":
"""Returns list of categories for a given boss.""" """Returns list of categories for a given boss."""
options: "list[nextcord.SelectOption]" = [] options: list[nextcord.SelectOption] = []
with Session(engine) as session: with Session(engine) as session:
subq = select(CategoryGroup).where(CategoryGroup.GroupId == group_id).subquery() subq = select(CategoryGroup).where(CategoryGroup.GroupId == group_id).subquery()
@@ -149,7 +147,6 @@ class SubmissionCreateButton(nextcord.ui.View):
@nextcord.ui.button(label="Submit", style=nextcord.ButtonStyle.primary, custom_id="submitcreate") @nextcord.ui.button(label="Submit", style=nextcord.ButtonStyle.primary, custom_id="submitcreate")
async def submitcreate_button(self, button: nextcord.ui.Button, interaction: nextcord.Interaction): # pylint: disable=unused-argument async def submitcreate_button(self, button: nextcord.ui.Button, interaction: nextcord.Interaction): # pylint: disable=unused-argument
"""Button for score submission.""" """Button for score submission."""
# Don't allow a user to start a new submission if one already exists # Don't allow a user to start a new submission if one already exists
if submission_objects.has_user(interaction.user.id): if submission_objects.has_user(interaction.user.id):
await interaction.send("Please finish in progress submission first.", ephemeral=True) await interaction.send("Please finish in progress submission first.", ephemeral=True)
@@ -196,7 +193,7 @@ class SubmissionView(nextcord.ui.View):
bot: commands.Bot, bot: commands.Bot,
submission_user: int, submission_user: int,
*, *,
timeout: Optional[float] = 300, timeout: float | None = 300,
): ):
super().__init__(timeout=timeout) super().__init__(timeout=timeout)
self._submission_user = submission_user self._submission_user = submission_user
+2 -7
View File
@@ -13,16 +13,12 @@ class VerificationView(nextcord.ui.View):
super().__init__(timeout=None) super().__init__(timeout=None)
@nextcord.ui.button(label="Deny", style=nextcord.ButtonStyle.red, custom_id="deny-button-submission") @nextcord.ui.button(label="Deny", style=nextcord.ButtonStyle.red, custom_id="deny-button-submission")
async def deny_button( async def deny_button(self, button: nextcord.ui.Button, interaction: nextcord.Interaction): # pylint: disable=unused-argument
self, button: nextcord.ui.Button, interaction: nextcord.Interaction
): # pylint: disable=unused-argument
"""Button for denying score submission.""" """Button for denying score submission."""
await interaction.message.delete() await interaction.message.delete()
@nextcord.ui.button(label="Approve", style=nextcord.ButtonStyle.green, custom_id="approve-button-submission") @nextcord.ui.button(label="Approve", style=nextcord.ButtonStyle.green, custom_id="approve-button-submission")
async def approve_button( async def approve_button(self, button: nextcord.ui.Button, interaction: nextcord.Interaction): # pylint: disable=unused-argument
self, button: nextcord.ui.Button, interaction: nextcord.Interaction
): # pylint: disable=unused-argument
"""Button for approving score submission.""" """Button for approving score submission."""
for embed in interaction.message.embeds: for embed in interaction.message.embeds:
content = embed.to_dict() content = embed.to_dict()
@@ -38,7 +34,6 @@ class VerificationView(nextcord.ui.View):
user = field["value"] user = field["value"]
try: try:
await submit_score(interaction, group_id, category_id, user, score) await submit_score(interaction, group_id, category_id, user, score)
content = interaction.message.content content = interaction.message.content
content += f"\nApproved by <@{interaction.user.id}>." content += f"\nApproved by <@{interaction.user.id}>."
+3 -2
View File
@@ -9,9 +9,10 @@ import sys
import nextcord import nextcord
from nextcord.ext import commands from nextcord.ext import commands
from classes.submission import SubmissionMap
# pylint: disable=import-error # pylint: disable=import-error
from utilities import config from utilities import config
from classes.submission import SubmissionMap
logger = logging.getLogger("nextcord") logger = logging.getLogger("nextcord")
logger.setLevel(logging.DEBUG) logger.setLevel(logging.DEBUG)
@@ -29,7 +30,7 @@ def main():
print("ERROR: Python version must be >= 3.8") print("ERROR: Python version must be >= 3.8")
sys.exit() sys.exit()
# Allows privledged intents for monitoring members joining, roles editing, and role assignments # Allows privileged intents for monitoring members joining, roles editing, and role assignments
# These need to be enabled in the developer portal as well # These need to be enabled in the developer portal as well
intents = nextcord.Intents.default() intents = nextcord.Intents.default()
+6 -6
View File
@@ -7,12 +7,12 @@ readme = "README.md"
[tool.poetry.dependencies] [tool.poetry.dependencies]
python = "^3.12" python = "^3.12"
nextcord = ">=2.3.2" nextcord = "^2.6.0"
python-dotenv = ">=0.20.0" python-dotenv = "^1.0.1"
pandas = ">=1.5.2" pandas = "^2.2.3"
sqlalchemy = "<2.0" sqlalchemy = "^2.0.35"
pymysql = ">=1.0.2" pymysql = "^1.1.1"
cryptography = ">=2.8" cryptography = "^43.0.1"
[tool.poetry.group.dev.dependencies] [tool.poetry.group.dev.dependencies]
+1
View File
@@ -1,6 +1,7 @@
""" """
Simple file to set up the sqlalchemy connection to database. Simple file to set up the sqlalchemy connection to database.
""" """
from sqlalchemy import create_engine from sqlalchemy import create_engine
from sqlalchemy.ext.automap import automap_base from sqlalchemy.ext.automap import automap_base
+28 -76
View File
@@ -16,9 +16,7 @@ from utilities.sql_data_classes import Category, CategoryGroup, Guild, Score, en
MAX_LINE_LENGTH = 142 MAX_LINE_LENGTH = 142
def autofill_categories( def autofill_categories(guild_id: int, group_id: int, category_begin: str) -> "dict[str, str]":
guild_id: int, group_id: int, category_begin: str
) -> "dict[str, str]":
""" """
Returns all category names that start with category_begin. Returns all category names that start with category_begin.
If category_begin is empty, returns list of all category names. If category_begin is empty, returns list of all category names.
@@ -31,7 +29,6 @@ def autofill_categories(
Returns: Returns:
dict[str, str]: List of all category names that start with category_begin. dict[str, str]: List of all category names that start with category_begin.
""" """
with Session(engine) as session: with Session(engine) as session:
# search database for all category names # search database for all category names
subq = ( subq = (
@@ -42,9 +39,7 @@ def autofill_categories(
) )
cat_group_alias = aliased(CategoryGroup, subq) cat_group_alias = aliased(CategoryGroup, subq)
stmt = select(Category).where(Category.GroupId == cat_group_alias.GroupId) stmt = select(Category).where(Category.GroupId == cat_group_alias.GroupId)
categories_dict: "dict[str, str]" = { categories_dict: dict[str, str] = {cat.CategoryName: str(cat.CategoryId) for cat in session.scalars(stmt)}
cat.CategoryName: str(cat.CategoryId) for cat in session.scalars(stmt)
}
# if string empty, return all category names # if string empty, return all category names
if not category_begin: if not category_begin:
@@ -52,10 +47,8 @@ def autofill_categories(
# else find all categories that start with category_begin # else find all categories that start with category_begin
# TODO is it better to have sql handle the starts with or python? # TODO is it better to have sql handle the starts with or python?
categories: "dict[str, str]" = { categories: dict[str, str] = {
cat_name: cat_id cat_name: cat_id for cat_name, cat_id in categories_dict.items() if category_begin.lower() in cat_name.lower()
for cat_name, cat_id in categories_dict.items()
if category_begin.lower() in cat_name.lower()
} }
return dict(list(categories.items())[:25]) return dict(list(categories.items())[:25])
@@ -71,33 +64,26 @@ def autofill_groups(guild_id: int, group_begin: str) -> "dict[str, str]":
Returns: Returns:
dict[str, str]: Dict of group names and ids where group name start with group_begin. Key is name, value is id. dict[str, str]: Dict of group names and ids where group name start with group_begin. Key is name, value is id.
""" """
with Session(engine) as session: with Session(engine) as session:
# Select all groups in guild. # Select all groups in guild.
stmt = select(CategoryGroup).where(CategoryGroup.GuildId == guild_id) stmt = select(CategoryGroup).where(CategoryGroup.GuildId == guild_id)
# dict has to have key values of same type for discord. # dict has to have key values of same type for discord.
group_dict: "dict[str, str]" = { group_dict: dict[str, str] = {group.GroupName: str(group.GroupId) for group in session.scalars(stmt)}
group.GroupName: str(group.GroupId) for group in session.scalars(stmt)
}
# if string empty, return all group names # if string empty, return all group names
if not group_begin: if not group_begin:
return dict(list(group_dict.items())[:25]) return dict(list(group_dict.items())[:25])
# else find all groups that start with group_begin # else find all groups that start with group_begin
groups: "dict[str, str]" = { groups: dict[str, str] = {
group_name: group_id group_name: group_id for group_name, group_id in group_dict.items() if group_begin.lower() in group_name.lower()
for group_name, group_id in group_dict.items()
if group_begin.lower() in group_name.lower()
} }
return dict(list(groups.items())[:25]) return dict(list(groups.items())[:25])
def autofill_users_in_group_category( def autofill_users_in_group_category(guild_id: int, group_id: int, category_id: int, user_begin: str) -> "list[str]":
guild_id: int, group_id: int, category_id: int, user_begin: str
) -> "list[str]":
""" """
Returns list of users that match user_begin. Either all users in group or just users in category if category is Returns list of users that match user_begin. Either all users in group or just users in category if category is
not None. Might be expensive due to unioning all users in group. not None. Might be expensive due to unioning all users in group.
@@ -111,7 +97,6 @@ def autofill_users_in_group_category(
Returns: Returns:
list[str]: List of users that start with user_begin. list[str]: List of users that start with user_begin.
""" """
with Session(engine) as session: with Session(engine) as session:
# get all categories in the group # get all categories in the group
subq = ( subq = (
@@ -124,9 +109,7 @@ def autofill_users_in_group_category(
# If no category id, get all users in group. # If no category id, get all users in group.
if category_id == 0: if category_id == 0:
subq = select(Category.CategoryId).where( subq = select(Category.CategoryId).where(Category.GroupId == category_group.GroupId)
Category.GroupId == category_group.GroupId
)
stmt = select(distinct(Score.UserName)).where(Score.CategoryId.in_(subq)) stmt = select(distinct(Score.UserName)).where(Score.CategoryId.in_(subq))
else: else:
@@ -138,20 +121,16 @@ def autofill_users_in_group_category(
.subquery() .subquery()
) )
category = aliased(Category, subq) category = aliased(Category, subq)
stmt = select(distinct(Score.UserName)).where( stmt = select(distinct(Score.UserName)).where(Score.CategoryId == category.CategoryId)
Score.CategoryId == category.CategoryId
)
users_list: "list[str]" = list(session.scalars(stmt)) users_list: list[str] = list(session.scalars(stmt))
# if string empty, return all users # if string empty, return all users
if not user_begin: if not user_begin:
return users_list[:25] return users_list[:25]
# else find all users that start with user_begin # else find all users that start with user_begin
users: "list[str]" = [ users: list[str] = [user for user in users_list if user_begin.lower() in user.lower()]
user for user in users_list if user_begin.lower() in user.lower()
]
return users[:25] return users[:25]
@@ -167,27 +146,18 @@ def autofill_users_in_guild(guild_id: int, user_begin: str) -> "list[str]":
Returns: Returns:
list[str]: All users with submissions in the guild. list[str]: All users with submissions in the guild.
""" """
with Session(engine) as session: with Session(engine) as session:
group_ids = select(CategoryGroup.GroupId).where( group_ids = select(CategoryGroup.GroupId).where(CategoryGroup.GuildId == guild_id)
CategoryGroup.GuildId == guild_id category_ids = select(Category.CategoryId).where(Category.GroupId.in_(group_ids))
) stmt = select(distinct(Score.UserName)).where(Score.CategoryId.in_(category_ids))
category_ids = select(Category.CategoryId).where( users_list: list[str] = list(session.scalars(stmt))
Category.GroupId.in_(group_ids)
)
stmt = select(distinct(Score.UserName)).where(
Score.CategoryId.in_(category_ids)
)
users_list: "list[str]" = list(session.scalars(stmt))
# if string empty, return all users # if string empty, return all users
if not user_begin: if not user_begin:
return users_list[:25] return users_list[:25]
# else find all users that start with user_begin # else find all users that start with user_begin
users: "list[str]" = [ users: list[str] = [user for user in users_list if user_begin.lower() in user.lower()]
user for user in users_list if user_begin.lower() in user.lower()
]
return users[:25] return users[:25]
@@ -205,7 +175,6 @@ def format_highscore_message(session: Session, group: CategoryGroup) -> str:
Returns: Returns:
str: Formatted string for discord message str: Formatted string for discord message
""" """
ret_string = f"**{group.GroupName}**\n```" ret_string = f"**{group.GroupName}**\n```"
stmt = select(Guild).where(Guild.GuildId == group.GuildId) stmt = select(Guild).where(Guild.GuildId == group.GuildId)
@@ -227,8 +196,7 @@ def format_highscore_message(session: Session, group: CategoryGroup) -> str:
max_cat_size: int = 0 max_cat_size: int = 0
pad_length = group.PadLength pad_length = group.PadLength
for cat in session.scalars(stmt): for cat in session.scalars(stmt):
if cat.Size > max_cat_size: max_cat_size = max(cat.Size, max_cat_size)
max_cat_size = cat.Size
categories.append(cat) categories.append(cat)
category_ids.append(cat.CategoryId) category_ids.append(cat.CategoryId)
@@ -240,7 +208,7 @@ def format_highscore_message(session: Session, group: CategoryGroup) -> str:
# this list is what will be printed in the discord message # this list is what will be printed in the discord message
# first string will be category list, others will be score printouts # first string will be category list, others will be score printouts
str_list: "list[str]" = ["" for i in range(max_cat_size + 1)] str_list: list[str] = ["" for i in range(max_cat_size + 1)]
groups_df = d_f.groupby("CategoryId") groups_df = d_f.groupby("CategoryId")
# TODO figure out how to have less nests. # TODO figure out how to have less nests.
@@ -254,15 +222,11 @@ def format_highscore_message(session: Session, group: CategoryGroup) -> str:
# process data. Want each group sorted correctly and only top X results # process data. Want each group sorted correctly and only top X results
processed_group = ( processed_group = (
group_df.sort_values("Score", ascending=category.IsAscending) group_df.sort_values("Score", ascending=category.IsAscending).head(category.Size).reset_index()
.head(category.Size)
.reset_index()
) )
# add each score to list # add each score to list
for ind in range( for ind in range(max_cat_size): # need to subtract to deal with category name string
max_cat_size
): # need to subtract to deal with category name string
if ind in processed_group.index: if ind in processed_group.index:
username = processed_group["UserName"][ind] username = processed_group["UserName"][ind]
score = processed_group["Score"][ind] score = processed_group["Score"][ind]
@@ -275,12 +239,11 @@ def format_highscore_message(session: Session, group: CategoryGroup) -> str:
ind + 1, ind + 1,
pad_length, pad_length,
) )
# if not enough entries, pad out
elif ind + 1 <= category.Size:
str_list[ind + 1] += f"{ind + 1}: ".ljust(pad_length, " ")
else: else:
# if not enough entries, pad out str_list[ind + 1] += "".ljust(pad_length, " ")
if ind + 1 <= category.Size:
str_list[ind + 1] += f"{ind + 1}: ".ljust(pad_length, " ")
else:
str_list[ind + 1] += "".ljust(pad_length, " ")
else: else:
for ind in range(max_cat_size): for ind in range(max_cat_size):
if ind + 1 <= category.Size: if ind + 1 <= category.Size:
@@ -364,9 +327,7 @@ def get_time_str(time_in: float) -> str:
return time.strftime(f"%-M:%S.{milliseconds}", time.gmtime(time_in)) return time.strftime(f"%-M:%S.{milliseconds}", time.gmtime(time_in))
async def send_highscore_message( async def send_highscore_message(session: Session, guild: nextcord.Guild, group: CategoryGroup) -> nextcord.Message:
session: Session, guild: nextcord.Guild, group: CategoryGroup
) -> nextcord.Message:
""" """
Function takes care of checking if a message exists for a group. Function takes care of checking if a message exists for a group.
Creates and sends/edits the message. Creates and sends/edits the message.
@@ -379,7 +340,6 @@ async def send_highscore_message(
Returns: Returns:
nextcord.Message: Message object that was posted with group contents. nextcord.Message: Message object that was posted with group contents.
""" """
highscore_string = format_highscore_message(session, group) highscore_string = format_highscore_message(session, group)
channel = guild.get_channel(group.ChannelId) channel = guild.get_channel(group.ChannelId)
@@ -413,7 +373,6 @@ async def submit_score(
username (str): The user who submitted the score. username (str): The user who submitted the score.
score (str): Score to submit. Need to convert to correct type. score (str): Score to submit. Need to convert to correct type.
""" """
with Session(engine) as session: with Session(engine) as session:
stmt = select(Category.IsTimeRecord).where(Category.CategoryId == category_id) stmt = select(Category.IsTimeRecord).where(Category.CategoryId == category_id)
is_time_record = session.scalar(stmt) is_time_record = session.scalar(stmt)
@@ -425,11 +384,7 @@ async def submit_score(
else: else:
score = float(score) score = float(score)
stmt = ( stmt = select(Score).where(Score.UserName == username).where(Score.CategoryId == category_id)
select(Score)
.where(Score.UserName == username)
.where(Score.CategoryId == category_id)
)
existing_score = session.scalar(stmt) existing_score = session.scalar(stmt)
if existing_score: if existing_score:
@@ -454,11 +409,8 @@ def add_preset_configuration(guild_id: int, config_str: str) -> None:
guild_id (int): Guild id to add config to guild_id (int): Guild id to add config to
config_str (str): Config name to add config_str (str): Config name to add
""" """
with Session(engine) as session: with Session(engine) as session:
with open( with open(f"example_config/{config_str}.json", encoding="utf-8") as file:
f"example_config/{config_str}.json", encoding="utf-8", mode="r"
) as file:
config = json.load(file) config = json.load(file)
for group_name, categorys in config["tables"].items(): for group_name, categorys in config["tables"].items():