Utils.py got formatted by ruff.
This commit is contained in:
+78
-23
@@ -16,7 +16,9 @@ from utilities.sql_data_classes import Category, CategoryGroup, Guild, Score, en
|
|||||||
MAX_LINE_LENGTH = 142
|
MAX_LINE_LENGTH = 142
|
||||||
|
|
||||||
|
|
||||||
def autofill_categories(guild_id: int, group_id: int, category_begin: str) -> "dict[str, str]":
|
def autofill_categories(
|
||||||
|
guild_id: int, group_id: int, category_begin: str
|
||||||
|
) -> "dict[str, str]":
|
||||||
"""
|
"""
|
||||||
Returns all category names that start with category_begin.
|
Returns all category names that start with category_begin.
|
||||||
If category_begin is empty, returns list of all category names.
|
If category_begin is empty, returns list of all category names.
|
||||||
@@ -40,16 +42,20 @@ def autofill_categories(guild_id: int, group_id: int, category_begin: str) -> "d
|
|||||||
)
|
)
|
||||||
cat_group_alias = aliased(CategoryGroup, subq)
|
cat_group_alias = aliased(CategoryGroup, subq)
|
||||||
stmt = select(Category).where(Category.GroupId == cat_group_alias.GroupId)
|
stmt = select(Category).where(Category.GroupId == cat_group_alias.GroupId)
|
||||||
categories_dict: "dict[str, str]" = {cat.CategoryName: str(cat.CategoryId) for cat in session.scalars(stmt)}
|
categories_dict: "dict[str, str]" = {
|
||||||
|
cat.CategoryName: str(cat.CategoryId) for cat in session.scalars(stmt)
|
||||||
|
}
|
||||||
|
|
||||||
# if string empty, return all category names
|
# if string empty, return all category names
|
||||||
if not category_begin:
|
if not category_begin:
|
||||||
return dict(list(category_begin.items())[:25])
|
return dict(list(categories_dict.items())[:25])
|
||||||
|
|
||||||
# else find all categories that start with category_begin
|
# else find all categories that start with category_begin
|
||||||
# TODO is it better to have sql handle the starts with or python?
|
# TODO is it better to have sql handle the starts with or python?
|
||||||
categories: "dict[str, str]" = {
|
categories: "dict[str, str]" = {
|
||||||
cat_name: cat_id for cat_name, cat_id in categories_dict.items() if category_begin.lower() in cat_name.lower()
|
cat_name: cat_id
|
||||||
|
for cat_name, cat_id in categories_dict.items()
|
||||||
|
if category_begin.lower() in cat_name.lower()
|
||||||
}
|
}
|
||||||
return dict(list(categories.items())[:25])
|
return dict(list(categories.items())[:25])
|
||||||
|
|
||||||
@@ -71,7 +77,9 @@ def autofill_groups(guild_id: int, group_begin: str) -> "dict[str, str]":
|
|||||||
stmt = select(CategoryGroup).where(CategoryGroup.GuildId == guild_id)
|
stmt = select(CategoryGroup).where(CategoryGroup.GuildId == guild_id)
|
||||||
|
|
||||||
# dict has to have key values of same type for discord.
|
# dict has to have key values of same type for discord.
|
||||||
group_dict: "dict[str, str]" = {group.GroupName: str(group.GroupId) for group in session.scalars(stmt)}
|
group_dict: "dict[str, str]" = {
|
||||||
|
group.GroupName: str(group.GroupId) for group in session.scalars(stmt)
|
||||||
|
}
|
||||||
|
|
||||||
# if string empty, return all group names
|
# if string empty, return all group names
|
||||||
if not group_begin:
|
if not group_begin:
|
||||||
@@ -79,13 +87,17 @@ def autofill_groups(guild_id: int, group_begin: str) -> "dict[str, str]":
|
|||||||
|
|
||||||
# else find all groups that start with group_begin
|
# else find all groups that start with group_begin
|
||||||
groups: "dict[str, str]" = {
|
groups: "dict[str, str]" = {
|
||||||
group_name: group_id for group_name, group_id in group_dict.items() if group_begin.lower() in group_name.lower()
|
group_name: group_id
|
||||||
|
for group_name, group_id in group_dict.items()
|
||||||
|
if group_begin.lower() in group_name.lower()
|
||||||
}
|
}
|
||||||
|
|
||||||
return dict(list(groups.items())[:25])
|
return dict(list(groups.items())[:25])
|
||||||
|
|
||||||
|
|
||||||
def autofill_users_in_group_category(guild_id: int, group_id: int, category_id: int, user_begin: str) -> "list[str]":
|
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
|
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.
|
not None. Might be expensive due to unioning all users in group.
|
||||||
@@ -112,7 +124,9 @@ def autofill_users_in_group_category(guild_id: int, group_id: int, category_id:
|
|||||||
|
|
||||||
# If no category id, get all users in group.
|
# If no category id, get all users in group.
|
||||||
if category_id == 0:
|
if category_id == 0:
|
||||||
subq = select(Category.CategoryId).where(Category.GroupId == category_group.GroupId)
|
subq = select(Category.CategoryId).where(
|
||||||
|
Category.GroupId == category_group.GroupId
|
||||||
|
)
|
||||||
stmt = select(distinct(Score.UserName)).where(Score.CategoryId.in_(subq))
|
stmt = select(distinct(Score.UserName)).where(Score.CategoryId.in_(subq))
|
||||||
|
|
||||||
else:
|
else:
|
||||||
@@ -124,7 +138,9 @@ def autofill_users_in_group_category(guild_id: int, group_id: int, category_id:
|
|||||||
.subquery()
|
.subquery()
|
||||||
)
|
)
|
||||||
category = aliased(Category, subq)
|
category = aliased(Category, subq)
|
||||||
stmt = select(distinct(Score.UserName)).where(Score.CategoryId == category.CategoryId)
|
stmt = select(distinct(Score.UserName)).where(
|
||||||
|
Score.CategoryId == category.CategoryId
|
||||||
|
)
|
||||||
|
|
||||||
users_list: "list[str]" = list(session.scalars(stmt))
|
users_list: "list[str]" = list(session.scalars(stmt))
|
||||||
|
|
||||||
@@ -133,7 +149,9 @@ def autofill_users_in_group_category(guild_id: int, group_id: int, category_id:
|
|||||||
return users_list[:25]
|
return users_list[:25]
|
||||||
|
|
||||||
# else find all users that start with user_begin
|
# else find all users that start with user_begin
|
||||||
users: "list[str]" = [user for user in users_list if user_begin.lower() in user.lower()]
|
users: "list[str]" = [
|
||||||
|
user for user in users_list if user_begin.lower() in user.lower()
|
||||||
|
]
|
||||||
|
|
||||||
return users[:25]
|
return users[:25]
|
||||||
|
|
||||||
@@ -151,9 +169,15 @@ def autofill_users_in_guild(guild_id: int, user_begin: str) -> "list[str]":
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
with Session(engine) as session:
|
with Session(engine) as session:
|
||||||
group_ids = select(CategoryGroup.GroupId).where(CategoryGroup.GuildId == guild_id)
|
group_ids = select(CategoryGroup.GroupId).where(
|
||||||
category_ids = select(Category.CategoryId).where(Category.GroupId.in_(group_ids))
|
CategoryGroup.GuildId == guild_id
|
||||||
stmt = select(distinct(Score.UserName)).where(Score.CategoryId.in_(category_ids))
|
)
|
||||||
|
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))
|
users_list: "list[str]" = list(session.scalars(stmt))
|
||||||
|
|
||||||
# if string empty, return all users
|
# if string empty, return all users
|
||||||
@@ -161,7 +185,9 @@ def autofill_users_in_guild(guild_id: int, user_begin: str) -> "list[str]":
|
|||||||
return users_list[:25]
|
return users_list[:25]
|
||||||
|
|
||||||
# else find all users that start with user_begin
|
# else find all users that start with user_begin
|
||||||
users: "list[str]" = [user for user in users_list if user_begin.lower() in user.lower()]
|
users: "list[str]" = [
|
||||||
|
user for user in users_list if user_begin.lower() in user.lower()
|
||||||
|
]
|
||||||
|
|
||||||
return users[:25]
|
return users[:25]
|
||||||
|
|
||||||
@@ -207,7 +233,10 @@ def format_highscore_message(session: Session, group: CategoryGroup) -> str:
|
|||||||
category_ids.append(cat.CategoryId)
|
category_ids.append(cat.CategoryId)
|
||||||
|
|
||||||
# get all scores in category group, load into data frame.
|
# 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)
|
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
|
# this list is what will be printed in the discord message
|
||||||
# first string will be category list, others will be score printouts
|
# first string will be category list, others will be score printouts
|
||||||
@@ -225,17 +254,26 @@ def format_highscore_message(session: Session, group: CategoryGroup) -> str:
|
|||||||
|
|
||||||
# process data. Want each group sorted correctly and only top X results
|
# process data. Want each group sorted correctly and only top X results
|
||||||
processed_group = (
|
processed_group = (
|
||||||
group_df.sort_values("Score", ascending=category.IsAscending).head(category.Size).reset_index()
|
group_df.sort_values("Score", ascending=category.IsAscending)
|
||||||
|
.head(category.Size)
|
||||||
|
.reset_index()
|
||||||
)
|
)
|
||||||
|
|
||||||
# add each score to list
|
# add each score to list
|
||||||
for ind in range(max_cat_size): # need to subtract to deal with category name string
|
for ind in range(
|
||||||
|
max_cat_size
|
||||||
|
): # need to subtract to deal with category name string
|
||||||
if ind in processed_group.index:
|
if ind in processed_group.index:
|
||||||
username = processed_group["UserName"][ind]
|
username = processed_group["UserName"][ind]
|
||||||
score = processed_group["Score"][ind]
|
score = processed_group["Score"][ind]
|
||||||
|
|
||||||
str_list[ind + 1] += format_score_string(
|
str_list[ind + 1] += format_score_string(
|
||||||
username, score, group.UsernameLength, category.IsTimeRecord, ind + 1, pad_length
|
username,
|
||||||
|
score,
|
||||||
|
group.UsernameLength,
|
||||||
|
category.IsTimeRecord,
|
||||||
|
ind + 1,
|
||||||
|
pad_length,
|
||||||
)
|
)
|
||||||
else:
|
else:
|
||||||
# if not enough entries, pad out
|
# if not enough entries, pad out
|
||||||
@@ -275,7 +313,12 @@ def format_highscore_message(session: Session, group: CategoryGroup) -> str:
|
|||||||
|
|
||||||
|
|
||||||
def format_score_string(
|
def format_score_string(
|
||||||
username: str, score: float, username_length: int, is_time_record: bool, rank: int, pad_length: int
|
username: str,
|
||||||
|
score: float,
|
||||||
|
username_length: int,
|
||||||
|
is_time_record: bool,
|
||||||
|
rank: int,
|
||||||
|
pad_length: int,
|
||||||
) -> str:
|
) -> str:
|
||||||
"""
|
"""
|
||||||
Builds the string to show a user's score.
|
Builds the string to show a user's score.
|
||||||
@@ -321,7 +364,9 @@ def get_time_str(time_in: float) -> str:
|
|||||||
return time.strftime(f"%-M:%S.{milliseconds}", time.gmtime(time_in))
|
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:
|
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.
|
Function takes care of checking if a message exists for a group.
|
||||||
Creates and sends/edits the message.
|
Creates and sends/edits the message.
|
||||||
@@ -351,7 +396,11 @@ async def send_highscore_message(session: Session, guild: nextcord.Guild, group:
|
|||||||
|
|
||||||
|
|
||||||
async def submit_score(
|
async def submit_score(
|
||||||
interaction: nextcord.Interaction, group_id: int, category_id: int, username: str, score: str
|
interaction: nextcord.Interaction,
|
||||||
|
group_id: int,
|
||||||
|
category_id: int,
|
||||||
|
username: str,
|
||||||
|
score: str,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""
|
"""
|
||||||
Adds passed score to group. Then updates group message by calling send_highscore_message.
|
Adds passed score to group. Then updates group message by calling send_highscore_message.
|
||||||
@@ -376,7 +425,11 @@ async def submit_score(
|
|||||||
else:
|
else:
|
||||||
score = float(score)
|
score = float(score)
|
||||||
|
|
||||||
stmt = select(Score).where(Score.UserName == username).where(Score.CategoryId == category_id)
|
stmt = (
|
||||||
|
select(Score)
|
||||||
|
.where(Score.UserName == username)
|
||||||
|
.where(Score.CategoryId == category_id)
|
||||||
|
)
|
||||||
existing_score = session.scalar(stmt)
|
existing_score = session.scalar(stmt)
|
||||||
|
|
||||||
if existing_score:
|
if existing_score:
|
||||||
@@ -403,7 +456,9 @@ def add_preset_configuration(guild_id: int, config_str: str) -> None:
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
with Session(engine) as session:
|
with Session(engine) as session:
|
||||||
with open(f"example_config/{config_str}.json", encoding="utf-8", mode="r") as file:
|
with open(
|
||||||
|
f"example_config/{config_str}.json", encoding="utf-8", mode="r"
|
||||||
|
) as file:
|
||||||
config = json.load(file)
|
config = json.load(file)
|
||||||
|
|
||||||
for group_name, categorys in config["tables"].items():
|
for group_name, categorys in config["tables"].items():
|
||||||
|
|||||||
Reference in New Issue
Block a user