From 18df783d86375074e9bcec6095259b788428a8ae Mon Sep 17 00:00:00 2001 From: alwayscompiling Date: Wed, 30 Nov 2022 20:52:16 -0500 Subject: [PATCH] Update utils to use new database. --- utilities/utils.py | 332 +++++++++++++++++++++++++++------------------ 1 file changed, 198 insertions(+), 134 deletions(-) diff --git a/utilities/utils.py b/utilities/utils.py index 5be80b8..0a10de0 100644 --- a/utilities/utils.py +++ b/utilities/utils.py @@ -2,103 +2,88 @@ Contains various utlity functions for highscores bot """ +import time + import nextcord +import pandas as pd +from sqlalchemy import distinct, select +from sqlalchemy.orm import aliased -from classes.category import Category # pylint: disable=import-error -from classes.guild import Guild # pylint: disable=import-error -from classes.score import Score # pylint: disable=import-error -from classes.table import Table # pylint: disable=import-error -from highscores import highscores_data, highscores_message_map # pylint: disable=import-error +from highscores import Category, CategoryGroup, Guild, Score, engine, session + +MAX_LINE_LENGTH = 142 -def autofill_categories(guild_id: int, table_name: str, category_begin: str) -> "list[str]": +def autofill_categories(guild_id: int, group_name: str, category_begin: str) -> "list[str]": """ Returns all category names that start with category_begin. If category_begin is empty, returns list of all category names. Args: - guild_id (int): Guild id that holds the table and categories. - table_name (str): Table name to search in. + guild_id (int): Guild id that holds the group and categories. + group_name (str): Group name to search in. category_begin (str): The beginning of a category name to search with. Can be empty. Returns: list[str]: List of all category names that start with category_begin. """ - categories_list = highscores_data.get_guild(guild_id).get_table(table_name).get_categories().keys() + # search database for all category names + subq = ( + select(CategoryGroup) + .where(CategoryGroup.GuildId == guild_id) + .where(CategoryGroup.GroupName == group_name) + .subquery() + ) + cat_group_alias = aliased(CategoryGroup, subq) + stmt = select(Category.CategoryName).where(Category.GroupId == cat_group_alias.GroupId) + categories_list: "list[str]" = list(session.scalars(stmt)) # if string empty, return all category names if not category_begin: return categories_list # else find all categories that start with category_begin - categories = [category for category in categories_list if category.lower().startswith(category_begin.lower())] + # TODO is it better to have sql handle the starts with or python? + categories: "list[str]" = [category for category in categories_list if category_begin.lower() in category.lower()] return categories -def autofill_category_users(guild_id: int, table_name: str, category_name: str, user_begin: str) -> "list[str]": +def autofill_groups(guild_id: int, group_begin: str) -> "list[str]": """ - Returns list of users that are in the table and category that also start with the user_begin string. + 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 corresponding to table and category. - table_name (str): Table to look in for category. - category_name (str): Category to look in for users. - user_begin (str): Beginning of user name to search with. Can be empty. + 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: - list[str]: List of users that start with user_begin. + list[str]: List of group names that start with group_begin. """ - try: - users_list: "list[str]" = ( - highscores_data.get_guild(guild_id).get_table(table_name).get_category(category_name).get_users() - ) + # Select all groups in guild. + stmt = select(CategoryGroup.GroupName).where(CategoryGroup.GuildId == guild_id) + group_list: "list[str]" = list(session.scalars(stmt)) - # if string empty, return all users - if not user_begin: - return users_list + # if string empty, return all group names + if not group_begin: + return group_list - # else find all users that start with user_begin - users = [user for user in users_list if user.lower().startswith(user_begin.lower())] - - return users - - except AttributeError: - return [] + # else find all groups that start with group_begin + groups: "list[str]" = [group for group in group_list if group_begin.lower() in group.lower()] + return groups -def autofill_tables(guild_id: int, table_begin: str) -> "list[str]": +def autofill_users_in_group_category( + guild_id: int, group_name: str, category_name: str, user_begin: str +) -> "list[str]": """ - Returns all table names that start with table_begin. If table_begin is empty, returns list of all table names. + 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 that holds the tables. - table_begin (str): The beginning of a table name to search with. Can be empty. - - Returns: - list[str]: List of table names that start with table_begin. - """ - - tables_list = highscores_data.get_guild(guild_id).get_tables().keys() - - # if string empty, return all table names - if not table_begin: - return tables_list - - # else find all tables that start with table_begin - tables = [table for table in tables_list if table.lower().startswith(table_begin.lower())] - return tables - - -def autofill_users(guild_id: int, table_name: str, category_name: str, user_begin: str) -> "list[str]": - """ - Returns list of users that match user_begin. Either all users in table or just users in category if category is - None. Might be sexpensive due to unioning all users in table. - - Args: - guild_id (int): Guild id corresponding to table and category. - table_name (str): Table to look in for category. + guild_id (int): Guild id corresponding to group and category. + group_name (str): group to look in for category. category_name (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. @@ -106,139 +91,218 @@ def autofill_users(guild_id: int, table_name: str, category_name: str, user_begi list[str]: List of users that start with user_begin. """ - try: - # If no category name, get all users in table. - if category_name is None: - user_set: set = set() + # get all categories in the group + subq = ( + select(CategoryGroup) + .where(CategoryGroup.GuildId == guild_id) + .where(CategoryGroup.GroupName == group_name) + .subquery() + ) + category_group = aliased(CategoryGroup, subq) - for category in highscores_data.get_guild(guild_id).get_table(table_name).get_categories().values(): - user_set = user_set | set(category.get_users()) + # If no category name, get all users in group. + if category_name is None: + subq = select(Category.CategoryId).where(Category.GroupId == category_group.GroupId).subquery() + stmt = select(distinct(Score.UserName)).where(Score.CategoryId.in_(subq)) - users_list = list(user_set) - else: - users_list: "list[str]" = ( - highscores_data.get_guild(guild_id).get_table(table_name).get_category(category_name).get_users() - ) + else: + # get all users in category_name + subq = ( + select(Category) + .where(Category.GroupId == category_group.GroupId) + .where(Category.CategoryName == category_name) + .subquery() + ) + category = aliased(Category, subq) + stmt = select(distinct(Score.UserName)).where(Score.CategoryId == category.CategoryId) - # if string empty, return all users - if not user_begin: - return users_list + users_list: "list[str]" = list(session.scalars(stmt)) - # else find all users that start with user_begin - users = [user for user in users_list if user.lower().startswith(user_begin.lower())] + # if string empty, return all users + if not user_begin: + return users_list - return users + # else find all users that start with user_begin + users = [user for user in users_list if user_begin.lower() in user.lower()] - except AttributeError: - return [] + return users -def format_highscore_message(guild: Guild, table: Table) -> str: +def format_highscore_message(group: CategoryGroup) -> str: """ Creates a formatted string for a boss's highscore discord message. This formatted string is in discord's code block to ensure monospace text. Loops through each category to build the message one place at a time. Args: - guild (Guild): Guild object that is posting the message. - table (Table): Table object that contains the contents of the message. + group (group): group object that contains the contents of the message. Returns: str: Formatted string for discord message """ - ret_string = f"**{table.name}**\n```" + ret_string = f"**{group.GroupName}**\n```" - # Gather all categories and add to string - for category in table.categories: - ret_string += f"{category.name}".ljust(guild.category_pad_length, " ") - ret_string += "\n" + stmt = select(Guild).where(Guild.GuildId == Category.GuildId) + guild = session.scalars(stmt).one() - # list rankings in a single line. Start with first place for each category and then continue - for i in range(table.size): - for category in table.categories: - if len(category.scores) > i: - # pad username to max length. - # have to convert category.scores into list to make it subscriptable. no idea why. - username: str = list(category.scores)[i].user.ljust(guild.username_length, " ") + # Get all categories in the group + subq = ( + select(CategoryGroup) + .where(CategoryGroup.GuildId == group.GuildId) + .where(CategoryGroup.GroupName == group.GroupName) + .subquery() + ) + cat_group_alias = aliased(CategoryGroup, subq) + stmt = select(Category).where(Category.GroupId == cat_group_alias.GroupId) - score: str = list(category.scores)[i].get_str() + # set up a map for the categories + # also grab the max category size + categories = [] + category_ids = [] + max_cat_size = 0 + pad_length = group.PadLength + for cat in session.scalars(stmt): + if cat.Size > max_cat_size: + max_cat_size = cat.Size + categories.append(cat) + category_ids.append(cat.CategoryId) - # pad score to correct length - # MM:SS.ms is 7 characters long - if category.time_record: - # if there are single digit minutes, add a space for formatting. - if score.find(":") and len(score.split(":", maxsplit=1)[0]) < 2: - score = " " + score + # get all scores in category group, load into data frame. + d_f = pd.read_sql_query(sql=session.query(Score).where(Score.CategoryId.in_(category_ids)).statement, con=engine) + + # this list is what will be printed in the discord message + # first string will be category list, others will be score printouts + str_list: "list[str]" = ["" for i in range(max_cat_size + 1)] + groups = 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.indices: + group = groups.get_group(category.CategoryId) + + # process data. Want each group sorted correctly and only top X results + group.sort_values("Score", ascending=category.IsAscending, inplace=True) + group = group.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 group.index: + username = group["UserName"][ind].ljust(guild.UsernameLength, " ") + + # 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(group["Score"][ind]) + if score.find(":") and len(score.split(":", maxsplit=1)[0]) < 2: + score = " " + score + else: + score = str(int(group["Score"][ind])).rjust(7, " ") + + str_list[ind + 1] += f"{ind + 1}: {username} - {score}".ljust(pad_length, " ") else: - score = score.rjust(7, " ") + # if not enough entries, pad out + if ind + 1 <= category.Size: + str_list[ind + 1] += f"{ind + 1}: ".ljust(pad_length, " ") + else: + str_list[ind + 1] += "".ljust(pad_length, " ") + else: + for ind in range(max_cat_size): + if ind + 1 <= category.Size: + str_list[ind + 1] += f"{ind + 1}: ".ljust(pad_length, " ") + else: + str_list[ind + 1] += "".ljust(pad_length, " ") - ret_string += f"{i+1}: {username} - {score}".ljust(guild.category_pad_length, " ") - else: - ret_string += f"{i+1}:".ljust(guild.category_pad_length, " ") - ret_string += "\n" + # next step is to cut down on the length of each line if required + # need to use index to allow for modifying the list + for ind, line in enumerate(str_list): + if len(line) > MAX_LINE_LENGTH: + split_loc = int(MAX_LINE_LENGTH / pad_length) * pad_length + new_line = line[split_loc:] + str_list[ind] = line[:split_loc] + str_list.append(new_line) + if ind + 1 / max_cat_size == 0: + str_list.append("\n") + + # this list can now be placed into the format message function in the bot + for string in str_list: + # rstrip to clear unneeded spaces + # add newline here for each string + print(string.rstrip()) return ret_string + "```" -async def send_highscore_message(channel: nextcord.TextChannel, guild: Guild, table: Table) -> nextcord.Message: +def get_time_str(time_in: float) -> str: """ - Function takes care of checking if a message exists for a table. + Formats a float value into the desired string format for a timestamp. + + Args: + time_in (float): A time record in seconds + + Returns: + str: Describes the time record in string format. + """ + # TODO see if theres a better way to do this + milliseconds = round((time_in - int(time_in)) * 10) + return time.strftime(f"%-M:%S.{milliseconds}", time.gmtime(time_in)) + + +async def send_highscore_message(channel: nextcord.TextChannel, group: CategoryGroup) -> nextcord.Message: + """ + Function takes care of checking if a message exists for a group. Creates and sends/edits the message. Args: channel (nextcord.TextChannel): Discord channel object where message is sent - guild (Guild): Guild object for Guild that is posting the message - table (Table): Table to post to the message + group (group): group to post to the message Returns: - nextcord.Message: Message object that was posted with table contents + nextcord.Message: Message object that was posted with group contents """ - highscore_string = format_highscore_message(guild, table) + highscore_string = format_highscore_message(group) - message_id = table.message_id try: - message: nextcord.Message = await channel.fetch_message(message_id) + message: nextcord.Message = await channel.fetch_message(group.MessageId) await message.edit(content=highscore_string) except nextcord.NotFound: # message doesn't exist. message: nextcord.Message = await channel.send(highscore_string) - message_id = message.id - table.message_id = message_id - # save message -> boss_name in message_map - # get guild id from channel - # channel should come from the guild that is requesting this highscores message anyways. - # check if guild is in message map already - if not channel.guild.id in highscores_message_map: - highscores_message_map[channel.guild.id] = {} - highscores_message_map[channel.guild.id][message_id] = table.name + group.MessageId = message.id + session.commit() return message async def submit_score( - interaction: nextcord.Interaction, guild: Guild, table: Table, category: Category, score: Score + interaction: nextcord.Interaction, group: CategoryGroup, category_id: int, username: str, score: float ) -> None: """ - Adds passed score to table. Then updates table message by calling send_highscore_message. + Adds passed score to group. Then updates group message by calling send_highscore_message. Ensures that a highscores channel exists. Args: - interaction (nextcord.Interaction): Interaction that caused the score submission - guild (Guild): Guild to submit score to - table (Table): Table to submit score to - category (Category): Category to submit score to - score (Score): Score to submit + interaction (nextcord.Interaction): Interaction that caused the score submission. + group (CategoryGroup): group to submit score to. + category_id (Category): Category to submit score to. + username (str): The user who submitted the score. + score (float): Score to submit. """ - category.add_score(score) - highscores_data.save_guild_data(guild) + score_obj = Score(UserName=username, Score=score, CategoryId=category_id) + session.add(score_obj) + session.commit() # edit message - print(f"Editting message {guild.highscore_channel}") - channel = interaction.guild.get_channel(guild.highscore_channel) + print(f"Editting message {group.MessageId}") + channel = interaction.guild.get_channel(group.ChannelId) error_response = "Registered Highscores channel does not exist or was never registered." error_response += 'Register with "?register" command.' assert channel is not None, await interaction.send(error_response, ephemeral=True) - await send_highscore_message(channel, guild, table) + await send_highscore_message(channel, group)