imcompiling/agon
1
0
Fork 0
Code Issues 17 Pull Requests 4 Actions Packages Activity

Update utils to use new database.

Browse Source
This commit is contained in:
alwayscompiling
2022-11-30 20:54:06 -05:00
parent 03d7485037
commit 18df783d86
1 changed files with 198 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utilities/utils.py
+190 -126
View File
@@ -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()
for category in highscores_data.get_guild(guild_id).get_table(table_name).get_categories().values():
user_set = user_set | set(category.get_users())
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()
# 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)
# 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))
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)
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.lower().startswith(user_begin.lower())]
users = [user for user in users_list if user_begin.lower() in user.lower()]
return users
except AttributeError:
return []
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
# 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 category.time_record:
# 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 = score.rjust(7, " ")
score = str(int(group["Score"][ind])).rjust(7, " ")
ret_string += f"{i+1}: {username} - {score}".ljust(guild.category_pad_length, " ")
str_list[ind + 1] += f"{ind + 1}: {username} - {score}".ljust(pad_length, " ")
else:
ret_string += f"{i+1}:".ljust(guild.category_pad_length, " ")
ret_string += "\n"
# 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)
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)