391 lines
11 KiB
Python
391 lines
11 KiB
Python
"""
|
|
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_current_cycle(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_current_cycle(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()
|