Compare commits
237
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a0a7e37921 | ||
|
|
c01c888503 | ||
|
|
58dbe547bb | ||
|
|
3cf4fc6b33 | ||
|
|
411f8ab275 | ||
|
|
8952c4a352 | ||
|
|
93248797a0 | ||
|
|
dace72f5cc | ||
|
|
f468a028d0 | ||
|
|
a9b889234a | ||
|
|
88cec73644 | ||
|
|
91e57c6555 | ||
|
|
6c5f40f788 | ||
|
|
a9920db381 | ||
|
|
d04b8c5e32 | ||
|
|
3627419e12 | ||
|
|
3446ce9ca2 | ||
|
|
582b88e248 | ||
|
|
2db744d170 | ||
|
|
fa33be1b4a | ||
|
|
296e14f0f6 | ||
|
|
4c8444d077 | ||
|
|
2e2030a533 | ||
|
|
f4ec272567 | ||
|
|
b7dd44a028 | ||
|
|
081e406d51 | ||
|
|
e3a937649f | ||
|
|
d09c9b2b03 | ||
|
|
8624285d09 | ||
|
|
f1cbe81699 | ||
|
|
aea2e0beb3 | ||
|
|
fdd08bc1a2 | ||
|
|
98dbd30327 | ||
|
|
2565b7689f | ||
|
|
52305c0107 | ||
|
|
bed07d440b | ||
|
|
4e832a084f | ||
|
|
cc20868bde | ||
|
|
14602282e9 | ||
|
|
2e3f298036 | ||
|
|
44cb9025b0 | ||
|
|
02145ba244 | ||
|
|
58e616ea47 | ||
|
|
2ec7296c0f | ||
|
|
d0f328dd2d | ||
|
|
2154d713cb | ||
|
|
ea0ebcc2e9 | ||
|
|
b9cc3b46bb | ||
|
|
439ad539a8 | ||
|
|
10e8e5a446 | ||
|
|
34e55b962d | ||
|
|
1fdb9ff1b6 | ||
|
|
4f255a32e2 | ||
|
|
814baa27d6 | ||
|
|
9531dca892 | ||
|
|
18df783d86 | ||
|
|
03d7485037 | ||
|
|
1a55ae0ac5 | ||
|
|
76d6ba7969 | ||
|
|
f95d3130ff | ||
|
|
09696071ef | ||
|
|
e5d11d200c | ||
|
|
6a87622b58 | ||
|
|
9bc88a7e03 | ||
|
|
4c0e6a6209 | ||
|
|
017cb7c9df | ||
|
|
17867671f1 | ||
|
|
8896e41726 | ||
|
|
12a55dd626 | ||
|
|
77a32b3adc | ||
|
|
2f10a1c7e6 | ||
|
|
a3bed52479 | ||
|
|
6b7dd6042d | ||
|
|
c4d5477f8b | ||
|
|
f5342f1541 | ||
|
|
1e336e290d | ||
|
|
9ebbca2557 | ||
|
|
8a2108fc14 | ||
|
|
03c7a9303a | ||
|
|
a0c5527bf6 | ||
|
|
8155fb0c8f | ||
|
|
06b46ef27c | ||
|
|
1858f0c37d | ||
|
|
792a225987 | ||
|
|
2e324dc9c5 | ||
|
|
09f7a4323c | ||
|
|
dd73e9b472 | ||
|
|
b4f5ddc822 | ||
|
|
d2f33959ed | ||
|
|
652533d071 | ||
|
|
73e350c845 | ||
|
|
9dec15dbfa | ||
|
|
41ba4dc23c | ||
|
|
790cff3406 | ||
|
|
7fb87ae12f | ||
|
|
2407e61913 | ||
|
|
6147263647 | ||
|
|
cb71cbe1d5 | ||
|
|
080356fe9c | ||
|
|
d4e19284f0 | ||
|
|
ae9153b78e | ||
|
|
ca58e9ac26 | ||
|
|
432888e468 | ||
|
|
c918b9492b | ||
|
|
caef6a3652 | ||
|
|
b2acafd6e7 | ||
|
|
349fdfe18d | ||
|
|
efed6e001f | ||
|
|
6b0e61cdf7 | ||
|
|
0acfa77492 | ||
|
|
ffefd08761 | ||
|
|
a61662429a | ||
|
|
57244bcbfe | ||
|
|
3fd9ff46a3 | ||
|
|
d4ad727f2a | ||
|
|
516ff7feeb | ||
|
|
0eb332aefd | ||
|
|
ff07787a9a | ||
|
|
b5959ccc33 | ||
|
|
b80f160914 | ||
|
|
3b866be3ab | ||
|
|
e55b22c821 | ||
|
|
f06e6f1e1e | ||
|
|
5f3e7361b4 | ||
|
|
835611a11e | ||
|
|
ee9a23ca79 | ||
|
|
448bed17ec | ||
|
|
b4d0f8a15e | ||
|
|
72a97f6cb7 | ||
|
|
8aadf07968 | ||
|
|
a6f98d11ab | ||
|
|
4ee0593619 | ||
|
|
d4db9b51f0 | ||
|
|
f67eb75c7b | ||
|
|
6f4322bf6a | ||
|
|
66ace947c6 | ||
|
|
4d9c70bf09 | ||
|
|
4c944c8323 | ||
|
|
b65254c2cc | ||
|
|
5b3c9ef247 | ||
|
|
027da38574 | ||
|
|
80e4602621 | ||
|
|
07310c804a | ||
|
|
4641ce4048 | ||
|
|
142842a36b | ||
|
|
6371fc4e8c | ||
|
|
3fac5f7a06 | ||
|
|
80f3966d39 | ||
|
|
69a439d13a | ||
|
|
251ac2cf47 | ||
|
|
2dc0350213 | ||
|
|
7e40937cdd | ||
|
|
761a7df29f | ||
|
|
b3a84fc3a8 | ||
|
|
262bcd07b9 | ||
|
|
8ca77d93b3 | ||
|
|
1b4f0362b5 | ||
|
|
ce60288892 | ||
|
|
2515b90497 | ||
|
|
3ce9f8090f | ||
|
|
68da97e138 | ||
|
|
ef28b97a1b | ||
|
|
e43a0a8977 | ||
|
|
2ee2d3405e | ||
|
|
af204a3156 | ||
|
|
109b22be62 | ||
|
|
aa4b06c125 | ||
|
|
96fce1d661 | ||
|
|
3d471198f5 | ||
|
|
05d0653a5b | ||
|
|
6758f4e90d | ||
|
|
6e6b57c49c | ||
|
|
b4f72b9528 | ||
|
|
cd81ee3140 | ||
|
|
6ddba14908 | ||
|
|
0fc0fc33da | ||
|
|
529761b45f | ||
|
|
bac64110c1 | ||
|
|
d04de253e7 | ||
|
|
8c7a801800 | ||
|
|
01df80d358 | ||
|
|
06248cfe6c | ||
|
|
a3af93d4bc | ||
|
|
646ee0274b | ||
|
|
fe0a74cfc4 | ||
|
|
5bed587da9 | ||
|
|
f7a40f83a9 | ||
|
|
a66bc4982f | ||
|
|
04051eeacc | ||
|
|
f33dc8ce37 | ||
|
|
306f983c8e | ||
|
|
9e9d616814 | ||
|
|
5a09beea31 | ||
|
|
8c2b3fa487 | ||
|
|
62e89960f7 | ||
|
|
e2b0c0ea89 | ||
|
|
3eed30567d | ||
|
|
565babd1d3 | ||
|
|
455b51a998 | ||
|
|
f136181216 | ||
|
|
463ef48052 | ||
|
|
64004329da | ||
|
|
f497928054 | ||
|
|
0911ebfa45 | ||
|
|
beedf946c9 | ||
|
|
5b31314f02 | ||
|
|
4b15fef30c | ||
|
|
e11a86af7b | ||
|
|
f4f8f426d0 | ||
|
|
b8eb84165e | ||
|
|
273ff757af | ||
|
|
f84e2eef29 | ||
|
|
b4eb5ab6d1 | ||
|
|
1ae6a4a223 | ||
|
|
e1a429c42a | ||
|
|
790e8ca50f | ||
|
|
7ba43ea4f4 | ||
|
|
39c858af64 | ||
|
|
093861b2e6 | ||
|
|
4ef487a8c4 | ||
|
|
849ebb966a | ||
|
|
f1b0e34194 | ||
|
|
7cdc7d2cef | ||
|
|
db4672dcda | ||
|
|
fc2b601687 | ||
|
|
493d44fa19 | ||
|
|
5d3a1568a2 | ||
|
|
be29194a57 | ||
|
|
d067da2300 | ||
|
|
6d0c10e1cc | ||
|
|
609bfa1277 | ||
|
|
158e53e799 | ||
|
|
3cc67a4e9e | ||
|
|
df80eb2952 | ||
|
|
b78eb7b9bf | ||
|
|
65ce143cfc | ||
|
|
56984b3df0 |
+4
-1
@@ -132,5 +132,8 @@ nextcord.log
|
||||
|
||||
# data file for bot
|
||||
highscores_data.json
|
||||
highscores_message_map.json
|
||||
highscores_config.json
|
||||
deleted_backups/
|
||||
highscores_data/
|
||||
TODO.txt
|
||||
sql_tester.py
|
||||
|
||||
@@ -0,0 +1,278 @@
|
||||
"""
|
||||
Contains classes and enum for tracking data on submissions.
|
||||
"""
|
||||
|
||||
from enum import Enum
|
||||
|
||||
from nextcord import Embed
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
# pylint: disable=import-error
|
||||
from utilities.sql_data_classes import Category, CategoryGroup, engine
|
||||
|
||||
|
||||
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)
|
||||
)
|
||||
with Session(engine) as session:
|
||||
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)
|
||||
@@ -0,0 +1,24 @@
|
||||
"""
|
||||
Cog loader. Adds all command cogs to the bot.
|
||||
"""
|
||||
|
||||
from nextcord.ext import commands
|
||||
|
||||
from .create import CreateCog
|
||||
from .delete import DeleteCog
|
||||
from .refresh import RefreshCog
|
||||
from .set import SetCog
|
||||
from .update import UpdateCog
|
||||
|
||||
|
||||
# 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(CreateCog(bot))
|
||||
bot.add_cog(DeleteCog())
|
||||
bot.add_cog(RefreshCog(bot))
|
||||
bot.add_cog(SetCog())
|
||||
bot.add_cog(UpdateCog(bot))
|
||||
@@ -0,0 +1,172 @@
|
||||
"""
|
||||
Commands for creating items in bot.
|
||||
"""
|
||||
|
||||
from typing import Optional
|
||||
|
||||
import nextcord
|
||||
from nextcord.ext import commands
|
||||
from sqlalchemy import exc, select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
# pylint: disable=import-error
|
||||
from cogs.submission.submission_views import SubmissionCreateButton
|
||||
from utilities.sql_data_classes import Category, CategoryGroup, engine
|
||||
from utilities.utils import MAX_LINE_LENGTH, autofill_groups, send_highscore_message
|
||||
|
||||
|
||||
class CreateCog(commands.Cog, name="Create Cog"):
|
||||
"""
|
||||
Class holds commands for creating things.
|
||||
"""
|
||||
|
||||
def __init__(self, bot: commands.Bot) -> None:
|
||||
"""
|
||||
Default constructor for CreateCog class.
|
||||
"""
|
||||
self._bot = bot
|
||||
|
||||
@nextcord.slash_command(default_member_permissions=nextcord.Permissions(administrator=True))
|
||||
async def create(self, interaction: nextcord.Interaction):
|
||||
"""
|
||||
Parent slash command for creating things. Never called.
|
||||
"""
|
||||
|
||||
# need all these arguments because discord command.
|
||||
# pylint: disable=too-many-arguments
|
||||
@create.subcommand(description="Create a new Category. Must not exist already in Group.")
|
||||
async def category(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
group_id: str = nextcord.SlashOption(name="group", description="Group to add Category to.", required=True),
|
||||
category_name: str = nextcord.SlashOption(description="Name of new Category.", required=True),
|
||||
is_ascending: bool = nextcord.SlashOption(description="If higher value is better score.", required=True),
|
||||
is_time_record: bool = nextcord.SlashOption(
|
||||
description="If the score should be shown in time format. Cannot be changed.", required=True
|
||||
),
|
||||
is_validated: bool = nextcord.SlashOption(
|
||||
description="If scores need to be validated before posted.", required=True
|
||||
),
|
||||
size: int = nextcord.SlashOption(
|
||||
description="How many scores to post in Category. Default is 3.", required=False, default=3, min_value=1
|
||||
),
|
||||
) -> None:
|
||||
"""
|
||||
Creates a new category to be placed into group_id group. Fails if the category name already exists within the
|
||||
group. Then calls send_highscore_message().
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Interaction for the command.
|
||||
group_id (str): Id of group to add category to. Is SlashOption.
|
||||
category_name (str): Name of new category. Is SlashOption.
|
||||
is_ascending (bool): If the scores are sorted ascending or not. Is SlashOption.
|
||||
is_time_record (bool): If the scores are a time record or not. Is SlashOption.
|
||||
is_validated (bool): If the scores should be validated or not. Is SlashOption.
|
||||
size (int): How many sores to show in the category message. Is SlashOption.
|
||||
"""
|
||||
|
||||
new_category = Category(
|
||||
CategoryName=category_name,
|
||||
GroupId=int(group_id),
|
||||
Size=size,
|
||||
IsAscending=is_ascending,
|
||||
IsTimeRecord=is_time_record,
|
||||
IsValidated=is_validated,
|
||||
)
|
||||
|
||||
# Open connection to engine and add new category.
|
||||
with Session(engine) as session:
|
||||
stmt = select(CategoryGroup).where(CategoryGroup.GroupId == int(group_id))
|
||||
group = session.scalar(stmt)
|
||||
try:
|
||||
session.add(new_category)
|
||||
session.commit()
|
||||
await send_highscore_message(session, interaction.guild, group)
|
||||
await interaction.response.send_message(
|
||||
f'"{category_name}" has been added to "{group.GroupName}"!', ephemeral=True
|
||||
)
|
||||
except exc.IntegrityError:
|
||||
# Category name already exists in group
|
||||
session.rollback()
|
||||
await interaction.response.send_message(
|
||||
f'"{category_name}" already exists in "{group.GroupName}". Please rename.', ephemeral=True
|
||||
)
|
||||
|
||||
# need all these arguments because discord command.
|
||||
# pylint: disable=too-many-arguments
|
||||
@create.subcommand(description="Create a new Category Group. Must not exist already for Guild.")
|
||||
async def group(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
group_name: str = nextcord.SlashOption(description="Name of new Group.", required=True),
|
||||
pad_length: int = nextcord.SlashOption(
|
||||
description="Number of characters wide a category is printed. Default is 26.",
|
||||
required=False,
|
||||
default=26,
|
||||
min_value=7,
|
||||
max_value=MAX_LINE_LENGTH,
|
||||
),
|
||||
username_length: int = nextcord.SlashOption(
|
||||
description="Number of characters of a username to display. Default is 12.",
|
||||
required=False,
|
||||
default=12,
|
||||
min_value=1,
|
||||
max_value=30,
|
||||
),
|
||||
channel: Optional[nextcord.abc.GuildChannel] = nextcord.SlashOption(
|
||||
description="Channel to post Group in. If left empty, defaults to current channel.",
|
||||
required=False,
|
||||
channel_types=[nextcord.ChannelType.text],
|
||||
default=None,
|
||||
),
|
||||
) -> None:
|
||||
"""
|
||||
Creates a new category group to be placed into the database with the guild's id. Fails if the category group
|
||||
name already exists within the guild. Then calls send_highscore_message().
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Interaction for the command.
|
||||
group_name (str): Name of new Group. Is SlashOption.
|
||||
pad_length (int): Number of characters wide a category is printed. Is SlashOption.
|
||||
username_length (int): Number of characters of a username to display. Is SlashOption.
|
||||
channel (Optional[nextcord.abc.GuildChannel]): Channel to post group in. If None, defaults to current
|
||||
channel. Is SlashOption.
|
||||
"""
|
||||
|
||||
# if no channel provided, use current channel.
|
||||
if channel is None:
|
||||
channel = interaction.channel
|
||||
|
||||
new_group = CategoryGroup(
|
||||
GroupName=group_name,
|
||||
ChannelId=channel.id,
|
||||
GuildId=interaction.guild_id,
|
||||
PadLength=pad_length,
|
||||
UsernameLength=username_length,
|
||||
)
|
||||
|
||||
with Session(engine) as session:
|
||||
try:
|
||||
session.add(new_group)
|
||||
session.commit()
|
||||
message: nextcord.Message = await send_highscore_message(session, interaction.guild, new_group)
|
||||
await message.edit(view=SubmissionCreateButton(self._bot))
|
||||
await interaction.response.send_message(f'"{group_name}" has been created!', ephemeral=True)
|
||||
except exc.IntegrityError:
|
||||
# Category Group already exists in guild
|
||||
session.rollback()
|
||||
await interaction.response.send_message(
|
||||
f'"{group_name}" already exists in "{interaction.guild.name}". Please rename.', ephemeral=True
|
||||
)
|
||||
|
||||
@category.on_autocomplete("group_id")
|
||||
async def group_autocomplete(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))
|
||||
@@ -0,0 +1,297 @@
|
||||
"""
|
||||
Commands for deleting groups/categories/scores in database and messages in discord..
|
||||
"""
|
||||
|
||||
from typing import Optional
|
||||
|
||||
import nextcord
|
||||
from nextcord.ext import commands
|
||||
from sqlalchemy import distinct, select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
# pylint: disable=import-error
|
||||
from utilities.sql_data_classes import Category, CategoryGroup, Score, engine
|
||||
from utilities.utils import (
|
||||
autofill_categories,
|
||||
autofill_groups,
|
||||
autofill_users_in_group_category,
|
||||
autofill_users_in_guild,
|
||||
send_highscore_message,
|
||||
)
|
||||
|
||||
from .views.delete_confirm import DeleteConfirmView, create_delete_id
|
||||
|
||||
|
||||
class DeleteCog(commands.Cog, name="Table Management"):
|
||||
"""
|
||||
Class holds commands for deleting.
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""
|
||||
Default constructor for DeleteCog class.
|
||||
"""
|
||||
|
||||
@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 all(self, interaction: nextcord.Interaction):
|
||||
"""
|
||||
Subcommand parent for deleting all of a thing. Never called.
|
||||
"""
|
||||
|
||||
@all.subcommand(description="Clears highscores channel of all table messages. Does not change data.")
|
||||
async def 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
|
||||
|
||||
with Session(engine) as session:
|
||||
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)
|
||||
|
||||
@all.subcommand(
|
||||
description="Removes all of a user's scores. Warning: data is destroyed, use with case.",
|
||||
)
|
||||
async def users(
|
||||
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
|
||||
"""
|
||||
|
||||
with Session(engine) as session:
|
||||
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(session, interaction.guild, group)
|
||||
|
||||
await interaction.send(f'User "{user_name}" was removed from all tables.', ephemeral=True)
|
||||
|
||||
@delete.subcommand(description="Delete a category from a group. Deletion is perminant.")
|
||||
async def category(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
group_id: str = nextcord.SlashOption(name="group", description="Group to delete from.", required=True),
|
||||
category_id: str = nextcord.SlashOption(name="category", description="Category to delete.", required=True),
|
||||
) -> None:
|
||||
"""
|
||||
Calls the delete confirmation view for the given group and category.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Interaction of command.
|
||||
group_id (str): Group to delete from. Is SlashOption.
|
||||
category_id (str): Category to delete. Is SlashOption.
|
||||
"""
|
||||
await interaction.response.send_modal(
|
||||
DeleteConfirmView(custom_id=create_delete_id(int(group_id), int(category_id)))
|
||||
)
|
||||
|
||||
@delete.subcommand(description="Delete a group. Deletion is perminant.")
|
||||
async def group(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
group_id: str = nextcord.SlashOption(name="group", description="Group to delete.", required=True),
|
||||
) -> None:
|
||||
"""
|
||||
Calls the delete confirmation view for the given group.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Interaction of command.
|
||||
group_id (str): Group to delete from. Is SlashOption.
|
||||
"""
|
||||
await interaction.response.send_modal(DeleteConfirmView(custom_id=create_delete_id(int(group_id))))
|
||||
|
||||
@delete.subcommand()
|
||||
async def user(self, interaction: nextcord.Interaction):
|
||||
"""
|
||||
Subcommand parent for deleting user information. Never called.
|
||||
"""
|
||||
|
||||
@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.
|
||||
"""
|
||||
|
||||
with Session(engine) as session:
|
||||
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(session, interaction.guild, 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.
|
||||
"""
|
||||
|
||||
with Session(engine) as session:
|
||||
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(session, interaction.guild, 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")
|
||||
@category.on_autocomplete("group_id")
|
||||
@group.on_autocomplete("group_id")
|
||||
async def group_autocomplete(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")
|
||||
@category.on_autocomplete("category_id")
|
||||
async def category_autocomplete(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 user_autocomplete(
|
||||
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
|
||||
)
|
||||
)
|
||||
|
||||
@users.on_autocomplete("user_name")
|
||||
async def all_users_autocomplete(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))
|
||||
@@ -0,0 +1,104 @@
|
||||
"""
|
||||
Commands for refreshing highscore messages.
|
||||
"""
|
||||
|
||||
import nextcord
|
||||
from nextcord.ext import commands
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
# pylint: disable=import-error
|
||||
from cogs.submission.submission_views import SubmissionCreateButton
|
||||
from utilities.sql_data_classes import CategoryGroup, engine
|
||||
from utilities.utils import autofill_groups, send_highscore_message
|
||||
|
||||
|
||||
class RefreshCog(commands.Cog, name="Refresh Cog"):
|
||||
"""
|
||||
Class holds commands for refreshing.
|
||||
"""
|
||||
|
||||
def __init__(self, bot: commands.Bot) -> None:
|
||||
"""
|
||||
Default constructor for RefreshCog class
|
||||
|
||||
Args:
|
||||
bot (commands.Bot): Bot object. Required to register submission button view.
|
||||
"""
|
||||
self._bot = bot
|
||||
|
||||
@nextcord.slash_command(default_member_permissions=nextcord.Permissions(administrator=True))
|
||||
async def refresh(self, interaction: nextcord.Interaction):
|
||||
"""
|
||||
Parent slash command for refreshing things. Never called.
|
||||
"""
|
||||
|
||||
@refresh.subcommand(description="Manually refresh a table. Use when changing settings or data manually changed.")
|
||||
async def 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.
|
||||
"""
|
||||
|
||||
with Session(engine) as session:
|
||||
stmt = (
|
||||
select(CategoryGroup)
|
||||
.where(CategoryGroup.GuildId == interaction.guild_id)
|
||||
.where(CategoryGroup.GroupId == int(group_id))
|
||||
)
|
||||
group = session.scalar(stmt)
|
||||
|
||||
message: nextcord.Message = await send_highscore_message(session, interaction.guild, group)
|
||||
await message.edit(view=SubmissionCreateButton(self._bot))
|
||||
|
||||
await interaction.response.send_message(f"Updated table {group.GroupName}!", ephemeral=True)
|
||||
|
||||
@refresh.subcommand()
|
||||
async def all(self, interaction: nextcord.Interaction):
|
||||
"""
|
||||
Subcommand parent for refreshing all things. Never called.
|
||||
"""
|
||||
|
||||
@all.subcommand(description="Triggers post/update of highscores to registered highscore channels.")
|
||||
async def 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.
|
||||
"""
|
||||
|
||||
with Session(engine) as session:
|
||||
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):
|
||||
message: nextcord.Message = await send_highscore_message(session, interaction.guild, group)
|
||||
await message.edit(view=SubmissionCreateButton(self._bot))
|
||||
|
||||
await interaction.send(f"Posted all groups in {interaction.guild.name}'s highscores.", ephemeral=True)
|
||||
|
||||
@group.on_autocomplete("group_id")
|
||||
async def group_autocomplete(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))
|
||||
@@ -0,0 +1,212 @@
|
||||
"""
|
||||
Set command heirarchy.
|
||||
"""
|
||||
|
||||
from typing import Optional
|
||||
|
||||
import nextcord
|
||||
from nextcord.ext import commands
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
# pylint: disable=import-error
|
||||
from utilities.sql_data_classes import CategoryGroup, Guild, engine
|
||||
from utilities.utils import MAX_LINE_LENGTH, autofill_groups
|
||||
|
||||
|
||||
class SetCog(commands.Cog, name="Set command."):
|
||||
"""
|
||||
Class holds commands for setting variables in guild's database.
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""
|
||||
Default constructor for SetCog class. Does nothing.
|
||||
"""
|
||||
|
||||
@nextcord.slash_command(default_member_permissions=nextcord.Permissions(administrator=True))
|
||||
async def set(self, interaction: nextcord.Interaction):
|
||||
"""
|
||||
Parent slash command for setting settings. Never called.
|
||||
"""
|
||||
|
||||
@set.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
|
||||
with Session(engine) as session:
|
||||
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)
|
||||
|
||||
@set.subcommand(
|
||||
description="Set character length for a category. Can cut off username/score. Affects all categories."
|
||||
)
|
||||
async def pad_length(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
group_id: str = nextcord.SlashOption(
|
||||
name="group", description="Category Group to assign pad length to.", required=True
|
||||
),
|
||||
length: int = nextcord.SlashOption(
|
||||
description="Number of characters alloted to a category.",
|
||||
required=True,
|
||||
default=26,
|
||||
min_value=7,
|
||||
max_value=MAX_LINE_LENGTH,
|
||||
),
|
||||
) -> 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.
|
||||
"""
|
||||
|
||||
with Session(engine) as session:
|
||||
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="Registers the channel for submission messages.")
|
||||
async def submission_channel(
|
||||
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
|
||||
|
||||
with Session(engine) as session:
|
||||
channel_id = channel.id
|
||||
stmt = select(Guild).where(Guild.GuildId == interaction.guild_id)
|
||||
guild = session.scalar(stmt)
|
||||
guild.SubmissionChannel = channel_id
|
||||
session.commit()
|
||||
await interaction.send(f"Registered <#{channel_id}>.", 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, min_value=1, max_value=30
|
||||
),
|
||||
) -> 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.
|
||||
"""
|
||||
|
||||
with Session(engine) as session:
|
||||
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)
|
||||
|
||||
@set.subcommand(description="Registers the channel for verification messages.")
|
||||
async def verify_channel(
|
||||
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
|
||||
with Session(engine) as session:
|
||||
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)
|
||||
|
||||
@group_output_channel.on_autocomplete("group_id")
|
||||
@pad_length.on_autocomplete("group_id")
|
||||
@username_length.on_autocomplete("group_id")
|
||||
async def group_autocomplete(self, interaction: nextcord.Interaction, group_id: str):
|
||||
"""
|
||||
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,84 @@
|
||||
"""
|
||||
Commands for updating information/variables held in database.
|
||||
"""
|
||||
|
||||
import nextcord
|
||||
from nextcord.ext import commands
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
# pylint: disable=import-error
|
||||
from utilities.sql_data_classes import Guild, engine
|
||||
from utilities.utils import autofill_users_in_guild
|
||||
|
||||
from .views.update_username_verify import UpdateUsernameVerifyView
|
||||
|
||||
|
||||
class UpdateCog(commands.Cog, name="Update commands."):
|
||||
"""
|
||||
Class holds commands and views needed for updating things in bot.
|
||||
"""
|
||||
|
||||
def __init__(self, bot: commands.Bot):
|
||||
"""
|
||||
Default constructor for Update class. Sets internal bot variable.
|
||||
|
||||
Args:
|
||||
bot (commands.Bot): Bot object. Required to register update username button view.
|
||||
"""
|
||||
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()
|
||||
async def update(self, interaction: nextcord.Interaction):
|
||||
"""
|
||||
Parent slash command for updating things. Never called.
|
||||
"""
|
||||
|
||||
@update.subcommand(description="Request a username update. Admins must approve.")
|
||||
async def 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.
|
||||
"""
|
||||
|
||||
with Session(engine) as session:
|
||||
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)
|
||||
|
||||
@username.on_autocomplete("old_username")
|
||||
async def all_users_autocomplete(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))
|
||||
@@ -0,0 +1,97 @@
|
||||
"""
|
||||
Contains view and helper functions for confirming and processing deletes.
|
||||
"""
|
||||
|
||||
from typing import Optional
|
||||
|
||||
import nextcord
|
||||
from sqlalchemy import delete, select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
# pylint: disable=import-error
|
||||
from utilities.sql_data_classes import Category, CategoryGroup, engine
|
||||
from utilities.utils import send_highscore_message
|
||||
|
||||
|
||||
class DeleteConfirmView(nextcord.ui.Modal):
|
||||
"""
|
||||
View modal popup for confirming deletions.
|
||||
"""
|
||||
|
||||
def __init__(self, custom_id: str):
|
||||
super().__init__(title="Warning, deletion is perminant.", custom_id=custom_id)
|
||||
|
||||
self.checker = nextcord.ui.TextInput(
|
||||
label='Type "delete" to confirm deletion.',
|
||||
required=True,
|
||||
placeholder="delete",
|
||||
style=nextcord.TextInputStyle.short,
|
||||
)
|
||||
self.add_item(self.checker)
|
||||
|
||||
async def callback(self, interaction: nextcord.Interaction):
|
||||
if self.checker.value.lower() != "delete":
|
||||
await interaction.response.send_message('Response was not "delete". Canceling delete.')
|
||||
return
|
||||
|
||||
del_vals = parse_delete_id(self.custom_id)
|
||||
|
||||
with Session(engine) as session:
|
||||
stmt = select(CategoryGroup).where(CategoryGroup.GroupId == del_vals["group_id"])
|
||||
group = session.scalar(stmt)
|
||||
if del_vals["category_id"] is None:
|
||||
# delete entire group
|
||||
session.delete(group)
|
||||
# delete the message related to group
|
||||
channel: nextcord.TextChannel = interaction.guild.get_channel(group.ChannelId)
|
||||
message: nextcord.Message = await channel.fetch_message(group.MessageId)
|
||||
await message.delete()
|
||||
|
||||
else:
|
||||
stmt = (
|
||||
delete(Category)
|
||||
.where(Category.GroupId == del_vals["group_id"])
|
||||
.where(Category.CategoryId == del_vals["category_id"])
|
||||
)
|
||||
session.execute(stmt)
|
||||
# update the message
|
||||
await send_highscore_message(session, interaction.guild, group)
|
||||
session.commit()
|
||||
|
||||
await interaction.response.send_message("Deletion complete.", ephemeral=True)
|
||||
|
||||
|
||||
def create_delete_id(group_id: int, category_id: Optional[int] = None) -> str:
|
||||
"""
|
||||
Creates a delete id string to tell the view what to delete.
|
||||
|
||||
Args:
|
||||
group_id (int): group id to delete
|
||||
category_id (Optional[int]): category id to delete. Can be None.
|
||||
|
||||
Returns:
|
||||
str: string representation of the delete id.
|
||||
"""
|
||||
return f"delete-{group_id}-{category_id}"
|
||||
|
||||
|
||||
def parse_delete_id(delete_id: str) -> dict:
|
||||
"""
|
||||
Returns a dict containing the values of a delete id.
|
||||
|
||||
Args:
|
||||
id (str): the delete id
|
||||
|
||||
Returns:
|
||||
dict: dict containing values from a delete id.
|
||||
"""
|
||||
vals = {}
|
||||
group_id = delete_id.split("-")[1]
|
||||
category_id = delete_id.split("-")[2]
|
||||
vals["group_id"] = group_id
|
||||
if category_id == "None":
|
||||
vals["category_id"] = None
|
||||
else:
|
||||
vals["category_id"] = category_id
|
||||
|
||||
return vals
|
||||
@@ -0,0 +1,76 @@
|
||||
"""View for allowing a username update."""
|
||||
|
||||
import nextcord
|
||||
from sqlalchemy import distinct, exc, select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
# pylint: disable=import-error
|
||||
from utilities.sql_data_classes import Category, CategoryGroup, Score, engine
|
||||
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
|
||||
"""
|
||||
|
||||
with Session(engine) as session:
|
||||
try:
|
||||
embed_content = interaction.message.embeds[0].to_dict()
|
||||
old_username: str = embed_content["fields"][0]["value"]
|
||||
new_username: str = embed_content["fields"][1]["value"]
|
||||
|
||||
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(session, interaction.guild, 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)
|
||||
@@ -0,0 +1,35 @@
|
||||
"""
|
||||
Defines cog for guild listeners. Contains functions for various guild events.
|
||||
"""
|
||||
|
||||
import nextcord
|
||||
from nextcord.ext import commands
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
# pylint: disable=import-error
|
||||
from utilities.sql_data_classes import Guild, engine
|
||||
from utilities.utils import add_preset_configuration
|
||||
|
||||
|
||||
class GuildListener(commands.Cog, name="Guild Listener"):
|
||||
"""Sets up message listener."""
|
||||
|
||||
def __init__(self, bot: commands.Bot):
|
||||
self.bot = bot
|
||||
|
||||
@commands.Cog.listener()
|
||||
async def on_guild_join(self, guild: nextcord.Guild):
|
||||
"""Listener for Guild Join. Adds guild to database."""
|
||||
|
||||
new_guild = Guild(GuildId=guild.id)
|
||||
|
||||
with Session(engine) as session:
|
||||
session.add(new_guild)
|
||||
session.commit()
|
||||
|
||||
|
||||
# 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(GuildListener(bot))
|
||||
@@ -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))
|
||||
@@ -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
|
||||
@@ -4,14 +4,15 @@ Defines cog for on message listener.
|
||||
|
||||
import re
|
||||
|
||||
from nextcord.ext import commands
|
||||
import nextcord
|
||||
from nextcord.ext import commands
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from cogs.submission.submission_views import SubmissionState # pylint: disable=import-error
|
||||
from cogs.submission.submission_views import get_submission_embed # pylint: disable=import-error
|
||||
from highscores import highscores_config # pylint: disable=import-error
|
||||
from highscores import highscores_data # pylint: disable=import-error
|
||||
from highscores import submission_objects # pylint: disable=import-error
|
||||
# pylint: disable=import-error
|
||||
from classes.submission import SubmissionState
|
||||
from highscores import submission_objects
|
||||
from utilities.sql_data_classes import Category, Guild, engine
|
||||
|
||||
|
||||
class SubmissionListener(commands.Cog, name="Message Listener"):
|
||||
@@ -23,63 +24,74 @@ class SubmissionListener(commands.Cog, name="Message Listener"):
|
||||
@commands.Cog.listener()
|
||||
async def on_message(self, message: nextcord.Message):
|
||||
"""Listener for message creation. Adds score and attachments to submission."""
|
||||
if message.author.id in submission_objects:
|
||||
message_id = submission_objects[message.author.id]["message_id"]
|
||||
submission_message: nextcord.Message = await self.bot.get_channel(
|
||||
highscores_data["submission_channel_id"]
|
||||
).fetch_message(message_id)
|
||||
# don't want to process any messages from bot.
|
||||
if message.author.id == self.bot.user.id:
|
||||
return
|
||||
|
||||
guild_id = message.guild.id
|
||||
|
||||
if submission_objects.has_user(message.author.id):
|
||||
submission = submission_objects.get_user_submission(message.author.id)
|
||||
message_id = submission.message_id
|
||||
|
||||
with Session(engine) as session:
|
||||
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)
|
||||
)
|
||||
|
||||
# need check if correct user and correct channel
|
||||
if message.channel.id == highscores_data["submission_channel_id"]:
|
||||
state = submission_objects[message.author.id]["submission_state"]
|
||||
if message.channel.id == submission_channel_id:
|
||||
state = submission.submission_state
|
||||
if state == SubmissionState.SCORE and message.content:
|
||||
|
||||
# check with regex if score matches correctly
|
||||
is_time_record = highscores_config["categories"][
|
||||
submission_objects[message.author.id]["category"]
|
||||
]["is_time_record"]
|
||||
stmt = select(Category.IsTimeRecord).where(Category.CategoryId == submission.category_id)
|
||||
|
||||
time_regex = re.search(
|
||||
"([0-9]{2}:)?[0-9]{2}(\.[0-9])?", # pylint: disable=anomalous-backslash-in-string
|
||||
message.content,
|
||||
)
|
||||
int_regex = re.search("[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 or SS.ms or SS.",
|
||||
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:
|
||||
# standardize time score format.
|
||||
# add colon if its not there
|
||||
if score.find(":") == -1:
|
||||
score = "00:" + score
|
||||
# add a 0 if there isn't 2 digits before a colon
|
||||
elif len(score.split(":", maxsplit=1)[0]) < 2:
|
||||
score = "0" + score
|
||||
submission_objects[message.author.id]["score"] = score
|
||||
embed = get_submission_embed(message.author.id)
|
||||
|
||||
if re.search("^[1-9]?[0-9]:[0-9]{2}(\\.[0-9])?$", score):
|
||||
# score is already fine.
|
||||
pass
|
||||
elif re.search("^[0-9]{2}(\\.[0-9])?$", score):
|
||||
# need to add leading colon and 0
|
||||
score = "0:" + score
|
||||
elif re.search("^[0-9](\\.[0-9])?$", score):
|
||||
# need to add 0 in seconds and leading colon and 0
|
||||
score = "0:0" + score
|
||||
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.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["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()
|
||||
|
||||
@@ -1,54 +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_config # pylint: disable=import-error
|
||||
from highscores import highscores_data # pylint: disable=import-error
|
||||
from utilities.data_storage import save_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
|
||||
```
|
||||
"""
|
||||
highscore_channel_id = highscores_data["highscore_channel_id"]
|
||||
print(f"Printing out highscores information in channel {highscore_channel_id}")
|
||||
channel = self.bot.get_channel(highscore_channel_id)
|
||||
|
||||
error_response = 'Registered Highscores channel does not exist or was never registered. \n\
|
||||
Register with "?register" command.'
|
||||
assert channel is not None, await ctx.send(error_response)
|
||||
|
||||
for boss, categories in highscores_config[ # pylint: disable=unused-variable
|
||||
"highscore_table"
|
||||
].items():
|
||||
message: nextcord.Message = await send_highscore_message(channel, boss)
|
||||
await message.edit(view=SubmissionCreateButton(self.bot))
|
||||
|
||||
# Saving in case new message was sent.
|
||||
save_highscores_data(highscores_data)
|
||||
|
||||
|
||||
# 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))
|
||||
@@ -0,0 +1,55 @@
|
||||
"""
|
||||
Functions for controlling the bot's presence.
|
||||
"""
|
||||
|
||||
import random
|
||||
|
||||
import nextcord
|
||||
from nextcord.ext import commands, tasks
|
||||
|
||||
|
||||
class BotPresence(commands.Cog, name="Bot Presence"):
|
||||
"""Class for bot presence functions."""
|
||||
|
||||
def __init__(self, bot: commands.Bot):
|
||||
self.bot = bot
|
||||
|
||||
@commands.Cog.listener()
|
||||
async def on_guild_join(self, guild: nextcord.Guild): # pylint: disable=unused-argument
|
||||
"""Listener for Guild Join. Updated presence on guild join."""
|
||||
await self.change_presence()
|
||||
|
||||
@commands.Cog.listener()
|
||||
async def on_ready(self):
|
||||
"""Starts the loop when bot is ready."""
|
||||
await self.bot.wait_until_ready()
|
||||
# pylint doesn't recognize that periodic_change_presense is subclassed from tasks.loop.
|
||||
self.periodic_change_presence.start() # pylint: disable=no-member
|
||||
|
||||
@tasks.loop(minutes=2)
|
||||
async def periodic_change_presence(self):
|
||||
"""Loops every 2 minutes and changes presence."""
|
||||
await self.change_presence()
|
||||
|
||||
async def change_presence(self):
|
||||
"""Randomly chooses a presence and updates bot's presence."""
|
||||
statuses = [
|
||||
"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))
|
||||
|
||||
|
||||
# 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(BotPresence(bot))
|
||||
@@ -1,67 +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
|
||||
from utilities.data_storage import save_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):
|
||||
self.bot = bot
|
||||
|
||||
@commands.command(name="register")
|
||||
@commands.has_permissions(administrator=True)
|
||||
async def register_highscores(self, ctx: commands.Context, registration_type: str):
|
||||
"""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
|
||||
|
||||
if registration_type == "verify":
|
||||
await ctx.send(f"Registered <#{channel_id}> as channel to post verifications in.")
|
||||
highscores_data["verification_channel_id"] = channel_id
|
||||
elif registration_type == "highscore":
|
||||
await ctx.send(f"Registered <#{channel_id}> as channel to post highscores in.")
|
||||
highscores_data["highscore_channel_id"] = channel_id
|
||||
elif registration_type == "submission":
|
||||
await ctx.send(f"Registered <#{channel_id}> as channel to post submissions in.")
|
||||
highscores_data["submission_channel_id"] = 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)
|
||||
|
||||
save_highscores_data(highscores_data)
|
||||
|
||||
|
||||
# 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(RegisterVerificationChannel(bot))
|
||||
@@ -1,52 +0,0 @@
|
||||
"""
|
||||
Bot command to submit a score
|
||||
"""
|
||||
|
||||
from nextcord.ext import commands
|
||||
|
||||
from highscores import highscores_data # pylint: disable=import-error
|
||||
from .verification_view import VerificationView
|
||||
|
||||
|
||||
class Score(commands.Cog, name="Score"):
|
||||
"""Submits a score."""
|
||||
|
||||
def __init__(self, bot: commands.Bot):
|
||||
self.bot = bot
|
||||
|
||||
@commands.Cog.listener()
|
||||
async def on_ready(self):
|
||||
"""Called when the cog is loaded"""
|
||||
self.bot.add_view(VerificationView())
|
||||
|
||||
@commands.command(name="score")
|
||||
async def score(self, ctx: commands.Context, boss_name: str, category: str, score: str):
|
||||
"""A command which calls a function to submit the score to highscores.
|
||||
Gathers required information from command and passes on.
|
||||
Usage:
|
||||
```
|
||||
?score "Boss Name" "Category" "score"
|
||||
This command still works if you prefer to use it over the ui.
|
||||
```
|
||||
"""
|
||||
verification_channel_id = highscores_data["verification_channel_id"]
|
||||
channel = self.bot.get_channel(verification_channel_id)
|
||||
|
||||
error_response = "Registered Verification channel does not exist or was never registered. \
|
||||
Contact server admins."
|
||||
assert channel is not None, await ctx.send(error_response)
|
||||
|
||||
verification_string = f"{ctx.author.display_name}:{boss_name}:{category}:{score}"
|
||||
|
||||
await channel.send(
|
||||
content=verification_string,
|
||||
files=[await attch.to_file() for attch in ctx.message.attachments],
|
||||
view=VerificationView(),
|
||||
)
|
||||
|
||||
|
||||
# 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(Score(bot))
|
||||
@@ -1,40 +0,0 @@
|
||||
"""View for verification buttons"""
|
||||
|
||||
import nextcord
|
||||
from utilities.utils import submit_score # pylint: disable=import-error
|
||||
|
||||
|
||||
class VerificationView(nextcord.ui.View):
|
||||
"""Defines views for verifying score submissions"""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__(timeout=None)
|
||||
|
||||
@nextcord.ui.button(
|
||||
label="Approve", style=nextcord.ButtonStyle.green, custom_id="approve-score-command"
|
||||
)
|
||||
async def approve_button(
|
||||
self, button: nextcord.ui.Button, interaction: nextcord.Interaction
|
||||
): # pylint: disable=unused-argument
|
||||
"""Button for approving score submission."""
|
||||
content = interaction.message.content.split(":")
|
||||
user = content[0]
|
||||
boss_name = content[1]
|
||||
category = content[2]
|
||||
score = content[3]
|
||||
|
||||
await submit_score(interaction, user, boss_name, category, score)
|
||||
await interaction.message.delete()
|
||||
await interaction.channel.send(
|
||||
f"Approved: {boss_name} {category} {user} {score}",
|
||||
files=[await attch.to_file() for attch in interaction.message.attachments],
|
||||
)
|
||||
|
||||
@nextcord.ui.button(
|
||||
label="Deny", style=nextcord.ButtonStyle.red, custom_id="deny-score-command"
|
||||
)
|
||||
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()
|
||||
@@ -1,63 +1,31 @@
|
||||
"""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 Session, aliased
|
||||
|
||||
# pylint: disable=import-error
|
||||
from classes.submission import Submission, SubmissionState
|
||||
from highscores import submission_objects
|
||||
from utilities.sql_data_classes import Category, CategoryGroup, Guild, engine
|
||||
|
||||
from highscores import highscores_data # pylint: disable=import-error
|
||||
from highscores import submission_objects # pylint: disable=import-error
|
||||
from highscores import highscores_message_map # pylint: disable=import-error
|
||||
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(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]" = []
|
||||
boss_dict = highscores_data["tables"][boss_name]
|
||||
for category in boss_dict["categories"].items():
|
||||
options.append(nextcord.SelectOption(label=category[0]))
|
||||
|
||||
with Session(engine) as session:
|
||||
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
|
||||
|
||||
@@ -73,16 +41,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,
|
||||
)
|
||||
|
||||
|
||||
@@ -94,11 +61,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):
|
||||
@@ -108,7 +76,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()
|
||||
|
||||
|
||||
@@ -121,23 +89,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):
|
||||
@@ -148,25 +112,28 @@ class SubmitButton(nextcord.ui.Button):
|
||||
self._bot = bot
|
||||
|
||||
async def callback(self, interaction: nextcord.Interaction):
|
||||
verification_channel_id = highscores_data["verification_channel_id"]
|
||||
channel = self._bot.get_channel(verification_channel_id)
|
||||
with Session(engine) as session:
|
||||
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 ############
|
||||
@@ -179,31 +146,36 @@ 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."""
|
||||
boss_name = highscores_message_map[interaction.message.id]
|
||||
channel = self._bot.get_channel(highscores_data["submission_channel_id"])
|
||||
|
||||
# Don't allow a user to start a new submission if one already exists
|
||||
if submission_objects.has_user(interaction.user.id):
|
||||
await interaction.send("Please finish in progress submission first.", ephemeral=True)
|
||||
return
|
||||
|
||||
with Session(engine) as session:
|
||||
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,
|
||||
"submission_state": SubmissionState.CATEGORY,
|
||||
"boss_name": boss_name,
|
||||
"score": "",
|
||||
"category": "",
|
||||
"username": interaction.user.display_name.split("|")[0][
|
||||
: highscores_data["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}>",
|
||||
@@ -211,11 +183,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):
|
||||
@@ -226,17 +198,18 @@ class SubmissionView(nextcord.ui.View):
|
||||
bot: commands.Bot,
|
||||
submission_user: int,
|
||||
*,
|
||||
timeout: Optional[float] = 180,
|
||||
timeout: Optional[float] = 300,
|
||||
):
|
||||
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._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._boss_name)))
|
||||
self.add_item(SubmissionDropdown(bot, category_select_options(self._group_id)))
|
||||
self.add_item(CancelButton())
|
||||
self.add_item(NextButton(self._bot))
|
||||
elif (
|
||||
@@ -251,4 +224,13 @@ class SubmissionView(nextcord.ui.View):
|
||||
self.add_item(SubmitButton(self._bot))
|
||||
|
||||
async def interaction_check(self, interaction: nextcord.Interaction) -> bool:
|
||||
return self._submission_user == interaction.user.id
|
||||
check = self._submission_user == interaction.user.id
|
||||
|
||||
if not check:
|
||||
await interaction.send("This is not your submission.", ephemeral=True)
|
||||
|
||||
return check
|
||||
|
||||
async def on_timeout(self) -> None:
|
||||
# remove user from submission objects after time out. Allows user to start new submission.
|
||||
submission_objects.remove_user_submission(self._submission_user)
|
||||
|
||||
@@ -1,7 +1,9 @@
|
||||
"""View for verification buttons"""
|
||||
|
||||
import nextcord
|
||||
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):
|
||||
@@ -10,37 +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"]
|
||||
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:
|
||||
await submit_score(interaction, user, boss_name, category, 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)
|
||||
|
||||
@@ -1,9 +0,0 @@
|
||||
import os
|
||||
|
||||
from dotenv.main import load_dotenv
|
||||
|
||||
load_dotenv()
|
||||
|
||||
# Discord config
|
||||
DISCORD_TOKEN = os.getenv("DISCORD_TOKEN", "")
|
||||
BOT_PREFIX = "?"
|
||||
@@ -0,0 +1,16 @@
|
||||
version: "3"
|
||||
services:
|
||||
sqldb:
|
||||
restart: unless-stopped
|
||||
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:
|
||||
@@ -0,0 +1,3 @@
|
||||
FROM mysql:latest
|
||||
|
||||
COPY highscores_database.sql /docker-entrypoint-initdb.d/
|
||||
@@ -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)
|
||||
);
|
||||
@@ -1,8 +1,8 @@
|
||||
{
|
||||
"highscore_table": {
|
||||
"tables": {
|
||||
"The Ambassador": [
|
||||
"Kill Count",
|
||||
"Kill Time",
|
||||
"Solo Kill Count",
|
||||
"Solo Kill Time",
|
||||
"Group Kill Count",
|
||||
"Group Kill Time",
|
||||
"Spooned Log"
|
||||
@@ -21,8 +21,8 @@
|
||||
"Spooned Log"
|
||||
],
|
||||
"Astellarn": [
|
||||
"Kill Count",
|
||||
"Kill Time",
|
||||
"Solo Kill Count",
|
||||
"Solo Kill Time",
|
||||
"Group Kill Count",
|
||||
"Group Kill Time",
|
||||
"Spooned Log"
|
||||
@@ -42,8 +42,8 @@
|
||||
"Spooned Log"
|
||||
],
|
||||
"Black Stone Dragon": [
|
||||
"Kill Count",
|
||||
"Kill Time",
|
||||
"Solo Kill Count",
|
||||
"Solo Kill Time",
|
||||
"Group Kill Count",
|
||||
"Group Kill Time",
|
||||
"Spooned Log"
|
||||
@@ -65,8 +65,8 @@
|
||||
"Spooned Log"
|
||||
],
|
||||
"Crassian Leviathan": [
|
||||
"Kill Count",
|
||||
"Kill Time",
|
||||
"Solo Kill Count",
|
||||
"Solo Kill Time",
|
||||
"Group Kill Count",
|
||||
"Group Kill Time",
|
||||
"Spooned Log"
|
||||
@@ -160,8 +160,8 @@
|
||||
"Spooned Log"
|
||||
],
|
||||
"Masuta the Ascended": [
|
||||
"Kill Count",
|
||||
"Kill Time",
|
||||
"Solo Kill Count",
|
||||
"Solo Kill Time",
|
||||
"Group Kill Count",
|
||||
"Group Kill Time",
|
||||
"Spooned Log"
|
||||
@@ -193,15 +193,15 @@
|
||||
"Spooned Log"
|
||||
],
|
||||
"The Sanctum Guardian": [
|
||||
"Kill Count",
|
||||
"Kill Time",
|
||||
"Solo Kill Count",
|
||||
"Solo Kill Time",
|
||||
"Group Kill Count",
|
||||
"Group Kill Time",
|
||||
"Spooned Log"
|
||||
],
|
||||
"Seiryu the Azure Serpent": [
|
||||
"Kill Count",
|
||||
"Kill Time",
|
||||
"Solo Kill Count",
|
||||
"Solo Kill Time",
|
||||
"Group Kill Count",
|
||||
"Group Kill Time",
|
||||
"Spooned Log"
|
||||
@@ -331,5 +331,5 @@
|
||||
"is_time_record": false
|
||||
}
|
||||
},
|
||||
"highscore_size": 3
|
||||
"table_size": 3
|
||||
}
|
||||
+11
-18
@@ -1,17 +1,17 @@
|
||||
"""
|
||||
main file for pvm-highscores bot
|
||||
main file for highscores bot
|
||||
"""
|
||||
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
|
||||
import nextcord
|
||||
from nextcord.ext import commands
|
||||
|
||||
import config
|
||||
from utilities.data_storage import open_highscores_config
|
||||
from utilities.data_storage import open_highscores_data
|
||||
from utilities.data_storage import create_message_map
|
||||
# pylint: disable=import-error
|
||||
from utilities import config
|
||||
from classes.submission import SubmissionMap
|
||||
|
||||
logger = logging.getLogger("nextcord")
|
||||
logger.setLevel(logging.DEBUG)
|
||||
@@ -19,23 +19,22 @@ 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_config = open_highscores_config()
|
||||
highscores_data = open_highscores_data()
|
||||
highscores_message_map = create_message_map(highscores_data)
|
||||
|
||||
submission_objects: dict = {}
|
||||
submission_objects: SubmissionMap = SubmissionMap()
|
||||
|
||||
|
||||
def main():
|
||||
"""Main function for bot."""
|
||||
min_python_version = (3, 8)
|
||||
if sys.version_info < min_python_version:
|
||||
print("ERROR: Python version must be >= 3.8")
|
||||
sys.exit()
|
||||
|
||||
# Allows privledged intents for monitoring members joining, roles editing, and role assignments
|
||||
# These need to be enabled in the developer portal as well
|
||||
intents = nextcord.Intents.default()
|
||||
|
||||
# Required in order to read messages (eg. prefix commands)
|
||||
intents.message_content = True # pylint: disable=assigning-non-slot
|
||||
intents.messages = True # pylint: disable=assigning-non-slot
|
||||
|
||||
# To enable the guilds priveleged intent:
|
||||
# intents.guilds = True
|
||||
@@ -43,15 +42,9 @@ def main():
|
||||
# To enable the members priveleged intent:
|
||||
# intents.members = True
|
||||
|
||||
# Set custom status to "Listening to ?help"
|
||||
activity = nextcord.Activity(
|
||||
type=nextcord.ActivityType.listening, name=f"{config.BOT_PREFIX}help"
|
||||
)
|
||||
|
||||
bot = commands.Bot(
|
||||
commands.when_mentioned_or(config.BOT_PREFIX),
|
||||
intents=intents,
|
||||
activity=activity,
|
||||
)
|
||||
|
||||
# Get the modules of all cogs whose directory structure is ./cogs/<module_name>
|
||||
|
||||
+4
-1
@@ -1,2 +1,5 @@
|
||||
nextcord>=2.0.0rc1
|
||||
nextcord>=2.3.2
|
||||
python-dotenv>=0.20.0
|
||||
pandas>=1.5.2
|
||||
sqlalchemy>=1.4.44
|
||||
mysqlclient>-2.1.1
|
||||
@@ -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", "")
|
||||
@@ -1,64 +0,0 @@
|
||||
"""Functions for retreiving and storing highscores data and configurations."""
|
||||
|
||||
import os
|
||||
import json
|
||||
|
||||
HIGHSCORES_CONFIG_FILE = "highscores_config.json"
|
||||
HIGHSCORES_DATA_FILE = "highscores_data.json"
|
||||
|
||||
|
||||
def open_highscores_config():
|
||||
"""
|
||||
Opens the highscores config file and returns.
|
||||
@return: dict of highscores config.
|
||||
"""
|
||||
# Load bot config
|
||||
with open(HIGHSCORES_CONFIG_FILE, encoding="utf-8", mode="r") as file:
|
||||
return json.load(file)
|
||||
|
||||
|
||||
def open_highscores_data():
|
||||
"""
|
||||
Opens a highscores_data.json file if one exists, otherwise creates a skeleton of the data.
|
||||
@return: dict of highscores data.
|
||||
"""
|
||||
if os.path.exists(HIGHSCORES_DATA_FILE):
|
||||
with open(HIGHSCORES_DATA_FILE, encoding="utf-8", mode="r") as file:
|
||||
data = json.load(file)
|
||||
else:
|
||||
# create the json object
|
||||
data = {"highscore_channel_id": -1, "verification_channel_id": -1, "username_length": 12}
|
||||
config: dict = open_highscores_config()
|
||||
for boss, categories in config["highscore_table"].items():
|
||||
data["tables"][boss] = {"message_id": 0}
|
||||
boss_categories = {}
|
||||
for category in categories: # pylint: disable=unused-variable
|
||||
rank_list = []
|
||||
boss_categories[category] = rank_list
|
||||
data["tables"][boss]["categories"] = boss_categories
|
||||
return data
|
||||
|
||||
|
||||
def create_message_map(data: dict):
|
||||
"""
|
||||
Creates a message map from input highscores data.
|
||||
|
||||
@param data: highscores_data dict to create map from.
|
||||
@return: dict message_map of message to boss.
|
||||
"""
|
||||
|
||||
message_map = {}
|
||||
|
||||
for boss, info in data["tables"].items():
|
||||
message_map[info["message_id"]] = boss
|
||||
|
||||
return message_map
|
||||
|
||||
|
||||
def save_highscores_data(data):
|
||||
"""
|
||||
Saves the highscores data dict into a json file.
|
||||
@param data: dict of highscores data.
|
||||
"""
|
||||
with open(HIGHSCORES_DATA_FILE, encoding="utf-8", mode="w") as file:
|
||||
file.write(json.dumps(data, indent=4))
|
||||
@@ -0,0 +1,20 @@
|
||||
"""
|
||||
Simple file to set up the sqlalchemy connection to database.
|
||||
"""
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.ext.automap import automap_base
|
||||
|
||||
# pylint: disable=import-error
|
||||
from utilities import 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)
|
||||
Base = automap_base()
|
||||
Base.prepare(autoload_with=engine)
|
||||
|
||||
# define the classes
|
||||
Guild = Base.classes.Guilds
|
||||
CategoryGroup = Base.classes.CategoryGroups
|
||||
Category = Base.classes.Categories
|
||||
Score = Base.classes.Scores
|
||||
+356
-111
@@ -2,152 +2,397 @@
|
||||
Contains various utlity functions for highscores bot
|
||||
"""
|
||||
|
||||
import json
|
||||
import time
|
||||
|
||||
import nextcord
|
||||
from highscores import highscores_config # pylint: disable=import-error
|
||||
from highscores import highscores_data # pylint: disable=import-error
|
||||
from highscores import highscores_message_map # pylint: disable=import-error
|
||||
from utilities.data_storage import save_highscores_data # pylint: disable=import-error
|
||||
import pandas as pd
|
||||
from sqlalchemy import distinct, select
|
||||
from sqlalchemy.orm import Session, aliased
|
||||
|
||||
# pylint: disable=import-error
|
||||
from utilities.sql_data_classes import Category, CategoryGroup, Guild, Score, engine
|
||||
|
||||
MAX_LINE_LENGTH = 142
|
||||
|
||||
|
||||
def format_highscore_message(boss_name: str):
|
||||
def autofill_categories(guild_id: int, group_id: int, category_begin: str) -> "dict[str, str]":
|
||||
"""
|
||||
returns a formatted string for a boss's highscore
|
||||
@param boss: the dictionary object for the boss
|
||||
@return the formatted string
|
||||
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.
|
||||
"""
|
||||
ret_string = f"**{boss_name}**\n```"
|
||||
boss_data = highscores_data["tables"][boss_name]
|
||||
|
||||
# Gather all categories and add to string
|
||||
for key, value in boss_data["categories"].items():
|
||||
ret_string += f"{key}".ljust(26, " ")
|
||||
ret_string += "\n"
|
||||
with Session(engine) as session:
|
||||
# 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)}
|
||||
|
||||
highscore_size = highscores_config["highscore_size"]
|
||||
# if string empty, return all category names
|
||||
if not category_begin:
|
||||
return categories_dict
|
||||
|
||||
# list rankings in a single line. Start with first place for each category and then continue
|
||||
for i in range(highscore_size):
|
||||
for key, value in boss_data["categories"].items():
|
||||
if len(value) > i:
|
||||
score_pair = value[i]
|
||||
# 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
|
||||
|
||||
# pad username to max length.
|
||||
username: str = score_pair[0].ljust(highscores_data["username_length"], " ")
|
||||
score: str = str(score_pair[1])
|
||||
|
||||
if highscores_config["categories"][key]["is_time_record"]:
|
||||
# pad score to correct length
|
||||
# MM:SS.ms is 7 characters long
|
||||
score = score.ljust(7, " ")
|
||||
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.
|
||||
"""
|
||||
|
||||
with Session(engine) as session:
|
||||
# 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.
|
||||
"""
|
||||
|
||||
with Session(engine) as session:
|
||||
# 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))
|
||||
|
||||
line_str = f"{i+1}: {username} - "
|
||||
ret_string += line_str + score.rjust(26 - len(line_str) - 1) + " "
|
||||
else:
|
||||
ret_string += f"{i+1}: submit your score".ljust(26, " ")
|
||||
# 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.
|
||||
"""
|
||||
|
||||
with Session(engine) as session:
|
||||
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(session: Session, 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:
|
||||
session (Session): the sqlalchemy session used for this function call.
|
||||
group (CategoryGroup): group object that contains the contents of the message.
|
||||
|
||||
Returns:
|
||||
str: Formatted string for discord message
|
||||
"""
|
||||
|
||||
ret_string = f"**{group.GroupName}**\n```"
|
||||
|
||||
stmt = select(Guild).where(Guild.GuildId == group.GuildId)
|
||||
|
||||
# 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)
|
||||
|
||||
# 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)
|
||||
|
||||
# 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:
|
||||
# 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, " ")
|
||||
|
||||
# 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, boss_name: str) -> nextcord.Message:
|
||||
def get_time_str(time_in: float) -> str:
|
||||
"""
|
||||
Function takes care of checking if a message exists for a boss, then
|
||||
creating and sending/editing the message
|
||||
@param channel: the channel to send the message in/search for message id
|
||||
@param boss: the boss to send the message about
|
||||
"""
|
||||
highscore_string = format_highscore_message(boss_name)
|
||||
Formats a float value into the desired string format for a timestamp.
|
||||
|
||||
message_id = highscores_data["tables"][boss_name]["message_id"]
|
||||
try:
|
||||
message: nextcord.Message = await channel.fetch_message(message_id)
|
||||
await message.edit(content=highscore_string)
|
||||
except nextcord.NotFound:
|
||||
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(session: Session, guild: nextcord.Guild, group: CategoryGroup) -> nextcord.Message:
|
||||
"""
|
||||
Function takes care of checking if a message exists for a group.
|
||||
Creates and sends/edits the message.
|
||||
|
||||
Args:
|
||||
session (Session): The sqlalchemy session being used for this function call.
|
||||
guild (nextcord.Guild): the nextcord.Guild object requesting this message.
|
||||
group (CategoryGroup): group to post to the message.
|
||||
|
||||
Returns:
|
||||
nextcord.Message: Message object that was posted with group contents.
|
||||
"""
|
||||
|
||||
highscore_string = format_highscore_message(session, group)
|
||||
channel = guild.get_channel(group.ChannelId)
|
||||
|
||||
if group.MessageId is None:
|
||||
# message doesn't exist.
|
||||
message: nextcord.Message = await channel.send(highscore_string)
|
||||
message_id = message.id
|
||||
highscores_data["tables"][boss_name]["message_id"] = message_id
|
||||
# save message -> boss_name in message_map
|
||||
highscores_message_map[str(message_id)] = boss_name
|
||||
group.MessageId = message.id
|
||||
session.commit()
|
||||
else:
|
||||
message: nextcord.Message = await channel.fetch_message(group.MessageId)
|
||||
await message.edit(content=highscore_string)
|
||||
|
||||
return message
|
||||
|
||||
|
||||
async def submit_score(
|
||||
interaction: nextcord.Interaction, user: str, boss_name: str, category: str, score: str
|
||||
):
|
||||
interaction: nextcord.Interaction, group_id: int, category_id: int, username: str, score: str
|
||||
) -> None:
|
||||
"""
|
||||
Adds a score to the high scores and sorts. Removes a score if larger than size limit.
|
||||
@param boss: the boss to add score to
|
||||
@param category: the category to add score to
|
||||
@param user: user who submitted score
|
||||
@param score: score to add
|
||||
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.
|
||||
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.
|
||||
"""
|
||||
|
||||
boss_category_config = highscores_config["categories"][category]
|
||||
with Session(engine) as session:
|
||||
stmt = select(Category.IsTimeRecord).where(Category.CategoryId == category_id)
|
||||
is_time_record = session.scalar(stmt)
|
||||
|
||||
# create score tuple for either time or int
|
||||
# define sort index
|
||||
if boss_category_config["is_time_record"]:
|
||||
if score.find(":") == -1:
|
||||
score_seconds = float(score)
|
||||
if is_time_record:
|
||||
minutes = float(score.split(":", maxsplit=1)[0])
|
||||
seconds = float(score.split(":", maxsplit=1)[1])
|
||||
score = (minutes * 60) + seconds
|
||||
else:
|
||||
minutes = score.split(":")[0]
|
||||
seconds = score.split(":")[1]
|
||||
score_seconds = int(minutes) * 60 + float(seconds)
|
||||
score_tuple = (user, score, score_seconds)
|
||||
sort_index = 2
|
||||
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_tuple = (user, int(score))
|
||||
sort_index = 1
|
||||
score_obj = Score(UserName=username, Score=score, CategoryId=category_id)
|
||||
session.add(score_obj)
|
||||
session.commit()
|
||||
|
||||
scores: "list[tuple]" = highscores_data["tables"][boss_name]["categories"][category]
|
||||
stmt = select(CategoryGroup).where(CategoryGroup.GroupId == group_id)
|
||||
group = session.scalar(stmt)
|
||||
|
||||
# if user is in an existing tuple, delete it.
|
||||
users_scores: "list[tuple]" = [score_tuple]
|
||||
for tup in scores:
|
||||
if user == tup[0]:
|
||||
# Add tup to users_scores, remove from scores
|
||||
users_scores.append(tup)
|
||||
await send_highscore_message(session, interaction.guild, group)
|
||||
|
||||
# sort users_scores
|
||||
users_scores.sort(
|
||||
key=lambda x: x[sort_index],
|
||||
reverse=not boss_category_config["ascending"],
|
||||
|
||||
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 Session(engine) as session:
|
||||
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)
|
||||
|
||||
for index, tup in enumerate(users_scores):
|
||||
if index == 0:
|
||||
# first score is the best score.
|
||||
if tup not in scores:
|
||||
scores.append(tup)
|
||||
else:
|
||||
# discard all other scores from scores list.
|
||||
if tup in scores:
|
||||
scores.remove(tup)
|
||||
|
||||
# sort scores.
|
||||
scores.sort(
|
||||
key=lambda x: x[sort_index],
|
||||
reverse=not boss_category_config["ascending"],
|
||||
)
|
||||
|
||||
# enforce highscore size
|
||||
while len(scores) > highscores_config["highscore_size"]:
|
||||
scores.pop()
|
||||
highscores_data["tables"][boss_name]["categories"][category] = scores
|
||||
|
||||
# edit message
|
||||
highscore_channel_id = highscores_data["highscore_channel_id"]
|
||||
print(f"Editting message {highscore_channel_id}")
|
||||
channel = interaction.guild.get_channel(highscore_channel_id)
|
||||
|
||||
error_response = 'Registered Highscores channel does not exist or was never registered. \
|
||||
Register with "?register" command.'
|
||||
assert channel is not None, await interaction.send(error_response, ephemeral=True)
|
||||
|
||||
await send_highscore_message(channel, boss_name)
|
||||
|
||||
# save data
|
||||
save_highscores_data(highscores_data)
|
||||
session.commit()
|
||||
|
||||
Reference in New Issue
Block a user