""" File contains util functions. """ from datetime import datetime import nextcord from sqlalchemy import select from sqlalchemy.orm import Session from config import DakinyConfig from database_setup.sql_data_classes import ( CouncilMembers, ParticipationPoints, Votes, engine, ) NOMINEE_SPLITTER = "========================================\n\n" def get_datetime(day: str) -> datetime: """ Takes in a string month and day and returns a datetime object of that Month/Day this year. Args: day (str): Month/Day in that format. Returns: datetime: Datetime representation of Month/Day. """ year = datetime.today().year month = int(day.split("/")[0]) day = int(day.split("/")[1]) # ensure that year rolls over if current month is December and the next month is less than 12. if datetime.today().month == 12 and month < 12: year += 1 if datetime.today().month == 1 and month > 1: year -= 1 return datetime(year, month, day) def get_election_cycle_name(config: DakinyConfig) -> str: """ Builds the unique name of the current election cycle Args: config (DakinyConfig): the config used to know the election cycle. Returns: str: the name of the election cycle """ cycle = get_current_election_cycle(config) current_cycle = config.election_cycles[cycle] return f"vote-{get_datetime(current_cycle.term_start).year}-{current_cycle.name}" def get_cycle_name(config: DakinyConfig) -> str: """ Builds the name of the current cycle/quarter Args: config (DakinyConfig): the config used to know the election cycle. Returns: str: the name of the cycle. """ cycle = get_current_cycle(config) current_cycle = config.election_cycles[cycle] return f"{get_datetime(current_cycle.term_start).year}-{current_cycle.name}" def get_dm_election_name(config: DakinyConfig) -> str: """ creates the name of the discord manager election Args: config (DakinyConfig): Config sent to get_election_cycle_name Returns: str: the name of the election """ return f"{get_election_cycle_name(config)}-dm" def get_cl_election_name(config: DakinyConfig) -> str: """ creates the name of the clan leader election Args: config (DakinyConfig): Config sent to get_election_cycle_name Returns: str: the name of the election """ return f"{get_election_cycle_name(config)}-cl" def get_current_election_cycle(config: DakinyConfig) -> int: """ Returns index for active election. An election is considered active between the election start and the term start. Args: config (DakinyConfig): Config object to use Returns: int: cycle index for active election. None for no active cycle """ today = datetime.today() for cycle, election in enumerate(config.election_cycles): if ( today.date() >= get_datetime(election.election_start).date() and today.date() <= get_datetime(election.term_start).date() ): return cycle return None def get_current_cycle(config: DakinyConfig) -> int: """ Returns the index for the current cycle/quarter. Defined as between term start and end. Shouldn't return none Args: config (DakinyConfig): Config object to use Returns: int: cycle index for current cycle. """ today = datetime.today() for cycle, election in enumerate(config.election_cycles): if ( today.date() >= get_datetime(election.term_start).date() and today.date() <= get_datetime(election.term_end).date() ): return cycle return None def build_votes_str(config: DakinyConfig, guild: nextcord.Guild) -> str: """ Creates a string that contains all nominees and their votes to be sent as a discord message. Args: config (DakinyConfig): Daking config object guild (nextcord.Guild): The guild object. Used to get display names for sorting. Returns: str: String that contains all nominees and their votes. """ votes: dict[int, int] = {} with Session(engine) as session: stmt = select(CouncilMembers).where( CouncilMembers.ElectionCycle == get_election_cycle_name(config) ) for nominee in sorted( session.scalars(stmt), key=lambda nominee: guild.get_member(nominee.MemberId).display_name, ): votes[nominee.MemberId] = 0 stmt = select(Votes).where( Votes.ElectionCycle == get_election_cycle_name(config) ) for vote in session.scalars(stmt): if vote.VotedForId is not None: votes[vote.VotedForId] += 1 votes_str: str = "====================" votes_str += user_vote_str(votes) return votes_str def build_highcouncil_votes_str(config: DakinyConfig, guild: nextcord.Guild) -> str: """ Creates a string that contains all high council nominees and their votes to be sent as a discord message. Args: config (DakinyConfig): Daking config object Returns: str: String that contains all nominees and their votes. guild (nextcord.Guild): The guild object. Used to get display names for sorting. """ cl_votes: dict[int, int] = {} dm_votes: dict[int, int] = {} with Session(engine) as session: stmt = ( select(CouncilMembers) .where(CouncilMembers.ElectionCycle == get_election_cycle_name(config)) .filter(CouncilMembers.ClanLeaderSpeechId.is_not(None)) ) for nominee in sorted( session.scalars(stmt), key=lambda nominee: guild.get_member(nominee.MemberId).display_name, ): cl_votes[nominee.MemberId] = 0 stmt = select(Votes).where(Votes.ElectionCycle == get_cl_election_name(config)) for vote in session.scalars(stmt): if vote.VotedForId is not None: cl_votes[vote.VotedForId] += 1 stmt = ( select(CouncilMembers) .where(CouncilMembers.ElectionCycle == get_election_cycle_name(config)) .filter(CouncilMembers.DiscordManagerSpeechId.is_not(None)) ) for nominee in sorted( session.scalars(stmt), key=lambda nominee: guild.get_member(nominee.MemberId).display_name, ): dm_votes[nominee.MemberId] = 0 stmt = select(Votes).where(Votes.ElectionCycle == get_dm_election_name(config)) for vote in session.scalars(stmt): if vote.VotedForId is not None: dm_votes[vote.VotedForId] += 1 # build the votes string votes_str: str = "Clan Leader Votes\n====================" votes_str += user_vote_str(cl_votes) votes_str += "\n\nDiscord Manager Votes\n====================" votes_str += user_vote_str(dm_votes) return votes_str def user_vote_str(votes: dict[int, int]) -> str: """ Creates the string representing how many votes each user has. Args: votes (dict[int, int]): Dict of all users and how many votes they have. Returns: str: The string showing names and number of votes. """ ret_str: str = "" for user_id, num_votes in votes.items(): ret_str += f"\n<@{user_id}>: {'☑️' * num_votes}" return ret_str def decorate_nomination(user_id: int, speech: str) -> str: """ Decorates a nomination speech so that it sits within multiple speeches better. Args: user_id (int): ID of user sending the speech speech (str): Speech from the user Returns: str: Decorated speech """ msg = f"**Clan Council Nominee: <@{user_id}>**\n" msg += NOMINEE_SPLITTER msg += speech return msg def decorate_cl_nomination(user_id: int, speech: str) -> str: """ Decorates a nomination speech so that it sits within multiple speeches better. Args: user_id (int): ID of user sending the speech speech (str): Speech from the user Returns: str: Decorated speech """ msg = f"**Clan Leader Nominee: <@{user_id}>**\n" msg += NOMINEE_SPLITTER msg += speech return msg def decorate_dm_nomination(user_id: int, speech: str) -> str: """ Decorates a nomination speech so that it sits within multiple speeches better. Args: user_id (int): ID of user sending the speech speech (str): Speech from the user Returns: str: Decorated speech """ msg = f"**Discord Manager Nominee: <@{user_id}>**\n" msg += NOMINEE_SPLITTER msg += speech return msg def record_event_participation(config: DakinyConfig, users: list[str], is_member: bool): """ Adds participation points to all users listed. Checks if they don't exist in order to make a new entry, otherwise adds points to the existing value. Args: config (DakinyConfig): Dakiny config for getting points users (list[int]): users that participated and need points. is_member (bool): whether the users are a discord member or not """ cycle = get_cycle_name(config) participation_point_value = config.params.event_participate_points with Session(engine) as session: for user in users: stmt = ( select(ParticipationPoints) .where(ParticipationPoints.UserId == user) .where(ParticipationPoints.Cycle == cycle) ) points = session.scalar(stmt) if points is None: new_points = ParticipationPoints( Cycle=cycle, UserId=user, DiscordMember=is_member, Points=participation_point_value, ) session.add(new_points) else: points.Points += participation_point_value session.commit() def record_event_host(config: DakinyConfig, user: int): """ Adds host points to the given user. Checks if they don't exist in order to make a new entry, otherwise adds points to the existing value. Args: config (DakinyConfig): Dakiny config for getting points user (int): Id of user that hosted. """ cycle = get_cycle_name(config) host_point_value = config.params.event_host_points with Session(engine) as session: stmt = ( select(ParticipationPoints) .where(ParticipationPoints.UserId == str(user)) .where(ParticipationPoints.Cycle == cycle) ) points = session.scalar(stmt) if points is None: new_points = ParticipationPoints( Cycle=cycle, UserId=str(user), DiscordMember=True, Points=host_point_value, ) session.add(new_points) else: points.Points += host_point_value session.commit()