552 lines
18 KiB
Python
552 lines
18 KiB
Python
"""File contains util functions."""
|
|
|
|
import calendar
|
|
import os
|
|
from collections.abc import Sequence
|
|
from datetime import UTC, datetime
|
|
from enum import StrEnum
|
|
from pathlib import Path
|
|
|
|
import aiofiles
|
|
import nextcord
|
|
from sqlalchemy import select, true, update
|
|
from sqlalchemy.orm import Session
|
|
|
|
from DakinySettings import settings
|
|
from database_setup.sql_data_classes import (
|
|
CouncilMembers,
|
|
EventPoints,
|
|
EventPointsLimit,
|
|
Messages,
|
|
Votes,
|
|
engine,
|
|
)
|
|
|
|
NOMINEE_SPLITTER = "========================================\n\n"
|
|
REPO_DIR = "./dakiny-messages"
|
|
MESSAGES_DIR = f"{REPO_DIR}/messages"
|
|
TEMPLATES_DIR = f"{REPO_DIR}/message_templates"
|
|
MAX_DISCORD_MESSAGE_LENGTH = 2000
|
|
GUILDS = [1000247361639030804, 239758371358638080]
|
|
|
|
VoteType = StrEnum("VoteType", ["COUNCIL", "DISCORDMANAGER", "CLANLEADER"])
|
|
|
|
|
|
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.
|
|
"""
|
|
today = datetime.now(tz=UTC).date()
|
|
year = 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 today.month == calendar.DECEMBER and month < calendar.DECEMBER:
|
|
year += 1
|
|
|
|
if today.month == calendar.JANUARY and month > calendar.JANUARY:
|
|
year -= 1
|
|
|
|
return datetime(year, month, day, tzinfo=UTC)
|
|
|
|
|
|
def get_election_name() -> str:
|
|
"""
|
|
Builds the unique name of the current election cycle.
|
|
|
|
Returns:
|
|
str: the name of the election cycle
|
|
"""
|
|
cycle = get_current_election_cycle()
|
|
current_cycle = settings.election_cycles[cycle]
|
|
return f"vote-{get_datetime(current_cycle.term_start).year}-{current_cycle.name}"
|
|
|
|
|
|
def get_quarter_name() -> 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_quarter_cycle(settings)
|
|
current_cycle = settings.election_cycles[cycle]
|
|
|
|
return f"{get_datetime(current_cycle.term_start).year}-{current_cycle.name}"
|
|
|
|
|
|
def get_current_election_cycle() -> 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.now(tz=UTC)
|
|
|
|
for cycle, election in enumerate(settings.election_cycles):
|
|
if (
|
|
today.date() >= get_datetime(election.election_start).date()
|
|
and today.date() <= get_datetime(election.term_start).date()
|
|
):
|
|
return cycle
|
|
|
|
# It isn't a current election, get the quarter
|
|
return get_quarter_cycle(settings)
|
|
|
|
|
|
def get_quarter_cycle(config: DakinySettings) -> 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.now(tz=UTC)
|
|
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: DakinySettings, guild: nextcord.Guild) -> str:
|
|
"""
|
|
Creates a string that contains all nominees and their votes to be sent as a discord message.
|
|
|
|
Args:
|
|
config (DakinyConfig): Dakiny 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] = {}
|
|
|
|
election_name = get_election_name()
|
|
|
|
with Session(engine) as session:
|
|
stmt = select(CouncilMembers).where(CouncilMembers.ElectionName == election_name)
|
|
|
|
for nominee in sorted(
|
|
session.scalars(stmt),
|
|
key=lambda nominee: guild.get_member(nominee.MemberId).display_name.lower(),
|
|
):
|
|
votes[nominee.MemberId] = 0
|
|
|
|
stmt = select(Votes).where(Votes.ElectionName == election_name).where(Votes.Type == VoteType.COUNCIL)
|
|
|
|
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: DakinySettings, 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): Dakiny 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.
|
|
"""
|
|
cl_votes: dict[int, int] = {}
|
|
dm_votes: dict[int, int] = {}
|
|
|
|
election_name = get_election_name()
|
|
|
|
with Session(engine) as session:
|
|
stmt = (
|
|
select(CouncilMembers)
|
|
.where(CouncilMembers.ElectionName == election_name)
|
|
.filter(CouncilMembers.ClanLeaderSpeechId.is_not(None))
|
|
)
|
|
|
|
for nominee in sorted(
|
|
session.scalars(stmt),
|
|
key=lambda nominee: guild.get_member(nominee.MemberId).display_name.lower(),
|
|
):
|
|
cl_votes[nominee.MemberId] = 0
|
|
|
|
stmt = select(Votes).where(Votes.ElectionName == election_name).where(Votes.Type == VoteType.CLANLEADER)
|
|
|
|
for vote in session.scalars(stmt):
|
|
if vote.VotedForId is not None:
|
|
cl_votes[vote.VotedForId] += 1
|
|
|
|
stmt = (
|
|
select(CouncilMembers)
|
|
.where(CouncilMembers.ElectionName == election_name)
|
|
.filter(CouncilMembers.DiscordManagerSpeechId.is_not(None))
|
|
)
|
|
|
|
for nominee in sorted(
|
|
session.scalars(stmt),
|
|
key=lambda nominee: guild.get_member(nominee.MemberId).display_name.lower(),
|
|
):
|
|
dm_votes[nominee.MemberId] = 0
|
|
|
|
stmt = select(Votes).where(Votes.ElectionName == election_name).where(Votes.Type == VoteType.DISCORDMANAGER)
|
|
|
|
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_points(config: DakinySettings, users: Sequence[str], add_points: int, *, is_member: bool) -> None:
|
|
"""
|
|
Adds event 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. Can be used for hosting or participating.
|
|
|
|
Args:
|
|
config (DakinyConfig): Dakiny config for getting points
|
|
users (Sequence[int]): users that participated and need points.
|
|
add_points (int): number of points to add
|
|
is_member (bool): whether the users are a discord member or not
|
|
"""
|
|
quarter_name = get_quarter_name()
|
|
with Session(engine) as session:
|
|
for user in users:
|
|
stmt = select(EventPoints).where(EventPoints.UserId == user).where(EventPoints.QuarterName == quarter_name)
|
|
|
|
points = session.scalar(stmt)
|
|
if points is None:
|
|
add_points = get_allowed_points(session, config.params.event_points_weekly_limit, user, add_points)
|
|
new_points = EventPoints(
|
|
QuarterName=quarter_name,
|
|
UserId=user,
|
|
DiscordMember=is_member,
|
|
Points=add_points,
|
|
)
|
|
session.add(new_points)
|
|
else:
|
|
add_points = get_allowed_points(session, config.params.event_points_weekly_limit, user, add_points)
|
|
points.Points += add_points
|
|
|
|
session.commit()
|
|
|
|
|
|
def get_allowed_points(session: Session, points_limit: int, user: str, points: int) -> int:
|
|
"""
|
|
Gets the maximum number of points a user can be given.
|
|
|
|
Searches how many points they have been given
|
|
this week and subtracts that from the weekly limit. If under, returns up to "points" number of points (will not
|
|
go over points_limit) otherwise returns 0. Updates the limit object or creates a new one if one isn't found.
|
|
|
|
Args:
|
|
session (Session): Open connection to database.
|
|
points_limit (int): Maximum number of points a user can be given in a week.
|
|
user (str): user to be given points. Used to check a user's limit.
|
|
points (int): maximum number of points to give to user.
|
|
|
|
Returns:
|
|
int: how many points are allowed to be given to the user.
|
|
"""
|
|
# check for limit
|
|
today = datetime.now(tz=UTC).date()
|
|
this_week = f"{today.isocalendar().year}-{today.isocalendar().week}"
|
|
stmt = select(EventPointsLimit).where(EventPointsLimit.UserId == user).where(EventPointsLimit.Week == this_week)
|
|
|
|
# if under limit, can add up to however many points would make the current limit equal to the
|
|
# allowed limit.
|
|
limit = session.scalar(stmt)
|
|
if limit is None:
|
|
new_limit = EventPointsLimit(UserId=user, Points=points, Week=this_week)
|
|
add_points = points
|
|
session.add(new_limit)
|
|
else:
|
|
add_points = points if limit.Points + points <= points_limit else (max(0, points_limit - limit.Points))
|
|
limit.Points += add_points
|
|
|
|
session.commit()
|
|
|
|
return add_points
|
|
|
|
|
|
async def update_clan_messages(guild: nextcord.Guild, channel_updates: Sequence[str]) -> None:
|
|
"""
|
|
Updates all clan messages to contents in MESSAGES_DIR.
|
|
|
|
Does not to any git work. Is expected to be called after git is pulled.
|
|
|
|
For every directory, sort the files by number then edits/sends the message.
|
|
|
|
Channel fetching/management is handled by another function.
|
|
Message editing and sending is handled by another function.
|
|
|
|
Args:
|
|
guild (nextcord.Guild): Guild object that all channels and messages belong to.
|
|
channel_updates: (Sequence[str]): a sequence of the channels to update.
|
|
"""
|
|
dirs = [dir_name for dir_name in channel_updates if Path(f"{MESSAGES_DIR}/{dir_name}").is_dir()]
|
|
|
|
with Session(engine) as session:
|
|
for dir_name in dirs:
|
|
dir_path = f"{MESSAGES_DIR}/{dir_name}"
|
|
channel = await get_clan_message_channel(session, dir_name, guild)
|
|
sorted_files = sorted(
|
|
os.listdir(dir_path),
|
|
key=lambda filename: int(filename.split(".")[0]),
|
|
)
|
|
|
|
for file_name in sorted_files:
|
|
message_name = f"{dir_name}/{file_name}"
|
|
file_path = f"{dir_path}/{file_name}"
|
|
async with aiofiles.open(file_path, encoding="utf8") as reader:
|
|
await edit_clan_message(
|
|
session, channel, message_name, await reader.read(MAX_DISCORD_MESSAGE_LENGTH)
|
|
)
|
|
|
|
|
|
async def get_clan_message_channel(session: Session, channel_name: str, guild: nextcord.Guild) -> nextcord.TextChannel:
|
|
"""
|
|
Used to obtain a channel from the guild object.
|
|
|
|
Searches database for an id to get the channel. If that is not
|
|
found, then searches through all channels in the guild to find one with a matching name. If that also doesn't
|
|
turn up a channel, then creates it. Adds an entry into database for last two paths.
|
|
Assumes that there is an open session to the database.
|
|
|
|
Args:
|
|
session (Session): Open connection to database.
|
|
channel_name (str): Name of channel to obtain.
|
|
guild (nextcord.Guild): Guild object to get channel from.
|
|
|
|
Returns:
|
|
nextcord.TextChannel: Channel object with name channel_name.
|
|
"""
|
|
stmt = select(Messages.Id).where(Messages.Path == channel_name)
|
|
channel_id = session.scalar(stmt)
|
|
if channel_id is None:
|
|
ret_channel = None
|
|
# channel is not tracked. see if it exists
|
|
channels = await guild.fetch_channels()
|
|
for channel in channels:
|
|
if channel.name == channel_name and channel.__class__ is nextcord.TextChannel:
|
|
ret_channel = channel
|
|
|
|
if ret_channel is None:
|
|
# channel doesn't exist, we must create it
|
|
ret_channel = await guild.create_text_channel(
|
|
channel_name,
|
|
reason="Creating channel as required for by design of Dakiny-Messages repo.",
|
|
)
|
|
|
|
# record channel information
|
|
new_item = Messages(Path=channel_name, Id=ret_channel.id)
|
|
session.add(new_item)
|
|
session.commit()
|
|
return ret_channel
|
|
|
|
try:
|
|
return await guild.fetch_channel(channel_id)
|
|
except nextcord.NotFound:
|
|
# the channel was deleted for some reason
|
|
ret_channel = await guild.create_text_channel(
|
|
channel_name,
|
|
reason="Creating channel as required for by design of Dakiny-Messages repo.",
|
|
)
|
|
stmt = update(Messages).where(Messages.Path == channel_name).values(Id=ret_channel.id)
|
|
session.execute(stmt)
|
|
session.commit()
|
|
return ret_channel
|
|
|
|
|
|
async def edit_clan_message(
|
|
session: Session,
|
|
channel: nextcord.TextChannel,
|
|
message_name: str,
|
|
message_content: str,
|
|
) -> None:
|
|
"""
|
|
Obtains a message id from database.
|
|
|
|
If no id, then need to send a new message and record the id.
|
|
If there is a message, then edit the message. If that message is not found in discord, then need to
|
|
send a new message and record the id in the database.
|
|
|
|
Args:
|
|
session (Session): Open connection to database.
|
|
channel (nextcord.TextChannel): Channel object that message belongs to.
|
|
message_name (str): Name of the message. Corresponds to the file that contains message content.
|
|
message_content (str): Content of the message to edit/send.
|
|
"""
|
|
stmt = select(Messages.Id).where(Messages.Path == message_name)
|
|
msg_id = session.scalar(stmt)
|
|
if msg_id is None:
|
|
# don't have a message for this file
|
|
msg = await channel.send(message_content)
|
|
new_message = Messages(Path=message_name, Id=msg.id)
|
|
session.add(new_message)
|
|
session.commit()
|
|
else:
|
|
try:
|
|
msg: nextcord.Message = await channel.fetch_message(msg_id)
|
|
await msg.edit(message_content)
|
|
except nextcord.NotFound:
|
|
msg = await channel.send(message_content)
|
|
stmt = update(Messages).where(Messages.Path == message_name).values(Id=msg.id)
|
|
session.execute(stmt)
|
|
session.commit()
|
|
|
|
|
|
async def edit_clan_council_info(guild: nextcord.Guild, config: DakinySettings) -> None:
|
|
"""
|
|
Gets the winners of the elections and updates the permanent message that states the current council.
|
|
|
|
Args:
|
|
guild (nextcord.Guild): Guild object that to use to get channel.
|
|
config (DakinyConfig): Configs
|
|
"""
|
|
clan_info_channel: nextcord.TextChannel = await guild.fetch_channel(config.params.clan_info_channel_id)
|
|
|
|
election_name = get_election_name()
|
|
with Session(engine) as session:
|
|
stmt = (
|
|
select(CouncilMembers.MemberId)
|
|
.where(CouncilMembers.ElectionName == election_name)
|
|
.where(CouncilMembers.DiscordManager == true())
|
|
)
|
|
discord_manager = session.scalar(stmt)
|
|
stmt = (
|
|
select(CouncilMembers.MemberId)
|
|
.where(CouncilMembers.ElectionName == election_name)
|
|
.where(CouncilMembers.ClanLeader == true())
|
|
)
|
|
clan_leader = session.scalar(stmt)
|
|
stmt = (
|
|
select(CouncilMembers.MemberId)
|
|
.where(CouncilMembers.ElectionName == election_name)
|
|
.where(CouncilMembers.Elected == true())
|
|
)
|
|
|
|
council_member_ids = sorted(
|
|
session.scalars(stmt), key=lambda user_id: guild.get_member(user_id).display_name.lower()
|
|
)
|
|
|
|
council_members_strs = [
|
|
f"<@{user_id}>" for user_id in council_member_ids if user_id not in (discord_manager, clan_leader)
|
|
]
|
|
|
|
quarter = get_quarter_name()
|
|
|
|
async with aiofiles.open(f"{TEMPLATES_DIR}/council_members.txt", encoding="utf8") as file:
|
|
message = await file.read()
|
|
message = message.format(
|
|
quarter=quarter,
|
|
council="\n".join(council_members_strs),
|
|
clan_leader=clan_leader,
|
|
discord_manager=discord_manager,
|
|
)
|
|
|
|
await edit_clan_message(session, clan_info_channel, "council_members.txt", message)
|