From 3dfdc10334ed3ff8ba613284c4f5f3d2db1e81c5 Mon Sep 17 00:00:00 2001 From: imcompiling Date: Thu, 24 Jul 2025 12:39:16 -0400 Subject: [PATCH] Feat: Swap Election to use new models. Address many pylance errors. Rewrite a few functions, will require testing to ensure functionality remains the same. --- .vscode/settings.json | 2 + dakiny_bot/classes/__init__.py | 1 + dakiny_bot/classes/election.py | 252 ++++++++++++++++++------------ dakiny_bot/classes/pyproject.toml | 4 - 4 files changed, 159 insertions(+), 100 deletions(-) create mode 100644 dakiny_bot/classes/__init__.py diff --git a/.vscode/settings.json b/.vscode/settings.json index ef5d734..de8c198 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -8,7 +8,9 @@ "CLANLEADER", "Dakiny", "DISCORDMANAGER", + "gitea", "highcouncil", + "imcompiling", "nextcord", "ondelete" ] diff --git a/dakiny_bot/classes/__init__.py b/dakiny_bot/classes/__init__.py new file mode 100644 index 0000000..7215812 --- /dev/null +++ b/dakiny_bot/classes/__init__.py @@ -0,0 +1 @@ +"""Empty init file for classes module.""" diff --git a/dakiny_bot/classes/election.py b/dakiny_bot/classes/election.py index 54cf7ce..a79f903 100644 --- a/dakiny_bot/classes/election.py +++ b/dakiny_bot/classes/election.py @@ -5,17 +5,17 @@ from datetime import UTC, datetime, timedelta from pathlib import Path import nextcord +from sqlalchemy import select, update +from sqlalchemy.orm import Session +from sqlmodel import col + from dakiny_bot.classes.nominate_highcouncil_views import NominateHighCouncilView from dakiny_bot.classes.nominate_views import NominateView from dakiny_bot.classes.voting_highcouncil_views import VoteHighcouncilView from dakiny_bot.classes.voting_views import VoteView from dakiny_bot.dakiny_settings import settings -from dakiny_bot.database.sql_data_classes import ( - CouncilMembers, - ElectionInformation, - Votes, - engine, -) +from dakiny_bot.database.db import engine +from dakiny_bot.database.models import CouncilMember, ElectionInformation, Vote from dakiny_bot.logger import dakiny_logger from dakiny_bot.utils import ( TEMPLATES_DIR, @@ -26,8 +26,6 @@ from dakiny_bot.utils import ( get_datetime, get_election_name, ) -from sqlalchemy import select, true, update -from sqlalchemy.orm import Session REQUIRED_VOTES_TO_BE_ELECTED = 3 @@ -70,7 +68,7 @@ class Election: list[str]: A list of message contents in the correct order. Must be a string because the message is too long. """ - with Path(f"{TEMPLATES_DIR}/election_announcement.txt", encoding="utf8").open() as file: + with Path(f"{TEMPLATES_DIR}/election_announcement.txt").open() as file: message = file.read() next_election = get_datetime( @@ -98,7 +96,7 @@ class Election: Returns: str: Message to send. """ - with Path(f"{TEMPLATES_DIR}/highcouncil_election.txt", encoding="utf8").open() as file: + with Path(f"{TEMPLATES_DIR}/highcouncil_election.txt").open() as file: message = file.read() return message.format( @@ -116,7 +114,7 @@ class Election: Returns: str: Message to send. """ - with Path(f"{TEMPLATES_DIR}/highcouncil_vote.txt", encoding="utf8").open() as file: + with Path(f"{TEMPLATES_DIR}/highcouncil_vote.txt").open() as file: message = file.read() return message.format( @@ -131,7 +129,7 @@ class Election: Returns: str: Message to send """ - with Path(f"{TEMPLATES_DIR}/vote_start.txt", encoding="utf8").open() as file: + with Path(f"{TEMPLATES_DIR}/vote_start.txt").open() as file: message = file.read() vote_end: datetime = get_datetime(settings.election_cycles[self.cycle].voting_end) @@ -171,7 +169,10 @@ class Election: async def election_start_action(self) -> None: """Performs required actions for when the election cycle starts.""" # create new channel for election - parent: nextcord.TextChannel = self.guild.get_channel(settings.channel_ids.voting) + parent = self.guild.get_channel(settings.channel_ids.voting) + if not isinstance(parent, nextcord.TextChannel): + dakiny_logger.error("Voting channel (id: %s) is not a text channel.", settings.channel_ids.voting) + return quarter_name: str = get_election_name() thread_msg: nextcord.Message = await parent.send(content=quarter_name) @@ -192,9 +193,9 @@ class Election: # create new entry into database. with Session(engine) as session: new_cycle = ElectionInformation( - ElectionName=quarter_name, - ThreadId=thread.id, - CouncilNominationMsgId=nominate_msg.id, + name=quarter_name, + thread_id=thread.id, + council_nomination_msg_id=nominate_msg.id, ) session.add(new_cycle) session.commit() @@ -207,64 +208,98 @@ class Election: Removes write permission from the general clan member role. """ message = self.build_vote_start() - election_thread: nextcord.Thread = self.guild.get_thread(self.get_election_info().ThreadId) - await election_thread.send(message) - vote_msg = await election_thread.send( - build_votes_str(self.guild), - view=VoteView(self.guild), - ) with Session(engine) as session: - stmt = select(ElectionInformation).where(ElectionInformation.ElectionName == get_election_name()) - info = session.scalar(stmt) - info.CouncilVoteMsgId = vote_msg.id + stmt = select(ElectionInformation).where(col(ElectionInformation.name == get_election_name())) + if (election_info := session.scalar(stmt)) is None: + dakiny_logger.error("Unable to find election %s", get_election_name()) + return + if (election_thread := self.guild.get_thread(election_info.thread_id)) is None: + dakiny_logger.error("Recorded thread id %s does not result in a thread.", election_info.thread_id) + return + + await election_thread.send(message) + + vote_msg = await election_thread.send( + build_votes_str(self.guild), + view=VoteView(self.guild), + ) + + election_info.council_vote_msg_id = vote_msg.id session.commit() - thread: nextcord.Thread = self.guild.get_thread(info.ThreadId) - nominate_msg: nextcord.Message = await thread.fetch_message(info.CouncilNominationMsgId) + if election_info.council_nomination_msg_id is None: + dakiny_logger.error("Election doesn't have a nomination message set.") + return + + nominate_msg: nextcord.Message = await election_thread.fetch_message( + election_info.council_nomination_msg_id + ) await nominate_msg.edit(view=None) async def voting_end_action(self) -> None: """Performs required actions for when voting time has ended.""" message = self.build_highcouncil_election_announce() - council_elect_role: nextcord.Role = self.guild.get_role(settings.role_ids.clan_council_elect) - election_thread: nextcord.Thread = self.guild.get_thread(self.get_election_info().ThreadId) - - nominate_msg: nextcord.Message = await election_thread.send(message, view=NominateHighCouncilView(self.guild)) + if (council_elect_role := self.guild.get_role(settings.role_ids.clan_council_elect)) is None: + dakiny_logger.error("Clan council elect role not found (id: %s).", settings.role_ids.clan_council_elect) + return election_name = get_election_name() with Session(engine) as session: + stmt = select(ElectionInformation).where(col(ElectionInformation.name == election_name)) + if (info := session.scalar(stmt)) is None: + dakiny_logger.error("Unable to find election %s", election_name) + return + + if (election_thread := self.guild.get_thread(info.thread_id)) is None: + dakiny_logger.error("Recorded thread id %s does not result in a thread.", info.thread_id) + return + + nominate_msg: nextcord.Message = await election_thread.send( + message, view=NominateHighCouncilView(self.guild) + ) + # remove vote button from vote message - stmt = select(ElectionInformation).where(ElectionInformation.ElectionName == election_name) - info = session.scalar(stmt) - vote_msg: nextcord.Message = await election_thread.fetch_message(info.CouncilVoteMsgId) + if not info.council_vote_msg_id: + dakiny_logger.error("Council vote message id not set.") + return + vote_msg: nextcord.Message = await election_thread.fetch_message(info.council_vote_msg_id) await vote_msg.edit(view=None) # add the highcouncil nomination message to election information - info.HighCouncilNominationMsgId = nominate_msg.id + info.high_council_nomination_msg_id = nominate_msg.id session.commit() # add the highcouncil role to everyone with 3 or more votes. set elected to true in database stmt = ( - select(Votes.VotedForId) - .where(Votes.ElectionName == election_name) - .where(Votes.Type == VoteType.COUNCIL) + select(col(Vote.voted_for_id)) + .where(col(Vote.election_name == election_name)) + .where(col(Vote.type == VoteType.COUNCIL)) ) vote_count = collections.Counter(session.scalars(stmt)) for user_id, votes in vote_count.items(): + if user_id is None: + # Abstained votes, ignore + continue if votes >= REQUIRED_VOTES_TO_BE_ELECTED: stmt = ( - select(CouncilMembers) - .where(CouncilMembers.ElectionName == election_name) - .where(CouncilMembers.MemberId == user_id) + select(CouncilMember) + .where(col(CouncilMember.election_name == election_name)) + .where(col(CouncilMember.member_id == user_id)) ) - council_member = session.scalar(stmt) - council_member.Elected = True + if (council_member := session.scalar(stmt)) is None: + dakiny_logger.error( + "Council member (id: %s) for election %s not found.", user_id, election_name + ) + return + council_member.elected = True session.commit() - member: nextcord.Member = self.guild.get_member(user_id) + if (member := self.guild.get_member(user_id)) is None: + dakiny_logger.error("Member (id: %s) not found in guild.", user_id) + return await member.add_roles( council_elect_role, @@ -274,42 +309,58 @@ class Election: async def highcouncil_election_start_action(self) -> None: """Performs required actions for when the highcouncil is elected.""" message = self.build_highcouncil_vote_start() - election_thread: nextcord.Thread = self.guild.get_thread(self.get_election_info().ThreadId) - await election_thread.send(message) - vote_msg = await election_thread.send( - build_highcouncil_votes_str(self.guild), - view=VoteHighcouncilView(self.guild), - ) - with Session(engine) as session: - stmt = select(ElectionInformation).where(ElectionInformation.ElectionName == get_election_name()) - info = session.scalar(stmt) - info.HighCouncilVoteMsgId = vote_msg.id + stmt = select(ElectionInformation).where(col(ElectionInformation.name == get_election_name())) + if (info := session.scalar(stmt)) is None: + dakiny_logger.error("Unknown election name %s.", get_election_name()) + return + + if (election_thread := self.guild.get_thread(info.thread_id)) is None: + dakiny_logger.error("Recorded thread id %s does not result in a thread.", info.thread_id) + return + + await election_thread.send(message) + vote_msg = await election_thread.send( + build_highcouncil_votes_str(self.guild), + view=VoteHighcouncilView(self.guild), + ) + + info.high_council_vote_msg_id = vote_msg.id session.commit() # remove highcouncil nomination view - nomination_msg: nextcord.Message = await election_thread.fetch_message(info.HighCouncilNominationMsgId) + if info.high_council_nomination_msg_id is None: + dakiny_logger.error("Election doesn't have a nomination message set.") + return + + nomination_msg: nextcord.Message = await election_thread.fetch_message(info.high_council_nomination_msg_id) await nomination_msg.edit(view=None) async def term_start_action(self) -> None: """Marks the election as over.""" self.election_over = True - elect_role: nextcord.Role = self.guild.get_role(settings.role_ids.clan_council_elect) - council_role: nextcord.Role = self.guild.get_role(settings.role_ids.clan_council) + if (elect_role := self.guild.get_role(settings.role_ids.clan_council_elect)) is None: + dakiny_logger.error("Cannot find Elect Role.") + return + if (council_role := self.guild.get_role(settings.role_ids.clan_council)) is None: + dakiny_logger.error("Cannot find Council Role.") + return election_name = get_election_name() with Session(engine) as session: # role management 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)) ) for member_id in session.scalars(stmt): - member = self.guild.get_member(member_id) + if (member := self.guild.get_member(member_id)) is None: + dakiny_logger.warning("Cannot find member (id: %s) to manage roles.", member_id) + continue await member.remove_roles( elect_role, reason=f"{member.display_name} no longer needs the elect role.", @@ -322,35 +373,36 @@ class Election: # announce high council members # ties are undefined behavior. stmt = ( - select(Votes.VotedForId) - .where(Votes.ElectionName == election_name) - .where(Votes.Type == VoteType.DISCORDMANAGER) + select(col(Vote.voted_for_id)) + .where(col(Vote.election_name == election_name)) + .where(col(Vote.type == VoteType.DISCORDMANAGER)) ) dm_votes = collections.Counter(session.scalars(stmt)) stmt = ( - select(Votes.VotedForId) - .where(Votes.ElectionName == election_name) - .where(Votes.Type == VoteType.CLANLEADER) + select(col(Vote.voted_for_id)) + .where(col(Vote.election_name == election_name)) + .where(col(Vote.type == VoteType.CLANLEADER)) ) cl_votes = collections.Counter(session.scalars(stmt)) + cl_votes.most_common() try: - dm_winner = max(dm_votes, key=dm_votes.get) + dm_winner = dm_votes.most_common(1).pop()[0] stmt = ( - update(CouncilMembers) - .where(CouncilMembers.ElectionName == election_name) - .where(CouncilMembers.MemberId == dm_winner) + update(CouncilMember) + .where(col(CouncilMember.election_name == election_name)) + .where(col(CouncilMember.member_id == dm_winner)) .values(DiscordManager=True) ) session.execute(stmt) except ValueError: dm_winner = "No winner" try: - cl_winner = max(cl_votes, key=cl_votes.get) + cl_winner = cl_votes.most_common(1).pop()[0] stmt = ( - update(CouncilMembers) - .where(CouncilMembers.ElectionName == election_name) - .where(CouncilMembers.MemberId == cl_winner) + update(CouncilMember) + .where(col(CouncilMember.election_name == election_name)) + .where(col(CouncilMember.member_id == cl_winner)) .values(ClanLeader=True) ) session.execute(stmt) @@ -359,14 +411,21 @@ class Election: session.commit() - election_thread: nextcord.Thread = self.guild.get_thread(self.get_election_info().ThreadId) + stmt = select(ElectionInformation).where(col(ElectionInformation.name == election_name)) + if (info := session.scalar(stmt)) is None: + dakiny_logger.error("Could not find information for this election (%s).", election_name) + return + + if (election_thread := self.guild.get_thread(info.thread_id)) is None: + dakiny_logger.error("Recorded thread id %s does not result in a thread.", info.thread_id) + return + + if info.high_council_vote_msg_id is None: + dakiny_logger.error("No recorded High Council Voting Message.") + return # clear view from vote message - stmt = select(ElectionInformation.HighCouncilVoteMsgId).where( - ElectionInformation.ElectionName == get_election_name() - ) - vote_msg_id = session.scalar(stmt) - vote_msg: nextcord.Message = await election_thread.fetch_message(vote_msg_id) + vote_msg: nextcord.Message = await election_thread.fetch_message(info.high_council_vote_msg_id) await vote_msg.edit(view=None) highcouncil_msg: str = f"Congratulations for our high council members!\nClan Leader: <@{cl_winner}>\n" @@ -378,22 +437,23 @@ class Election: await election_thread.edit(archived=True, locked=True) await edit_clan_council_info(self.guild) - def get_election_info(self) -> ElectionInformation: - """ - Gets information for current election. - - Returns: - ElectionInformation: The information from the database about the current election. - """ - with Session(engine) as session: - stmt = select(ElectionInformation).where(ElectionInformation.ElectionName == get_election_name()) - return session.scalar(stmt) - async def election_reminder(self) -> None: """Sends a message to general chat reminder clan about election.""" - general: nextcord.TextChannel = self.guild.get_channel(settings.channel_ids.general) + general = self.guild.get_channel(settings.channel_ids.general) - message = f"Reminder to <@&{settings.role_ids.clan_member}>! Clan elections are happening! Head over " - message += f"to <#{self.get_election_info().ThreadId}> to cast your vote!" + if not isinstance(general, nextcord.TextChannel): + dakiny_logger.error( + "Recorded general channel (id: %s) is not a text channel.", settings.channel_ids.general + ) + return - await general.send(content=message) + with Session(engine) as session: + stmt = select(ElectionInformation).where(col(ElectionInformation.name == get_election_name())) + if (info := session.scalar(stmt)) is None: + dakiny_logger.error("Could not find election %s.", get_election_name()) + return + + message = f"Reminder to <@&{settings.role_ids.clan_member}>! Clan elections are happening! Head over " + message += f"to <#{info.thread_id}> to cast your vote!" + + await general.send(content=message) diff --git a/dakiny_bot/classes/pyproject.toml b/dakiny_bot/classes/pyproject.toml index 5962016..5b0f99f 100644 --- a/dakiny_bot/classes/pyproject.toml +++ b/dakiny_bot/classes/pyproject.toml @@ -4,7 +4,3 @@ extend = "../../pyproject.toml" # Exclude a variety of commonly ignored directories. exclude = ["gitea_webhook.py"] - -[tool.ruff.lint] -# ignore implicit-namespace-package in this directory. -ignore = ["INP001"]