diff --git a/dakiny_bot/utils.py b/dakiny_bot/utils.py index ec9199c..b13907d 100644 --- a/dakiny_bot/utils.py +++ b/dakiny_bot/utils.py @@ -1,7 +1,7 @@ """File contains util functions.""" import calendar -import os +import pathlib from collections.abc import Sequence from datetime import UTC, datetime from enum import StrEnum @@ -9,17 +9,12 @@ from pathlib import Path import aiofiles import nextcord -from sqlalchemy import select, true, update +from sqlalchemy import select, update from sqlalchemy.orm import Session +from sqlmodel import col -from dakiny_bot.database.sql_data_classes import ( - CouncilMembers, - Messages, - Points, - PointsLimit, - Votes, - engine, -) +from dakiny_bot.database.db import engine +from dakiny_bot.database.models import CouncilMember, Message, Points, PointsLimit, Vote from dakiny_bot.logger import dakiny_logger from .dakiny_settings import settings @@ -35,7 +30,9 @@ VoteType = StrEnum("VoteType", ["COUNCIL", "DISCORDMANAGER", "CLANLEADER"]) PointType = StrEnum("PointType", ["CLAN", "EVENT"]) -def get_datetime(month_day: str) -> datetime: +# TODO(imcompiling): resolve returning None from this function +# https://gitea.cheezers.org/imcompiling/dakiny-bot/issues/102 +def get_datetime(month_day: str) -> datetime: # pyright: ignore[reportReturnType] """ Takes in a string month and day and returns a datetime object of that Month/Day this year. @@ -129,7 +126,9 @@ def get_quarter_cycle() -> int: return cycle dakiny_logger.debug("Quarter cycle is None.") - return None + # TODO(imcompiling) Add None to return signature or find better return value + # https://gitea.cheezers.org/imcompiling/dakiny-bot/issues/102 + return None # pyright: ignore[reportReturnType] def build_votes_str(guild: nextcord.Guild) -> str: @@ -147,19 +146,20 @@ def build_votes_str(guild: nextcord.Guild) -> str: election_name = get_election_name() with Session(engine) as session: - stmt = select(CouncilMembers).where(CouncilMembers.ElectionName == election_name) + stmt = select(CouncilMember).where(col(CouncilMember.election_name == election_name)) for nominee in sorted( session.scalars(stmt), - key=lambda nominee: guild.get_member(nominee.MemberId).display_name.lower(), + # Unsure how to None check in the lambda. + key=lambda nominee: guild.get_member(nominee.member_id).display_name.lower(), # pyright: ignore[reportOptionalMemberAccess] ): - votes[nominee.MemberId] = 0 + votes[nominee.member_id] = 0 - stmt = select(Votes).where(Votes.ElectionName == election_name).where(Votes.Type == VoteType.COUNCIL) + stmt = select(Vote).where(col(Vote.election_name == election_name)).where(col(Vote.type == VoteType.COUNCIL)) for vote in session.scalars(stmt): - if vote.VotedForId is not None: - votes[vote.VotedForId] += 1 + if vote.voted_for_id is not None: + votes[vote.voted_for_id] += 1 votes_str: str = "====================" votes_str += user_vote_str(votes) @@ -184,46 +184,52 @@ def build_highcouncil_votes_str(guild: nextcord.Guild) -> str: with Session(engine) as session: stmt = ( - select(CouncilMembers) - .where(CouncilMembers.ElectionName == election_name) - .filter(CouncilMembers.ClanLeaderSpeechId.is_not(None)) + select(CouncilMember) + .where(col(CouncilMember.election_name == election_name)) + .filter(col(CouncilMember.clan_leader_speech_id is not None)) ) for nominee in sorted( session.scalars(stmt), - key=lambda nominee: guild.get_member(nominee.MemberId).display_name.lower(), + # Unsure how to None check in the lambda. + key=lambda nominee: guild.get_member(nominee.member_id).display_name.lower(), # pyright: ignore[reportOptionalMemberAccess] ): - cl_votes[nominee.MemberId] = 0 + cl_votes[nominee.member_id] = 0 - stmt = select(Votes).where(Votes.ElectionName == election_name).where(Votes.Type == VoteType.CLANLEADER) + stmt = select(Vote).where(col(Vote.election_name == election_name)).where(col(Vote.type == VoteType.CLANLEADER)) for vote in session.scalars(stmt): - if vote.VotedForId is not None: - cl_votes[vote.VotedForId] += 1 + if vote.voted_for_id is not None: + cl_votes[vote.voted_for_id] += 1 stmt = ( - select(CouncilMembers) - .where(CouncilMembers.ElectionName == election_name) - .filter(CouncilMembers.DiscordManagerSpeechId.is_not(None)) + select(CouncilMember) + .where(col(CouncilMember.election_name == election_name)) + .filter(col(CouncilMember.discord_manager_speech_id is not None)) ) for nominee in sorted( session.scalars(stmt), - key=lambda nominee: guild.get_member(nominee.MemberId).display_name.lower(), + # Unsure how to None check in the lambda. + key=lambda nominee: guild.get_member(nominee.member_id).display_name.lower(), # pyright: ignore[reportOptionalMemberAccess] ): - dm_votes[nominee.MemberId] = 0 + dm_votes[nominee.member_id] = 0 - stmt = select(Votes).where(Votes.ElectionName == election_name).where(Votes.Type == VoteType.DISCORDMANAGER) + stmt = ( + select(Vote) + .where(col(Vote.election_name == election_name)) + .where(col(Vote.type == VoteType.DISCORDMANAGER)) + ) for vote in session.scalars(stmt): - if vote.VotedForId is not None: - dm_votes[vote.VotedForId] += 1 + if vote.voted_for_id is not None: + dm_votes[vote.voted_for_id] += 1 # build the votes string - votes_str: str = "Clan Leader Votes\n====================" + votes_str: str = "Clan Leader Vote\n====================" votes_str += user_vote_str(cl_votes) - votes_str += "\n\nDiscord Manager Votes\n====================" + votes_str += "\n\nDiscord Manager Vote\n====================" votes_str += user_vote_str(dm_votes) return votes_str @@ -317,7 +323,7 @@ def record_points(users: Sequence[str], add_points: int, point_type: PointType, quarter_name = get_quarter_name() with Session(engine) as session: for user in users: - stmt = select(Points).where(Points.UserId == user).where(Points.QuarterName == quarter_name) + stmt = select(Points).where(col(Points.user_id == user)).where(col(Points.quarter_name == quarter_name)) points_obj = session.scalar(stmt) add_points = get_allowed_points(session, point_type, user, add_points) @@ -325,16 +331,16 @@ def record_points(users: Sequence[str], add_points: int, point_type: PointType, add_event_points = add_points if point_type == PointType.EVENT else 0 if points_obj is None: new_points = Points( - QuarterName=quarter_name, - UserId=user, - DiscordMember=is_member, - EventPoints=add_event_points, - ClanPoints=add_clan_points, + quarter_name=quarter_name, + user_id=user, + discord_member=is_member, + event_points=add_event_points, + clan_points=add_clan_points, ) session.add(new_points) else: - points_obj.EventPoints += add_event_points - points_obj.ClanPoints += add_clan_points + points_obj.event_points += add_event_points + points_obj.clan_points += add_clan_points session.commit() @@ -364,7 +370,7 @@ def get_allowed_points(session: Session, point_type: PointType, user: str, point if point_type == PointType.EVENT else settings.point_settings.clan_points_weekly_limit ) - stmt = select(PointsLimit).where(PointsLimit.UserId == user).where(PointsLimit.Week == this_week) + stmt = select(PointsLimit).where(col(PointsLimit.user_id == user)).where(col(PointsLimit.week == this_week)) # if under limit, can add up to however many points would make the current limit equal to the # allowed limit. @@ -372,16 +378,16 @@ def get_allowed_points(session: Session, point_type: PointType, user: str, point if limit_obj is None: clan_points = points if point_type == PointType.CLAN else 0 event_points = points if point_type == PointType.EVENT else 0 - new_limit = PointsLimit(UserId=user, Week=this_week, EventPoints=event_points, ClanPoints=clan_points) + new_limit = PointsLimit(user_id=user, week=this_week, event_points=event_points, clan_points=clan_points) add_points = points session.add(new_limit) else: - curr_points = limit_obj.EventPoints if point_type == PointType.EVENT else limit_obj.ClanPoints + curr_points = limit_obj.event_points if point_type == PointType.EVENT else limit_obj.clan_points add_points = points if curr_points + points <= points_limit else (max(0, points_limit - curr_points)) if point_type == PointType.EVENT: - limit_obj.EventPoints += add_points + limit_obj.event_points += add_points else: - limit_obj.ClanPoints += add_points + limit_obj.clan_points += add_points session.commit() @@ -410,8 +416,8 @@ async def update_clan_messages(guild: nextcord.Guild, channel_updates: Sequence[ dir_path = f"{MESSAGES_DIR}/{dir_name}" channel = await get_clan_message_channel(session, dir_name, guild) sorted_files = sorted( - os.listdir(dir_path), - key=lambda filename: int(filename.split(".")[0]), + pathlib.Path(dir_path).iterdir(), + key=lambda file: int(file.name.split(".")[0]), ) for file_name in sorted_files: @@ -440,41 +446,49 @@ async def get_clan_message_channel(session: Session, channel_name: str, guild: n Returns: nextcord.TextChannel: Channel object with name channel_name. """ - stmt = select(Messages.Id).where(Messages.Path == channel_name) + stmt = select(col(Message.id)).where(col(Message.path == channel_name)) channel_id = session.scalar(stmt) + ret_channel = None if channel_id is None: - ret_channel = None # channel is not tracked. see if it exists channels = await guild.fetch_channels() for channel in channels: - if channel.name == channel_name and channel.__class__ is nextcord.TextChannel: + if channel.name == channel_name and isinstance(channel, nextcord.TextChannel): ret_channel = channel if ret_channel is None: # channel doesn't exist, we must create it ret_channel = await guild.create_text_channel( channel_name, - reason="Creating channel as required for by design of Dakiny-Messages repo.", + reason="Creating channel as required for by design of Dakiny-Message repo.", ) # record channel information - new_item = Messages(Path=channel_name, Id=ret_channel.id) + new_item = Message(path=channel_name, id=ret_channel.id) session.add(new_item) session.commit() - return ret_channel - try: - return await guild.fetch_channel(channel_id) - except nextcord.NotFound: - # the channel was deleted for some reason - ret_channel = await guild.create_text_channel( - channel_name, - reason="Creating channel as required for by design of Dakiny-Messages repo.", - ) - stmt = update(Messages).where(Messages.Path == channel_name).values(Id=ret_channel.id) - session.execute(stmt) - session.commit() - return ret_channel + else: + try: + ret_channel = await guild.fetch_channel(channel_id) + if not isinstance(ret_channel, nextcord.TextChannel): + dakiny_logger.warning( + "Channel %s is not a text channel. Does not work for clan messages.", ret_channel.name + ) + # TODO(imcompiling) Figure out how to handle returning None here. Maybe delete channel and remake it? + # https://gitea.cheezers.org/imcompiling/dakiny-bot/issues/102 + return None # pyright: ignore[reportReturnType] + except nextcord.NotFound: + # the channel was deleted for some reason + ret_channel = await guild.create_text_channel( + channel_name, + reason="Creating channel as required for by design of Dakiny-Message repo.", + ) + stmt = update(Message).where(col(Message.path == channel_name)).values(id=ret_channel.id) + session.execute(stmt) + session.commit() + + return ret_channel async def edit_clan_message( @@ -496,21 +510,21 @@ async def edit_clan_message( message_name (str): Name of the message. Corresponds to the file that contains message content. message_content (str): Content of the message to edit/send. """ - stmt = select(Messages.Id).where(Messages.Path == message_name) + stmt = select(col(Message.id)).where(col(Message.path == message_name)) msg_id = session.scalar(stmt) if msg_id is None: # don't have a message for this file msg = await channel.send(message_content) - new_message = Messages(Path=message_name, Id=msg.id) + new_message = Message(path=message_name, id=msg.id) session.add(new_message) session.commit() else: try: msg: nextcord.Message = await channel.fetch_message(msg_id) - await msg.edit(message_content) + await msg.edit(content=message_content) except nextcord.NotFound: msg = await channel.send(message_content) - stmt = update(Messages).where(Messages.Path == message_name).values(Id=msg.id) + stmt = update(Message).where(col(Message.path == message_name)).values(id=msg.id) session.execute(stmt) session.commit() @@ -522,30 +536,35 @@ async def edit_clan_council_info(guild: nextcord.Guild) -> None: Args: guild (nextcord.Guild): Guild object that to use to get channel. """ - clan_info_channel: nextcord.TextChannel = await guild.fetch_channel(settings.channel_ids.clan_info) + clan_info_channel = await guild.fetch_channel(settings.channel_ids.clan_info) + + if not isinstance(clan_info_channel, nextcord.TextChannel): + return election_name = get_election_name() with Session(engine) as session: stmt = ( - select(CouncilMembers.MemberId) - .where(CouncilMembers.ElectionName == election_name) - .where(CouncilMembers.DiscordManager == true()) + select(col(CouncilMember.member_id)) + .where(col(CouncilMember.election_name == election_name)) + .where(col(CouncilMember.discord_manager is True)) ) discord_manager = session.scalar(stmt) stmt = ( - select(CouncilMembers.MemberId) - .where(CouncilMembers.ElectionName == election_name) - .where(CouncilMembers.ClanLeader == true()) + select(col(CouncilMember.member_id)) + .where(col(CouncilMember.election_name == election_name)) + .where(col(CouncilMember.clan_leader is True)) ) clan_leader = session.scalar(stmt) stmt = ( - select(CouncilMembers.MemberId) - .where(CouncilMembers.ElectionName == election_name) - .where(CouncilMembers.Elected == true()) + select(col(CouncilMember.member_id)) + .where(col(CouncilMember.election_name == election_name)) + .where(col(CouncilMember.elected is True)) ) council_member_ids = sorted( - session.scalars(stmt), key=lambda user_id: guild.get_member(user_id).display_name.lower() + session.scalars(stmt), + # Unsure how to None check in the lambda. + key=lambda user_id: guild.get_member(user_id).display_name.lower(), # pyright: ignore[reportOptionalMemberAccess] ) council_members_strs = [