dakiny-bot/dakiny_bot/utils.py

"""File contains util functions."""

import calendar
import pathlib
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, update
from sqlalchemy.orm import Session
from sqlmodel import col

from dakiny_bot.database.db import engine
from dakiny_bot.database.models import CouncilMember, Message, Points, PointsLimit, Vote
from dakiny_bot.logger import dakiny_logger

from .dakiny_settings import settings

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"])
PointType = StrEnum("PointType", ["CLAN", "EVENT"])


# TODO(imcompiling): resolve returning None from this function
# https://gitea.cheezers.org/imcompiling/dakiny-bot/issues/102
def get_datetime(month_day: str) -> datetime:  # pyright: ignore[reportReturnType]
    """
    Takes in a string month and day and returns a datetime object of that Month/Day this year.

    Args:
        month_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(month_day.split("/")[0])
    day = int(month_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.OCTOBER:
        year += 1

    if today.month == calendar.JANUARY and month > calendar.MARCH:
        year -= 1
    try:
        return datetime(year, month, day, tzinfo=UTC)
    except ValueError:
        dakiny_logger.warning("Invalid month day: %s.", month_day)


def get_election_name() -> str:
    """
    Builds the unique name of the current election cycle.

    Returns:
        str: the name of the election cycle
    """
    cycle = get_active_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.

    Returns:
        str: the name of the cycle.
    """
    cycle = get_quarter_cycle()
    current_cycle = settings.election_cycles[cycle]

    return f"{get_datetime(current_cycle.term_start).year}-{current_cycle.name}"


def get_active_election_cycle() -> int:
    """
    Returns index for active election. An election is considered active between the election start and the term start.

    Returns:
        int: cycle index for active election. None for no active cycle
    """
    today = datetime.now(tz=UTC)
    dakiny_logger.debug("Today is %s", today)

    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()
        ):
            dakiny_logger.debug("Cycle is %d", cycle)
            return cycle

    # It isn't a current election, get the quarter
    return get_quarter_cycle()


def get_quarter_cycle() -> int:
    """
    Returns the index for the current cycle/quarter. Defined as between term start and end.

    Can return None if entire year isn't covered by cycles.

    Returns:
        int: cycle index for current cycle.
    """
    today = datetime.now(tz=UTC)
    dakiny_logger.debug("Today is %s", today)
    for cycle, election in enumerate(settings.election_cycles):
        if (
            today.date() >= get_datetime(election.term_start).date()
            and today.date() <= get_datetime(election.term_end).date()
        ):
            dakiny_logger.debug("Cycle is %d", cycle)
            return cycle

    dakiny_logger.debug("Quarter cycle is None.")
    # TODO(imcompiling) Add None to return signature or find better return value
    # https://gitea.cheezers.org/imcompiling/dakiny-bot/issues/102
    return None  # pyright: ignore[reportReturnType]


def build_votes_str(guild: nextcord.Guild) -> str:
    """
    Creates a string that contains all nominees and their votes to be sent as a discord message.

    Args:
        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(CouncilMember).where(col(CouncilMember.election_name == election_name))

        for nominee in sorted(
            session.scalars(stmt),
            # Unsure how to None check in the lambda.
            key=lambda nominee: guild.get_member(nominee.member_id).display_name.lower(),  # pyright: ignore[reportOptionalMemberAccess]
        ):
            votes[nominee.member_id] = 0

        stmt = select(Vote).where(col(Vote.election_name == election_name)).where(col(Vote.type == VoteType.COUNCIL))

        for vote in session.scalars(stmt):
            if vote.voted_for_id is not None:
                votes[vote.voted_for_id] += 1

    votes_str: str = "===================="
    votes_str += user_vote_str(votes)

    return votes_str


def build_highcouncil_votes_str(guild: nextcord.Guild) -> str:
    """
    Creates a string that contains all high council nominees and their votes to be sent as a discord message.

    Args:
        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(CouncilMember)
            .where(col(CouncilMember.election_name == election_name))
            .filter(col(CouncilMember.clan_leader_speech_id is not None))
        )

        for nominee in sorted(
            session.scalars(stmt),
            # Unsure how to None check in the lambda.
            key=lambda nominee: guild.get_member(nominee.member_id).display_name.lower(),  # pyright: ignore[reportOptionalMemberAccess]
        ):
            cl_votes[nominee.member_id] = 0

        stmt = select(Vote).where(col(Vote.election_name == election_name)).where(col(Vote.type == VoteType.CLANLEADER))

        for vote in session.scalars(stmt):
            if vote.voted_for_id is not None:
                cl_votes[vote.voted_for_id] += 1

        stmt = (
            select(CouncilMember)
            .where(col(CouncilMember.election_name == election_name))
            .filter(col(CouncilMember.discord_manager_speech_id is not None))
        )

        for nominee in sorted(
            session.scalars(stmt),
            # Unsure how to None check in the lambda.
            key=lambda nominee: guild.get_member(nominee.member_id).display_name.lower(),  # pyright: ignore[reportOptionalMemberAccess]
        ):
            dm_votes[nominee.member_id] = 0

        stmt = (
            select(Vote)
            .where(col(Vote.election_name == election_name))
            .where(col(Vote.type == VoteType.DISCORDMANAGER))
        )

        for vote in session.scalars(stmt):
            if vote.voted_for_id is not None:
                dm_votes[vote.voted_for_id] += 1

    # build the votes string
    votes_str: str = "Clan Leader Vote\n===================="
    votes_str += user_vote_str(cl_votes)

    votes_str += "\n\nDiscord Manager Vote\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_points(users: Sequence[str], add_points: int, point_type: PointType, *, is_member: bool) -> None:
    """
    Adds 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 in events or for the clan
    points.

    Args:
        users (Sequence[int]): users that participated and need points.
        add_points (int): number of points to add
        point_type (PointType): which point type to add to
        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(Points).where(col(Points.user_id == user)).where(col(Points.quarter_name == quarter_name))

            points_obj = session.scalar(stmt)
            add_points = get_allowed_points(session, point_type, user, add_points)
            add_clan_points = add_points if point_type == PointType.CLAN else 0
            add_event_points = add_points if point_type == PointType.EVENT else 0
            if points_obj is None:
                new_points = Points(
                    quarter_name=quarter_name,
                    user_id=user,
                    discord_member=is_member,
                    event_points=add_event_points,
                    clan_points=add_clan_points,
                )
                session.add(new_points)
            else:
                points_obj.event_points += add_event_points
                points_obj.clan_points += add_clan_points

        session.commit()


def get_allowed_points(session: Session, point_type: PointType, 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.
        point_type (PointType): Which type of points to check the limit for.
        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}"
    points_limit = (
        settings.point_settings.events_weekly_limit
        if point_type == PointType.EVENT
        else settings.point_settings.clan_points_weekly_limit
    )
    stmt = select(PointsLimit).where(col(PointsLimit.user_id == user)).where(col(PointsLimit.week == this_week))

    # if under limit, can add up to however many points would make the current limit equal to the
    # allowed limit.
    limit_obj = session.scalar(stmt)
    if limit_obj is None:
        clan_points = points if point_type == PointType.CLAN else 0
        event_points = points if point_type == PointType.EVENT else 0
        new_limit = PointsLimit(user_id=user, week=this_week, event_points=event_points, clan_points=clan_points)
        add_points = points
        session.add(new_limit)
    else:
        curr_points = limit_obj.event_points if point_type == PointType.EVENT else limit_obj.clan_points
        add_points = points if curr_points + points <= points_limit else (max(0, points_limit - curr_points))
        if point_type == PointType.EVENT:
            limit_obj.event_points += add_points
        else:
            limit_obj.clan_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(
                pathlib.Path(dir_path).iterdir(),
                key=lambda file: int(file.name.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(col(Message.id)).where(col(Message.path == channel_name))
    channel_id = session.scalar(stmt)
    ret_channel = None
    if channel_id is None:
        # channel is not tracked. see if it exists
        channels = await guild.fetch_channels()
        for channel in channels:
            if channel.name == channel_name and isinstance(channel, 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-Message repo.",
            )

        # record channel information
        new_item = Message(path=channel_name, id=ret_channel.id)
        session.add(new_item)
        session.commit()

    else:
        try:
            ret_channel = await guild.fetch_channel(channel_id)
            if not isinstance(ret_channel, nextcord.TextChannel):
                dakiny_logger.warning(
                    "Channel %s is not a text channel. Does not work for clan messages.", ret_channel.name
                )
                # TODO(imcompiling) Figure out how to handle returning None here. Maybe delete channel and remake it?
                # https://gitea.cheezers.org/imcompiling/dakiny-bot/issues/102
                return None  # pyright: ignore[reportReturnType]
        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-Message repo.",
            )
            stmt = update(Message).where(col(Message.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(col(Message.id)).where(col(Message.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 = Message(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(content=message_content)
        except nextcord.NotFound:
            msg = await channel.send(message_content)
            stmt = update(Message).where(col(Message.path == message_name)).values(id=msg.id)
            session.execute(stmt)
            session.commit()


async def edit_clan_council_info(guild: nextcord.Guild) -> 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.
    """
    clan_info_channel = await guild.fetch_channel(settings.channel_ids.clan_info)

    if not isinstance(clan_info_channel, nextcord.TextChannel):
        return

    election_name = get_election_name()
    with Session(engine) as session:
        stmt = (
            select(col(CouncilMember.member_id))
            .where(col(CouncilMember.election_name == election_name))
            .where(col(CouncilMember.discord_manager is True))
        )
        discord_manager = session.scalar(stmt)
        stmt = (
            select(col(CouncilMember.member_id))
            .where(col(CouncilMember.election_name == election_name))
            .where(col(CouncilMember.clan_leader is True))
        )
        clan_leader = session.scalar(stmt)
        stmt = (
            select(col(CouncilMember.member_id))
            .where(col(CouncilMember.election_name == election_name))
            .where(col(CouncilMember.elected is True))
        )

        council_member_ids = sorted(
            session.scalars(stmt),
            # Unsure how to None check in the lambda.
            key=lambda user_id: guild.get_member(user_id).display_name.lower(),  # pyright: ignore[reportOptionalMemberAccess]
        )

        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)