bot_utils.py

import discord
import random
from datetime import UTC, datetime
from discord.ext import commands
from enum import Enum
from operator import attrgetter
from src.bot.constants import messages, variables
from src.bot.tools import chat_formatting
from src.bot.tools.background_tasks import BackGroundTasks
from src.database.dal.bot.servers_dal import ServersDal


class Colors(Enum):
    black = discord.Color.default()
    teal = discord.Color.teal()
    dark_teal = discord.Color.dark_teal()
    green = discord.Color.green()
    dark_green = discord.Color.dark_green()
    blue = discord.Color.blue()
    dark_blue = discord.Color.dark_blue()
    purple = discord.Color.purple()
    dark_purple = discord.Color.dark_purple()
    magenta = discord.Color.magenta()
    dark_magenta = discord.Color.dark_magenta()
    gold = discord.Color.gold()
    dark_gold = discord.Color.dark_gold()
    orange = discord.Color.orange()
    dark_orange = discord.Color.dark_orange()
    red = discord.Color.red()
    dark_red = discord.Color.dark_red()
    lighter_grey = discord.Color.lighter_grey()
    dark_grey = discord.Color.dark_grey()
    light_grey = discord.Color.light_grey()
    darker_grey = discord.Color.darker_grey()
    blurple = discord.Color.blurple()
    greyple = discord.Color.greyple()


async def insert_server(bot: commands.Bot, server: discord.Guild) -> None:
    """Insert server information into database and initialize GW2 configs."""
    servers_dal = ServersDal(bot.db_session, bot.log)
    await servers_dal.insert_server(server.id, server.name)

    from src.gw2.tools import gw2_utils

    await gw2_utils.insert_gw2_server_configs(bot, server)


def init_background_tasks(bot: commands.Bot) -> None:
    """Initialize bot background tasks if configured."""
    bg_activity_timer = bot.settings["bot"]["BGActivityTimer"]
    if bg_activity_timer and bg_activity_timer > 0:
        bg_tasks = BackGroundTasks(bot)
        bot.loop.create_task(bg_tasks.change_presence_task(bg_activity_timer))


async def load_cogs(bot):
    bot.log.debug(messages.LOADING_EXTENSIONS)
    for ext in variables.ALL_COGS:
        cog_name = ext.replace("/", ".").replace(".py", "")
        try:
            await bot.load_extension(cog_name)
            bot.log.debug(f"{cog_name}")
        except Exception as e:
            bot.log.error(f"{messages.LOADING_EXTENSION_FAILED}: {cog_name}")
            bot.log.error(f"{e.__class__.__name__}: {e}\n")


async def invoke_subcommand(ctx, command_name: str):
    await ctx.message.channel.typing()
    if ctx.invoked_subcommand:
        return ctx.invoked_subcommand
    else:
        if ctx.command is not None:
            cmd = ctx.command
        else:
            cmd = ctx.bot.get_command(command_name)
        await send_help_msg(ctx, cmd)
        return None


def get_embed(ctx, description=None, color=None):
    if not color:
        color = ctx.bot.settings["bot"]["EmbedColor"]
    ebd = discord.Embed(color=color)
    if description:
        ebd.description = description
    return ebd


async def send_msg(ctx, description=None, dm=False, color=None):
    embed = get_embed(ctx, description, color)
    await send_embed(ctx, embed, dm)


async def send_warning_msg(ctx, description, dm=False):
    embed = get_embed(ctx, chat_formatting.warning(description), color=discord.Color.orange())
    await send_embed(ctx, embed, dm)


async def send_info_msg(ctx, description, dm=False):
    embed = get_embed(ctx, chat_formatting.info(description), discord.Color.blue())
    await send_embed(ctx, embed, dm)


async def send_error_msg(ctx, description, dm=False):
    embed = get_embed(ctx, chat_formatting.error(description), discord.Color.red())
    await send_embed(ctx, embed, dm)


async def send_help_msg(ctx, cmd):
    if ctx.bot.help_command.dm_help:
        await ctx.author.send(chat_formatting.box(cmd.help))
    else:
        await ctx.send(chat_formatting.box(cmd.help))


async def send_embed(ctx, embed, dm=False):
    try:
        if not embed.color:
            embed.color = ctx.bot.settings["bot"]["EmbedColor"]
        if not embed.author:
            embed.set_author(name=ctx.message.author.display_name, icon_url=ctx.message.author.display_avatar.url)

        if is_private_message(ctx):
            # Already in DM, just send the embed
            await ctx.author.send(embed=embed)
        elif dm:
            # Send to DM and notify in channel
            try:
                await ctx.author.send(embed=embed)
                notification_embed = discord.Embed(
                    description="📬 Response sent to your DM", color=discord.Color.green()
                )
                notification_embed.set_author(
                    name=ctx.author.display_name,
                    icon_url=ctx.author.avatar.url if ctx.author.avatar else ctx.author.default_avatar.url,
                )
                await ctx.send(embed=notification_embed)
            except discord.Forbidden, discord.HTTPException:
                # DM failed, fall back to sending in the channel
                await ctx.send(embed=embed)
        else:
            # Send to channel
            await ctx.send(embed=embed)
    except (discord.Forbidden, discord.HTTPException) as e:
        ctx.bot.log.error(f"Failed to send message: {e}")
        if dm or is_private_message(ctx):
            # Only show DM disabled message when we were actually trying to DM
            try:
                await ctx.send(
                    embed=discord.Embed(
                        description=chat_formatting.error(messages.DISABLED_DM),
                        color=discord.Color.red(),
                    )
                )
            except discord.Forbidden, discord.HTTPException:
                pass  # Can't send to channel either, nothing we can do
        else:
            # Channel send failed — notify the user with a simple embed
            try:
                await ctx.send(
                    embed=discord.Embed(
                        description=chat_formatting.error(messages.SEND_MESSAGE_FAILED),
                        color=discord.Color.red(),
                    )
                )
            except discord.Forbidden, discord.HTTPException:
                pass  # Can't send anything, nothing we can do
    except Exception as e:
        ctx.bot.log.error(f"Unexpected error sending message: {e}")


class EmbedPaginatorView(discord.ui.View):
    """Persistent pagination view for embed pages with Previous/Next buttons.

    Pages are stored in the database so pagination survives bot restarts.
    """

    def __init__(self, pages: list[discord.Embed] | None = None, author_id: int = 0):
        super().__init__(timeout=None)
        self.pages = pages or []
        self.current_page = 0
        self.author_id = author_id
        self.message: discord.Message | None = None
        self._update_buttons()

    def _update_buttons(self):
        self.previous_button.disabled = self.current_page == 0
        self.page_indicator.label = f"{self.current_page + 1}/{len(self.pages)}" if self.pages else "0/0"
        self.next_button.disabled = self.current_page >= len(self.pages) - 1

    @staticmethod
    def _embed_to_dict(embed: discord.Embed) -> dict:
        return embed.to_dict()

    @staticmethod
    def _dict_to_embed(data: dict) -> discord.Embed:
        return discord.Embed.from_dict(data)

    async def send_and_save(self, ctx) -> None:
        """Send the first page and save all pages to the database."""
        msg = await ctx.send(embed=self.pages[0], view=self)
        self.message = msg
        from src.database.dal.bot.embed_pages_dal import EmbedPagesDal

        dal = EmbedPagesDal(ctx.bot.db_session, ctx.bot.log)
        pages_data = [self._embed_to_dict(p) for p in self.pages]
        await dal.insert_embed_pages(msg.id, msg.channel.id, self.author_id, pages_data)

    async def _load_from_db(self, interaction: discord.Interaction) -> bool:
        """Load pages from database if not in memory. Returns True if loaded."""
        if self.pages:
            return True
        from src.database.dal.bot.embed_pages_dal import EmbedPagesDal

        bot = interaction.client
        dal = EmbedPagesDal(bot.db_session, bot.log)
        record = await dal.get_embed_pages(interaction.message.id)
        if not record:
            await interaction.response.send_message("This pagination has expired.", ephemeral=True)
            return False
        self.pages = [self._dict_to_embed(p) for p in record["pages"]]
        self.current_page = record["current_page"]
        self.author_id = record["author_id"]
        self._update_buttons()
        return True

    async def _save_current_page(self, interaction: discord.Interaction) -> None:
        """Update current page in database."""
        from src.database.dal.bot.embed_pages_dal import EmbedPagesDal

        bot = interaction.client
        dal = EmbedPagesDal(bot.db_session, bot.log)
        await dal.update_current_page(interaction.message.id, self.current_page)

    @discord.ui.button(label="\u25c0", style=discord.ButtonStyle.secondary, custom_id="paginator:prev")
    async def previous_button(self, interaction: discord.Interaction, button: discord.ui.Button):
        if not await self._load_from_db(interaction):
            return
        if interaction.user.id != self.author_id:
            return await interaction.response.send_message(
                "Only the command invoker can use these buttons.", ephemeral=True
            )
        self.current_page -= 1
        self._update_buttons()
        await interaction.response.edit_message(embed=self.pages[self.current_page], view=self)
        await self._save_current_page(interaction)

    @discord.ui.button(label="1/1", style=discord.ButtonStyle.secondary, disabled=True, custom_id="paginator:page")
    async def page_indicator(self, interaction: discord.Interaction, button: discord.ui.Button):
        await interaction.response.defer()

    @discord.ui.button(label="\u25b6", style=discord.ButtonStyle.secondary, custom_id="paginator:next")
    async def next_button(self, interaction: discord.Interaction, button: discord.ui.Button):
        if not await self._load_from_db(interaction):
            return
        if interaction.user.id != self.author_id:
            return await interaction.response.send_message(
                "Only the command invoker can use these buttons.", ephemeral=True
            )
        self.current_page += 1
        self._update_buttons()
        await interaction.response.edit_message(embed=self.pages[self.current_page], view=self)
        await self._save_current_page(interaction)


async def send_paginated_embed(ctx, embed: discord.Embed, max_fields: int = 25) -> None:
    """Send an embed with pagination if it exceeds max_fields.

    Splits the embed's fields across multiple pages with navigation buttons.
    Non-field properties (color, author, description, thumbnail) are preserved on each page.
    """
    if len(embed.fields) <= max_fields:
        await send_embed(ctx, embed)
        return

    color = embed.color or ctx.bot.settings["bot"]["EmbedColor"]
    total_fields = len(embed.fields)
    pages = []

    for i in range(0, total_fields, max_fields):
        page_embed = discord.Embed(color=color, description=embed.description)
        if embed.author:
            page_embed.set_author(name=embed.author.name, icon_url=embed.author.icon_url)
        if embed.thumbnail:
            page_embed.set_thumbnail(url=embed.thumbnail.url)

        for field in embed.fields[i : i + max_fields]:
            page_embed.add_field(name=field.name, value=field.value, inline=field.inline)

        page_number = (i // max_fields) + 1
        total_pages = (total_fields + max_fields - 1) // max_fields
        page_embed.set_footer(text=f"Page {page_number}/{total_pages}")
        pages.append(page_embed)

    if len(pages) == 1:
        await send_embed(ctx, pages[0])
        return

    view = EmbedPaginatorView(pages, ctx.message.author.id)
    await view.send_and_save(ctx)


async def delete_message(ctx, warning=False):
    if not is_private_message(ctx):
        color = None
        msg = messages.MESSAGE_REMOVED_FOR_PRIVACY

        try:
            await ctx.message.delete()
        except Exception as e:
            color = discord.Color.red()
            msg = messages.DELETE_MESSAGE_NO_PERMISSION
            ctx.bot.log.error(f"{e}: {msg}")
        finally:
            if warning:
                await send_msg(ctx, msg, False, color)


def is_member_admin(member: discord.Member | None) -> bool:
    """Check if a member has administrator permissions."""
    return member is not None and hasattr(member, "guild_permissions") and member.guild_permissions.administrator


def is_bot_owner(ctx: commands.Context, member: discord.Member) -> bool:
    """Check if a member is the bot owner."""
    return ctx.bot.owner_id == member.id


def is_server_owner(ctx: commands.Context, member: discord.Member) -> bool:
    """Check if a member is the server owner."""
    return member.id == ctx.guild.owner_id


def is_private_message(ctx: commands.Context) -> bool:
    """Check if the context is a private/DM message."""
    return isinstance(ctx.channel, discord.DMChannel)


def get_current_date_time():
    return datetime.now(UTC)


def get_current_date_time_str_long():
    return convert_datetime_to_str_long(get_current_date_time())


def convert_datetime_to_str_long(date: datetime):
    return date.strftime(variables.DATE_TIME_FORMATTER_STR)


def convert_datetime_to_str_short(date: datetime):
    return date.strftime(f"{variables.DATE_FORMATTER} {variables.TIME_FORMATTER}")


def convert_str_to_datetime_short(date_str: str) -> datetime:
    """Convert short format string to datetime."""
    return datetime.strptime(date_str, f"{variables.DATE_FORMATTER} {variables.TIME_FORMATTER}")


def get_object_member_by_str(ctx: commands.Context, member_str: str) -> discord.Member | None:
    """Find a guild member by name, display name, or nickname."""
    if is_private_message(ctx):
        return None

    for member in ctx.guild.members:
        if member_str in (member.name, member.display_name) or (
            member.nick is not None and member.nick.lower() == member_str.lower()
        ):
            return member

    return None


def get_user_by_id(bot: commands.Bot, user_id: int) -> discord.User | None:
    """Get a user by their ID."""
    return bot.get_user(int(user_id))


def get_member_by_id(guild: discord.Guild, member_id: int) -> discord.Member | None:
    """Get a guild member by their ID."""
    return guild.get_member(int(member_id))


async def send_msg_to_system_channel(log, server, embed, plain_msg=None):
    channel_to_send_msg = get_server_system_channel(server)
    if channel_to_send_msg:
        try:
            await channel_to_send_msg.send(embed=embed)
        except discord.HTTPException as e:
            log.error(e)
            if plain_msg:
                await channel_to_send_msg.send(plain_msg)


def get_server_system_channel(server: discord.Guild) -> discord.TextChannel | None:
    """Get the server's system channel or find the first readable text channel."""
    if server.system_channel:
        return server.system_channel

    # Find the first publicly readable text channel
    sorted_channels = sorted(server.text_channels, key=attrgetter("position"))

    for channel in sorted_channels:
        if not hasattr(channel, "overwrites"):
            continue

        for target, permissions in channel.overwrites.items():
            if hasattr(target, "name") and target.name == "@everyone" and permissions.read_messages in (True, None):
                return channel

    return None


def get_color_settings(color: str) -> discord.Color | None:
    """Get a Discord color from string name or generate random color."""
    color_lower = color.lower()

    if color_lower == "random":
        system_random = random.SystemRandom()
        hex_color = "".join(system_random.choice("0123456789ABCDEF") for _ in range(6))
        return discord.Color(int(hex_color, 16))

    for color_enum in Colors:
        if color_enum.name.lower() == color_lower:
            return color_enum.value

    return None


def get_bot_stats(bot: commands.Bot) -> dict[str, str | datetime]:
    """Get comprehensive bot statistics."""
    unique_users = sum(1 for user in bot.users if not user.bot)
    bot_users = sum(1 for user in bot.users if user.bot)

    text_channels = sum(
        1 for guild in bot.guilds for channel in guild.channels if isinstance(channel, discord.TextChannel)
    )

    voice_channels = sum(
        1 for guild in bot.guilds for channel in guild.channels if isinstance(channel, discord.VoiceChannel)
    )

    total_channels = text_channels + voice_channels

    return {
        "servers": f"{len(bot.guilds)} servers",
        "users": f"({unique_users} users)({bot_users} bots)[{len(bot.users)} total]",
        "channels": f"({text_channels} text)({voice_channels} voice)[{total_channels} total]",
        "start_time": getattr(bot, "start_time", None) or get_current_date_time(),
    }