390 lines
16 KiB
Python
390 lines
16 KiB
Python
"""Contains the election cycle class."""
|
|
|
|
import collections
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
import nextcord
|
|
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 DakinySettings import settings
|
|
from database_setup.sql_data_classes import (
|
|
CouncilMembers,
|
|
ElectionInformation,
|
|
Votes,
|
|
engine,
|
|
)
|
|
from sqlalchemy import select, true, update
|
|
from sqlalchemy.orm import Session
|
|
from 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.
|
|
config (DakinyConfig): The entire config object.
|
|
guild (nextcord.Guild): The guild that this election is running in.
|
|
"""
|
|
self.cycle = cycle
|
|
self.election_cycle = settings.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 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_parent)
|
|
quarter_name: str = get_election_name(settings)
|
|
|
|
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(settings, self.guild),
|
|
view=VoteView(self.guild),
|
|
)
|
|
|
|
with Session(engine) as session:
|
|
stmt = select(ElectionInformation).where(ElectionInformation.ElectionName == get_election_name(settings))
|
|
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(settings)
|
|
|
|
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(settings, self.guild),
|
|
view=VoteHighcouncilView(self.guild),
|
|
)
|
|
|
|
with Session(engine) as session:
|
|
stmt = select(ElectionInformation).where(ElectionInformation.ElectionName == get_election_name(settings))
|
|
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(settings)
|
|
|
|
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(settings)
|
|
)
|
|
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, settings)
|
|
|
|
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(settings))
|
|
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)
|