Compare commits

..
16 Commits
11 changed files with 473 additions and 71 deletions
+2 -2
View File
@@ -10,8 +10,8 @@
"ondelete"
],
"python.analysis.diagnosticSeverityOverrides": {
"reportOptionalMemberAccess": "warning",
"reportOptionalMemberAccess": false,
"reportArgumentType": "information",
"reportAttributeAccessIssue": "warning"
}
}
}
+10 -8
View File
@@ -2,14 +2,15 @@
from nextcord.ext import commands
from .create import CreateCog
from .delete import DeleteCog
from .help import HelpCog
from .refresh import RefreshCog
from .set import SetCog
from .submission import SubmissionCog
from .update import UpdateCog
from .view import ViewCog
from cogs.commands.create import CreateCog
from cogs.commands.delete import DeleteCog
from cogs.commands.get import GetCog
from cogs.commands.help import HelpCog
from cogs.commands.refresh import RefreshCog
from cogs.commands.set import SetCog
from cogs.commands.submission import SubmissionCog
from cogs.commands.update import UpdateCog
from cogs.commands.view import ViewCog
# This function will be called when this extension is loaded.
@@ -18,6 +19,7 @@ def setup(bot: commands.Bot) -> None:
"""Adds all command classes to bot."""
bot.add_cog(CreateCog(bot))
bot.add_cog(DeleteCog())
bot.add_cog(GetCog())
bot.add_cog(HelpCog())
bot.add_cog(RefreshCog(bot))
bot.add_cog(SetCog())
+148
View File
@@ -0,0 +1,148 @@
"""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 to register channel to. Is SlashOption.
category_id (str): String representation of Category id to register channel to. Is SlashOption.
"""
stmt = (
select(Category).where(col(CategoryGroup.id) == int(group_id)).where(col(Category.id) == int(category_id))
)
with Session(engine) as session:
category = session.scalar(stmt)
if category 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 to register channel to. Is SlashOption.
category_id (str): String representation of Category id to register channel to. Is SlashOption.
"""
stmt = (
select(Category).where(col(CategoryGroup.id) == int(group_id)).where(col(Category.id) == int(category_id))
)
with Session(engine) as session:
category = session.scalar(stmt)
if category 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 to register channel to. Is SlashOption.
category_id (str): String representation of Category id to register channel to. Is SlashOption.
"""
stmt = (
select(Category).where(col(CategoryGroup.id) == int(group_id)).where(col(Category.id) == int(category_id))
)
with Session(engine) as session:
category = session.scalar(stmt)
if category 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."""
@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")
@size.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))
+275 -34
View File
@@ -7,8 +7,9 @@ from sqlalchemy.orm import Session
from sqlmodel import col
from database.db import engine
from database.models import CategoryGroup, Guild
from utilities.utils import MAX_LINE_LENGTH, autofill_groups
from database.models import Category, CategoryGroup, Guild
from utilities.logger import agon_logger
from utilities.utils import MAX_LINE_LENGTH, autofill_categories, autofill_groups, send_highscore_message
class SetCog(commands.Cog, name="Set command."):
@@ -21,13 +22,241 @@ class SetCog(commands.Cog, name="Set command."):
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(
@set.subcommand("Commands for editing a category.")
async def category(self, interaction: nextcord.Interaction) -> None:
"""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,
interaction: nextcord.Interaction,
group_id: str = nextcord.SlashOption(
name="group", description="Category Group to assign to channel.", required=True
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 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(CategoryGroup.id) == int(group_id)).where(col(Category.id) == int(category_id))
)
with Session(engine) as session:
category = session.scalar(stmt)
if category 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.
group = category.category_group
if group is None:
agon_logger.warning(f"{category.name}:{category.id} group relationship was None.")
stmt = select(CategoryGroup).where(col(CategoryGroup.id) == group_id)
group = session.scalar(stmt)
if group 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(CategoryGroup.id) == int(group_id)).where(col(Category.id) == int(category_id))
)
with Session(engine) as session:
category = session.scalar(stmt)
if category 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.
group = category.category_group
if group is None:
agon_logger.warning(f"{category.name}:{category.id} group relationship was None.")
stmt = select(CategoryGroup).where(col(CategoryGroup.id) == group_id)
group = session.scalar(stmt)
if group 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(CategoryGroup.id) == int(group_id)).where(col(Category.id) == int(category_id))
)
with Session(engine) as session:
category = session.scalar(stmt)
if category 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()
# Use the relationship to get group category is in.
group = category.category_group
if group is None:
agon_logger.warning(f"{category.name}:{category.id} group relationship was None.")
stmt = select(CategoryGroup).where(col(CategoryGroup.id) == group_id)
group = session.scalar(stmt)
if group is None:
await interaction.response.send_message("Group was not found.", ephemeral=True)
return
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(CategoryGroup.id) == int(group_id)).where(col(Category.id) == int(category_id))
)
with Session(engine) as session:
category = session.scalar(stmt)
if category 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.
group = category.category_group
if group is None:
agon_logger.warning(f"{category.name}:{category.id} group relationship was None.")
stmt = select(CategoryGroup).where(col(CategoryGroup.id) == group_id)
group = session.scalar(stmt)
if group 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:
group = session.scalar(stmt)
if group 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(
description="Channel to register",
required=False,
@@ -41,7 +270,7 @@ 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 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.
"""
channel_id = interaction.channel.id if not channel else channel.id
@@ -52,22 +281,18 @@ class SetCog(commands.Cog, name="Set command."):
.where(col(CategoryGroup.guild_id) == interaction.guild_id)
.where(col(CategoryGroup.id) == int(group_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
group = session.scalar(stmt)
group.ChannelId = channel_id
session.commit()
await interaction.send(f"Registered <#{channel_id}>.", ephemeral=True)
@set.subcommand(
@group.subcommand(
description="Set character length for a category. Can cut off username/score. Affects all categories."
)
async def pad_length(
self,
interaction: nextcord.Interaction,
group_id: str = nextcord.SlashOption(
name="group", description="Category Group to assign pad length to.", required=True
),
group_id: str = nextcord.SlashOption(name="group", description="Group to assign pad length to.", required=True),
length: int = nextcord.SlashOption(
description="Number of characters alloted to a category.",
required=True,
@@ -90,17 +315,18 @@ class SetCog(commands.Cog, name="Set command."):
.where(col(CategoryGroup.guild_id) == interaction.guild_id)
.where(col(CategoryGroup.id) == int(group_id))
)
if (group := session.scalar(stmt)) is None:
await interaction.response.send_message("Group or Category was not found.", ephemeral=True)
group = session.scalar(stmt)
if group is None:
await interaction.response.send_message("Group was not found.", ephemeral=True)
return
group.pad_length = length
group.PadLength = length
session.commit()
await interaction.response.send_message(
f"Category pad length set to {length} for {group.name}", ephemeral=True
)
await send_highscore_message(session, interaction.guild, group)
@set.subcommand(
@group.subcommand(
description="Registers the channel for submission messages. Currently this information is deprecated."
)
async def submission_channel(
@@ -125,14 +351,12 @@ class SetCog(commands.Cog, name="Set command."):
with Session(engine) as session:
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 = session.scalar(stmt)
guild.submission_channel = channel_id
session.commit()
await interaction.send(f"Registered <#{channel_id}>.", ephemeral=True)
@set.subcommand(description="Set username length for a Group of Categories.")
@group.subcommand(description="Set username length for a Group of Categories.")
async def username_length(
self,
interaction: nextcord.Interaction,
@@ -157,11 +381,8 @@ class SetCog(commands.Cog, name="Set command."):
.where(col(CategoryGroup.guild_id) == interaction.guild_id)
.where(col(CategoryGroup.id) == int(group_id))
)
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
group = session.scalar(stmt)
group.UsernameLength = length
session.commit()
await interaction.response.send_message(f"Username length set to {length}.", ephemeral=True)
@@ -188,14 +409,34 @@ class SetCog(commands.Cog, name="Set command."):
with Session(engine) as session:
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
guild = session.scalar(stmt)
guild.VerificationChannel = channel_id
session.commit()
await interaction.send(f"Registered <#{channel_id}>.", ephemeral=True)
@group_output_channel.on_autocomplete("group_id")
@category_name.on_autocomplete("category_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")
@username_length.on_autocomplete("group_id")
async def group_autocomplete(self, interaction: nextcord.Interaction, group_id: str) -> None:
+14 -11
View File
@@ -11,7 +11,7 @@ from sqlmodel import col
from cogs.submission.verification_view import VerificationView
from database.db import engine
from database.models import Category, CategoryGroup, Guild, Submission
from main import logger
from main import agon_logger
from utilities.utils import autofill_categories, autofill_groups, submit_score
@@ -64,7 +64,10 @@ class SubmissionCog(commands.Cog):
.where(col(Category.group_id) == int(group_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)
return
@@ -76,13 +79,10 @@ class SubmissionCog(commands.Cog):
return
# 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.
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)
if verified_score is None:
return
# Create submission verification request or submit score if verification not needed.
@@ -92,20 +92,23 @@ class SubmissionCog(commands.Cog):
return
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 := session.scalar(stmt)) is None:
if verification_channel_id is None:
error_response = "No verification channel was registered. Contact server admins."
await interaction.response.send_message(error_response, ephemeral=True)
return
if (channel := interaction.guild.get_channel(verification_channel_id)) is None:
channel = interaction.guild.get_channel(verification_channel_id)
if channel is None:
error_response = "Registered Verification channel does not exist. Contact server admins."
await interaction.response.send_message(error_response, ephemeral=True)
return
if not isinstance(channel, nextcord.TextChannel):
await interaction.response.send_message("An error happened with submission.")
logger.error(
agon_logger.error(
"Channel id %s in guild %s is does not correspond to a TextChannel.", channel.id, channel.guild.id
)
return
+2 -1
View File
@@ -26,9 +26,10 @@ 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 := session.scalar(stmt)) is None:
if guild_data 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.name = guild.name
guild_data.GuildName = guild.name
session.commit()
+3 -1
View File
@@ -39,7 +39,9 @@ class VerificationView(nextcord.ui.View):
.where(col(Submission.message_id) == interaction.message.id)
)
if (submission := session.scalar(stmt)) is None:
submission = session.scalar(stmt)
if submission is None:
await interaction.response.send_message("Could not find submission data in database.", ephemeral=True)
await interaction.message.delete()
return
+4 -9
View File
@@ -8,6 +8,7 @@ import nextcord
from nextcord.ext import commands
from utilities import config
from utilities.logger import agon_logger
nextcord_logger = logging.getLogger("nextcord")
nextcord_logger.setLevel(logging.DEBUG)
@@ -15,18 +16,12 @@ nextcord_handler = logging.FileHandler(filename="nextcord.log", encoding="utf-8"
nextcord_handler.setFormatter(logging.Formatter("%(asctime)s:%(levelname)s:%(name)s: %(message)s"))
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:
"""Main function for bot."""
min_python_version = (3, 12)
if sys.version_info < min_python_version:
logger.error("ERROR: Python version must be >= 3.12")
agon_logger.error("ERROR: Python version must be >= 3.12")
sys.exit()
# Allows privileged intents for monitoring members joining, roles editing, and role assignments
@@ -46,9 +41,9 @@ def main() -> None:
async def on_ready() -> None:
"""When Discord is connected."""
if bot.user is None:
logger.error("Discord connection failed.")
agon_logger.error("Discord connection failed.")
return
logger.info("%s has connected to Discord!", bot.user.name)
agon_logger.info("%s has connected to Discord!", bot.user.name)
# Run Discord bot
bot.run(config.DISCORD_TOKEN)
+10
View File
@@ -0,0 +1,10 @@
"""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:
# Select all groups in guild.
stmt = select(CategoryGroup).where(col(CategoryGroup.guild_id) == guild_id)
stmt = select(CategoryGroup).wherecol((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.message_id = message.id
group.MessageId = 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)