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

Begin updating utils.

Browse Source
This commit is contained in:
imcompiling
2025-03-03 15:54:28 -05:00
parent 1c3fb231b7
commit 157b8d4f3f
1 changed files with 31 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utilities/utils.py
+31 -28
View File
@@ -8,7 +8,8 @@ from sqlalchemy import distinct, select
from sqlalchemy.orm import Session, aliased
# pylint: disable=import-error
from database.db import Category, CategoryGroup, Guild, Score, engine
from database.db import engine
from database.models import Category, CategoryGroup, Guild, Score
MAX_LINE_LENGTH = 142
@@ -31,13 +32,13 @@ def autofill_categories(guild_id: int, group_id: int, category_begin: str) -> "d
# search database for all category names
subq = (
select(CategoryGroup)
.where(CategoryGroup.GuildId == guild_id)
.where(CategoryGroup.GroupId == group_id)
.where(CategoryGroup.guild_id == guild_id)
.where(CategoryGroup.id == group_id)
.subquery()
)
cat_group_alias = aliased(CategoryGroup, subq)
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)}
stmt = select(Category).where(Category.group_id == cat_group_alias.id)
categories_dict: dict[str, str] = {cat.name: str(cat.id) for cat in session.scalars(stmt)}
# if string empty, return all category names
if not category_begin:
@@ -64,10 +65,10 @@ def autofill_groups(guild_id: int, group_begin: str) -> "dict[str, str]":
"""
with Session(engine) as session:
# Select all groups in guild.
stmt = select(CategoryGroup).where(CategoryGroup.GuildId == guild_id)
stmt = select(CategoryGroup).where(CategoryGroup.guild_id == guild_id)
# 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.name: str(group.id) for group in session.scalars(stmt)}
# if string empty, return all group names
if not group_begin:
@@ -83,8 +84,10 @@ def autofill_groups(guild_id: int, group_begin: str) -> "dict[str, 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
not None. Might be expensive due to unioning all users in group.
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 union-ing all
users in group.
Args:
guild_id (int): Guild id corresponding to group and category.
@@ -99,27 +102,27 @@ def autofill_users_in_group_category(guild_id: int, group_id: int, category_id:
# get all categories in the group
subq = (
select(CategoryGroup)
.where(CategoryGroup.GuildId == guild_id)
.where(CategoryGroup.GroupId == group_id)
.where(CategoryGroup.guild_id == guild_id)
.where(CategoryGroup.id == 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))
subq = select(Category.id).where(Category.group_id == category_group.id)
stmt = select(distinct(Score.username)).where(Score.category_id.in_(subq))
else:
# get all users in category_id
subq = (
select(Category)
.where(Category.GroupId == category_group.GroupId)
.where(Category.CategoryId == category_id)
.where(Category.group_id == category_group.id)
.where(Category.id == category_id)
.subquery()
)
category = aliased(Category, subq)
stmt = select(distinct(Score.UserName)).where(Score.CategoryId == category.CategoryId)
stmt = select(distinct(Score.username)).where(Score.category == category.id)
users_list: list[str] = list(session.scalars(stmt))
@@ -145,9 +148,9 @@ def autofill_users_in_guild(guild_id: int, user_begin: str) -> "list[str]":
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))
group_ids = select(CategoryGroup.id).where(CategoryGroup.guild_id == guild_id)
category_ids = select(Category.id).where(Category.group_id.in_(group_ids))
stmt = select(distinct(Score.username)).where(Score.category_id.in_(category_ids))
users_list: list[str] = list(session.scalars(stmt))
# if string empty, return all users
@@ -176,34 +179,34 @@ def format_highscore_message(session: Session, group: CategoryGroup) -> str:
Returns:
str: Formatted string for discord message
"""
ret_string = f"**{group.GroupName}**\n```"
ret_string = f"**{group.name}**\n```"
stmt = select(Guild).where(Guild.GuildId == group.GuildId)
stmt = select(Guild).where(Guild.id == group.guild_id)
# Get all categories in the group
subq = (
select(CategoryGroup)
.where(CategoryGroup.GuildId == group.GuildId)
.where(CategoryGroup.GroupName == group.GroupName)
.where(CategoryGroup.guild_id == group.guild_id)
.where(CategoryGroup.name == group.name)
.subquery()
)
cat_group_alias = aliased(CategoryGroup, subq)
stmt = select(Category).where(Category.GroupId == cat_group_alias.GroupId)
stmt = select(Category).where(Category.group_id == cat_group_alias.id)
# 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
pad_length = group.pad_length
for cat in session.scalars(stmt):
max_cat_size = max(cat.Size, max_cat_size)
max_cat_size = max(cat.size, max_cat_size)
categories.append(cat)
category_ids.append(cat.CategoryId)
category_ids.append(cat.id)
# 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,
sql=session.query(Score).where(Score.category_id.in_(category_ids)).statement,
con=engine,
)