"""Contains the election cycle class.""" import collections from datetime import UTC, datetime, timedelta from pathlib import Path import nextcord from sqlalchemy import select, true, update from sqlalchemy.orm import Session 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.DakinySettings import settings from dakiny_bot.database.sql_data_classes import ( CouncilMembers, ElectionInformation, Votes, engine, ) from dakiny_bot.utils import ( TEMPLATES_DIR, VoteType, build_highcouncil_votes_str, build_votes_str, edit_clan_council_info, get_datetime, get_election_name, ) REQUIRED_VOTES_TO_BE_ELECTED = 3 class Election: """ Election class. This class represents all the dates required for an election cycle along with all the functions needed to operate the cycle. """ def __init__(self, cycle: int, guild: nextcord.Guild) -> None: """ Generates all dates relevant to the election cycle determined by the config and cycle. Args: cycle (int): Which election cycle this cycle is. guild (nextcord.Guild): The guild that this election is running in. """ self.cycle = cycle self.election_cycle = settings.election_cycles[self.cycle] self.active = self.election_cycle.active self.election_start: datetime = get_datetime(self.election_cycle.election_start) self.term_start: datetime = get_datetime(self.election_cycle.term_start) self.term_end: datetime = get_datetime(self.election_cycle.term_end) self.voting_start: datetime = get_datetime(self.election_cycle.voting_start) self.voting_end: datetime = get_datetime(self.election_cycle.voting_end) self.highcouncil_election_announce: datetime = get_datetime(self.election_cycle.highcouncil_election_announce) self.highcouncil_election_start: datetime = get_datetime(self.election_cycle.highcouncil_election_start) self.highcouncil_election_end: datetime = get_datetime(self.election_cycle.highcouncil_election_end) self.guild: nextcord.Guild = guild self.election_over: bool = False def build_election_announcement(self) -> list[str]: """ Builds the messages to announce the election is starting. Returns: 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: message = file.read() next_election = get_datetime( settings.election_cycles[(self.cycle + 1) % len(settings.election_cycles)].election_start ) message = message.format( clan_council_id=settings.role_ids.clan_council, high_council_id=settings.role_ids.high_council, election_start=self.election_start.strftime("%B %d"), term_start=self.term_start.strftime("%B %d"), term_end=self.term_end.strftime("%B %d"), voting_start=self.voting_start.strftime("%B %d"), voting_end=self.voting_end.strftime("%B %d"), highcouncil_election=self.highcouncil_election_start.strftime("%B %d"), next_election_start=next_election.strftime("%B %d"), ) return message.split("---message_break---") def build_highcouncil_election_announce(self) -> str: """ Builds message to announce highcouncil election start. Returns: str: Message to send. """ with Path(f"{TEMPLATES_DIR}/highcouncil_election.txt", encoding="utf8").open() as file: message = file.read() return message.format( clan_council_elect_role_id=settings.role_ids.clan_council_elect, highcouncil_election_announce=self.highcouncil_election_announce.strftime("%B %d"), highcouncil_election_start=self.highcouncil_election_start.strftime("%B %d"), highcouncil_election_end=self.highcouncil_election_end.strftime("%B %d"), term_start=self.term_start.strftime("%B %d"), ) def build_highcouncil_vote_start(self) -> str: """ Builds the message to announce that highcouncil voting has started. Returns: str: Message to send. """ with Path(f"{TEMPLATES_DIR}/highcouncil_vote.txt", encoding="utf8").open() as file: message = file.read() return message.format( clan_council_elect_role_id=settings.role_ids.clan_council_elect, term_start=self.term_start.strftime("%B %d"), ) def build_vote_start(self) -> str: """ Builds the voting start message out of template and configurations. Returns: str: Message to send """ with Path(f"{TEMPLATES_DIR}/vote_start.txt", encoding="utf8").open() as file: message = file.read() vote_end: datetime = get_datetime(settings.election_cycles[self.cycle].voting_end) return message.format( clan_member_id=settings.role_ids.clan_member, voting_end=vote_end.strftime("%B %d"), voting_end_day=vote_end.strftime("%A"), ) async def check_date(self) -> None: """Function checks current date to see if something needs to be run.""" today = datetime.now(tz=UTC) # check all dates if action is needed if today.date() == self.election_start.date(): await self.election_start_action() elif today.date() == self.voting_start.date(): await self.voting_start_action() elif (self.voting_end - timedelta(3)).date() <= today.date() < self.voting_end.date(): await self.election_reminder() elif today.date() == self.voting_end.date(): await self.voting_end_action() elif today.date() == self.highcouncil_election_start.date(): await self.highcouncil_election_start_action() elif today.date() == self.term_start.date(): await self.term_start_action() 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) quarter_name: str = get_election_name() thread_msg: nextcord.Message = await parent.send(content=quarter_name) thread: nextcord.Thread = await parent.create_thread(name=quarter_name, message=thread_msg) # post message announcing election # election start message is returned as an array of string because it is too long. send each string # individually. messages = self.build_election_announcement() for index, message in enumerate(messages): # if last item, add the nominate button if index + 1 == len(messages): nominate_msg: nextcord.Message = await thread.send(message, view=NominateView(self.guild)) else: await thread.send(message) # create new entry into database. with Session(engine) as session: new_cycle = ElectionInformation( ElectionName=quarter_name, ThreadId=thread.id, CouncilNominationMsgId=nominate_msg.id, ) session.add(new_cycle) session.commit() async def voting_start_action(self) -> None: """ Actions performed when starting a vote. Sends message informing members that voting has started. 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 session.commit() thread: nextcord.Thread = self.guild.get_thread(info.ThreadId) nominate_msg: nextcord.Message = await thread.fetch_message(info.CouncilNominationMsgId) 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)) election_name = get_election_name() with Session(engine) as session: # 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) await vote_msg.edit(view=None) # add the highcouncil nomination message to election information info.HighCouncilNominationMsgId = 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) ) vote_count = collections.Counter(session.scalars(stmt)) for user_id, votes in vote_count.items(): if votes >= REQUIRED_VOTES_TO_BE_ELECTED: stmt = ( select(CouncilMembers) .where(CouncilMembers.ElectionName == election_name) .where(CouncilMembers.MemberId == user_id) ) council_member = session.scalar(stmt) council_member.Elected = True session.commit() member: nextcord.Member = self.guild.get_member(user_id) await member.add_roles( council_elect_role, reason=f"{member.display_name} has won a nomination and become a Council Elect.", ) 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 session.commit() # remove highcouncil nomination view nomination_msg: nextcord.Message = await election_thread.fetch_message(info.HighCouncilNominationMsgId) 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) 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()) ) for member_id in session.scalars(stmt): member = self.guild.get_member(member_id) await member.remove_roles( elect_role, reason=f"{member.display_name} no longer needs the elect role.", ) await member.add_roles( council_role, reason=f"{member.display_name} has been elected to council.", ) # announce high council members # ties are undefined behavior. stmt = ( select(Votes.VotedForId) .where(Votes.ElectionName == election_name) .where(Votes.Type == VoteType.DISCORDMANAGER) ) dm_votes = collections.Counter(session.scalars(stmt)) stmt = ( select(Votes.VotedForId) .where(Votes.ElectionName == election_name) .where(Votes.Type == VoteType.CLANLEADER) ) cl_votes = collections.Counter(session.scalars(stmt)) try: dm_winner = max(dm_votes, key=dm_votes.get) stmt = ( update(CouncilMembers) .where(CouncilMembers.ElectionName == election_name) .where(CouncilMembers.MemberId == dm_winner) .values(DiscordManager=True) ) session.execute(stmt) except ValueError: dm_winner = "No winner" try: cl_winner = max(cl_votes, key=cl_votes.get) stmt = ( update(CouncilMembers) .where(CouncilMembers.ElectionName == election_name) .where(CouncilMembers.MemberId == cl_winner) .values(ClanLeader=True) ) session.execute(stmt) except ValueError: cl_winner = "No winner" session.commit() election_thread: nextcord.Thread = self.guild.get_thread(self.get_election_info().ThreadId) # 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) await vote_msg.edit(view=None) highcouncil_msg: str = f"Congratulations for our high council members!\nClan Leader: <@{cl_winner}>\n" highcouncil_msg += f"Discord Manager: <@{dm_winner}>" await election_thread.send(highcouncil_msg) await election_thread.send("This election cycle is now over. Thank you everyone for participating!") 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) 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!" await general.send(content=message)