Getch class (Written by Soheab_)

getch.py

from __future__ import annotations

from typing import TYPE_CHECKING, Any, Callable, TypeVar

import discord

if TYPE_CHECKING:
    from index import Akiko
    from Manager.database import AkikoRecordClass

# GuildChannel doesn't cover PrivateChannel and Thread
Channel = discord.abc.GuildChannel | discord.abc.PrivateChannel | discord.Thread

# All possible classes that got a getcher method and can be used in the from_cls method or returned by the any method
# except the db classes (AkikoRecordClass) for now.
PossibleGetchCls = (
    type[discord.User]
    | type[discord.Member]
    | type[discord.Guild]
    | type[discord.Role]
    | type[Channel]
    | type[discord.Message]
    | type[discord.Emoji]
    | type[discord.Invite]
)
# TypeVar for the from_cls method
# so the return type is the same as the cls given
ClsT = TypeVar("ClsT", bound=PossibleGetchCls)


class Getcher:
    """A class that contains methods to get objects from the bot cache or fetch them from the api.

    Parameters
    ----------
    bot: Akiko
        The bot instance. Used to get the cache and fetch objects from the api.

        This bot instance cannot be accessed publicly from the instance with a
        property or something like that because that would be weird since this
        class is supposed to be attached and accessed from the bot instance.

    Attributes
    ----------
    channel_cls: dict[object, Callable[..., Any]]
        A dict that contains all channel classes as keys and the channel method as value.
        This is used in the from_cls/any method to get the right method for the given class.
    cls_to_method: dict[object, Callable[..., Any]]
        A dict that contains all classes that got a getcher method as keys and the method as value.
        This is used in the from_cls/any method to get the right method for the given class.
    """

    def __init__(self, bot: Akiko) -> None:
        self.__bot: Akiko = bot

        self.channel_cls = {
            discord.abc.GuildChannel: self.channel,
            discord.abc.PrivateChannel: self.channel,
            discord.Thread: self.channel,
            discord.TextChannel: self.channel,
            discord.VoiceChannel: self.channel,
            discord.CategoryChannel: self.channel,
            discord.StageChannel: self.channel,
            discord.ForumChannel: self.channel,
        }
        self.cls_to_method = {
            discord.User: self.user,
            discord.Member: self.member,
            discord.Guild: self.guild,
            discord.Role: self.role,
            discord.Message: self.message,
            discord.Emoji: self.emoji,
            discord.Invite: self.invite,
        }

    async def __maybe_found(
        self,
        to_call: Callable[..., Any],
        *args: Any,
        ignore_exceptions: tuple[type[Exception]] | None = None,
        **kwargs: Any,
    ) -> Any:
        """A method that calls, awaits a function and returns None if the method raised any of the given exceptions.

        Parameters
        ----------
        to_call: Callable[..., Any]
            The method to call and await.
        *args: Any
            The args to pass to the method.
        ignore_exceptions: tuple[type[Exception]] | None
            The exceptions to ignore. :exc:`discord.NotFound` is always ignored.
        **kwargs: Any
            The kwargs to pass to the method.

        Returns
        -------
        Any | None
            The return value of the method or None if it raised one of the given exceptions.
        """
        try:
            return await to_call(*args, **kwargs)
        except (ignore_exceptions or ()) + (discord.NotFound,):
            return None

    async def __maybe_fetch_guild(
        self, guild: int | discord.abc.Snowflake | None, /, fetch: bool = True
    ) -> discord.Guild | None:
        """A method that returns a guild if it's already cached or fetches it from the api if it's not.
        This is a separate method because it's used in various methods.

        Parameters
        ----------
        guild: int | discord.abc.Snowflake | None
            The guild id or the guild instance or an object with a ``.id`` attribute or None.
        fetch: bool
            Whether to fetch the guild from the api if it's not cached. Defaults to True.

        Returns
        -------
        discord.Guild | None
            The guild if it's cached or fetched from the api or None if it's not cached and fetch is False.
        """
        if not guild:
            return None

        if isinstance(guild, discord.Guild):
            return guild

        guild_id = guild.id if not isinstance(guild, int) else guild

        cached = self.__bot.get_guild(int(guild_id))
        if cached:
            return cached
        if fetch:
            return await self.__maybe_found(self.__bot.fetch_guild, int(guild_id))

        return None

    async def from_cls(self, cls: ClsT, id: int | str, /, **kwargs: Any) -> ClsT | None:
        """A method that calls the right method for the given class and returns the result.

        Parameters
        ----------
        cls: ClsT
            The class to call the method with the id and kwargs on.
        id: int | str
            The id to call the method with.
        **kwargs: Any
            The kwargs to call the method with.

        Returns
        -------
        ClsT | None
            The result of the method or None if the class doesn't have a getcher method.
        """
        cls_to_method: dict[PossibleGetchCls, Callable[..., Any]] = self.cls_to_method | self.channel_cls

        if method := cls_to_method.get(cls):
            return await method(id, **kwargs)

        return None

    async def any(self, id: int) -> tuple[Any, PossibleGetchCls | None]:
        """A method that calls all getcher methods and returns the first result that isn't None.

        Parameters
        ----------
        id: int
            The id to call the methods with.

        Returns
        -------
        tuple[Any, PossibleGetchCls | None]
            A tuple containing the result of the method and the class that got the result from or
            ``(None, None)`` if all methods returned None.
        """
        cls_to_method: dict[PossibleGetchCls, Callable[..., Any]] = self.cls_to_method | self.channel_cls
        for kls in cls_to_method:
            if found := await self.from_cls(kls, id):
                return found, kls

        return None, None

    async def user(self, user_id: int | str, /) -> discord.User | None:
        """A method that returns a user if it's already cached or fetches it from the api if it's not.

        Parameters
        ----------
        user_id: int | str
            ID of the user to get.

        Returns
        -------
        discord.User | None
            The user or None if not found.
        """
        return self.__bot.get_user(int(user_id)) or await self.__maybe_found(
            self.__bot.fetch_user, int(user_id)
        )

    async def user_named(
        self, name: str, /, guild: int | discord.abc.Snowflake | None = None, fetch_guild: bool = False
    ) -> discord.User | discord.Member | None:
        """A method that gets a user by name. This cannot get users that don't share a guild with the bot.

        This searches the following (in order) and returns the first user it finds:

        - username
        - username#discriminator
        - global_name
        - nickname (if guild)

        It first searches through all cached users the bot can see (``bot.users``) and returns the first user it finds that matches.

        Else it uses ``guild.query_members`` on the guild given or all guilds if none is given or when it's not found in that one
        and returns the first member that discord returned.

        Parameters
        ----------
        name: str
            The name of the user to get.
        guild: int | discord.abc.Snowflake | None
            The guild id or the guild instance or an object with a ``.id`` attribute or None.
            This can be used to get the user from a specific guild or else it loops through all guilds
            and returns the first user it finds. This is slower than passing the guild directly for
            obvious reasons. Defaults to None.
        fetch_guild: bool
            Whether to fetch the guild from the api if it's not cached. Defaults to False.
            This fetches all guilds and loops through them to find the user.
            This is slower than passing the guild directly for obvious reasons.

        Returns
        -------
        discord.User | discord.Member | None
            The user or member or None if not found.
        """
        users = self.__bot.users

        username, _, discriminator = name.rpartition("#")

        # If # isn't found then "discriminator" actually has the username
        if not username:
            discriminator, username = username, discriminator

        if discriminator == "0" or (len(discriminator) == 4 and discriminator.isdigit()):
            pred = lambda u: u.name == username and u.discriminator == discriminator
        else:
            pred = lambda u: u.name == name or u.global_name == name

        found_user = discord.utils.find(pred, users)
        if found_user:
            return found_user

        async def query_guild_members(guild: discord.Guild) -> list[discord.Member]:
            """A function inside the method that actually queries the guild members."""
            return await guild.query_members(query=name, limit=5)

        guild = await self.__maybe_fetch_guild(guild, fetch_guild)
        if guild and (members := await query_guild_members(guild)):
            return members[0]

        for _guild in await self.guilds(force_fetch=fetch_guild):
            if _guild == guild:
                continue

            if members := await query_guild_members(_guild):
                return members[0]

        return None

    async def member(
        self,
        member_id: int | str,
        /,
        guild: int | discord.abc.Snowflake | None = None,
        *,
        fetch_guild: bool = True,
    ) -> discord.Member | None:
        """A method that returns a member if it's already cached or fetches it from the api if it's not.

        Parameters
        ----------
        member_id: int | str
            ID of the member to get.
        guild: int | discord.abc.Snowflake | None
            The guild id or the guild instance or an object with a ``.id`` attribute or None.
            This can be used to get the member from a specific guild or else it loops through all guilds
            and returns the first member it finds. This is slower than passing the guild directly for
            obvious reasons. Defaults to None.
        fetch_guild: bool
            Whether to fetch the guild from the api if it's not cached. Defaults to True.

        Returns
        -------
        discord.Member | None
            The member or None if not found.
        """

        async def get_member(guild: discord.Guild) -> discord.Member | None:
            """A function inside the method that actually get or fetches the member.
            Because it's called multiple times in the method.
            """
            return guild.get_member(int(member_id)) or await self.__maybe_found(
                guild.fetch_member, int(member_id)
            )

        guild = await self.__maybe_fetch_guild(guild, fetch_guild)
        if not guild:
            for _guild in await self.guilds(force_fetch=fetch_guild):
                if member := await get_member(_guild):
                    return member
        else:
            #  if not guild.chunked:
            #   await guild.chunk()
            return await get_member(guild)

        return None

    async def guilds(self, force_fetch: bool = False, limit: int | None = None) -> list[discord.Guild]:
        """A method that returns all guilds the bot is in.

        This method fetches if ``force_fetch`` is ``True``, doesn't check cache.

        Parameters
        ----------
        force_fetch: bool
            Whether to fetch the guilds from the api if they're not cached. Defaults to ``False``.
            Default to ``False`` because it's kinda slow to fetch.
        limit: int | None
            The limit to fetch. Defaults to ``None`` which means no limit.
            This is only used if ``force_fetch`` is ``True`` and is passed to the ``fetch_guilds`` method.

        Returns
        -------
        list[discord.Guild]
            A list of all guilds the bot is in.
        """
        cached = self.__bot.guilds
        if cached and not force_fetch:
            guilds = list(cached)
        else:
            guilds = [guild async for guild in self.__bot.fetch_guilds(limit=limit)]

        # await asyncio.gather(*(guild.chunk() for guild in guilds if not guild.chunked))
        return guilds

    async def guild(self, guild_id: int | str, /) -> discord.Guild | None:
        """A method that returns a guild if it's already cached or fetches it from the api if it's not.

        Parameters
        ----------
        guild_id: int | str
            ID of the guild to get.

        Returns
        -------
        discord.Guild | None
            The guild or None if not found.
        """
        guild = self.__bot.get_guild(int(guild_id)) or await self.__maybe_found(
            self.__bot.fetch_guild, int(guild_id)
        )
        # if guild and not guild.chunked:
        # await guild.chunk()

        return guild

    async def all_channels(
        self,
        force_fetch: bool = False,
    ) -> list[Channel]:
        """A method that returns all channels the bot can see.

        This method fetches if ``force_fetch`` is ``True``, doesn't check cache.

        Parameters
        ----------
        force_fetch: bool
            Whether to fetch the channels from the api if they're not cached. Defaults to ``False``.
            Default to ``False`` because it's kinda slow to fetch EVERY guild and channels in them.

        Returns
        -------
        list[Channel]
            A list of all channels the bot can see.
        """
        cached = self.__bot.get_all_channels()
        if cached and not force_fetch:
            return list(cached)

        guilds = await self.guilds(force_fetch)
        return [channel for guild in guilds for channel in await guild.fetch_channels()]

    async def role(
        self,
        role_id: int | str,
        /,
        guild: int | discord.abc.Snowflake | None = None,
        *,
        fetch_guild: bool = True,
        fetch_roles: bool = True,
    ) -> discord.Role | None:
        """A method that returns a role if it's already cached or fetches it from the api if it's not.

        Parameters
        ----------
        role_id: int | str
            ID of the role to get.
        guild: int | discord.abc.Snowflake | None
            The guild id or the guild instance or an object with a ``.id`` attribute or None.
            This can be used to get the role from a specific guild or else it loops through all guilds
            and returns the first role it finds. This is slower than passing the guild directly for
            obvious reasons. Defaults to None.
        fetch_guild: bool
            Whether to fetch the guild from the api if it's not cached. Defaults to True.
            This is only used if ``guild`` is None.
            This fetches all guilds and loops through them to find the role.
        fetch_roles: bool
            Whether to fetch the roles from the api if they're not cached. Defaults to True.
            You cannot fetch individual roles from the api so keep that in mind.

        Returns
        -------
        discord.Role | None
            The role or None if not found.
        """

        async def get_role(guild: discord.Guild) -> discord.Role | None:
            """A function inside the method that actually get or fetches the role.
            Because it's called multiple times in the method.
            """
            cached = guild.get_role(int(role_id))
            if cached:
                return cached

            if fetch_roles:
                return discord.utils.get(await guild.fetch_roles(), id=int(role_id))

            return None

        guild = await self.__maybe_fetch_guild(guild, fetch_guild)
        if not guild:
            for _guild in await self.guilds(force_fetch=fetch_guild):
                if role := await get_role(_guild):
                    return role
        else:
            return await get_role(guild)

        return None

    async def channel(
        self,
        channel_id: int | str,
        /,
        guild: int | discord.abc.Snowflake | None = None,
        *,
        fetch_guild: bool = True,
    ) -> Channel | None:
        """A method that returns a channel if it's already cached or fetches it from the api if it's not.

        Parameters
        ----------
        channel_id: int | str
            ID of the channel to get.
        guild: int | discord.abc.Snowflake | None
            The guild id or the guild instance or an object with a ``.id`` attribute or None.
            This can be used to get the channel from a specific guild or else it loops through all guilds
            and returns the first channel it finds. This is slower than passing the guild directly for
            obvious reasons. Defaults to None.
        fetch_guild: bool
            Whether to fetch the guild from the api if it's not cached. Defaults to True.
            This is only used if ``guild`` is None.
            This fetches all guilds and loops through them to find the channel.
        """
        bot_cached = self.__bot.get_channel(int(channel_id))
        if bot_cached:
            return bot_cached

        async def get_channel(guild: discord.Guild) -> Channel | None:
            """A function inside the method that actually get or fetches the channel.
            Because it's called multiple times in the method.
            """
            return guild.get_channel(int(channel_id)) or await self.__maybe_found(
                guild.fetch_channel, int(channel_id)
            )

        guild = await self.__maybe_fetch_guild(guild, fetch_guild)
        if not guild:
            for _guild in await self.guilds(force_fetch=fetch_guild):
                if channel := await get_channel(_guild):
                    return channel
        else:
            return await get_channel(guild)

        return None

    async def message(
        self,
        message_id: int | str,
        /,
        channel: int | discord.abc.Snowflake | None = None,
        *,
        fetch_channel_and_guild: bool = False,
    ) -> discord.Message | None:
        """A method that returns a message if it's already cached or fetches it from the api if it's not.

        This method ignores :exc:`discord.Forbidden` because the bot might not have access to the channel.
        And we don't want it to raise an error because of that. It just returns None. Keep that in mind.

        This first tries to get the message from cache. If it's not found in cache, it tries to get it from
        the given channel. If it's not found in the given channel, it loops through all channels.

        Parameters
        ----------
        message_id: int | str
            ID of the message to get.
        channel: int | discord.abc.Snowflake | None
            The channel id or the channel instance or an object with a ``.id`` attribute or None.
            This can be used to get the message from a specific channel or else it loops through all channels
            and returns the first message it finds. This is slower than passing the channel directly for
            obvious reasons. Defaults to None.

            I recommend passing the channel directly because it's much faster and less rate limited.
        fetch_channel_and_guild: bool
            Whether to fetch the channel and guild from the api if they're not cached. Defaults to False.
            This is only used if ``channel`` is None.
            This fetches all channels and loops through them to find the message.
            This is slower than passing the channel directly for obvious reasons.
            This is also slower than passing the channel directly because it fetches the guild too.
        """
        from_cache = discord.utils.get(self.__bot.cached_messages, id=int(message_id))
        if from_cache:
            return from_cache

        async def fetch_message(channel: discord.abc.Messageable) -> discord.Message | None:
            """ "A function inside the method that actually fetches the message.
            Because it's called multiple times in the method.

            This also ignores :exc:`discord.Forbidden` because the bot might not have access to the channel.
            And we don't want it to raise an error because of that.
            """
            return await self.__maybe_found(
                channel.fetch_message,
                int(message_id),
                ignore_exceptions=(discord.Forbidden,),
            )

        if channel and isinstance(channel, discord.abc.Messageable):
            in_given_channel = await fetch_message(channel)
            if in_given_channel:
                return in_given_channel

        for _channel in await self.all_channels(fetch_channel_and_guild):
            if isinstance(_channel, discord.abc.Messageable) and (message := await fetch_message(_channel)):
                return message

        return None

    async def emojis(self, force_fetch: bool = False) -> list[discord.Emoji]:
        """A method that returns all emojis the bot can see.

        This method fetches if ``force_fetch`` is ``True``, doesn't check cache.

        Parameters
        ----------
        force_fetch: bool
            Whether to fetch the emojis from the api if they're not cached. Defaults to ``False``.
            Default to ``False`` because it's kinda slow to fetch EVERY guild and emojis in them.

            Also fetches all guilds and loops through them to find the emojis.
        """
        cached = self.__bot.emojis
        if cached and not force_fetch:
            return list(cached)

        guilds = await self.guilds(force_fetch)
        return [emoji for guild in guilds for emoji in await guild.fetch_emojis()]

    async def emoji(
        self,
        emoji: int | str,
        /,
        by_name: bool = False,
        by_format: bool = False,
        by_id: bool = True,
        force_fetch: bool = True,
    ) -> discord.Emoji | None:
        """A method that returns an emoji if it's already cached or fetches it from the api if it's not.

        Hierachy of the parameters:
        1. by_name
        2. by_format
        3. by_id

        Parameters
        ----------
        emoji: int | str
            The emoji name, id or format (<:name:id>) to get.
        by_name: bool
            Whether to get the emoji by name. Defaults to False.
        by_format: bool
            Whether to get the emoji by format. Defaults to False.
        by_id: bool
            Whether to get the emoji by id. Defaults to True.

            This is ignored if ``by_name`` or ``by_format`` is True.
        force_fetch: bool
            Whether to fetch the emojis from the api if they're not cached. Defaults to ``True``.
            Default to ``True`` but it's kinda slow to fetch EVERY guild and emojis in them.

            Also fetches all guilds and loops through them to find the emojis.

        Returns
        -------
        discord.Emoji | None
            The emoji or None if not found.
        """
        emojis = await self.emojis(force_fetch)

        if by_name:
            return discord.utils.get(emojis, name=str(emoji))
        elif by_format:
            return discord.utils.find(lambda e: str(e) == str(emoji), emojis)
        elif by_id:
            return discord.utils.get(emojis, id=int(emoji))

        return None

    async def invite(
        self,
        invite_code: str,
    ) -> discord.Invite | None:
        """A method that fetches an invite from the api.
        Invites are not cached so this always fetches from the API.

        Parameters
        ----------
        invite_code: str
            The invite code to get.

        Returns
        -------
        discord.Invite | None
            The invite or None if not found.
        """
        return await self.__maybe_found(self.__bot.fetch_invite, invite_code)

    async def database_user(self, user_id: int | str, /) -> AkikoRecordClass | None:
        """A method that returns a user from the database if it exists.

        This is an experimental method and more will probably be added in the future.
        The database impl also has a getch method but it's not used here because
        experimental yes.

        Parameters
        ----------
        user_id: int | str
            The user id to get.

        Returns
        -------
        AkikoRecordClass | None
            The user or None if not found.
        """
        return await self.__bot.db.fetchrow("SELECT * FROM users WHERE userid = $1", int(user_id))