""" Contains the election cycle class """ import collections from datetime import datetime, timedelta import nextcord from sqlalchemy import select, true, update from sqlalchemy.orm import Session from classes.nominate_highcouncil_views import NominateHighCouncilView from classes.nominate_views import NominateView from classes.voting_highcouncil_views import VoteHighcouncilView from classes.voting_views import VoteView from config import DakinyConfig from database_setup.sql_data_classes import ( CouncilMembers, ElectionInformation, Votes, engine, ) from utils import ( TEMPLATES_DIR, VoteType, build_highcouncil_votes_str, build_votes_str, edit_clan_council_info, get_datetime, get_election_name, ) class ElectionCycle: """ 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, config: DakinyConfig, 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. config (DakinyConfig): The entire config object. guild (nextcord.Guild): The guild that this election is running in. """ self.config: DakinyConfig = config self.cycle = cycle self.election_cycle = self.config.election_cycles[self.cycle] 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 open( f"{TEMPLATES_DIR}/election_announcement.txt", mode="r", encoding="utf8" ) as file: message = file.read() next_election = get_datetime( self.config.election_cycles[ (self.cycle + 1) % len(self.config.election_cycles) ].election_start ) message = message.format( clan_council_id=self.config.params.clan_council_role_id, high_council_id=self.config.params.high_council_role_id, 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 open( f"{TEMPLATES_DIR}/highcouncil_election.txt", mode="r", encoding="utf8" ) as file: message = file.read() return message.format( clan_council_elect_role_id=self.config.params.clan_council_elect_role_id, 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 open( f"{TEMPLATES_DIR}/highcouncil_vote.txt", mode="r", encoding="utf8" ) as file: message = file.read() return message.format( clan_council_elect_role_id=self.config.params.clan_council_elect_role_id, 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 open(f"{TEMPLATES_DIR}/vote_start.txt", mode="r", encoding="utf8") as file: message = file.read() vote_end: datetime = get_datetime( self.config.election_cycles[self.cycle].voting_end ) return message.format( clan_member_id=self.config.params.clan_member_role_id, 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.today() # 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( self.config.params.voting_parent_id ) quarter_name: str = get_election_name(self.config) 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.config, self.guild) ) else: await thread.send(message) # create new entry into database. with Session(engine) as session: new_cycle = ElectionInformation( ElectionCycle=quarter_name, ThreadId=thread.id, CouncilNominationMsgId=nominate_msg.id, ) session.add(new_cycle) session.commit() async def voting_start_action(self) -> None: """ 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.config, self.guild), view=VoteView(self.config, self.guild), ) with Session(engine) as session: stmt = select(ElectionInformation).where( ElectionInformation.ElectionCycle == get_election_name(self.config) ) 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( self.config.params.clan_council_elect_role_id ) 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.config, self.guild) ) election_name = get_election_name(self.config) with Session(engine) as session: # remove vote button from vote message stmt = select(ElectionInformation).where( ElectionInformation.ElectionCycle == 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.ElectionCycle == election_name) .where(Votes.Type == VoteType.COUNCIL) ) vote_count = collections.Counter(session.scalars(stmt)) for user_id, votes in vote_count.items(): if votes >= 3: stmt = ( select(CouncilMembers) .where(CouncilMembers.ElectionCycle == 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.config, self.guild), view=VoteHighcouncilView(self.config, self.guild), ) with Session(engine) as session: stmt = select(ElectionInformation).where( ElectionInformation.ElectionCycle == get_election_name(self.config) ) 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( self.config.params.clan_council_elect_role_id ) council_role: nextcord.Role = self.guild.get_role( self.config.params.clan_council_role_id ) election_cycle = get_election_name(self.config) with Session(engine) as session: # role management stmt = ( select(CouncilMembers.MemberId) .where(CouncilMembers.ElectionCycle == election_cycle) .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.ElectionCycle == election_cycle) .where(Votes.Type == VoteType.DISCORDMANAGER) ) dm_votes = collections.Counter(session.scalars(stmt)) stmt = ( select(Votes.VotedForId) .where(Votes.ElectionCycle == election_cycle) .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.ElectionCycle == election_cycle) .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.ElectionCycle == election_cycle) .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.ElectionCycle == get_election_name(self.config) ) 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, self.config) 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.ElectionCycle == get_election_name(self.config) ) 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( self.config.params.general_channel_id ) message = f"Reminder to <@&{self.config.params.clan_member_role_id}>! Clan elections are happening! Head over " message += f"to <#{self.get_election_info().ThreadId}> to cast your vote!" await general.send(content=message)