"""Contains the election cycle class.""" import collections 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.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, 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").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").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").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").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) dakiny_logger.debug("Checking date.") # check all dates if action is needed if today.date() == self.election_start.date(): dakiny_logger.debug("Election start.") await self.election_start_action() elif today.date() == self.voting_start.date(): dakiny_logger.debug("Voting start.") await self.voting_start_action() elif (self.voting_end - timedelta(3)).date() <= today.date() < self.voting_end.date(): dakiny_logger.debug("Election reminder.") await self.election_reminder() elif today.date() == self.voting_end.date(): dakiny_logger.debug("Voting end.") await self.voting_end_action() elif today.date() == self.highcouncil_election_start.date(): dakiny_logger.debug("Highcouncil election start.") await self.highcouncil_election_start_action() elif today.date() == self.term_start.date(): dakiny_logger.debug("Term start") await self.term_start_action() else: dakiny_logger.warning("Check date didn't do anything. This should never happen.") async def election_start_action(self) -> None: """Performs required actions for when the election cycle starts.""" # create new channel for election 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) 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( name=quarter_name, thread_id=thread.id, council_nomination_msg_id=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() with Session(engine) as session: 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() 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() 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 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.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(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(CouncilMember) .where(col(CouncilMember.election_name == election_name)) .where(col(CouncilMember.member_id == user_id)) ) 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() 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, 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() 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("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 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 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(col(CouncilMember.member_id)) .where(col(CouncilMember.election_name == election_name)) .where(col(CouncilMember.elected)) ) for member_id in session.scalars(stmt): 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.", ) 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(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(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 = dm_votes.most_common(1).pop()[0] stmt = ( 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 = cl_votes.most_common(1).pop()[0] stmt = ( update(CouncilMember) .where(col(CouncilMember.election_name == election_name)) .where(col(CouncilMember.member_id == cl_winner)) .values(ClanLeader=True) ) session.execute(stmt) except ValueError: cl_winner = "No winner" session.commit() 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 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" 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) async def election_reminder(self) -> None: """Sends a message to general chat reminder clan about election.""" general = self.guild.get_channel(settings.channel_ids.general) if not isinstance(general, nextcord.TextChannel): dakiny_logger.error( "Recorded general channel (id: %s) is not a text channel.", settings.channel_ids.general ) return 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)