Compare commits

..
14 Commits
Author SHA1 Message Date
imcompiling f735e6d846 Merge branch '70-sqlmodel'
build_develop / build (push) Successful in 34s
build_release / build (push) Successful in 22s
2025-03-14 16:43:28 -04:00
imcompiling 9fbaf7bcf6 Merge branch 'main' of https://gitea.cheezers.org/imcompiling/agon
build_develop / build (push) Has been cancelled
2025-03-14 16:42:48 -04:00
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 f52b2fe49f Merge pull request 'Update ghcr.io/astral-sh/uv Docker tag to v0.6.5' (#81) from renovate/ghcr.io-astral-sh-uv-0.x into main
build_develop / build (push) Successful in 48s
Reviewed-on: #81
Reviewed-by: imcompiling <[email protected]>
2025-03-10 08:40:41 -04:00
renovate-bot 24a9fa998a Update ghcr.io/astral-sh/uv Docker tag to v0.6.5 2025-03-09 05:01:04 +00:00
imcompiling 63db4c1dd2 Merge pull request 'Update ghcr.io/astral-sh/uv Docker tag to v0.6.3' (#79) from renovate/ghcr.io-astral-sh-uv-0.x into main
build_develop / build (push) Successful in 51s
Reviewed-on: #79
Reviewed-by: imcompiling <[email protected]>
2025-03-03 23:03:42 -05:00
renovate-bot 8604ba2547 Update ghcr.io/astral-sh/uv Docker tag to v0.6.3 2025-03-02 05:01:27 +00:00
12 changed files with 71 additions and 499 deletions
+1 -1
View File
@@ -10,7 +10,7 @@
"ondelete" "ondelete"
], ],
"python.analysis.diagnosticSeverityOverrides": { "python.analysis.diagnosticSeverityOverrides": {
"reportOptionalMemberAccess": false, "reportOptionalMemberAccess": "warning",
"reportArgumentType": "information", "reportArgumentType": "information",
"reportAttributeAccessIssue": "warning" "reportAttributeAccessIssue": "warning"
} }
+1 -1
View File
@@ -1,7 +1,7 @@
FROM python:3.12.3-slim FROM python:3.12.3-slim
# install uv # install uv
COPY --from=ghcr.io/astral-sh/uv:0.6.2 /uv /uvx /bin/ COPY --from=ghcr.io/astral-sh/uv:0.6.5 /uv /uvx /bin/
RUN apt-get update && apt-get -y upgrade && apt-get -y install sqlite3 RUN apt-get update && apt-get -y upgrade && apt-get -y install sqlite3
+8 -10
View File
@@ -2,15 +2,14 @@
from nextcord.ext import commands from nextcord.ext import commands
from cogs.commands.create import CreateCog from .create import CreateCog
from cogs.commands.delete import DeleteCog from .delete import DeleteCog
from cogs.commands.get import GetCog from .help import HelpCog
from cogs.commands.help import HelpCog from .refresh import RefreshCog
from cogs.commands.refresh import RefreshCog from .set import SetCog
from cogs.commands.set import SetCog from .submission import SubmissionCog
from cogs.commands.submission import SubmissionCog from .update import UpdateCog
from cogs.commands.update import UpdateCog from .view import ViewCog
from cogs.commands.view import ViewCog
# This function will be called when this extension is loaded. # This function will be called when this extension is loaded.
@@ -19,7 +18,6 @@ def setup(bot: commands.Bot) -> None:
"""Adds all command classes to bot.""" """Adds all command classes to bot."""
bot.add_cog(CreateCog(bot)) bot.add_cog(CreateCog(bot))
bot.add_cog(DeleteCog()) bot.add_cog(DeleteCog())
bot.add_cog(GetCog())
bot.add_cog(HelpCog()) bot.add_cog(HelpCog())
bot.add_cog(RefreshCog(bot)) bot.add_cog(RefreshCog(bot))
bot.add_cog(SetCog()) bot.add_cog(SetCog())
-197
View File
@@ -1,197 +0,0 @@
"""Get command hierarchy."""
import nextcord
from nextcord.ext import commands
from sqlmodel import Session, col, select
from database.db import engine
from database.models import Category, CategoryGroup
from utilities.utils import autofill_categories, autofill_groups
class GetCog(commands.Cog, name="Get command."):
"""Class holds commands for setting variables in guild's database."""
def __init__(self) -> None:
"""Default constructor for SetCog class. Does nothing."""
@nextcord.slash_command(default_member_permissions=nextcord.Permissions(administrator=True))
async def get(self, interaction: nextcord.Interaction) -> None:
"""Parent slash command for getting. Never called."""
@get.subcommand(description="Commands for getting category information.")
async def category(self, interaction: nextcord.Interaction) -> None:
"""Parent slash subcommand for getting category information. Never called."""
@category.subcommand(description="Get whether a category is sorted in ascending or descending order.")
async def is_ascending(
self,
interaction: nextcord.Interaction,
group_id: str = nextcord.SlashOption(name="group", description="Group that Category is in.", required=True),
category_id: str = nextcord.SlashOption(
name="category", description="Category to information from.", required=True
),
) -> None:
"""
Command to view how a category is sorted.
Args:
interaction (nextcord.Interaction): Interaction that invoked the command.
group_id (str): String representation of Group id get information from. Is SlashOption.
category_id (str): String representation of Category id to get information from. Is SlashOption.
"""
stmt = (
select(Category).where(col(Category.group_id) == int(group_id)).where(col(Category.id) == int(category_id))
)
with Session(engine) as session:
if (category := session.scalar(stmt)) is None:
await interaction.response.send_message("Group or Category was not found.", ephemeral=True)
return
await interaction.response.send_message(
f"{category.name} is sorted in {'ascending' if category.is_ascending else 'descending'} order.",
ephemeral=True,
)
@category.subcommand(description="Get whether a category required validation.")
async def is_validated(
self,
interaction: nextcord.Interaction,
group_id: str = nextcord.SlashOption(name="group", description="Group that Category is in.", required=True),
category_id: str = nextcord.SlashOption(
name="category", description="Category to information from.", required=True
),
) -> None:
"""
Command to view if a category required photo validation.
Args:
interaction (nextcord.Interaction): Interaction that invoked the command.
group_id (str): String representation of Group id get information from. Is SlashOption.
category_id (str): String representation of Category id to get information from. Is SlashOption.
"""
stmt = (
select(Category).where(col(Category.group_id) == int(group_id)).where(col(Category.id) == int(category_id))
)
with Session(engine) as session:
if (category := session.scalar(stmt)) is None:
await interaction.response.send_message("Group or Category was not found.", ephemeral=True)
return
await interaction.response.send_message(
f"{category.name} does {'not ' if not category.is_validated else ''} require validation.",
ephemeral=True,
)
@category.subcommand(description="Get the size for a category.")
async def size(
self,
interaction: nextcord.Interaction,
group_id: str = nextcord.SlashOption(name="group", description="Group that Category is in.", required=True),
category_id: str = nextcord.SlashOption(name="category", description="Category to get size of.", required=True),
) -> None:
"""
Command to view a category's size.
Args:
interaction (nextcord.Interaction): Interaction that invoked the command.
group_id (str): String representation of Group id get information from. Is SlashOption.
category_id (str): String representation of Category id to get information from. Is SlashOption.
"""
stmt = (
select(Category).where(col(Category.group_id) == int(group_id)).where(col(Category.id) == int(category_id))
)
with Session(engine) as session:
if (category := session.scalar(stmt)) is None:
await interaction.response.send_message("Group or Category was not found.", ephemeral=True)
return
await interaction.response.send_message(f"{category.name} has a size of {category.size}.", ephemeral=True)
@get.subcommand(description="Commands for getting group information.")
async def group(self, interaction: nextcord.Interaction) -> None:
"""Parent slash subcommand for getting group information. Never called."""
@group.subcommand(description="Get the pad length of a group.")
async def pad_length(
self,
interaction: nextcord.Interaction,
group_id: str = nextcord.SlashOption(name="group", description="Group to get pad length of.", required=True),
) -> None:
"""
Command to view a group's pad length.
Args:
interaction (nextcord.Interaction): Interaction that invoked the command.
group_id (str): String representation of Group id get information from. Is SlashOption.
"""
stmt = select(CategoryGroup).where(col(CategoryGroup.id) == int(group_id))
with Session(engine) as session:
if (group := session.scalar(stmt)) is None:
await interaction.response.send_message("Group was not found.", ephemeral=True)
return
await interaction.response.send_message(
f"{group.name} has a pad length of {group.pad_length}.", ephemeral=True
)
@group.subcommand(description="Get the username length of a group.")
async def username_length(
self,
interaction: nextcord.Interaction,
group_id: str = nextcord.SlashOption(
name="group", description="Group to get username length of.", required=True
),
) -> None:
"""
Command to view a group's username length.
Args:
interaction (nextcord.Interaction): Interaction that invoked the command.
group_id (str): String representation of Group id get information from. Is SlashOption.
"""
stmt = select(CategoryGroup).where(col(CategoryGroup.id) == int(group_id))
with Session(engine) as session:
if (group := session.scalar(stmt)) is None:
await interaction.response.send_message("Group was not found.", ephemeral=True)
return
await interaction.response.send_message(
f"{group.name} has a username length of {group.username_length}.", ephemeral=True
)
@is_ascending.on_autocomplete("category_id")
@is_validated.on_autocomplete("category_id")
@size.on_autocomplete("category_id")
async def category_autocomplete(self, interaction: nextcord.Interaction, category_id: str, group_id: str) -> None:
"""
Autocomplete function for category names.
Args:
interaction (nextcord.Interaction): Interaction that invoked the command.
category_id (str): String to be autocompleted with Category name.
group_id (str): Group to find Category from.
"""
await interaction.response.send_autocomplete(
autofill_categories(interaction.guild_id, int(group_id), category_id)
)
@is_ascending.on_autocomplete("group_id")
@is_validated.on_autocomplete("group_id")
@pad_length.on_autocomplete("group_id")
@size.on_autocomplete("group_id")
@username_length.on_autocomplete("group_id")
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.
Args:
interaction (nextcord.Interaction): Interaction with command.
group_id (str): String to run autocompletion against.
"""
await interaction.response.send_autocomplete(autofill_groups(interaction.guild_id, group_id))
+33 -251
View File
@@ -7,9 +7,8 @@ from sqlalchemy.orm import Session
from sqlmodel import col from sqlmodel import col
from database.db import engine from database.db import engine
from database.models import Category, CategoryGroup, Guild from database.models import CategoryGroup, Guild
from utilities.logger import agon_logger from utilities.utils import MAX_LINE_LENGTH, autofill_groups
from utilities.utils import MAX_LINE_LENGTH, autofill_categories, autofill_groups, send_highscore_message
class SetCog(commands.Cog, name="Set command."): class SetCog(commands.Cog, name="Set command."):
@@ -22,219 +21,13 @@ class SetCog(commands.Cog, name="Set command."):
async def set(self, interaction: nextcord.Interaction) -> None: async def set(self, interaction: nextcord.Interaction) -> None:
"""Parent slash command for setting settings. Never called.""" """Parent slash command for setting settings. Never called."""
@set.subcommand("Commands for editing a category.") @set.subcommand(description="Registers the channel for a Category Group's messages.")
async def category(self, interaction: nextcord.Interaction) -> None: async def group_output_channel(
"""Parent slash subcommand for setting category settings. Never called."""
@category.subcommand(description="Update the number of entries to show for a category.", name="size")
async def category_size(
self, self,
interaction: nextcord.Interaction, interaction: nextcord.Interaction,
group_id: str = nextcord.SlashOption(name="group", description="Group that Category is in.", required=True), group_id: str = nextcord.SlashOption(
category_id: str = nextcord.SlashOption( name="group", description="Category Group to assign to channel.", required=True
name="category", description="Category to update size of.", required=True
), ),
size: int = nextcord.SlashOption(description="Number of entries to show for the category.", required=True),
) -> None:
"""
Command to update a category's size.
Args:
interaction (nextcord.Interaction): Interaction that invoked the command.
group_id (str): String representation of Group id to register channel to. Is SlashOption.
category_id (str): String representation of Category id to register channel to. Is SlashOption.
size (int): Size to set. Is SlashOption.
"""
stmt = (
select(Category).where(col(Category.group_id) == int(group_id)).where(col(Category.id) == int(category_id))
)
with Session(engine) as session:
if (category := session.scalar(stmt)) is None:
await interaction.response.send_message("Category or Group was not found.", ephemeral=True)
return
category.size = size
# session.add(category) test if this is needed. SQLAlchemy can do the update without adding.
session.commit()
# Use the relationship to get group category is in.
if (group := category.category_group) is None:
agon_logger.warning(f"{category.name}:{category.id} group relationship was None.")
stmt = select(CategoryGroup).where(col(CategoryGroup.id) == group_id)
if (group := session.scalar(stmt)) is None:
await interaction.response.send_message("Group was not found.", ephemeral=True)
return
await send_highscore_message(session, interaction.guild, group)
await interaction.response.send_message(f"{category.name} size has been updated to {size}.", ephemeral=True)
@category.subcommand(description="Update the is_ascending value for a category.")
async def is_ascending(
self,
interaction: nextcord.Interaction,
group_id: str = nextcord.SlashOption(name="group", description="Group that Category is in.", required=True),
category_id: str = nextcord.SlashOption(
name="category", description="Category to update is_ascending of.", required=True
),
ascending: bool = nextcord.SlashOption( # noqa: FBT001
description="Whether the scores are sorted ascending or not.", required=True
),
) -> None:
"""
Command to update a category's is_ascending boolean.
Args:
interaction (nextcord.Interaction): Interaction that invoked the command.
group_id (str): String representation of Group id to register channel to. Is SlashOption.
category_id (str): String representation of Category id to register channel to. Is SlashOption.
ascending (bool): Boolean to set. Determines whether scores are sorted ascending or not. Is SlashOption.
"""
stmt = (
select(Category).where(col(Category.group_id) == int(group_id)).where(col(Category.id) == int(category_id))
)
with Session(engine) as session:
if (category := session.scalar(stmt)) is None:
await interaction.response.send_message("Category or Group was not found.", ephemeral=True)
return
category.is_ascending = ascending
# session.add(category) test if this is needed. SQLAlchemy can do the update without adding.
session.commit()
# Use the relationship to get group category is in.
if (group := category.category_group) is None:
agon_logger.warning(f"{category.name}:{category.id} group relationship was None.")
stmt = select(CategoryGroup).where(col(CategoryGroup.id) == group_id)
if (group := session.scalar(stmt)) is None:
await interaction.response.send_message("Group was not found.", ephemeral=True)
return
await send_highscore_message(session, interaction.guild, group)
await interaction.response.send_message(
f"{category.name} is_ascending has been updated to {ascending}.", ephemeral=True
)
@category.subcommand(description="Update the is_validated value for a category.")
async def is_validated(
self,
interaction: nextcord.Interaction,
group_id: str = nextcord.SlashOption(name="group", description="Group that Category is in.", required=True),
category_id: str = nextcord.SlashOption(
name="category", description="Category to update is_ascending of.", required=True
),
validated: bool = nextcord.SlashOption( # noqa: FBT001
description="Whether the scores need to be validated.", required=True
),
) -> None:
"""
Command to update a category's is_validated boolean.
Args:
interaction (nextcord.Interaction): Interaction that invoked the command.
group_id (str): String representation of Group id to register channel to. Is SlashOption.
category_id (str): String representation of Category id to register channel to. Is SlashOption.
validated (bool): Boolean to set. Determines whether scores are validated or not. Is SlashOption.
"""
stmt = (
select(Category).where(col(Category.group_id) == int(group_id)).where(col(Category.id) == int(category_id))
)
with Session(engine) as session:
if (category := session.scalar(stmt)) is None:
await interaction.response.send_message("Category or Group was not found.", ephemeral=True)
return
category.is_validated = validated
# session.add(category) test if this is needed. SQLAlchemy can do the update without adding.
session.commit()
await interaction.response.send_message(
f"{category.name} is_validated has been updated to {validated}.", ephemeral=True
)
@category.subcommand(name="name", description="Change the name of a category.")
async def category_name(
self,
interaction: nextcord.Interaction,
group_id: str = nextcord.SlashOption(name="group", description="Group that Category is in.", required=True),
category_id: str = nextcord.SlashOption(
name="category", description="Category to update the name of.", required=True
),
name: str = nextcord.SlashOption(description="Name to update to.", required=True, min_length=1, max_length=30),
) -> None:
"""
Command to update a category's name.
Args:
interaction (nextcord.Interaction): Interaction that invoked the command.
group_id (str): String representation of Group id to register channel to. Is SlashOption.
category_id (str): String representation of Category id to register channel to. Is SlashOption.
name (str): The name to update the category's name to. Is SlashOption.
"""
stmt = (
select(Category).where(col(Category.group_id) == int(group_id)).where(col(Category.id) == int(category_id))
)
with Session(engine) as session:
if (category := session.scalar(stmt)) is None:
await interaction.response.send_message("Category or Group was not found.", ephemeral=True)
return
old_name = category.name
category.name = name
# session.add(category) test if this is needed. SQLAlchemy can do the update without adding.
session.commit()
# Use the relationship to get group category is in.
if (group := category.category_group) is None:
agon_logger.warning(f"{category.name}:{category.id} group relationship was None.")
stmt = select(CategoryGroup).where(col(CategoryGroup.id) == group_id)
if (group := session.scalar(stmt)) is None:
await interaction.response.send_message("Group was not found.", ephemeral=True)
return
await send_highscore_message(session, interaction.guild, group)
await interaction.response.send_message(f"{old_name} has been updated to {category.name}.", ephemeral=True)
@set.subcommand(description="Commands for editing a group.")
async def group(self, interaction: nextcord.Interaction) -> None:
"""Parent slash subcommand for setting group settings. Never called."""
@group.subcommand(name="name", description="Change the name of a Group.")
async def group_name(
self,
interaction: nextcord.Interaction,
group_id: str = nextcord.SlashOption(name="group", description="Group that Category is in.", required=True),
name: str = nextcord.SlashOption(description="Name to update to.", required=True, min_length=1, max_length=50),
) -> None:
"""
Command to update a group's name.
Args:
interaction (nextcord.Interaction): Interaction that invoked the command.
group_id (str): String representation of Group id to register channel to. Is SlashOption.
name (str): The name to update the group's name to. Is SlashOption.
"""
stmt = select(CategoryGroup).where(col(CategoryGroup.id) == int(group_id))
with Session(engine) as session:
if (group := session.scalar(stmt)) is None:
await interaction.response.send_message("Group was not found.", ephemeral=True)
return
old_name = group.name
group.name = name
# session.add(group) test if this is needed. SQLAlchemy can do the update without adding.
session.commit()
await send_highscore_message(session, interaction.guild, group)
await interaction.response.send_message(f"{old_name} has been updated to {group.name}.", ephemeral=True)
@group.subcommand(description="Registers the channel for a Group's messages.")
async def output_channel(
self,
interaction: nextcord.Interaction,
group_id: str = nextcord.SlashOption(name="group", description="Group to assign to channel.", required=True),
channel: nextcord.abc.GuildChannel | None = nextcord.SlashOption( channel: nextcord.abc.GuildChannel | None = nextcord.SlashOption(
description="Channel to register", description="Channel to register",
required=False, required=False,
@@ -248,7 +41,7 @@ class SetCog(commands.Cog, name="Set command."):
Args: Args:
interaction (nextcord.Interaction): Interaction that invoked the command. interaction (nextcord.Interaction): Interaction that invoked the command.
group_id (str): String representation of Group id to register channel to. Is SlashOption. group_id (str): String representation of Group id to register channel to. Is SlashOption.
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.
""" """
channel_id = interaction.channel.id if not channel else channel.id channel_id = interaction.channel.id if not channel else channel.id
@@ -259,18 +52,22 @@ class SetCog(commands.Cog, name="Set command."):
.where(col(CategoryGroup.guild_id) == interaction.guild_id) .where(col(CategoryGroup.guild_id) == interaction.guild_id)
.where(col(CategoryGroup.id) == int(group_id)) .where(col(CategoryGroup.id) == int(group_id))
) )
group = session.scalar(stmt) if (group := session.scalar(stmt)) is None:
group.ChannelId = channel_id await interaction.response.send_message("Group or Category was not found.", ephemeral=True)
return
group.channel_id = channel_id
session.commit() session.commit()
await interaction.send(f"Registered <#{channel_id}>.", ephemeral=True) await interaction.send(f"Registered <#{channel_id}>.", ephemeral=True)
@group.subcommand( @set.subcommand(
description="Set character length for a category. Can cut off username/score. Affects all categories." description="Set character length for a category. Can cut off username/score. Affects all categories."
) )
async def pad_length( async def pad_length(
self, self,
interaction: nextcord.Interaction, interaction: nextcord.Interaction,
group_id: str = nextcord.SlashOption(name="group", description="Group to assign pad length to.", required=True), group_id: str = nextcord.SlashOption(
name="group", description="Category Group to assign pad length to.", required=True
),
length: int = nextcord.SlashOption( length: int = nextcord.SlashOption(
description="Number of characters alloted to a category.", description="Number of characters alloted to a category.",
required=True, required=True,
@@ -294,16 +91,16 @@ class SetCog(commands.Cog, name="Set command."):
.where(col(CategoryGroup.id) == int(group_id)) .where(col(CategoryGroup.id) == int(group_id))
) )
if (group := session.scalar(stmt)) is None: if (group := session.scalar(stmt)) is None:
await interaction.response.send_message("Group was not found.", ephemeral=True) await interaction.response.send_message("Group or Category was not found.", ephemeral=True)
return return
group.PadLength = length
group.pad_length = length
session.commit() session.commit()
await interaction.response.send_message( await interaction.response.send_message(
f"Category pad length set to {length} for {group.name}", ephemeral=True f"Category pad length set to {length} for {group.name}", ephemeral=True
) )
await send_highscore_message(session, interaction.guild, group)
@group.subcommand( @set.subcommand(
description="Registers the channel for submission messages. Currently this information is deprecated." description="Registers the channel for submission messages. Currently this information is deprecated."
) )
async def submission_channel( async def submission_channel(
@@ -328,12 +125,14 @@ class SetCog(commands.Cog, name="Set command."):
with Session(engine) as session: with Session(engine) as session:
stmt = select(Guild).where(col(Guild.id) == interaction.guild_id) stmt = select(Guild).where(col(Guild.id) == interaction.guild_id)
guild = session.scalar(stmt) if (guild := session.scalar(stmt)) is None:
await interaction.response.send_message("Guild was not found.", ephemeral=True)
return
guild.submission_channel = channel_id guild.submission_channel = channel_id
session.commit() session.commit()
await interaction.send(f"Registered <#{channel_id}>.", ephemeral=True) await interaction.send(f"Registered <#{channel_id}>.", ephemeral=True)
@group.subcommand(description="Set username length for a Group of Categories.") @set.subcommand(description="Set username length for a Group of Categories.")
async def username_length( async def username_length(
self, self,
interaction: nextcord.Interaction, interaction: nextcord.Interaction,
@@ -358,8 +157,11 @@ class SetCog(commands.Cog, name="Set command."):
.where(col(CategoryGroup.guild_id) == interaction.guild_id) .where(col(CategoryGroup.guild_id) == interaction.guild_id)
.where(col(CategoryGroup.id) == int(group_id)) .where(col(CategoryGroup.id) == int(group_id))
) )
group = session.scalar(stmt) if (group := session.scalar(stmt)) is None:
group.UsernameLength = length await interaction.response.send_message("Group or Category was not found.", ephemeral=True)
return
group.username_length = length
session.commit() session.commit()
await interaction.response.send_message(f"Username length set to {length}.", ephemeral=True) await interaction.response.send_message(f"Username length set to {length}.", ephemeral=True)
@@ -386,34 +188,14 @@ class SetCog(commands.Cog, name="Set command."):
with Session(engine) as session: with Session(engine) as session:
stmt = select(Guild).where(col(Guild.id) == interaction.guild_id) stmt = select(Guild).where(col(Guild.id) == interaction.guild_id)
guild = session.scalar(stmt) if (guild := session.scalar(stmt)) is None:
guild.VerificationChannel = channel_id await interaction.response.send_message("Guild was not found.", ephemeral=True)
return
guild.verification_channel = channel_id
session.commit() session.commit()
await interaction.send(f"Registered <#{channel_id}>.", ephemeral=True) await interaction.send(f"Registered <#{channel_id}>.", ephemeral=True)
@category_name.on_autocomplete("category_id") @group_output_channel.on_autocomplete("group_id")
@category_size.on_autocomplete("category_id")
@is_ascending.on_autocomplete("category_id")
@is_validated.on_autocomplete("category_id")
async def category_autocomplete(self, interaction: nextcord.Interaction, category_id: str, group_id: str) -> None:
"""
Autocomplete function for category names.
Args:
interaction (nextcord.Interaction): Interaction that invoked the command.
category_id (str): String to be autocompleted with Category name.
group_id (str): Group to find Category from.
"""
await interaction.response.send_autocomplete(
autofill_categories(interaction.guild_id, int(group_id), category_id)
)
@category_name.on_autocomplete("group_id")
@category_size.on_autocomplete("group_id")
@group_name.on_autocomplete("group_id")
@is_ascending.on_autocomplete("group_id")
@is_validated.on_autocomplete("group_id")
@output_channel.on_autocomplete("group_id")
@pad_length.on_autocomplete("group_id") @pad_length.on_autocomplete("group_id")
@username_length.on_autocomplete("group_id") @username_length.on_autocomplete("group_id")
async def group_autocomplete(self, interaction: nextcord.Interaction, group_id: str) -> None: async def group_autocomplete(self, interaction: nextcord.Interaction, group_id: str) -> None:
+11 -14
View File
@@ -11,7 +11,7 @@ from sqlmodel import col
from cogs.submission.verification_view import VerificationView from cogs.submission.verification_view import VerificationView
from database.db import engine from database.db import engine
from database.models import Category, CategoryGroup, Guild, Submission from database.models import Category, CategoryGroup, Guild, Submission
from main import agon_logger from main import logger
from utilities.utils import autofill_categories, autofill_groups, submit_score from utilities.utils import autofill_categories, autofill_groups, submit_score
@@ -64,10 +64,7 @@ class SubmissionCog(commands.Cog):
.where(col(Category.group_id) == int(group_id)) .where(col(Category.group_id) == int(group_id))
.where(col(Category.id) == int(category_id)) .where(col(Category.id) == int(category_id))
) )
if (category := session.scalar(stmt)) is None:
category: Category | None = session.scalar(stmt)
if category is None:
await interaction.response.send_message("Category not found.", ephemeral=True) await interaction.response.send_message("Category not found.", ephemeral=True)
return return
@@ -79,10 +76,13 @@ class SubmissionCog(commands.Cog):
return return
# check with regex if score is in correct format # check with regex if score is in correct format
verified_score = await self.verify_score(interaction, score, is_time_record=category.is_time_record)
# check if verify_score was successful. It replies with corrections required itself. # 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 return
# Create submission verification request or submit score if verification not needed. # Create submission verification request or submit score if verification not needed.
@@ -92,23 +92,20 @@ class SubmissionCog(commands.Cog):
return return
stmt = select(col(Guild.verification_channel)).where(col(Guild.id) == interaction.guild_id) stmt = select(col(Guild.verification_channel)).where(col(Guild.id) == interaction.guild_id)
verification_channel_id: int | None = session.scalar(stmt)
if verification_channel_id is None: if (verification_channel_id := session.scalar(stmt)) is None:
error_response = "No verification channel was registered. Contact server admins." error_response = "No verification channel was registered. Contact server admins."
await interaction.response.send_message(error_response, ephemeral=True) await interaction.response.send_message(error_response, ephemeral=True)
return return
channel = interaction.guild.get_channel(verification_channel_id) if (channel := interaction.guild.get_channel(verification_channel_id)) is None:
if channel is None:
error_response = "Registered Verification channel does not exist. Contact server admins." error_response = "Registered Verification channel does not exist. Contact server admins."
await interaction.response.send_message(error_response, ephemeral=True) await interaction.response.send_message(error_response, ephemeral=True)
return return
if not isinstance(channel, nextcord.TextChannel): if not isinstance(channel, nextcord.TextChannel):
await interaction.response.send_message("An error happened with submission.") await interaction.response.send_message("An error happened with submission.")
agon_logger.error( logger.error(
"Channel id %s in guild %s is does not correspond to a TextChannel.", channel.id, channel.guild.id "Channel id %s in guild %s is does not correspond to a TextChannel.", channel.id, channel.guild.id
) )
return return
+1 -2
View File
@@ -26,10 +26,9 @@ class GuildListener(commands.Cog, name="Guild Listener"):
""" """
with Session(engine) as session: with Session(engine) as session:
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 check if the guild already exists # 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) new_guild = Guild(id=guild.id, name=guild.name)
session.add(new_guild) session.add(new_guild)
else: else:
+2 -2
View File
@@ -31,7 +31,7 @@ class GuildData(commands.Cog, name="Guild Data"):
with Session(engine) as session: with Session(engine) as session:
for guild in self.bot.guilds: for guild in self.bot.guilds:
# update given guild name for guild id # 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) guild_data = session.scalar(stmt)
# just in case checking if the guild doesn't exist. # 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) new_guild = Guild(id=guild.id, name=guild.name)
session.add(new_guild) session.add(new_guild)
else: else:
guild_data.GuildName = guild.name guild_data.name = guild.name
session.commit() session.commit()
+1 -3
View File
@@ -39,9 +39,7 @@ class VerificationView(nextcord.ui.View):
.where(col(Submission.message_id) == interaction.message.id) .where(col(Submission.message_id) == interaction.message.id)
) )
submission = session.scalar(stmt) if (submission := session.scalar(stmt)) is None:
if submission is None:
await interaction.response.send_message("Could not find submission data in database.", ephemeral=True) await interaction.response.send_message("Could not find submission data in database.", ephemeral=True)
await interaction.message.delete() await interaction.message.delete()
return return
+9 -4
View File
@@ -8,7 +8,6 @@ import nextcord
from nextcord.ext import commands from nextcord.ext import commands
from utilities import config from utilities import config
from utilities.logger import agon_logger
nextcord_logger = logging.getLogger("nextcord") nextcord_logger = logging.getLogger("nextcord")
nextcord_logger.setLevel(logging.DEBUG) nextcord_logger.setLevel(logging.DEBUG)
@@ -16,12 +15,18 @@ nextcord_handler = logging.FileHandler(filename="nextcord.log", encoding="utf-8"
nextcord_handler.setFormatter(logging.Formatter("%(asctime)s:%(levelname)s:%(name)s: %(message)s")) nextcord_handler.setFormatter(logging.Formatter("%(asctime)s:%(levelname)s:%(name)s: %(message)s"))
nextcord_logger.addHandler(nextcord_handler) nextcord_logger.addHandler(nextcord_handler)
logger = logging.getLogger("default")
logger.setLevel(logging.INFO)
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(logging.Formatter("%(asctime)s:%(levelname)s:%(name)s: %(message)s"))
logger.addHandler(handler)
def main() -> None: def main() -> None:
"""Main function for bot.""" """Main function for bot."""
min_python_version = (3, 12) min_python_version = (3, 12)
if sys.version_info < min_python_version: if sys.version_info < min_python_version:
agon_logger.error("ERROR: Python version must be >= 3.12") logger.error("ERROR: Python version must be >= 3.12")
sys.exit() sys.exit()
# Allows privileged intents for monitoring members joining, roles editing, and role assignments # Allows privileged intents for monitoring members joining, roles editing, and role assignments
@@ -41,9 +46,9 @@ def main() -> None:
async def on_ready() -> None: async def on_ready() -> None:
"""When Discord is connected.""" """When Discord is connected."""
if bot.user is None: if bot.user is None:
agon_logger.error("Discord connection failed.") logger.error("Discord connection failed.")
return return
agon_logger.info("%s has connected to Discord!", bot.user.name) logger.info("%s has connected to Discord!", bot.user.name)
# Run Discord bot # Run Discord bot
bot.run(config.DISCORD_TOKEN) bot.run(config.DISCORD_TOKEN)
-10
View File
@@ -1,10 +0,0 @@
"""Creates logger for Agon bot."""
import logging
import sys
agon_logger = logging.getLogger("default")
agon_logger.setLevel(logging.INFO)
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(logging.Formatter("%(asctime)s:%(levelname)s:%(name)s: %(message)s"))
agon_logger.addHandler(handler)
+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: with Session(engine) as session:
# Select all groups in guild. # 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. # 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)} 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: if group.message_id is None:
# message doesn't exist. # message doesn't exist.
message: nextcord.Message = await channel.send(highscore_string) message: nextcord.Message = await channel.send(highscore_string)
group.MessageId = message.id group.message_id = message.id
session.commit() session.commit()
else: else:
message: nextcord.Message = await channel.fetch_message(group.message_id) message: nextcord.Message = await channel.fetch_message(group.message_id)
@@ -390,7 +390,7 @@ async def submit_score(
existing_score = session.scalar(stmt) existing_score = session.scalar(stmt)
if existing_score: if existing_score:
existing_score.Score = score_f existing_score.score = score_f
else: else:
score_obj = Score(username=username, score=score_f, category_id=category_id) score_obj = Score(username=username, score=score_f, category_id=category_id)
session.add(score_obj) session.add(score_obj)