SeaCogs/aurora/utilities/utils.py

# pylint: disable=cyclic-import
from datetime import datetime, timedelta
from typing import Optional, Union

from dateutil.relativedelta import relativedelta as rd
from discord import File, Guild, Interaction, Member, SelectOption, TextChannel, User
from discord.errors import Forbidden
from redbot.core import commands, data_manager
from redbot.core.utils.chat_formatting import error

from ..utilities.config import config


def check_permissions(
    user: User,
    permissions: list,
    ctx: Union[commands.Context, Interaction] | None = None,
    guild: Guild | None = None,
) -> Union[bool, str]:
    """Checks if a user has a specific permission (or a list of permissions) in a channel."""
    if ctx:
        member = ctx.guild.get_member(user.id)
        resolved_permissions = ctx.channel.permissions_for(member)

    elif guild:
        member = guild.get_member(user.id)
        resolved_permissions = member.guild_permissions

    else:
        raise (KeyError)

    for permission in permissions:
        if (
            not getattr(resolved_permissions, permission, False)
            and resolved_permissions.administrator is not True
        ):
            return permission

    return False


async def check_moddable(
    target: Union[User, Member, TextChannel], interaction: Interaction, permissions: list
) -> bool:
    """Checks if a moderator can moderate a target."""
    is_channel = isinstance(target, TextChannel)
    if check_permissions(interaction.client.user, permissions, guild=interaction.guild):
        await interaction.response.send_message(
            error(
                f"I do not have the `{permissions}` permission, required for this action."
            ),
            ephemeral=True,
        )
        return False

    if await config.guild(interaction.guild).use_discord_permissions() is True:
        if check_permissions(interaction.user, permissions, guild=interaction.guild):
            await interaction.response.send_message(
                error(
                    f"You do not have the `{permissions}` permission, required for this action."
                ),
                ephemeral=True,
            )
            return False

    if interaction.user.id == target.id:
        await interaction.response.send_message(
            content="You cannot moderate yourself!", ephemeral=True
        )
        return False

    if not is_channel and target.bot:
        await interaction.response.send_message(
            content="You cannot moderate bots!", ephemeral=True
        )
        return False

    if isinstance(target, Member):
        if interaction.user.top_role <= target.top_role and await config.guild(interaction.guild).respect_hierarchy() is True:
            await interaction.response.send_message(
                content=error(
                    "You cannot moderate members with a higher role than you!"
                ),
                ephemeral=True,
            )
            return False

        if (
            interaction.guild.get_member(interaction.client.user.id).top_role
            <= target.top_role
        ):
            await interaction.response.send_message(
                content=error(
                    "You cannot moderate members with a role higher than the bot!"
                ),
                ephemeral=True,
            )
            return False

        immune_roles = await config.guild(target.guild).immune_roles()

        for role in target.roles:
            if role.id in immune_roles:
                await interaction.response.send_message(
                    content=error("You cannot moderate members with an immune role!"),
                    ephemeral=True,
                )
                return False

    return True


def get_next_case_number(guild_id: str, cursor=None) -> int:
    """This function returns the next case number from the MySQL table for a specific guild."""
    from .database import connect

    if not cursor:
        database = connect()
        cursor = database.cursor()
    cursor.execute(
        f"SELECT moderation_id FROM `moderation_{guild_id}` ORDER BY moderation_id DESC LIMIT 1"
    )
    result = cursor.fetchone()
    return (result[0] + 1) if result else 1


async def log(interaction: Interaction, moderation_id: int, resolved: bool = False) -> None:
    """This function sends a message to the guild's configured logging channel when an infraction takes place."""
    from ..models.moderation import Moderation
    from .factory import log_factory

    logging_channel_id = await config.guild(interaction.guild).log_channel()
    if logging_channel_id != " ":
        logging_channel = interaction.guild.get_channel(logging_channel_id)

        try:
            moderation = Moderation.from_sql(interaction.client, moderation_id, interaction.guild_id)
            embed = await log_factory(
                interaction=interaction, moderation=moderation, resolved=resolved
            )
            try:
                await logging_channel.send(embed=embed)
            except Forbidden:
                return
        except ValueError:
            return


async def send_evidenceformat(interaction: Interaction, moderation_id: int) -> None:
    """This function sends an ephemeral message to the moderator who took the moderation action, with a pre-made codeblock for use in the mod-evidence channel."""
    from ..models.moderation import Moderation
    from .factory import evidenceformat_factory

    send_evidence_bool = (
        await config.user(interaction.user).auto_evidenceformat()
        or await config.guild(interaction.guild).auto_evidenceformat()
        or False
    )
    if send_evidence_bool is False:
        return

    moderation = Moderation.from_sql(interaction.client, moderation_id, interaction.guild.id)
    content = await evidenceformat_factory(moderation=moderation)
    await interaction.followup.send(content=content, ephemeral=True)


def get_bool_emoji(value: Optional[bool]) -> str:
    """Returns a unicode emoji based on a boolean value."""
    if value is True:
        return "\N{WHITE HEAVY CHECK MARK}"
    if value is False:
        return "\N{NO ENTRY SIGN}"
    return "\N{BLACK QUESTION MARK ORNAMENT}\N{VARIATION SELECTOR-16}"


def get_pagesize_str(value: Union[int, None]) -> str:
    """Returns a string based on a pagesize value."""
    if value is None:
        return "\N{BLACK QUESTION MARK ORNAMENT}\N{VARIATION SELECTOR-16}"
    return str(value) + " cases per page"


def create_pagesize_options() -> list[SelectOption]:
    """Returns a list of SelectOptions for pagesize configuration."""
    options = []
    options.append(
        SelectOption(
            label="Default",
            value="default",
            description="Reset the pagesize to the default value.",
        )
    )
    for i in range(1, 21):
        options.append(
            SelectOption(
                label=str(i),
                value=str(i),
                description=f"Set the pagesize to {i}.",
            )
        )
    return options

def timedelta_from_relativedelta(relativedelta: rd) -> timedelta:
    """Converts a relativedelta object to a timedelta object."""
    now = datetime.now()
    then = now - relativedelta
    return now - then

def get_footer_image(coginstance: commands.Cog) -> File:
    """Returns the footer image for the embeds."""
    image_path = data_manager.bundled_data_path(coginstance) / "arrow.png"
    return File(image_path, filename="arrow.png", description="arrow")