Compare commits

...
106 Commits
Author SHA1 Message Date
alwayscompiling 296e14f0f6 Add docker compose file. 2022-12-04 11:57:23 -05:00
alwayscompiling 4c8444d077 Move config file into utilities. 2022-12-04 11:00:22 -05:00
alwayscompiling 2e2030a533 Move mysql docker stuff into own dir. 2022-12-04 10:56:44 -05:00
alwayscompiling f4ec272567 Update functions for username length being in group. 2022-12-04 10:54:54 -05:00
alwayscompiling b7dd44a028 Change database structure. UsernameLength belong in CategoryGroup. 2022-12-03 20:56:57 -05:00
alwayscompiling 081e406d51 Allow user to update score. 2022-12-03 20:49:17 -05:00
alwayscompiling e3a937649f Fix string formatting for splitting categories into multiple rows. 2022-12-02 22:08:01 -05:00
alwayscompiling d09c9b2b03 Clean up username update verify message on approval. 2022-12-02 20:57:42 -05:00
alwayscompiling 8624285d09 Remove another subquery. Enforce username length. 2022-12-02 20:41:10 -05:00
alwayscompiling f1cbe81699 Missed incorrect subquery. 2022-12-02 18:29:23 -05:00
alwayscompiling aea2e0beb3 Remove subqueries when sending to IN call. 2022-12-02 18:27:50 -05:00
alwayscompiling fdd08bc1a2 bug fixes in from_group and autocomplete user. 2022-12-02 18:17:06 -05:00
alwayscompiling 98dbd30327 Only delete specified user. cat_name is already the string. 2022-12-02 17:49:11 -05:00
alwayscompiling 2565b7689f Correct arg names. 2022-12-02 17:42:59 -05:00
alwayscompiling 52305c0107 Clarify parameter name for users, correct id usage for autofills. 2022-12-02 17:02:44 -05:00
alwayscompiling bed07d440b Merge branch '13-use-a-database' of github.com:alwayscompiling/pvm-highscores into 13-use-a-database 2022-12-02 16:53:51 -05:00
alwayscompiling 4e832a084f Corrected autofill function return structures. 2022-12-02 16:32:22 -05:00
alwayscompiling cc20868bde Corrected autofill function return structures. 2022-12-02 16:08:36 -05:00
alwayscompiling 14602282e9 Clarify optional SlashOptions 2022-12-02 15:28:53 -05:00
alwayscompiling 2e3f298036 Correct score submission value. Parse times correctly. 2022-12-01 23:30:21 -05:00
alwayscompiling 44cb9025b0 Edit embeds to provide bot information for submission. 2022-12-01 23:26:04 -05:00
alwayscompiling 02145ba244 Create new python file for handling sqlalchemy. Removes circular import. 2022-12-01 22:58:39 -05:00
alwayscompiling 58e616ea47 Put pandas group into separate variable to remove warning. 2022-12-01 22:42:43 -05:00
alwayscompiling 2ec7296c0f Bug fixes from testing post_highscore_groups. Command name change. 2022-12-01 20:34:12 -05:00
alwayscompiling d0f328dd2d Variable name changes for clarification and fixing runtime issues. 2022-12-01 20:21:36 -05:00
alwayscompiling 2154d713cb Update username verify view to use new database. 2022-12-01 18:14:23 -05:00
alwayscompiling ea0ebcc2e9 Update data management to use new database. 2022-12-01 18:06:43 -05:00
alwayscompiling b9cc3b46bb Add autofill to get all users in guild. 2022-12-01 17:42:29 -05:00
alwayscompiling 439ad539a8 Updated message management to use new database. 2022-12-01 17:07:50 -05:00
alwayscompiling 10e8e5a446 Update settings management to new database. 2022-12-01 16:40:06 -05:00
alwayscompiling 34e55b962d Change autofill functions to return dicts of name id pairs. 2022-12-01 16:39:32 -05:00
alwayscompiling 1fdb9ff1b6 Update the verification view to use database. 2022-11-30 22:58:27 -05:00
alwayscompiling 4f255a32e2 Change submission to use database. Also use ids rather than names. 2022-11-30 22:36:45 -05:00
alwayscompiling 814baa27d6 Update message listener to new database. 2022-11-30 21:54:05 -05:00
alwayscompiling 9531dca892 Update guild listener to use new database. Add default config function to utils. 2022-11-30 21:45:46 -05:00
alwayscompiling 18df783d86 Update utils to use new database. 2022-11-30 20:54:06 -05:00
alwayscompiling 03d7485037 ignore the todo text file 2022-11-29 22:46:13 -05:00
alwayscompiling 1a55ae0ac5 Move category size to Categories table. 2022-11-29 21:11:44 -05:00
alwayscompiling 76d6ba7969 Change to sqlalchemy and orm scheme. Remove classes due to orm. Edit requirements. 2022-11-27 22:20:29 -05:00
alwayscompiling f95d3130ff Updating requirements. 2022-11-27 09:26:44 -05:00
alwayscompiling 09696071ef Correct guild id property name. 2022-11-27 00:43:24 -05:00
alwayscompiling e5d11d200c Update score to mainly hold sql query information. 2022-11-27 00:35:42 -05:00
alwayscompiling 6a87622b58 Update category class to just hold sql data. 2022-11-27 00:23:15 -05:00
alwayscompiling 9bc88a7e03 Do not need highscores_data class. 2022-11-27 00:07:42 -05:00
alwayscompiling 4c0e6a6209 Change table class to guild class. Change to just store sql table information. 2022-11-27 00:07:15 -05:00
alwayscompiling 017cb7c9df Update Guild class to hold sql Guilds table information. 2022-11-27 00:02:30 -05:00
alwayscompiling 17867671f1 Change order of database columns for easier reading. 2022-11-26 23:49:02 -05:00
alwayscompiling 8896e41726 Revert "Remove all data storage classes."
This reverts commit 77a32b3adc.
2022-11-26 23:20:35 -05:00
alwayscompiling 12a55dd626 Connect to db on bot start. Remove message map. 2022-11-26 21:36:38 -05:00
alwayscompiling 77a32b3adc Remove all data storage classes. 2022-11-26 20:59:44 -05:00
alwayscompiling 2f10a1c7e6 Final changes to correct primary key and uniqueness rules. 2022-11-26 20:52:54 -05:00
alwayscompiling a3bed52479 Remove password in dockerfile, change guild/channel ids to bigint unsigned. 2022-11-26 17:29:04 -05:00
alwayscompiling 6b7dd6042d Add files for docker sql server. 2022-11-25 14:15:55 -05:00
alwayscompiling c4d5477f8b Merge branch '1-add-more-administrative-commands' into develop 2022-11-24 17:55:06 -05:00
alwayscompiling f5342f1541 Fill functionality for delete user from_table and from_category. 2022-11-24 17:47:21 -05:00
alwayscompiling 1e336e290d Catching AttributeError in autofill functions when invalid variables are sent. 2022-11-24 17:45:40 -05:00
alwayscompiling 9ebbca2557 Fill out docstrings. 2022-11-24 17:05:45 -05:00
alwayscompiling 8a2108fc14 Create shell commands and autofill functions. Fix set initializer in utils.autofill_users. 2022-11-22 17:12:26 -05:00
alwayscompiling 03c7a9303a Small rewrite to consolidate autofill_user function. Sort imports in message_management.py. 2022-11-22 17:02:18 -05:00
alwayscompiling a0c5527bf6 Create autofill functions for users in category and in table. Add function in category class to return a list of all users. 2022-11-22 16:26:31 -05:00
alwayscompiling 8155fb0c8f Reorganized commands to facilitate new commands. 2022-11-22 16:01:55 -05:00
alwayscompiling 06b46ef27c Create general autofill table and category util functions. Update table class to have get categories function. 2022-11-17 14:50:48 -05:00
alwayscompiling 1858f0c37d Add refresh table command. 2022-11-16 16:30:34 -05:00
alwayscompiling 792a225987 Correct docstring 2022-11-15 16:17:22 -05:00
alwayscompiling 2e324dc9c5 Use the new category pad length variable. 2022-11-15 15:52:58 -05:00
alwayscompiling 09f7a4323c Correct category pad default. 2022-11-15 15:50:12 -05:00
alwayscompiling dd73e9b472 Add commands to edit username length and category pad. 2022-11-15 15:48:17 -05:00
alwayscompiling b4f5ddc822 Update to nextcord 2.3.2. 2022-11-12 01:16:04 -05:00
alwayscompiling d2f33959ed Create remove_user command. 2022-08-13 21:46:41 -04:00
alwayscompiling 652533d071 Should have handled username updating code in the command, didn't need to create new functions for the command. 2022-08-13 21:26:50 -04:00
alwayscompiling 73e350c845 Save takes guild object, not id. 2022-08-13 20:46:59 -04:00
alwayscompiling 9dec15dbfa Correct table update_username. 2022-08-13 20:45:29 -04:00
alwayscompiling 41ba4dc23c Save data after modified. 2022-08-13 20:40:06 -04:00
alwayscompiling 790cff3406 Clear view on approval. 2022-08-13 20:33:14 -04:00
alwayscompiling 7fb87ae12f Add username setter. 2022-08-13 20:30:21 -04:00
alwayscompiling 2407e61913 Correcting embed access. 2022-08-13 20:30:03 -04:00
alwayscompiling 6147263647 Add UpdateUsernameVerify view to MessageManagementCog. 2022-08-13 20:23:56 -04:00
alwayscompiling cb71cbe1d5 Create update_username functions for category and table. 2022-08-13 20:08:57 -04:00
alwayscompiling 080356fe9c Create Guild.update_username(). 2022-08-13 19:48:29 -04:00
alwayscompiling d4e19284f0 Create view for verifyng username update request. 2022-08-12 23:18:21 -04:00
alwayscompiling ae9153b78e Change submission views to be more informative. 2022-08-12 22:49:54 -04:00
alwayscompiling ca58e9ac26 Create initial update_username command. 2022-08-12 22:39:12 -04:00
alwayscompiling 432888e468 Tag user on submission verify. 2022-08-12 22:28:14 -04:00
alwayscompiling c918b9492b Renamed to data_management. Better description. 2022-08-11 23:02:44 -04:00
alwayscompiling caef6a3652 Created command to clear table messages. 2022-08-11 23:00:11 -04:00
alwayscompiling b2acafd6e7 Changed design to be one class per file. 2022-08-11 22:44:22 -04:00
alwayscompiling 349fdfe18d Renamed python files to more describe their contents. Intend on having more than 1 class per file. 2022-08-11 22:17:52 -04:00
alwayscompiling efed6e001f Moved all commands into single commands cog. 2022-08-11 21:55:06 -04:00
alwayscompiling 6b0e61cdf7 Update nextcord. 2022-08-08 22:12:41 -04:00
alwayscompiling 0acfa77492 Change delete_scores to a slash command. 2022-08-08 22:07:12 -04:00
alwayscompiling ffefd08761 Updated presense to account for slash commands. 2022-08-08 21:54:27 -04:00
alwayscompiling a61662429a Remove help command. Doesn't work with slash commands. 2022-08-08 21:51:25 -04:00
alwayscompiling 57244bcbfe Update post highscores command to slash command. 2022-08-08 21:51:06 -04:00
alwayscompiling 3fd9ff46a3 imports cleanup. 2022-08-08 21:36:45 -04:00
alwayscompiling d4ad727f2a Correcting error statement. 2022-08-06 22:58:43 -04:00
alwayscompiling 516ff7feeb Update minimum python version. Nextcord v2.0 requires python 3.8. 2022-08-06 22:26:55 -04:00
alwayscompiling 0eb332aefd Added verify and submission register subcommands. Removed bot argument from Register class. 2022-08-06 22:21:08 -04:00
alwayscompiling ff07787a9a Create register group and highscore slash subcommand. 2022-08-06 20:14:30 -04:00
alwayscompiling b5959ccc33 Change constructor of score. Create new function to add score value from a string. 2022-08-06 17:32:17 -04:00
alwayscompiling b80f160914 Update verification view to use Table instead of Boss. 2022-08-06 16:55:06 -04:00
alwayscompiling 3b866be3ab Corrected typo in submission_state setter. 2022-08-06 16:45:05 -04:00
alwayscompiling e55b22c821 Easier to read submission access. 2022-08-06 16:41:50 -04:00
alwayscompiling f06e6f1e1e Create function to know if a user has a submission. 2022-08-06 16:30:49 -04:00
alwayscompiling 5f3e7361b4 Convert old submission_objects to new class. 2022-08-06 16:30:31 -04:00
alwayscompiling 835611a11e Created submission classes. 2022-08-06 16:10:19 -04:00
alwayscompiling ee9a23ca79 Renamed data folder to classes. 2022-08-06 14:59:58 -04:00
31 changed files with 1539 additions and 1412 deletions
+3 -1
View File
@@ -134,4 +134,6 @@ nextcord.log
highscores_data.json
highscores_config.json
deleted_backups/
highscores_data/
highscores_data/
TODO.txt
sql_tester.py
+276
View File
@@ -0,0 +1,276 @@
"""
Contains classes and enum for tracking data on submissions.
"""
from enum import Enum
from nextcord import Embed
from sqlalchemy import select
# pylint: disable=import-error
from utilities.sql_alchemy_connection import Category, CategoryGroup, session
class SubmissionState(Enum):
"""
Enum for which state a submission is in
"""
CATEGORY = 1
SCORE = 2
PROOF = 3
NAME = 4
SUBMIT = 5
VERIFY = 6
# TODO remake into a data class
class Submission:
"""
Class for an individual submission. Contains submission information such as
guild id, message id, table, category, username, score and submission state.
"""
def __init__(self, guild_id: int, group_id: int, username: str) -> None:
"""
Default constructor for Submission. Just need to know guild_id, table, and username.
Score and category are defaulted at empty, message id is 0, and submission state
is SubmissionState.CATEGORY.
"""
self._category_id: int = 0
self._guild_id: int = guild_id
self._message_id: int = 0
self._score: str = ""
self._submission_state: SubmissionState = SubmissionState.CATEGORY
self._group_id: int = group_id
self._username: str = username
@property
def category_id(self) -> int:
"""
Getter for Submission's Category id.
Returns:
int: Category id.
"""
return self._category_id
@category_id.setter
def category_id(self, category: int) -> None:
"""
Setter for Submission's Category id.
Args:
category (int): Category id
"""
self._category_id = category
@property
def guild_id(self) -> int:
"""
Getter for guild id Submission is happening in. Will not change.
Returns:
int: guild id
"""
return self._guild_id
@property
def message_id(self) -> int:
"""
Getter for message id that Submission has a view on.
Returns:
int: message id
"""
return self._message_id
@message_id.setter
def message_id(self, m_id: int) -> None:
"""
Setter for message id that Submission has a view on.
Args:
m_id (int): message id
"""
self._message_id = m_id
@property
def score(self) -> str:
"""
Getter for score.
Returns:
str: score
"""
return self._score
@score.setter
def score(self, score: str) -> None:
"""
Setter for score.
Args:
score (str): score
"""
self._score = score
@property
def submission_state(self) -> SubmissionState:
"""
Getter for state Submission is in.
Returns:
SubmissionState: State the Submission is in
"""
return self._submission_state
@submission_state.setter
def submission_state(self, state: SubmissionState) -> None:
"""
Setter for state Submission is in.
Args:
state (SubmissionState): State to change to
"""
self._submission_state = state
@property
def group_id(self) -> str:
"""
Getter for group Submission is for.
Returns:
str: group id
"""
return self._group_id
@group_id.setter
def group_id(self, table: str) -> None:
"""
Getter for table Submission is for.
Args:
table (str): group id
"""
self._group_id = table
@property
def username(self) -> str:
"""
Getter for Submission owner's username.
Returns:
str: Submission owner's username
"""
return self._username
@username.setter
def username(self, username: str) -> None:
"""
Setter for Submission owner's username.
Args:
username (str): Submission owner's username
"""
self._username = username
def get_embeds(self) -> "list[Embed]":
"""Creates an embed based off of the submission state and fields filled out."""
state: SubmissionState = self.submission_state
if state is SubmissionState.CATEGORY:
description = "Step 1: Choose your category!"
elif state is SubmissionState.SCORE:
description = "Step 2: Input your score!"
elif state is SubmissionState.PROOF:
description = "Step 3: Submit your proof!"
elif state is SubmissionState.NAME:
description = "Step 4 (Optional): Change name on submission! (proceed to next if skipping)"
elif state is SubmissionState.SUBMIT:
description = "Step 5: Submit if everything looks correct!"
elif state is SubmissionState.VERIFY:
description = "Review fields and accept."
embed = Embed(title="Highscores Submission", description=description)
embed.add_field(name="User Name", value=self.username, inline=False)
stmt = (
select(CategoryGroup.GroupName)
.where(CategoryGroup.GuildId == self.guild_id)
.where(CategoryGroup.GroupId == self.group_id)
)
embed.add_field(name="Group", value=session.scalar(stmt), inline=False)
if self.category_id:
stmt = (
select(Category.CategoryName)
.where(Category.GroupId == self.group_id)
.where(Category.CategoryId == self.category_id)
)
embed.add_field(name="Category", value=session.scalar(stmt), inline=False)
if self.score:
embed.add_field(name="Score", value=self.score, inline=False)
embeds = [embed]
if state is SubmissionState.VERIFY:
embed = Embed(title="Bot Information", description="Please disregard.")
embed.add_field(name="GroupId", value=self.group_id)
embed.add_field(name="CategoryId", value=self.category_id)
embeds.append(embed)
return embeds
class SubmissionMap:
"""
Tracks all active submission.
"""
def __init__(self) -> None:
"""
Default constructor for SubmissionMap. Created empty map.
"""
self._submission_map: dict = {}
def add_submission(self, user_id: int, submission: Submission) -> None:
"""
Adds new submission to submission map. User id is used as a key, a user should only
have one submission at a time.
Args:
user_id (int): User id of submitter. Unique, used as key.
submission (Submission): Submission object
"""
self._submission_map[user_id] = submission
def get_user_submission(self, user_id: int) -> Submission:
"""
Getter for a user's Submission.
Args:
user_id (int): user id
Returns:
Submission: user's submission
"""
return self._submission_map[user_id]
def has_user(self, user_id: int) -> bool:
"""
Tells if user has an active submission in submission map.
Args:
user_id (int): user id
Returns:
bool: bool indicating if user id is in submission map
"""
return user_id in self._submission_map
def remove_user_submission(self, user_id: int) -> None:
"""
Removes a user's Submission. Use when Submission is complete.
Args:
user_id (int): user id to identify user's Submission
"""
self._submission_map.pop(user_id)
+21
View File
@@ -0,0 +1,21 @@
"""
Cog loader. Adds all command cogs to the bot.
"""
from nextcord.ext import commands
# Command classes
from .data_management import DataManagementCog
from .message_management import MessageManagementCog
from .settings_management import SettingsManagementCog
# This function will be called when this extension is loaded.
# It is necessary to add these functions to the bot.
def setup(bot: commands.Bot):
"""
Adds all command classes to bot.
"""
bot.add_cog(DataManagementCog(bot))
bot.add_cog(MessageManagementCog(bot))
bot.add_cog(SettingsManagementCog())
+241
View File
@@ -0,0 +1,241 @@
"""
Commands for managing the tables/categories/scores for a Guild.
"""
from typing import Optional
import nextcord
from nextcord.ext import commands
from sqlalchemy import distinct, select
# pylint: disable=import-error
from utilities.sql_alchemy_connection import Category, CategoryGroup, Guild, Score, session
from utilities.utils import (
autofill_categories,
autofill_groups,
autofill_users_in_group_category,
autofill_users_in_guild,
send_highscore_message,
)
from .views.update_username_verify import UpdateUsernameVerifyView
class DataManagementCog(commands.Cog, name="Table Management"):
"""
Class holds commands for managing the data storage of the highscores.
"""
def __init__(self, bot: commands.Bot):
"""
Default constructor for DataManagement class. Sets internal bot variable.
"""
self._bot: commands.Bot = bot
@commands.Cog.listener()
async def on_ready(self):
"""
Called when cog is loaded. Adds views required for DataManagement commands.
"""
self._bot.add_view(UpdateUsernameVerifyView())
@nextcord.slash_command(default_member_permissions=nextcord.Permissions(administrator=True))
async def delete(self, interaction: nextcord.Interaction):
"""
Parent slash command for deleting things. Never called.
"""
@delete.subcommand()
async def user(self, interaction: nextcord.Interaction):
"""
Subcommand parent for deleting user information. Never called.
"""
@user.subcommand(
description="Removes all of a user's scores. Warning: data is destroyed, use with case.",
)
async def from_all(
self,
interaction: nextcord.Interaction,
user_name: str = nextcord.SlashOption(
description="User to remove scores for. Note: This is not always a user's discord name.", required=True
),
):
"""
Searches all of guild's scores data for any score from the user. Removes. Updates table
message is user was removed from that table.
Args:
interaction (nextcord.Interaction): Interaction that invoked the command.
user_name (str): User to remove from highscores
"""
stmt = select(distinct(Score.CategoryId)).where(Score.UserName == user_name)
category_ids: "list[int]" = list(session.scalars(stmt))
stmt = select(Score).where(Score.UserName == user_name)
for score in session.scalars(stmt):
session.delete(score)
session.commit()
group_ids = select(Category.GroupId).where(Category.CategoryId.in_(category_ids))
stmt = select(CategoryGroup).where(CategoryGroup.GroupId.in_(group_ids))
for group in session.scalars(stmt):
await send_highscore_message(interaction.guild.get_channel(group.ChannelId), group)
await interaction.send(f'User "{user_name}" was removed from all tables.', ephemeral=True)
@user.subcommand(description="Deletes a user's scores from a category.")
async def from_category(
self,
interaction: nextcord.Interaction,
group_id: str = nextcord.SlashOption(name="group", description="Group to delete user from.", required=True),
category_id: str = nextcord.SlashOption(
name="category", description="Category to delete user from.", required=True
),
user_name: str = nextcord.SlashOption(description="User to delete.", required=True),
):
"""
Removes a user from a category. Uses autofill functions to fill out group_id, category_id, and user_name.
Args:
interaction (nextcord.Interaction): Interaction that invoked the command.
group_id (str): String representation of Group 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.
"""
stmt = select(CategoryGroup).where(CategoryGroup.GroupId == int(group_id))
group = session.scalar(stmt)
stmt = select(Category.CategoryName).where(Category.CategoryId == int(category_id))
category_name = session.scalar(stmt)
stmt = select(Score).where(Score.CategoryId == int(category_id)).where(Score.UserName == user_name)
# there should only be one because category id and username are a composite primary key in score
session.delete(session.scalar(stmt))
session.commit()
await send_highscore_message(interaction.guild.get_channel(group.ChannelId), group)
await interaction.response.send_message(f'User "{user_name}" removed from "{category_name}".', ephemeral=True)
@user.subcommand(description="Deletes a user's scores from entire group.")
async def from_group(
self,
interaction: nextcord.Interaction,
group_id: str = nextcord.SlashOption(name="group", description="Group to delete user from.", required=True),
user_name: str = nextcord.SlashOption(description="User to delete.", required=True),
):
"""
Removes user from a goup. Uses autofill functions to fill out group_id and user_name.
Args:
interaction (nextcord.Interaction): Interaction that invoked the command.
group_id (str): String representation of Group id to remove user from. Is SlashOption.
user_name (str): The user to remove. Is SlashOption.
"""
stmt = select(CategoryGroup).where(CategoryGroup.GroupId == int(group_id))
group = session.scalar(stmt)
category_ids = select(Category.CategoryId).where(Category.GroupId == int(group_id))
stmt = select(Score).where(Score.CategoryId.in_(category_ids)).where(Score.UserName == user_name)
for score in session.scalars(stmt):
session.delete(score)
session.commit()
await send_highscore_message(interaction.guild.get_channel(group.ChannelId), group)
await interaction.response.send_message(
f'User "{user_name}" removed from all categories in {group.GroupName}.', ephemeral=True
)
@from_category.on_autocomplete("group_id")
@from_group.on_autocomplete("group_id")
async def autocomplete_group(self, interaction: nextcord.Interaction, group_id: str):
"""
Autocomplete function for table names.
Args:
interaction (nextcord.Interaction): Interaction that invoked the command.
group_id (str): String to be autocompleted with Group name.
"""
await interaction.response.send_autocomplete(autofill_groups(interaction.guild_id, group_id))
@from_category.on_autocomplete("category_id")
async def autocomplete_category(self, interaction: nextcord.Interaction, category_id: str, group_id: str):
"""
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)
)
@from_category.on_autocomplete("user_name")
@from_group.on_autocomplete("user_name")
async def autocomplete_user(
self, interaction: nextcord.Interaction, user_name: str, group_id: str, category_id: Optional[str] = None
):
"""
Autocomplete functions for user names.
Args:
interaction (nextcord.Interaction): Interaction that invoked the command.
user_name (str): String to be autocompleted with User name.
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.
"""
await interaction.response.send_autocomplete(
autofill_users_in_group_category(
interaction.guild_id, int(group_id), category_id if category_id else 0, user_name
)
)
@nextcord.slash_command(description="Request a username update. Admins must approve.")
async def update_username(
self,
interaction: nextcord.Interaction,
old_username: str = nextcord.SlashOption(description="Username currently stored in highscores.", required=True),
new_username: str = nextcord.SlashOption(description="Username to update to.", required=True),
):
"""
Creates a request to send to admins for changing a username. Uses
UpdateUsernameVerifyView to do the work.
Args:
interaction (nextcord.Interaction): Interaction that invoked the command
old_username (str): Username to replace.
new_username (str): Username to replace old with.
"""
stmt = select(Guild.VerificationChannel).where(Guild.GuildId == interaction.guild_id)
verify_channel = interaction.guild.get_channel(session.scalar(stmt))
embed: nextcord.Embed = nextcord.Embed(title="Username change request.")
embed.add_field(name="Old Username", value=old_username)
embed.add_field(name="New Username", value=new_username)
await verify_channel.send(
content=f"Username change request from <@{interaction.user.id}>.",
embed=embed,
view=UpdateUsernameVerifyView(),
)
await interaction.send("Request sent to admins for verification.", ephemeral=True)
@update_username.on_autocomplete("old_username")
@from_all.on_autocomplete("user_name")
async def username_autofill(self, interaction: nextcord.Interaction, username: str):
"""
Sends response from autofill_users_in_guild back to update_username for autocompletion of old_username.
Args:
interaction (nextcord.Interaction): Interaction with command.
old_username (str): String to run autocompletion against.
"""
await interaction.response.send_autocomplete(autofill_users_in_guild(interaction.guild_id, username))
+152
View File
@@ -0,0 +1,152 @@
"""
Commands for managing/displaying the highscores messages.
"""
from typing import Optional
import nextcord
from nextcord.ext import commands
from sqlalchemy import select
# pylint: disable=import-error
from cogs.submission.submission_views import SubmissionCreateButton
from utilities.sql_alchemy_connection import CategoryGroup, session
from utilities.utils import autofill_groups, send_highscore_message
class MessageManagementCog(commands.Cog, name="Message Management"):
"""
Class holds commands for managing the messages that display highscores.
"""
def __init__(self, bot: commands.Bot) -> None:
"""
Default constructor for MessageManagement class
Args:
bot (commands.Bot): Bot object. Required to register submission button view.
"""
self.bot = bot
@nextcord.slash_command(
description="Clears highscores channel of all table messages. Does not change data.",
default_member_permissions=nextcord.Permissions(administrator=True),
)
async def delete_channel_messages(
self,
interaction: nextcord.Interaction,
channel: Optional[nextcord.abc.GuildChannel] = nextcord.SlashOption(
description="Channel to clear of messages.",
required=False,
channel_types=[nextcord.ChannelType.text],
default=None,
),
) -> None:
"""
Deletes all group messages from channel. Does not change data.
Does not delete highscores channel id. Uses current channel if not supplied a channel.
Args:
interaction (nextcord.Interaction): Interaction that invoked the command
channel (Optional[nextcord.abc.GuildChannel]): Channel to clear of messages.
Is a SlashOption. Limited to text channels. Defaults to None.
"""
if not channel:
channel = interaction.channel
stmt = (
select(CategoryGroup)
.where(CategoryGroup.GuildId == interaction.guild_id)
.where(CategoryGroup.ChannelId == channel.id)
)
# TODO test when nothing in channel
groups = session.scalars(stmt)
await interaction.response.defer(ephemeral=True)
for group in groups:
try:
table_message: nextcord.Message = await channel.fetch_message(group.MessageId)
await table_message.delete()
except nextcord.NotFound:
# silently ignore that messages are not found. They got deleted already.
pass
await interaction.send("Table messages deleted.", ephemeral=True)
@nextcord.slash_command(
description="Triggers post/update of highscores to registered highscore channel.",
default_member_permissions=nextcord.Permissions(administrator=True),
)
async def post_highscore_groups(self, interaction: nextcord.Interaction) -> None:
"""
A command which posts the highscore in the registered channel.
Sends warning if no channel has been registered.
Either creates new messages for each table if one doesn't exist
or edits the existing message.
Args:
interaction (nextcord.Interaction): Interaction that invoked the command.
"""
stmt = select(CategoryGroup).where(CategoryGroup.GuildId == interaction.guild_id)
# awaiting response because posting scores takes a while
await interaction.response.defer(ephemeral=True)
for group in session.scalars(stmt):
channel = interaction.guild.get_channel(group.ChannelId)
# Error checking for registered highscore channel
error_response = "Registered Highscores channel does not exist or was never registered. "
error_response += 'Use "/register" to register channels.'
assert channel is not None, await interaction.response(error_response, ephemeral=True)
message: nextcord.Message = await send_highscore_message(channel, group)
await message.edit(view=SubmissionCreateButton(self.bot))
await interaction.send(f"Posted all groups in {interaction.guild.name}'s highscores.", ephemeral=True)
@nextcord.slash_command(
description="Manually refresh a table. Use when changing settings or data manually changed.",
default_member_permissions=nextcord.Permissions(administrator=True),
)
async def refresh_group(
self,
interaction: nextcord.Interaction,
group_id: str = nextcord.SlashOption(name="group", description="Name of Group.", required=True),
) -> None:
"""
Refreshes a table given by user.
Args:
interaction (nextcord.Interaction): Interaction that invoked the command.
group_id (str): Str representation of Group id to refresh. Is SlashOption.
"""
stmt = (
select(CategoryGroup)
.where(CategoryGroup.GuildId == interaction.guild_id)
.where(CategoryGroup.GroupId == int(group_id))
)
group = session.scalar(stmt)
channel = interaction.guild.get_channel(group.ChannelId)
message: nextcord.Message = await send_highscore_message(channel, group)
await message.edit(view=SubmissionCreateButton(self.bot))
await interaction.response.send_message(f"Updated table {group.GroupName}!", ephemeral=True)
@refresh_group.on_autocomplete("group_id")
async def refresh_table_autofill(self, interaction: nextcord.Interaction, group_id: str):
"""
Collects list of table names and compares to input table name for autocomplete assistance.
Args:
interaction (nextcord.Interaction): Interaction with command.
group_id (str): String to run autocomplete compared to.
"""
await interaction.response.send_autocomplete(autofill_groups(interaction.guild_id, group_id))
+200
View File
@@ -0,0 +1,200 @@
"""
Commands for managing the Guild's settings.
"""
from typing import Optional
import nextcord
from nextcord.ext import commands
from sqlalchemy import select
# pylint: disable=import-error
from utilities.sql_alchemy_connection import CategoryGroup, Guild, session
from utilities.utils import autofill_groups
class SettingsManagementCog(commands.Cog, name="Settings Management"):
"""
Class holds commands for managing Guild's settings.
"""
def __init__(self) -> None:
"""
Default constructor for SettingsManagement class. Does nothing.
"""
@nextcord.slash_command(default_member_permissions=nextcord.Permissions(administrator=True))
async def register(self, interaction: nextcord.Interaction):
"""
Parent slash command, never called.
"""
@register.subcommand(description="Registers the channel for a Category Group's messages.")
async def group_output_channel(
self,
interaction: nextcord.Interaction,
group_id: str = nextcord.SlashOption(
name="group", description="Category Group to assign to channel.", required=True
),
channel: Optional[nextcord.abc.GuildChannel] = nextcord.SlashOption(
description="Channel to register",
required=False,
channel_types=[nextcord.ChannelType.text],
default=None,
),
) -> None:
"""
Command function to register a CategoryGroup's output channel.
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
Is a SlashOption. Limited to text channels, Defaults to None.
"""
if not channel:
channel = interaction.channel
channel_id = channel.id
stmt = (
select(CategoryGroup)
.where(CategoryGroup.GuildId == interaction.guild_id)
.where(CategoryGroup.GroupId == int(group_id))
)
group = session.scalar(stmt)
group.ChannelId = channel_id
session.commit()
await interaction.send(f"Registered <#{channel_id}>.", ephemeral=True)
@register.subcommand(description="Registers the channel for submission messages.")
async def submission(
self,
interaction: nextcord.Interaction,
channel: Optional[nextcord.abc.GuildChannel] = nextcord.SlashOption(
description="Channel to register",
required=False,
channel_types=[nextcord.ChannelType.text],
default=None,
),
) -> None:
"""
Command function to register the submission message channel.
Args:
interaction (nextcord.Interaction): Interaction that invoked the command
channel Optional[nextcord.abc.GuildChannel], optional): Channel to register
Is a SlashOption. Limited to text channels, Defaults to None.
"""
if not channel:
channel = interaction.channel
channel_id = channel.id
stmt = select(Guild).where(Guild.GuildId == interaction.guild_id)
guild = session.scalar(stmt)
guild.SubmissionChannel = channel_id
session.commit()
await interaction.send(f"Registered <#{channel_id}>.", ephemeral=True)
@register.subcommand(description="Registers the channel for verification messages.")
async def verify(
self,
interaction: nextcord.Interaction,
channel: Optional[nextcord.abc.GuildChannel] = nextcord.SlashOption(
description="Channel to register",
required=False,
channel_types=[nextcord.ChannelType.text],
default=None,
),
) -> None:
"""
Command function to register the verification message channel.
Args:
interaction (nextcord.Interaction): Interaction that invoked the command
channel (Optional[nextcord.abc.GuildChannel]): Channel to register
Is a SlashOption. Limited to text channels, Defaults to None.
"""
if not channel:
channel = interaction.channel
channel_id = channel.id
stmt = select(Guild).where(Guild.GuildId == interaction.guild_id)
guild = session.scalar(stmt)
guild.VerificationChannel = channel_id
session.commit()
await interaction.send(f"Registered <#{channel_id}>.", ephemeral=True)
@nextcord.slash_command(default_member_permissions=nextcord.Permissions(administrator=True))
async def set(self, interaction: nextcord.Interaction):
"""
Parent slash command for setting settings. Never called.
"""
@set.subcommand(
description="Set character length for a category. Can cut off username/score. Affects all categories."
)
async def category_pad_length(
self,
interaction: nextcord.Interaction,
group_id: str = nextcord.SlashOption(
name="group", description="Category Group to assign pad length to.", required=True
),
length: int = nextcord.SlashOption(description="Number of characters.", required=True, default=26),
) -> None:
"""
Changes guild's category pad length.
Args:
interaction (nextcord.Interaction): Interaction that invoked the command
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.
"""
stmt = (
select(CategoryGroup)
.where(CategoryGroup.GuildId == interaction.guild_id)
.where(CategoryGroup.GroupId == int(group_id))
)
group = session.scalar(stmt)
group.PadLength = length
session.commit()
await interaction.response.send_message(
f"Category pad length set to {length} for {group.GroupName}", ephemeral=True
)
@set.subcommand(description="Set username length for a Group of Categories.")
async def username_length(
self,
interaction: nextcord.Interaction,
group_id: str = nextcord.SlashOption(
name="group", description="Group to change username length of.", required=True
),
length: int = nextcord.SlashOption(description="Number of characters.", required=True, default=12),
) -> None:
"""
Changes guild's username length limit.
Args:
interaction (nextcord.Interaction): Interaction that invoked the command
length (int): Number of characters. Is SlashOption. Defaults to 12.
"""
stmt = (
select(CategoryGroup)
.where(CategoryGroup.GuildId == interaction.guild_id)
.where(CategoryGroup.GroupId == int(group_id))
)
group = session.scalar(stmt)
group.UsernameLength = length
session.commit()
await interaction.response.send_message(f"Username length set to {length}.", ephemeral=True)
@group_output_channel.on_autocomplete("group_id")
@category_pad_length.on_autocomplete("group_id")
@username_length.on_autocomplete("group_id")
async def group_id_autocomplete(self, interaction: nextcord.Interaction, group_id: str):
"""
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))
@@ -0,0 +1,74 @@
"""View for allowing a username update."""
import nextcord
from sqlalchemy import distinct, exc, select
# pylint: disable=import-error
from utilities.sql_alchemy_connection import Category, CategoryGroup, Score, session
from utilities.utils import send_highscore_message
class UpdateUsernameVerifyView(nextcord.ui.View):
"""Defines view for allowing a username update."""
def __init__(self):
"""
Default constructor for UpdateUsernameVerifyView. Calls super constructor with no timeout.
"""
super().__init__(timeout=None)
@nextcord.ui.button(label="Deny", style=nextcord.ButtonStyle.red, custom_id="deny-username-update")
async def deny_button(
self, button: nextcord.ui.Button, interaction: nextcord.Interaction
): # pylint: disable=unused-argument
"""
Button for denying username update. Deletes message.
Args:
button (nextcord.ui.Button): unused
interaction (nextcord.Interaction): Interaction with the button
"""
await interaction.message.delete()
@nextcord.ui.button(label="Approve", style=nextcord.ButtonStyle.green, custom_id="approve-username-update")
async def approve_button(
self, button: nextcord.ui.Button, interaction: nextcord.Interaction
): # pylint: disable=unused-argument
"""
Button for approving username update. Tells guild data to update username, then updates
message to show it is complete.
Args:
button (nextcord.ui.Button): unused
interaction (nextcord.Interaction): Interaction with the button
"""
try:
embed_content = interaction.message.embeds[0].to_dict()
old_username: str = embed_content["fields"][0]["value"]
new_username: str = embed_content["fields"][1]["value"]
stmt = select(distinct(Score.CategoryId)).where(Score.UserName == old_username)
category_ids: "list[int]" = list(session.scalars(stmt))
stmt = select(Score).where(Score.UserName == old_username)
for score in session.scalars(stmt):
score.UserName = new_username
session.commit()
group_ids = select(Category.GroupId).where(Category.CategoryId.in_(category_ids))
stmt = select(CategoryGroup).where(CategoryGroup.GroupId.in_(group_ids))
for group in session.scalars(stmt):
await send_highscore_message(interaction.guild.get_channel(group.ChannelId), group)
content = interaction.message.content
content += f"\nApproved by <@{interaction.user.id}>."
await interaction.message.edit(content=content, embeds=interaction.message.embeds, view=None)
except exc.DataError:
session.rollback()
content = interaction.message.content
content += "\nError, UserName is too long."
await interaction.message.edit(content=content, embeds=interaction.message.embeds, view=None)
await interaction.response.send_message(content="User must resubmit with shorter name.", ephemeral=True)
-42
View File
@@ -1,42 +0,0 @@
"""
Bot command for deleting the guild's scores.
"""
from nextcord import Embed
from nextcord.ext import commands
from highscores import highscores_data # pylint: disable=import-error
class DeleteScores(commands.Cog, name="Delete Scores"):
"""
Warning: This command only works for developers.
Deletes this guild's scores.
"""
def __init__(self, bot: commands.Bot):
self.bot = bot
@commands.command(name="delete-scores")
@commands.has_permissions(administrator=True)
async def delete_scores(self, ctx: commands.Context):
"""A command which deletes all scores for this guild.
Usage:
```
?delete_scores
```
"""
# only allow AlwaysCompiling to use.
# If you are a developer and want to use command, can add name to list
if ctx.author.id == 206602713511886849:
for table in highscores_data.get_guild(ctx.guild.id).tables:
for category in table.categories:
category.clear()
highscores_data.save_guild_data(highscores_data.get_guild(ctx.guild.id))
# This function will be called when this extension is loaded.
# It is necessary to add these functions to the bot.
def setup(bot: commands.Bot):
"""Adds function to bot"""
bot.add_cog(DeleteScores(bot))
+8 -11
View File
@@ -2,10 +2,12 @@
Defines cog for guild listeners. Contains functions for various guild events.
"""
from nextcord.ext import commands
import nextcord
from nextcord.ext import commands
from highscores import highscores_data # pylint: disable=import-error
# pylint: disable=import-error
from utilities.sql_alchemy_connection import Guild, session
from utilities.utils import add_preset_configuration
class GuildListener(commands.Cog, name="Guild Listener"):
@@ -17,10 +19,11 @@ class GuildListener(commands.Cog, name="Guild Listener"):
@commands.Cog.listener()
async def on_guild_join(self, guild: nextcord.Guild):
"""Listener for Guild Join. Creates skeleton for new guild joined and saves it."""
highscores_data.add_guild(guild.id)
new_guild = Guild(GuildId=guild.id)
session.add(new_guild)
session.commit()
# This add config will be removed eventually. Config adding will have to be controlled by commands.
highscores_data.add_configuration(guild.id, "runescape")
highscores_data.save_guild_data(highscores_data.get_guild(guild.id))
add_preset_configuration(guild.id, "runescape")
welcome_message: str = "Hello, I keep track of Highscores for you!"
welcome_message += " Currently only supporting RuneScape boss highscores,"
@@ -29,12 +32,6 @@ class GuildListener(commands.Cog, name="Guild Listener"):
await guild.system_channel.send(content=welcome_message)
@commands.Cog.listener()
async def on_guild_remove(self, guild: nextcord.Guild):
"""Listener for Guild Remove. Removes data from highscores_data."""
print(f"Removing data for guild {guild.name} (id {guild.id}).")
highscores_data.remove_guild(guild.id)
# This function will be called when this extension is loaded.
# It is necessary to add these functions to the bot.
-26
View File
@@ -1,26 +0,0 @@
"""
Help command
"""
from nextcord.ext import commands
from .help_command import NewHelpCommand
class HelpCog(commands.Cog):
"""Help Command"""
def __init__(self, bot: commands.Bot):
self.bot = bot
self._original_help_command = bot.help_command
bot.help_command = NewHelpCommand()
bot.help_command.cog = self
def cog_unload(self):
self.bot.help_command = self._original_help_command
# setup functions for bot
def setup(bot: commands.Bot):
"""Adds function to bot"""
bot.add_cog(HelpCog(bot))
-84
View File
@@ -1,84 +0,0 @@
"""
Help command. Copied from discord-bot-template.
"""
import nextcord
from nextcord.embeds import EmptyEmbed
from nextcord.ext import commands
import config # pylint: disable=import-error
class NewHelpCommand(commands.MinimalHelpCommand):
"""Custom help command override using embeds"""
COLOUR = nextcord.Colour.blurple()
def get_ending_note(self):
"""Returns note to display at the bottom"""
invoked_with = self.invoked_with
return f"Use {config.BOT_PREFIX}{invoked_with} [command] for more info on a command."
def get_command_signature(self, command: commands.core.Command):
"""Retrieves the signature portion of the help page."""
return f"{command.qualified_name} {command.signature}"
async def send_bot_help(self, mapping: dict):
"""implements bot command help page"""
embed = nextcord.Embed(title="Bot Commands", colour=self.COLOUR)
avatar = self.context.bot.user.avatar
avatar_url = avatar.url if avatar else EmptyEmbed
embed.set_author(name=self.context.bot.user.name, icon_url=avatar_url)
description = self.context.bot.description
if description:
embed.description = description
for cog, commands in mapping.items(): # pylint: disable=redefined-outer-name
name = "No Category" if cog is None else cog.qualified_name
filtered = await self.filter_commands(commands, sort=True)
if filtered:
# \u2002 = middle dot
value = "\u2002".join(f"{config.BOT_PREFIX}{c.name}" for c in filtered)
if cog and cog.description:
value = f"{cog.description}\n{value}"
embed.add_field(name=name, value=value)
embed.set_footer(text=self.get_ending_note())
await self.get_destination().send(embed=embed)
async def send_cog_help(self, cog: commands.Cog):
"""implements cog help page"""
embed = nextcord.Embed(title=f"{cog.qualified_name} Commands", colour=self.COLOUR)
if cog.description:
embed.description = cog.description
filtered = await self.filter_commands(cog.get_commands(), sort=True)
for command in filtered:
embed.add_field(
name=self.get_command_signature(command),
value=command.short_doc or "...",
inline=False,
)
embed.set_footer(text=self.get_ending_note())
await self.get_destination().send(embed=embed)
async def send_group_help(self, group: commands.Group):
"""implements group help page and command help page"""
embed = nextcord.Embed(title=group.qualified_name, colour=self.COLOUR)
if group.help:
embed.description = group.help
if isinstance(group, commands.Group):
filtered = await self.filter_commands(group.commands, sort=True)
for command in filtered:
embed.add_field(
name=self.get_command_signature(command),
value=command.short_doc or "...",
inline=False,
)
embed.set_footer(text=self.get_ending_note())
await self.get_destination().send(embed=embed)
# Use the same function as group help for command help
send_command_help = send_group_help # type: ignore
+29 -45
View File
@@ -2,18 +2,17 @@
Defines cog for on message listener.
"""
from datetime import date
import random
import re
from nextcord.ext import commands
import nextcord
from nextcord.ext import commands
from sqlalchemy import select
from sqlalchemy.orm import aliased
from cogs.submission.submission_views import ( # pylint: disable=import-error
SubmissionState,
get_submission_embed,
)
from highscores import highscores_data, submission_objects # pylint: disable=import-error
# pylint: disable=import-error
from classes.submission import SubmissionState
from highscores import submission_objects
from utilities.sql_alchemy_connection import Category, CategoryGroup, Guild, session
class SubmissionListener(commands.Cog, name="Message Listener"):
@@ -31,48 +30,36 @@ class SubmissionListener(commands.Cog, name="Message Listener"):
guild_id = message.guild.id
# troll addition for friends birthday
if guild_id == 239758371358638080 and date.today() == date(2022, 7, 31):
# 5% chance to send the message
chance = random.randint(1, 20)
if chance == 1:
await message.channel.send("Happy Birthday Chris <@642330132958281750>")
if message.author.id in submission_objects:
message_id = submission_objects[message.author.id]["message_id"]
submission_channel_id = highscores_data.get_guild(guild_id).submission_channel
if submission_objects.has_user(message.author.id):
submission = submission_objects.get_user_submission(message.author.id)
message_id = submission.message_id
stmt = select(Guild.SubmissionChannel).where(Guild.GuildId == guild_id)
submission_channel_id = session.scalar(stmt)
submission_message: nextcord.Message = (
await self.bot.get_guild(guild_id)
.get_channel(submission_channel_id)
.fetch_message(message_id)
await self.bot.get_guild(guild_id).get_channel(submission_channel_id).fetch_message(message_id)
)
# need check if correct user and correct channel
if message.channel.id == submission_channel_id:
state = submission_objects[message.author.id]["submission_state"]
state = submission.submission_state
if state == SubmissionState.SCORE and message.content:
# check with regex if score matches correctly
is_time_record = (
highscores_data.get_guild(guild_id)
.get_table(submission_objects[message.author.id]["boss_name"])
.get_category(submission_objects[message.author.id]["category"])
.time_record
)
stmt = select(Category.IsTimeRecord).where(Category.CategoryId == submission.category_id)
time_regex = re.search(
"^([0-9]{1,2}:[0-9]|[0-9])?[0-9](\\.[0-9])?$", message.content
)
# check with regex if score matches correctly
is_time_record = session.scalar(stmt)
time_regex = re.search("^([0-9]{1,2}:[0-9]|[0-9])?[0-9](\\.[0-9])?$", message.content)
int_regex = re.search("^[0-9]+$", message.content)
if time_regex is None and is_time_record:
embed = submission_message.embeds[0]
embed.add_field(
embeds = submission_message.embeds[0]
embeds.add_field(
name="Error",
value="Time record must follow format MM:SS.ms, SS.ms, or SS.",
)
elif int_regex is None and not is_time_record:
embed = submission_message.embeds[0]
embed.add_field(name="Error", value="Number record must only be number.")
embeds = submission_message.embeds[0]
embeds.add_field(name="Error", value="Number record must only be number.")
else:
score = message.content
if is_time_record:
@@ -89,23 +76,20 @@ class SubmissionListener(commands.Cog, name="Message Listener"):
elif re.search("^[0]?[0-9]:[0-9]{2}(\\.[0-9])?$", score):
# if there is a minute 10ths place 0, remove
score = score.lstrip("0")
submission_objects[message.author.id]["score"] = score
embed = get_submission_embed(message.author.id)
submission.score = score
embeds = submission.get_embeds()
await submission_message.edit(
embed=embed,
embeds=embeds,
)
elif state == SubmissionState.PROOF and len(message.attachments) > 0:
await submission_message.edit(
files=[await attch.to_file() for attch in message.attachments],
)
elif state == SubmissionState.NAME and message.content:
submission_objects[message.author.id]["username"] = message.content[
: highscores_data.get_guild(guild_id).username_length
]
embed = get_submission_embed(message.author.id)
submission.username = message.content
embeds = submission.get_embeds()
await submission_message.edit(
embed=embed,
embeds=embeds,
)
await message.delete()
-57
View File
@@ -1,57 +0,0 @@
"""
Bot command to post the highscores
"""
from nextcord.ext import commands
import nextcord
from cogs.submission.submission_views import SubmissionCreateButton # pylint: disable=import-error
from highscores import highscores_data # pylint: disable=import-error
from utilities.utils import send_highscore_message # pylint: disable=import-error
class PostHighscores(commands.Cog, name="Post Highscores"):
"""Posts Highscores."""
def __init__(self, bot: commands.Bot):
self.bot = bot
@commands.command(name="post-scores")
@commands.has_permissions(administrator=True)
async def post_highscores(self, ctx: commands.Context):
"""
A command which posts the highscore in the designated channel.
Sends warning if no channel has been registered.
Usage:
```
?post-scores
```
"""
guild = highscores_data.get_guild(ctx.guild.id)
print(f"Printing out highscores information in channel {guild.highscore_channel}")
channel = ctx.guild.get_channel(guild.highscore_channel)
error_response = "Registered Highscores channel does not exist or was never registered. "
error_response += 'Register with "?register" command.'
assert channel is not None, await ctx.send(error_response)
if len(highscores_data.get_guild(ctx.guild.id).tables) < 1:
await channel.send(
"No tables have been created for this guild. Please create some tables."
)
return
for table in highscores_data.get_guild(ctx.guild.id).tables:
message: nextcord.Message = await send_highscore_message(channel, guild, table)
await message.edit(view=SubmissionCreateButton(self.bot))
# have to save data here in case message id was changed. rather save here after all
# messages have been sent.
highscores_data.save_guild_data(guild)
# This function will be called when this extension is loaded.
# It is necessary to add these functions to the bot.
def setup(bot: commands.Bot):
"""Adds function to bot"""
bot.add_cog(PostHighscores(bot))
+8 -6
View File
@@ -4,9 +4,8 @@ Functions for controlling the bot's presence.
import random
from nextcord.ext import commands
from nextcord.ext import tasks
import nextcord
from nextcord.ext import commands, tasks
class BotPresence(commands.Cog, name="Bot Presence"):
@@ -35,14 +34,17 @@ class BotPresence(commands.Cog, name="Bot Presence"):
async def change_presence(self):
"""Randomly chooses a presence and updates bot's presence."""
statuses = [
f"?help | {len(self.bot.guilds)} servers",
f"Vuln Pls. | {len(self.bot.guilds)} servers",
f"All your base are belong to us. | {len(self.bot.guilds)} servers",
f"Double plus good. | {len(self.bot.guilds)} servers",
"Now with slash commands!",
"Vuln Pls.",
"All your base are belong to us.",
"Double plus good.",
"That's a lot of damage!",
]
status = random.choice(statuses)
status += f" | {len(self.bot.guilds)} servers"
await self.bot.change_presence(activity=nextcord.Game(status))
-69
View File
@@ -1,69 +0,0 @@
"""
Bot command for registering various channels to the bot.
"""
from nextcord import Embed
from nextcord.ext import commands
from highscores import highscores_data # pylint: disable=import-error
class RegisterVerificationChannel(commands.Cog, name="Register"):
"""Bot command for registering various channels to the bot."""
def __init__(self, bot: commands.Bot) -> None:
self.bot = bot
@commands.command(name="register")
@commands.has_permissions(administrator=True)
async def register_highscores(self, ctx: commands.Context, registration_type: str) -> None:
"""A command which registers current channel as for various uses.
Usage:
```
?register verify (Use in channel you want to verify submissions in.)
?register highscore (Use in channel you want to show highscore messages in.)
?register submission (Use in channel you want submission messages to be send it.)
?register help (Shows help message.)
```
"""
# respond to the message
channel_id = ctx.channel.id
guild = highscores_data.get_guild(ctx.guild.id)
if registration_type == "verify":
await ctx.send(f"Registered <#{channel_id}> as channel to post verifications in.")
guild.verification_channel = channel_id
elif registration_type == "highscore":
await ctx.send(f"Registered <#{channel_id}> as channel to post highscores in.")
guild.highscore_channel = channel_id
elif registration_type == "submission":
await ctx.send(f"Registered <#{channel_id}> as channel to post submissions in.")
guild.submission_channel = channel_id
else:
embed = Embed(
title="Register command help", description="How to use the register command."
)
embed.add_field(
name="?register verify", value="Use in channel you want to verify submissions in."
)
embed.add_field(
name="?register highscore",
value="Use in channel you want to show highscore messages in.",
)
embed.add_field(
name="?register submission",
value="Use in channel you want submission messages to be send it.",
)
embed.add_field(name="?register help", value="Use to show this message.")
await ctx.reply(embed=embed)
# return, don't need to save data
return
highscores_data.save_guild_data(guild)
# This function will be called when this extension is loaded.
# It is necessary to add these functions to the bot.
def setup(bot: commands.Bot) -> None:
"""Adds function to bot"""
bot.add_cog(RegisterVerificationChannel(bot))
+63 -110
View File
@@ -1,65 +1,29 @@
"""Views used for submission process."""
from typing import Optional
from enum import Enum
import nextcord
from nextcord import Embed
from nextcord.ext import commands
from sqlalchemy import select
from sqlalchemy.orm import aliased
# pylint: disable=import-error
from classes.submission import Submission, SubmissionState
from utilities.sql_alchemy_connection import Category, Guild, CategoryGroup, session
from highscores import submission_objects
from highscores import ( # pylint: disable=import-error
highscores_data,
highscores_message_map,
submission_objects,
)
from .verification_view import VerificationView
class SubmissionState(Enum):
"""Enum for which state a submission is in"""
CATEGORY = 1
SCORE = 2
PROOF = 3
NAME = 4
SUBMIT = 5
VERIFY = 6
def get_submission_embed(userid: int) -> nextcord.Embed:
"""Creates an embed based off of the submission state and fields filled out."""
usersubmission: dict = submission_objects[userid]
state: SubmissionState = usersubmission["submission_state"]
if state is SubmissionState.CATEGORY:
description = "Step 1: Choose your category!"
elif state is SubmissionState.SCORE:
description = "Step 2: Input your score!"
elif state is SubmissionState.PROOF:
description = "Step 3: Submit your proof!"
elif state is SubmissionState.NAME:
description = "Step 4 (Optional): Change name on submission! (proceed to next if skipping)"
elif state is SubmissionState.SUBMIT:
description = "Step 5: Submit if everything looks correct!"
elif state is SubmissionState.VERIFY:
description = "Review fields and accept."
embed = Embed(title="Highscores Submission", description=description)
embed.add_field(name="User Name", value=usersubmission["username"], inline=False)
embed.add_field(name="Boss", value=usersubmission["boss_name"], inline=False)
if usersubmission["category"]:
embed.add_field(name="Category", value=usersubmission["category"], inline=False)
if usersubmission["score"]:
embed.add_field(name="Score", value=usersubmission["score"], inline=False)
return embed
def category_select_options(guild_id: int, boss_name: str) -> "list[nextcord.SelectOption]":
def category_select_options(group_id: int) -> "list[nextcord.SelectOption]":
"""Returns list of categories for a given boss."""
options: "list[nextcord.SelectOption]" = []
category_list = highscores_data.get_guild(guild_id).get_table(boss_name).categories
for category in category_list:
options.append(nextcord.SelectOption(label=category.name))
subq = select(CategoryGroup).where(CategoryGroup.GroupId == group_id).subquery()
cat_group_alias = aliased(CategoryGroup, subq)
stmt = select(Category).where(Category.GroupId == cat_group_alias.GroupId)
categories_list = session.scalars(stmt)
for category in categories_list:
options.append(nextcord.SelectOption(label=category.CategoryName, value=str(category.CategoryId)))
return options
@@ -75,16 +39,15 @@ class SubmissionDropdown(nextcord.ui.Select):
bot: commands.Bot,
options: "list[nextcord.SelectOption]",
):
super().__init__(
placeholder="Choose submission category.", min_values=1, max_values=1, options=options
)
super().__init__(placeholder="Choose submission category.", min_values=1, max_values=1, options=options)
self._bot = bot
async def callback(self, interaction: nextcord.Interaction):
submission_objects[interaction.user.id]["category"] = self.values[0]
embed = get_submission_embed(interaction.user.id)
submission = submission_objects.get_user_submission(interaction.user.id)
submission.category_id = self.values[0]
embeds = submission.get_embeds()
await interaction.message.edit(
embed=embed,
embeds=embeds,
)
@@ -96,11 +59,12 @@ class BackButton(nextcord.ui.Button):
self._bot = bot
async def callback(self, interaction: nextcord.Interaction):
state: SubmissionState = submission_objects[interaction.user.id]["submission_state"]
submission = submission_objects.get_user_submission(interaction.user.id)
state: SubmissionState = submission.submission_state
state = SubmissionState(state.value - 1)
submission_objects[interaction.user.id]["submission_state"] = state
embed = get_submission_embed(interaction.user.id)
await interaction.edit(embed=embed, view=SubmissionView(self._bot, interaction.user.id))
submission.submission_state = state
embeds = submission.get_embeds()
await interaction.edit(embeds=embeds, view=SubmissionView(self._bot, interaction.user.id))
class CancelButton(nextcord.ui.Button):
@@ -110,7 +74,7 @@ class CancelButton(nextcord.ui.Button):
super().__init__(label="Cancel", style=nextcord.ButtonStyle.red, custom_id="cancel")
async def callback(self, interaction: nextcord.Interaction):
submission_objects.pop(interaction.user.id)
submission_objects.remove_user_submission(interaction.user.id)
await interaction.message.delete()
@@ -123,23 +87,19 @@ class NextButton(nextcord.ui.Button):
async def callback(self, interaction: nextcord.Interaction):
# check if fields have been filled out for each state
state: SubmissionState = submission_objects[interaction.user.id]["submission_state"]
if (
state == SubmissionState.CATEGORY
and not submission_objects[interaction.user.id]["category"]
):
submission = submission_objects.get_user_submission(interaction.user.id)
state: SubmissionState = submission.submission_state
if state == SubmissionState.CATEGORY and not submission.category_id:
await interaction.send("Please choose a category.", ephemeral=True)
elif (
state == SubmissionState.SCORE and not submission_objects[interaction.user.id]["score"]
):
elif state == SubmissionState.SCORE and not submission.score:
await interaction.send("Please send a score.", ephemeral=True)
elif state == SubmissionState.PROOF and not len(interaction.message.attachments) > 0:
await interaction.send("Please attach picture for proof.", ephemeral=True)
else:
state = SubmissionState(state.value + 1)
submission_objects[interaction.user.id]["submission_state"] = state
embed = get_submission_embed(interaction.user.id)
await interaction.edit(embed=embed, view=SubmissionView(self._bot, interaction.user.id))
submission.submission_state = state
embeds = submission.get_embeds()
await interaction.edit(embeds=embeds, view=SubmissionView(self._bot, interaction.user.id))
class SubmitButton(nextcord.ui.Button):
@@ -150,27 +110,27 @@ class SubmitButton(nextcord.ui.Button):
self._bot = bot
async def callback(self, interaction: nextcord.Interaction):
verification_channel_id = highscores_data.get_guild(
interaction.guild_id
).verification_channel
stmt = select(Guild.VerificationChannel).where(Guild.GuildId == interaction.guild_id)
verification_channel_id = session.scalar(stmt)
channel = interaction.guild.get_channel(verification_channel_id)
submission = submission_objects.get_user_submission(interaction.user.id)
error_response = "Registered Verification channel does not exist or was never registered. \
Contact server admins."
assert channel is not None, await channel.send(error_response)
submission_objects[interaction.user.id]["submission_state"] = SubmissionState.VERIFY
submission.submission_state = SubmissionState.VERIFY
embed = get_submission_embed(interaction.user.id)
embeds = submission.get_embeds()
await channel.send(
files=[await attch.to_file() for attch in interaction.message.attachments],
view=VerificationView(),
embed=embed,
content=f"Submission from {interaction.user.display_name}",
embeds=embeds,
content=f"Submission from <@{interaction.user.id}>.",
)
await interaction.message.edit(content=f"<@{interaction.user.id}> Submitted!", view=None)
submission_objects.pop(interaction.user.id)
submission_objects.remove_user_submission(interaction.user.id)
############ Submission Views ############
@@ -183,39 +143,34 @@ class SubmissionCreateButton(nextcord.ui.View):
super().__init__(timeout=None)
self._bot = bot
@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
"""Button for score submission."""
# Don't allow a user to start a new submission if one already exists
if interaction.user.id in submission_objects:
if submission_objects.has_user(interaction.user.id):
await interaction.send("Please finish in progress submission first.", ephemeral=True)
return
boss_name = highscores_message_map[interaction.guild.id][interaction.message.id]
channel = interaction.guild.get_channel(
highscores_data.get_guild(interaction.guild_id).submission_channel
stmt = (
select(CategoryGroup.GroupId)
.where(CategoryGroup.GuildId == interaction.guild_id)
.where(CategoryGroup.MessageId == interaction.message.id)
)
group_id = session.scalar(stmt)
stmt = select(Guild.SubmissionChannel).where(Guild.GuildId == interaction.guild_id)
channel = interaction.guild.get_channel(session.scalar(stmt))
# a user should only have 1 submission active at once.
# submitting with dummy message_id until message created
# split username on pipe.
submission_objects[interaction.user.id] = {
"message_id": 0,
"guild_id": interaction.guild_id,
"submission_state": SubmissionState.CATEGORY,
"boss_name": boss_name,
"score": "",
"category": "",
"username": interaction.user.display_name.split("|")[0][
: highscores_data.get_guild(interaction.guild_id).username_length
],
}
# split username on pipe. Decided to store whole username...
username: str = interaction.user.display_name.split("|")[0]
submission: Submission = Submission(interaction.guild_id, group_id, username)
embed = get_submission_embed(interaction.user.id)
submission_objects.add_submission(interaction.user.id, submission)
embeds = submission.get_embeds()
submission_message = await channel.send(
f"<@{interaction.user.id}>",
@@ -223,11 +178,11 @@ class SubmissionCreateButton(nextcord.ui.View):
self._bot,
interaction.user.id,
),
embed=embed,
embeds=embeds,
)
# update the message_id once its created
submission_objects[interaction.user.id]["message_id"] = submission_message.id
submission.message_id = submission_message.id
class SubmissionView(nextcord.ui.View):
@@ -243,15 +198,13 @@ class SubmissionView(nextcord.ui.View):
super().__init__(timeout=timeout)
self._submission_user = submission_user
self._bot = bot
self._state = submission_objects[self._submission_user]["submission_state"]
self._boss_name = submission_objects[self._submission_user]["boss_name"]
self._guild_id = submission_objects[self._submission_user]["guild_id"]
self._state = submission_objects.get_user_submission(self._submission_user).submission_state
self._group_id = submission_objects.get_user_submission(self._submission_user).group_id
self._guild_id = submission_objects.get_user_submission(self._submission_user).guild_id
# add items depending on given state
if self._state is SubmissionState.CATEGORY:
self.add_item(
SubmissionDropdown(bot, category_select_options(self._guild_id, self._boss_name))
)
self.add_item(SubmissionDropdown(bot, category_select_options(self._group_id)))
self.add_item(CancelButton())
self.add_item(NextButton(self._bot))
elif (
@@ -275,4 +228,4 @@ class SubmissionView(nextcord.ui.View):
async def on_timeout(self) -> None:
# remove user from submission objects after time out. Allows user to start new submission.
submission_objects.pop(self._submission_user)
submission_objects.remove_user_submission(self._submission_user)
+20 -34
View File
@@ -1,10 +1,9 @@
"""View for verification buttons"""
import nextcord
from data.score import Score, ScoreTime # pylint: disable=import-error
from highscores import highscores_data # pylint: disable=import-error
from utilities.utils import submit_score # pylint: disable=import-error
# pylint: disable=import-error
from utilities.utils import submit_score
class VerificationView(nextcord.ui.View):
@@ -13,51 +12,38 @@ class VerificationView(nextcord.ui.View):
def __init__(self):
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(
self, button: nextcord.ui.Button, interaction: nextcord.Interaction
): # pylint: disable=unused-argument
"""Button for denying score submission."""
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(
self, button: nextcord.ui.Button, interaction: nextcord.Interaction
): # pylint: disable=unused-argument
"""Button for approving score submission."""
content = interaction.message.embeds[0].to_dict()
for embed in interaction.message.embeds:
content = embed.to_dict()
for field in content["fields"]:
if field["name"] == "Boss":
boss_name = field["value"]
elif field["name"] == "Category":
category = field["value"]
elif field["name"] == "Score":
score = field["value"]
elif field["name"] == "User Name":
user = field["value"]
for field in content["fields"]:
if field["name"] == "GroupId":
group_id = int(field["value"])
elif field["name"] == "CategoryId":
category_id = int(field["value"])
elif field["name"] == "Score":
score = field["value"]
elif field["name"] == "User Name":
user = field["value"]
try:
submission_guild = highscores_data.get_guild(interaction.guild_id)
submission_table = submission_guild.get_table(boss_name)
submission_category = submission_table.get_category(category)
if submission_category.time_record:
submission_score: ScoreTime = ScoreTime(user, float(score))
else:
submission_score: Score = Score(user, int(score))
await submit_score(
interaction,
submission_guild,
submission_table,
submission_category,
submission_score,
)
await interaction.message.edit(f"Approved by <@{interaction.user.id}>.", view=None)
await submit_score(interaction, group_id, category_id, user, score)
content = interaction.message.content
content += f"\nApproved by <@{interaction.user.id}>."
# remove the bot information embed. it should always be the first one.
await interaction.message.edit(content=content, view=None, embed=interaction.message.embeds[0])
except UnboundLocalError:
await interaction.send("Instruct user to submit with all fields.")
await interaction.message.edit(view=None)
-9
View File
@@ -1,9 +0,0 @@
import os
from dotenv.main import load_dotenv
load_dotenv()
# Discord config
DISCORD_TOKEN = os.getenv("DISCORD_TOKEN", "")
BOT_PREFIX = "?"
-186
View File
@@ -1,186 +0,0 @@
"""File for Category class."""
import json
from data.score import Score, ScoreTime # pylint: disable=import-error
class Category:
"""Stores a category's information and relative scores."""
def __init__(self, name: str) -> None:
"""
Default constructor for Category class. Creates empty category and populates with defaults.
Args:
name (str): Category's name
"""
self._name: str = name
self._ascending: bool = False
self._time_record: bool = False
self._scores_map: dict = {}
def __str__(self) -> str:
"""
Returns Category class as a string for printing.
Returns:
str: Category class as a string
"""
return self.to_str()
@classmethod
def from_dict(cls, name: str, data: dict):
"""
Creates a Category class from the passed data.
Args:
name (str): Category name
data (dict): Rest of Category data in dictionary format
Returns:
Category: Populated Category object
"""
category = cls(name)
category.ascending = data["ascending"]
category.time_record = data["time_record"]
for user, score in data["scores"].items():
if category.time_record:
category.add_score(ScoreTime(user, score))
else:
category.add_score(Score(user, score))
return category
@property
def ascending(self) -> bool:
"""
Getter for is ascending.
Returns:
bool: is ascending
"""
return self._ascending
@ascending.setter
def ascending(self, ascending: bool) -> None:
"""
Setter for is ascending.
Args:
ascending (bool): is ascending
"""
self._ascending = ascending
@property
def name(self) -> str:
"""
Getter for name.
Returns:
str: name
"""
return self._name
@property
def time_record(self) -> bool:
"""
Getter for is time record.
Returns:
bool: is time record
"""
return self._time_record
@time_record.setter
def time_record(self, time_record: bool) -> None:
"""
Setter for is time record.
Args:
time_record (bool): is time record
"""
self._time_record = time_record
@property
def scores(self) -> "list[Score]":
"""
Getter for Scores list.
Returns:
list[Score]: List of Scores in this Category
"""
return self._scores_map.values()
def add_score(self, score: Score) -> None:
"""
Adds score to dictionary storage map. Sorts dictionary by score value.
Args:
score (Score): Score to be added
"""
if score.user in self._scores_map:
self._scores_map[score.user].value = score.value
else:
self._scores_map[score.user] = score
self._scores_map = dict(
sorted(self._scores_map.items(), key=lambda x: x[1].value, reverse=not self.ascending)
)
def clear(self) -> None:
"""
Clears all Scores tracked in score map.
This is a test function and shouldn't really be used.
"""
self._scores_map.clear()
def get_user_score(self, user: str) -> Score:
"""
Gets a user's score. Returns None if user is not found.
Args:
user (str): Name of score to find
Returns:
Score: User's Score or None
"""
if user in self._scores_map:
return self._scores_map[user]
return None
def remove_score(self, score: str) -> None:
"""
Removes a user's score from dictionary storage.
Args:
score (str): user whose score to remove
"""
score = self._scores_map.pop(score)
del score
def to_dict(self) -> dict:
"""
Saves category data as a python dictionary.
Does not include category name. It should be the key of above data structure.
Returns:
dict: Category class in dictionary format
"""
data = {}
data["ascending"] = self.ascending
data["time_record"] = self.time_record
data["scores"] = {}
for score in self.scores:
data["scores"][score.user] = score.value
return data
def to_str(self) -> str:
"""
Returns Category object as a string.
Returns:
str: string of json representation of Category
"""
return json.dumps(self.to_dict(), indent=4)
-211
View File
@@ -1,211 +0,0 @@
"""File for Guild class."""
import json
from data.table import Table # pylint: disable=import-error
class Guild:
"""Stores a guild's information."""
def __init__(self, guild_id: int) -> None:
"""
Default constructor for Guild class. Initializes channel ids as 0 as they won't be
registered yet. Sets default username length of 12. Default highscores size to 3.
Args:
guild_id (int): Discord Guild id of new Guild
"""
self._guild_id: int = guild_id
self._highscore_channel_id: int = 0
self._submission_channel_id: int = 0
self._table_map: dict = {}
self._username_length: int = 12
self._verification_channel_id: int = 0
def __str__(self) -> str:
"""
Returns Guild class as a string for printing.
Returns:
str: Guild class as a string
"""
return self.to_str()
@classmethod
def from_dict(cls, guild_id: int, data: dict):
"""
Creates a Guild class from the passed data.
Args:
guild_id (int): Guild's id
data (dict): Rest of Guild data in dictionary format
Returns:
Guild: Populated Guild object
"""
guild = cls(guild_id)
guild.highscore_channel = data["highscore_channel_id"]
guild.submission_channel = data["submission_channel_id"]
guild.username_length = data["username_length"]
guild.verification_channel = data["verification_channel_id"]
for table_name, table_data in data["tables"].items():
table = Table.from_dict(table_name, table_data)
guild.add_table(table)
return guild
@property
def guild_id(self) -> int:
"""
Getter for Guild id
Returns:
int: Guild id
"""
return self._guild_id
@property
def highscore_channel(self) -> int:
"""
Getter for highscore channel id.
Returns:
int: highscore channel id
"""
return self._highscore_channel_id
@highscore_channel.setter
def highscore_channel(self, channel: int) -> None:
"""
Setter for highscore channel id.
Args:
channel (int): highscore channel id
"""
self._highscore_channel_id = channel
@property
def submission_channel(self) -> int:
"""
Getter for submission channel id.
Returns:
int: submission channel id
"""
return self._submission_channel_id
@submission_channel.setter
def submission_channel(self, channel: id) -> None:
"""
Setter fro submission channel id.
Args:
channel (id): submission channel id
"""
self._submission_channel_id = channel
@property
def tables(self) -> "list[Table]":
"""
Getter for Tables list.
Returns:
list[Table]: List of Tables in this Guild
"""
return self._table_map.values()
@property
def username_length(self) -> int:
"""
Getter for username length.
Returns:
int: username length
"""
return self._username_length
@username_length.setter
def username_length(self, length: int) -> None:
"""
Setter for username length.
Args:
length (int): username length
"""
self._username_length = length
@property
def verification_channel(self) -> int:
"""
Getter for verification channel id.
Returns:
int: verification channel id
"""
return self._verification_channel_id
@verification_channel.setter
def verification_channel(self, channel: int) -> None:
"""
Setter for verification channel id
Args:
channel (int): verification channel id
"""
self._verification_channel_id = channel
def add_table(self, table: Table) -> None:
"""
Adds a Table to table map. Shouldn't exist already.
Args:
table (Table): new Table to add
"""
if table.name not in self._table_map:
self._table_map[table.name] = table
def get_table(self, table_name: str) -> Table:
"""
Getter for specific Table. Returns None if Table is not found.
Args:
table_name (str): Name of Table to find
Returns:
Table: Table corresponding to Name. None if not found.
"""
if table_name in self._table_map:
return self._table_map[table_name]
return None
def to_dict(self) -> dict:
"""
Creates python dictionary object from class data.
Does not include guild_id in dict, it should be the key of above data structure.
Returns:
dict: Guild class in dictionary format
"""
data: dict = {}
data["highscore_channel_id"] = self.highscore_channel
data["submission_channel_id"] = self.submission_channel
data["tables"] = {}
data["username_length"] = self.username_length
data["verification_channel_id"] = self.verification_channel
for table in self.tables:
data["tables"][table.name] = table.to_dict()
return data
def to_str(self) -> str:
"""
Returns Guild object as a string.
Returns:
str: string of json representation of Guild
"""
return json.dumps(self.to_dict(), indent=4)
-153
View File
@@ -1,153 +0,0 @@
"""Class and functions for storing and retrieving highscores data."""
import os
import json
from data.guild import Guild # pylint: disable=import-error
from data.table import Table # pylint: disable=import-error
from data.category import Category # pylint: disable=import-error
HIGHSCORES_BACKUP_DIR = "deleted_backups"
HIGHSCORES_CONFIG_DIR = "example_config"
HIGHSCORES_DATA_DIR = "highscores_data"
class HighscoresData:
"""Class for storing and retrieving highscores data."""
def __init__(self) -> None:
"""
Default constructor for Highscores Data. Opens up known highscores data on creation.
"""
self._guild_map: dict = {}
# Ensure that data directories exist.
if not os.path.exists(HIGHSCORES_BACKUP_DIR):
os.mkdir(HIGHSCORES_BACKUP_DIR)
if not os.path.exists(HIGHSCORES_DATA_DIR):
os.mkdir(HIGHSCORES_DATA_DIR)
self.open_highscores_data()
def __str__(self) -> str:
"""
Returns Highscores class as a string for printing.
Returns:
str: Highscores class as a string
"""
data: dict = {}
for guild_id, guild_obj in self._guild_map.items():
data[guild_id] = guild_obj.to_dict()
return "Highscores Data\n" + json.dumps(data, indent=4)
@property
def guilds(self) -> "list[Guild]":
"""
Getter for Guilds list.
Returns:
list[Guild]: List of Guilds tracked by Highscores Bot
"""
return self._guild_map.values()
def add_configuration(self, guild_id: int, config_str: str) -> None:
"""
Adds an example configuration to a Guild. Uses guild id to know which guild and
config name to open a file in example_config.
Args:
guild_id (int): Guild id to add config to
config_str (str): Config name to add
"""
config = self._open_highscores_config(config_str)
guild = self.get_guild(guild_id)
for table_name, categorys in config["tables"].items():
table_obj = Table(table_name)
table_obj.size = config["table_size"]
for category_name in categorys:
category = Category(category_name)
category.ascending = config["categories"][category.name]["ascending"]
category.time_record = config["categories"][category.name]["is_time_record"]
table_obj.add_category(category)
guild.add_table(table_obj)
def add_guild(self, guild_id: int) -> None:
"""
Creates a new Guild and adds to Guild map.
Args:
guild_id (int): Guild Id to add
"""
guild: Guild = Guild(guild_id)
self._guild_map[guild_id] = guild
def get_guild(self, guild_id: int) -> Guild:
"""
Getter for Guild object that corresponds with the guild_id.
Args:
guild_id (int): id of Guild to retrieve.
Returns:
Guild: Guild object for guild_id
"""
return self._guild_map[guild_id]
def remove_guild(self, guild_id: int) -> None:
"""
Removes a guild's data from object. Saves backup.
Args:
guild_id (int): guild id to remove
"""
self._guild_map.pop(guild_id)
os.rename(
f"{HIGHSCORES_DATA_DIR}/{guild_id}.json", f"{HIGHSCORES_BACKUP_DIR}/{guild_id}.json"
)
def _open_highscores_config(self, config: str) -> dict:
"""
Opens the highscores config file and returns in dictionary format.
Will be replaced when config class is created.
Args:
config (str): Name of config to load.
Returns:
dict: dict of highscores config
"""
with open(f"{HIGHSCORES_CONFIG_DIR}/{config}.json", encoding="utf-8", mode="r") as file:
return json.load(file)
def open_highscores_data(self):
"""
Opens all saved highscores data in the HIGHSCORES_DATA_DIR dir. Creates a Guild object from
each file and adds to guild map.
"""
for data_file in os.listdir(HIGHSCORES_DATA_DIR):
guild_id: int = int(data_file.split(".", maxsplit=1)[0])
with open(f"{HIGHSCORES_DATA_DIR}/{data_file}", encoding="utf-8", mode="r") as file:
guild_obj: Guild = Guild.from_dict(guild_id, json.load(file))
self._guild_map[guild_obj.guild_id] = guild_obj
def save_all_data(self) -> None:
"""
Calls save_guild_data for each guild tracked in guild map.
"""
for guild in self.guilds:
self.save_guild_data(guild)
def save_guild_data(self, guild: Guild) -> None:
"""
Saves a guild to its own json file.
Args:
guild (Guild): Guild to save
"""
with open(
f"{HIGHSCORES_DATA_DIR}/{guild.guild_id}.json", encoding="utf-8", mode="w"
) as file:
file.write(guild.to_str()) # should use str function
-123
View File
@@ -1,123 +0,0 @@
"""File for Score classes"""
import time
class Score:
"""Parent class for Scores. Stores common information for a Score."""
# no user setter, it should never change.
def __init__(self, user: str, value: int) -> None:
"""
Default constructor for Score class.
Args:
user (str): Score's user
value (int): Score's value
"""
self._user: str = user
self._value: int = value
def __str__(self) -> str:
"""
Returns Score class as a string for printing.
Returns:
str: Score class as a string
"""
return self.to_str()
@property
def user(self) -> str:
"""
Getter for score user.
Returns:
str: user
"""
return self._user
@property
def value(self) -> int:
"""
Getter for score value.
Returns:
int: value
"""
return self._value
@value.setter
def value(self, value: int) -> None:
"""
Setter for score value.
Args:
value (int): score value input
"""
self._value = value
def get_str(self) -> str:
"""
Gets score's value as string.
Returns:
str: Score's value
"""
return str(self._value)
def to_str(self) -> str:
"""
Returns Score object as a string.
Returns:
str: string of json representation of Score
"""
return f"{self.user}: {self.get_str()}"
class ScoreTime(Score):
"""Class for a time based score. Inherits from Score."""
def __init__(self, user: str, score: float) -> None: # pylint: disable=super-init-not-called
"""
Default constructor for ScoreTime class.
Args:
user (str): Score's user
score (float): Score's value
"""
# store time score as seconds.
self._user: str = user
self._score: float = score
@property
def value(self) -> float:
"""
Getter for score value.
Returns:
float: value
"""
return self._score
@value.setter
def value(self, value: float) -> None:
"""
Setter for score value.
Args:
value (float): value
"""
self._score = value
def get_str(self) -> str:
"""
Gets score's value as string. Turns seconds into MM:SS.ms format.
Returns:
str: value in time format.
"""
milliseconds = round((self.value - int(self.value)) * 10)
return time.strftime(f"%-M:%S.{milliseconds}", time.gmtime(self.value))
-164
View File
@@ -1,164 +0,0 @@
"""File for Table class."""
import json
from data.category import Category # pylint: disable=import-error
class Table:
"""Stores a table's information a list of categories."""
def __init__(self, name: str) -> None:
"""
Default constructor for Table class.
Creates object and fills out class variables with defaults.
Args:
name (str): Table's name
"""
self._name: str = name
self._message_id: int = 0
self._categories_map: dict = {}
self._size = 3
def __str__(self) -> str:
"""
Returns Table class as a string for printing.
Returns:
str: Table class as a string
"""
return self.to_str()
@classmethod
def from_dict(cls, table_name: str, data: dict):
"""
Creates a Table class from the passed data.
Args:
table_name (str): Table name
data (dict): Rest of Table data in dictionary format
Returns:
Table: Populated Table object
"""
table = cls(table_name)
table.message_id = data["message_id"]
table.size = data["table_size"]
for cat_name, cat in data["categories"].items():
category = Category.from_dict(cat_name, cat)
table.add_category(category)
return table
@property
def categories(self) -> "list[Category]":
"""
Getter for Categories list.
Returns:
list[Category]: List of Categories in this Table
"""
return self._categories_map.values()
@property
def message_id(self) -> int:
"""
Getter for message id
Returns:
int: message id
"""
return self._message_id
@message_id.setter
def message_id(self, m_id: int) -> None:
"""
Setter for message id.
Args:
m_id (int): message id
"""
self._message_id = m_id
@property
def name(self) -> str:
"""
Getter for name.
Returns:
str: name
"""
return self._name
@property
def size(self) -> int:
"""
Getter for size of table.
Returns:
int: Size of table
"""
return self._size
@size.setter
def size(self, size: int) -> None:
"""
Setter for size of table.
Args:
size (int): New size of table
"""
self._size = size
def add_category(self, category: Category) -> None:
"""
Adds a Category to categories map. Shouldn't exist already.
Args:
category (Category): New Category to add
"""
if category.name not in self._categories_map:
self._categories_map[category.name] = category
def get_category(self, category_name: str) -> Category:
"""
Getter for specific category. Returns None if not found.
Args:
category_name (str): name of category to return
Returns:
Category: Category corresponding to name. None if not found.
"""
if category_name in self._categories_map:
return self._categories_map[category_name]
return None
def to_dict(self) -> dict:
"""
Saves Table as a python dictionary.
Does not include table name. It should be the key of above data structure.
Returns:
dict: Table class in dictionary format
"""
data = {}
data["message_id"] = self.message_id
data["table_size"] = self.size
data["categories"] = {}
for category in self.categories:
data["categories"][category.name] = category.to_dict()
return data
def to_str(self) -> str:
"""
Returns Table object as a string.
Returns:
str: string of json representation of Table
"""
return json.dumps(self.to_dict(), indent=4)
+3
View File
@@ -0,0 +1,3 @@
FROM mysql:latest
COPY highscores_database.sql /docker-entrypoint-initdb.d/
+43
View File
@@ -0,0 +1,43 @@
CREATE DATABASE highscores;
use highscores;
CREATE TABLE Guilds(
GuildId bigint unsigned not null unique,
SubmissionChannel bigint unsigned,
VerificationChannel bigint unsigned,
PRIMARY KEY (GuildId)
);
CREATE TABLE CategoryGroups(
GroupId int unsigned not null unique AUTO_INCREMENT,
GroupName varchar(50) not null,
ChannelId bigint unsigned,
GuildId bigint unsigned not null,
MessageId bigint unsigned,
PadLength smallint not null DEFAULT 26,
UsernameLength smallint DEFAULT 12,
CONSTRAINT UniqueGuildCategoryGroup UNIQUE (GuildId, GroupName),
FOREIGN KEY (GuildId) REFERENCES Guilds(GuildId),
PRIMARY KEY (GroupId)
);
CREATE TABLE Categories(
CategoryId int unsigned not null unique AUTO_INCREMENT,
CategoryName varchar(30) not null,
GroupId int unsigned not null,
Size smallint not null DEFAULT 3,
IsAscending boolean not null DEFAULT true,
IsTimeRecord boolean not null DEFAULT false,
IsValidated boolean not null DEFAULT true,
CONSTRAINT UniqueCategoryInGroup UNIQUE (GroupId, CategoryName),
FOREIGN KEY (GroupId) REFERENCES CategoryGroups(GroupId),
PRIMARY Key (CategoryId)
);
CREATE TABLE Scores(
UserName varchar(30) not null,
Score float not null, -- Maybe should have time and int scores?? timestamp type?
CategoryId int unsigned not null,
FOREIGN KEY (CategoryId) REFERENCES Categories(CategoryId),
PRIMARY KEY (CategoryId, UserName)
);
+6 -15
View File
@@ -1,5 +1,5 @@
"""
main file for pvm-highscores bot
main file for highscores bot
"""
import logging
@@ -9,8 +9,8 @@ import sys
import nextcord
from nextcord.ext import commands
import config
from data.highscores_data import HighscoresData
import utilities.config as config
from classes.submission import SubmissionMap
logger = logging.getLogger("nextcord")
logger.setLevel(logging.DEBUG)
@@ -18,23 +18,14 @@ handler = logging.FileHandler(filename="nextcord.log", encoding="utf-8", mode="w
handler.setFormatter(logging.Formatter("%(asctime)s:%(levelname)s:%(name)s: %(message)s"))
logger.addHandler(handler)
highscores_data: HighscoresData = HighscoresData()
highscores_message_map = {}
for guild in highscores_data.guilds:
highscores_message_map[guild.guild_id] = {}
for table in guild.tables:
highscores_message_map[guild.guild_id][table.message_id] = table.name
submission_objects: dict = {}
submission_objects: SubmissionMap = SubmissionMap()
def main():
"""Main function for bot."""
min_python_version = (3, 7)
min_python_version = (3, 8)
if sys.version_info < min_python_version:
print("ERROR: Python version must be >= 3.7")
print("ERROR: Python version must be >= 3.8")
sys.exit()
# Allows privledged intents for monitoring members joining, roles editing, and role assignments
+15
View File
@@ -0,0 +1,15 @@
version: "3"
services:
sqldb:
image: "highscores_db"
container_name: "Highscores-DataBase"
volumes:
- highscores_data:/var/lib/mysql
ports:
- 3306:3306
environment:
- MYSQL_ROOT_PASSWORD=${SQL_PASS}
- MYSQL_ROOT_USERNAME=${SQL_USER}
volumes:
highscores_data:
+5 -2
View File
@@ -1,2 +1,5 @@
nextcord>=2.0.0rc1
python-dotenv>=0.20.0
nextcord>=2.3.2
python-dotenv>=0.20.0
pandas>=1.5.2
sqlalchemy>=1.4.44
mysqlclient>-2.1.1
+17
View File
@@ -0,0 +1,17 @@
"""
Loads environment variables for use in bot.
"""
import os
from dotenv.main import load_dotenv
load_dotenv()
# Discord config
BOT_PREFIX = "?"
DISCORD_TOKEN = os.getenv("DISCORD_TOKEN", "")
SQL_DB = os.getenv("SQL_DB", "")
SQL_HOST = os.getenv("SQL_HOST", "")
SQL_PASS = os.getenv("SQL_PASS", "")
SQL_USER = os.getenv("SQL_USER", "")
+22
View File
@@ -0,0 +1,22 @@
"""
Simple file to set up the sqlalchemy connection to database.
"""
from sqlalchemy import create_engine
from sqlalchemy.ext.automap import automap_base
from sqlalchemy.orm import Session
import utilities.config as config
# connect with sql alchemy and autoload schema
url = f"mysql://{config.SQL_USER}:{config.SQL_PASS}@{config.SQL_HOST}/{config.SQL_DB}"
engine = create_engine(url)
connection = engine.connect()
Base = automap_base()
Base.prepare(autoload_with=engine)
session = Session(engine)
# define the classes
Guild = Base.classes.Guilds
CategoryGroup = Base.classes.CategoryGroups
Category = Base.classes.Categories
Score = Base.classes.Scores
+333 -64
View File
@@ -2,122 +2,391 @@
Contains various utlity functions for highscores bot
"""
import json
import time
import nextcord
from highscores import highscores_data, highscores_message_map # pylint: disable=import-error
from data.score import Score # pylint: disable=import-error
from data.guild import Guild # pylint: disable=import-error
from data.category import Category # pylint: disable=import-error
from data.table import Table # pylint: disable=import-error
import pandas as pd
from sqlalchemy import distinct, select
from sqlalchemy.orm import aliased
# pylint: disable=import-error
from utilities.sql_alchemy_connection import Category, CategoryGroup, Guild, Score, engine, session
MAX_LINE_LENGTH = 142
def format_highscore_message(guild: Guild, table: Table) -> str:
def autofill_categories(guild_id: int, group_id: int, category_begin: str) -> "dict[str, str]":
"""
Returns all category names that start with category_begin.
If category_begin is empty, returns list of all category names.
Args:
guild_id (int): Guild id that holds the group and categories.
group_id (int): Group id to search in.
category_begin (str): The beginning of a category name to search with. Can be empty.
Returns:
dict[str, str]: List of all category names that start with category_begin.
"""
# search database for all category names
subq = (
select(CategoryGroup)
.where(CategoryGroup.GuildId == guild_id)
.where(CategoryGroup.GroupId == group_id)
.subquery()
)
cat_group_alias = aliased(CategoryGroup, subq)
stmt = select(Category).where(Category.GroupId == cat_group_alias.GroupId)
categories_dict: dict = {cat.CategoryName: str(cat.CategoryId) for cat in session.scalars(stmt)}
# if string empty, return all category names
if not category_begin:
return categories_dict
# else find all categories that start with category_begin
# TODO is it better to have sql handle the starts with or python?
categories: dict = {
cat_name: cat_id for cat_name, cat_id in categories_dict.items() if category_begin.lower() in cat_name.lower()
}
return categories
def autofill_groups(guild_id: int, group_begin: str) -> "dict[str, str]":
"""
Returns all group names that start with group_begin. If group_begin is empty, returns list of all group names.
Args:
guild_id (int): Guild id that holds the groups.
group_begin (str): The beginning of a group name to search with. Can be empty.
Returns:
dict[str, str]: Dict of group names and ids where group name start with group_begin. Key is name, value is id.
"""
# Select all groups in guild.
stmt = select(CategoryGroup).where(CategoryGroup.GuildId == guild_id)
# dict has to have key values of same type for discord.
group_dict: dict = {group.GroupName: str(group.GroupId) for group in session.scalars(stmt)}
# if string empty, return all group names
if not group_begin:
return group_dict
# else find all groups that start with group_begin
groups: dict = {
group_name: group_id for group_name, group_id in group_dict.items() if group_begin.lower() in group_name.lower()
}
return groups
def autofill_users_in_group_category(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
not None. Might be expensive due to unioning all users in group.
Args:
guild_id (int): Guild id corresponding to group and category.
group_id (str): group to look in for category.
category_id (str): Category to look in for users. Can be None if not looking for category.
user_begin (str): Beginning of user name to search with. Can be empty.
Returns:
list[str]: List of users that start with user_begin.
"""
# get all categories in the group
subq = (
select(CategoryGroup)
.where(CategoryGroup.GuildId == guild_id)
.where(CategoryGroup.GroupId == group_id)
.subquery()
)
category_group = aliased(CategoryGroup, subq)
# If no category id, get all users in group.
if category_id == 0:
subq = select(Category.CategoryId).where(Category.GroupId == category_group.GroupId)
stmt = select(distinct(Score.UserName)).where(Score.CategoryId.in_(subq))
else:
# get all users in category_id
subq = (
select(Category)
.where(Category.GroupId == category_group.GroupId)
.where(Category.CategoryId == category_id)
.subquery()
)
category = aliased(Category, subq)
stmt = select(distinct(Score.UserName)).where(Score.CategoryId == category.CategoryId)
users_list: "list[str]" = list(session.scalars(stmt))
# if string empty, return all users
if not user_begin:
return users_list
# else find all users that start with user_begin
users = [user for user in users_list if user_begin.lower() in user.lower()]
return users
def autofill_users_in_guild(guild_id: int, user_begin: str) -> "list[str]":
"""
Obtains and returns a list of all the users who have submitted to the guild's highscores.
Args:
guild_id (int): Id of the guild.
user_begin (str): Beginning of user name to search with. Can be empty.
Returns:
list[str]: All users with submissions in the guild.
"""
group_ids = select(CategoryGroup.GroupId).where(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))
users_list: "list[str]" = list(session.scalars(stmt))
# if string empty, return all users
if not user_begin:
return users_list
# else find all users that start with user_begin
users = [user for user in users_list if user_begin.lower() in user.lower()]
return users
def format_highscore_message(group: CategoryGroup) -> str:
"""
Creates a formatted string for a boss's highscore discord message.
This formatted string is in discord's code block to ensure monospace text.
Loops through each category to build the message one place at a time.
Args:
guild (Guild): Guild object that is posting the message.
table (Table): Table object that contains the contents of the message.
group (group): group object that contains the contents of the message.
Returns:
str: Formatted string for discord message
"""
ret_string = f"**{table.name}**\n```"
ret_string = f"**{group.GroupName}**\n```"
# Gather all categories and add to string
for category in table.categories:
ret_string += f"{category.name}".ljust(26, " ")
ret_string += "\n"
stmt = select(Guild).where(Guild.GuildId == group.GuildId)
# list rankings in a single line. Start with first place for each category and then continue
for i in range(table.size):
for category in table.categories:
if len(category.scores) > i:
# pad username to max length.
# have to convert category.scores into list to make it subscriptable. no idea why.
username: str = list(category.scores)[i].user.ljust(guild.username_length, " ")
# Get all categories in the group
subq = (
select(CategoryGroup)
.where(CategoryGroup.GuildId == group.GuildId)
.where(CategoryGroup.GroupName == group.GroupName)
.subquery()
)
cat_group_alias = aliased(CategoryGroup, subq)
stmt = select(Category).where(Category.GroupId == cat_group_alias.GroupId)
score: str = list(category.scores)[i].get_str()
# set up a map for the categories
# also grab the max category size
categories = []
category_ids = []
max_cat_size: int = 0
pad_length = group.PadLength
for cat in session.scalars(stmt):
if cat.Size > max_cat_size:
max_cat_size = cat.Size
categories.append(cat)
category_ids.append(cat.CategoryId)
# pad score to correct length
# MM:SS.ms is 7 characters long
if category.time_record:
# if there are single digit minutes, add a space for formatting.
if score.find(":") and len(score.split(":", maxsplit=1)[0]) < 2:
score = " " + score
# get all scores in category group, load into data frame.
d_f = pd.read_sql_query(sql=session.query(Score).where(Score.CategoryId.in_(category_ids)).statement, con=engine)
# this list is what will be printed in the discord message
# first string will be category list, others will be score printouts
str_list: "list[str]" = ["" for i in range(max_cat_size + 1)]
groups_df = d_f.groupby("CategoryId")
# TODO figure out how to have less nests.
for category in categories:
# category name string
str_list[0] += f"{category.CategoryName}".ljust(pad_length, " ")
# ensure there is data to print
if category.CategoryId in groups_df.indices:
group_df = groups_df.get_group(category.CategoryId)
# process data. Want each group sorted correctly and only top X results
processed_group = (
group_df.sort_values("Score", ascending=category.IsAscending).head(category.Size).reset_index()
)
# add each score to list
for ind in range(max_cat_size): # need to subtract to deal with category name string
if ind in processed_group.index:
user_length = group.UsernameLength
username = processed_group["UserName"][ind][:user_length].ljust(user_length, " ")
# get score, pad to correct length
if category.IsTimeRecord:
# MM:SS.ms is 7 characters long
# if there are single digit minutes, add a space for formatting.
score = get_time_str(processed_group["Score"][ind])
if score.find(":") and len(score.split(":", maxsplit=1)[0]) < 2:
score = " " + score
else:
score = str(int(processed_group["Score"][ind])).rjust(7, " ")
str_list[ind + 1] += f"{ind + 1}: {username} - {score}".ljust(pad_length, " ")
else:
score = score.rjust(7, " ")
# if not enough entries, pad out
if ind + 1 <= category.Size:
str_list[ind + 1] += f"{ind + 1}: ".ljust(pad_length, " ")
else:
str_list[ind + 1] += "".ljust(pad_length, " ")
else:
for ind in range(max_cat_size):
if ind + 1 <= category.Size:
str_list[ind + 1] += f"{ind + 1}: ".ljust(pad_length, " ")
else:
str_list[ind + 1] += "".ljust(pad_length, " ")
ret_string += f"{i+1}: {username} - {score}".ljust(26, " ")
else:
ret_string += f"{i+1}: submit your score".ljust(26, " ")
ret_string += "\n"
# next step is to cut down on the length of each line if required
# need to use index to allow for modifying the list
for ind, line in enumerate(str_list):
if len(line) > MAX_LINE_LENGTH:
split_loc = int(MAX_LINE_LENGTH / pad_length) * pad_length
new_line = line[split_loc:]
str_list[ind] = line[:split_loc]
str_list.append(new_line)
# this list can now be placed into the format message function in the bot
for ind, string in enumerate(str_list):
# rstrip to clear unneeded spaces
# add newline here for each string
ret_string += string.rstrip() + "\n"
# add extra newline at category split unless there is already a space.
if ((ind + 1) % (max_cat_size + 1)) == 0:
# checking inside the other if to ensure called as little as possible.
if not ret_string.endswith("\n\n"):
ret_string += "\n"
return ret_string + "```"
async def send_highscore_message(
channel: nextcord.TextChannel, guild: Guild, table: Table
) -> nextcord.Message:
def get_time_str(time_in: float) -> str:
"""
Function takes care of checking if a message exists for a table.
Formats a float value into the desired string format for a timestamp.
Args:
time_in (float): A time record in seconds
Returns:
str: Describes the time record in string format.
"""
# TODO see if theres a better way to do this
milliseconds = round((time_in - int(time_in)) * 10)
return time.strftime(f"%-M:%S.{milliseconds}", time.gmtime(time_in))
async def send_highscore_message(channel: nextcord.TextChannel, group: CategoryGroup) -> nextcord.Message:
"""
Function takes care of checking if a message exists for a group.
Creates and sends/edits the message.
Args:
channel (nextcord.TextChannel): Discord channel object where message is sent
guild (Guild): Guild object for Guild that is posting the message
table (Table): Table to post to the message
group (CategoryGroup): group to post to the message
Returns:
nextcord.Message: Message object that was posted with table contents
nextcord.Message: Message object that was posted with group contents
"""
highscore_string = format_highscore_message(guild, table)
highscore_string = format_highscore_message(group)
message_id = table.message_id
try:
message: nextcord.Message = await channel.fetch_message(message_id)
message: nextcord.Message = await channel.fetch_message(group.MessageId)
await message.edit(content=highscore_string)
except nextcord.NotFound:
# message doesn't exist.
message: nextcord.Message = await channel.send(highscore_string)
message_id = message.id
table.message_id = message_id
# save message -> boss_name in message_map
# get guild id from channel
# channel should come from the guild that is requesting this highscores message anyways.
# check if guild is in message map already
if not channel.guild.id in highscores_message_map:
highscores_message_map[channel.guild.id] = {}
highscores_message_map[channel.guild.id][message_id] = table.name
group.MessageId = message.id
session.commit()
return message
async def submit_score(
interaction: nextcord.Interaction, guild: Guild, table: Table, category: Category, score: Score
interaction: nextcord.Interaction, group_id: int, category_id: int, username: str, score: str
) -> None:
"""
Adds passed score to table. Then updates table message by calling send_highscore_message.
Adds passed score to group. Then updates group message by calling send_highscore_message.
Ensures that a highscores channel exists.
Args:
interaction (nextcord.Interaction): Interaction that caused the score submission
guild (Guild): Guild to submit score to
table (Table): Table to submit score to
category (Category): Category to submit score to
score (Score): Score to submit
interaction (nextcord.Interaction): Interaction that caused the score submission.
group (CategoryGroup): group to submit score to.
category_id (Category): Category to submit score to.
username (str): The user who submitted the score.
score (str): Score to submit. Need to convert to correct type.
"""
category.add_score(score)
highscores_data.save_guild_data(guild)
stmt = select(Category.IsTimeRecord).where(Category.CategoryId == category_id)
is_time_record = session.scalar(stmt)
if is_time_record:
minutes = float(score.split(":", maxsplit=1)[0])
seconds = float(score.split(":", maxsplit=1)[1])
score = (minutes * 60) + seconds
else:
score = float(score)
stmt = select(Score).where(Score.UserName == username).where(Score.CategoryId == category_id)
existing_score = session.scalar(stmt)
if existing_score:
existing_score.Score = score
else:
score_obj = Score(UserName=username, Score=score, CategoryId=category_id)
session.add(score_obj)
session.commit()
stmt = select(CategoryGroup).where(CategoryGroup.GroupId == group_id)
group = session.scalar(stmt)
# edit message
print(f"Editting message {guild.highscore_channel}")
channel = interaction.guild.get_channel(guild.highscore_channel)
channel = interaction.guild.get_channel(group.ChannelId)
error_response = "Registered Highscores channel does not exist or was never registered."
error_response += 'Register with "?register" command.'
error_response += 'Register with "/register" command.'
assert channel is not None, await interaction.send(error_response, ephemeral=True)
await send_highscore_message(channel, guild, table)
await send_highscore_message(channel, group)
def add_preset_configuration(guild_id: int, config_str: str) -> None:
"""
Adds an example configuration to a Guild. Uses guild id to know which guild and
config name to open a file in example_config.
Args:
guild_id (int): Guild id to add config to
config_str (str): Config name to add
"""
with open(f"example_config/{config_str}.json", encoding="utf-8", mode="r") as file:
config = json.load(file)
for group_name, categorys in config["tables"].items():
group_obj = CategoryGroup(GroupName=group_name, GuildId=guild_id)
session.add(group_obj)
for category_name in categorys:
cat_obj = Category(
CategoryName=category_name,
Size=config["table_size"],
IsTimeRecord=config["categories"][category_name]["is_time_record"],
IsAscending=config["categories"][category_name]["ascending"],
)
session.add(cat_obj)
session.commit()