bot.py

"""
The MIT License (MIT)

Copyright (c) 2015-2021 Rapptz
Copyright (c) 2021-present Pycord Development

Permission is hereby granted, free of charge, to any person obtaining a
copy of this software and associated documentation files (the "Software"),
to deal in the Software without restriction, including without limitation
the rights to use, copy, modify, merge, publish, distribute, sublicense,
and/or sell copies of the Software, and to permit persons to whom the
Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
DEALINGS IN THE SOFTWARE.
"""

from __future__ import annotations

import asyncio
import collections
import collections.abc
import copy
import inspect
import logging
import sys
import traceback
from abc import ABC, abstractmethod
from typing import (
    TYPE_CHECKING,
    Any,
    Callable,
    Coroutine,
    Generator,
    Literal,
    Mapping,
    TypeVar,
)

from .client import Client
from .cog import CogMixin
from .commands import (
    ApplicationCommand,
    ApplicationContext,
    AutocompleteContext,
    MessageCommand,
    SlashCommand,
    SlashCommandGroup,
    UserCommand,
    command,
)
from .enums import IntegrationType, InteractionContextType, InteractionType
from .errors import CheckFailure, DiscordException
from .interactions import Interaction
from .shard import AutoShardedClient
from .types import interactions
from .user import User
from .utils import MISSING, async_all, find, get

if TYPE_CHECKING:
    from .member import Member

CoroFunc = Callable[..., Coroutine[Any, Any, Any]]
CFT = TypeVar("CFT", bound=CoroFunc)

__all__ = (
    "ApplicationCommandMixin",
    "Bot",
    "AutoShardedBot",
)

_log = logging.getLogger(__name__)


class ApplicationCommandMixin(ABC):
    """A mixin that implements common functionality for classes that need
    application command compatibility.

    Attributes
    ----------
    application_commands: :class:`dict`
        A mapping of command id string to :class:`.ApplicationCommand` objects.
    pending_application_commands: :class:`list`
        A list of commands that have been added but not yet registered. This is read-only and is modified via other
        methods.
    """

    def __init__(self, *args, **kwargs) -> None:
        super().__init__(*args, **kwargs)
        self._pending_application_commands = []
        self._application_commands = {}

    @property
    def all_commands(self):
        return self._application_commands

    @property
    def pending_application_commands(self):
        return self._pending_application_commands

    @property
    def commands(self) -> list[ApplicationCommand | Any]:
        commands = self.application_commands
        if self._bot._supports_prefixed_commands and hasattr(
            self._bot, "prefixed_commands"
        ):
            commands += getattr(self._bot, "prefixed_commands")
        return commands

    @property
    def application_commands(self) -> list[ApplicationCommand]:
        return list(self._application_commands.values())

    def add_application_command(self, command: ApplicationCommand) -> None:
        """Adds an :class:`.ApplicationCommand` into the internal list of commands.

        This is usually not called, instead the :meth:`command` or
        other shortcut decorators are used instead.

        .. versionadded:: 2.0

        Parameters
        ----------
        command: :class:`.ApplicationCommand`
            The command to add.
        """
        if isinstance(command, SlashCommand) and command.is_subcommand:
            raise TypeError("The provided command is a sub-command of group")

        if self._bot.debug_guilds and command.guild_ids is None:
            command.guild_ids = self._bot.debug_guilds
        if self._bot.default_command_contexts and command.contexts is None:
            command.contexts = self._bot.default_command_contexts
        if (
            self._bot.default_command_integration_types
            and command.integration_types is None
        ):
            command.integration_types = self._bot.default_command_integration_types

        for cmd in self.pending_application_commands:
            if cmd == command:
                command.id = cmd.id
                self._application_commands[command.id] = command
                break
        self._pending_application_commands.append(command)

    def remove_application_command(
        self, command: ApplicationCommand
    ) -> ApplicationCommand | None:
        """Remove an :class:`.ApplicationCommand` from the internal list
        of commands.

        .. versionadded:: 2.0

        Parameters
        ----------
        command: :class:`.ApplicationCommand`
            The command to remove.

        Returns
        -------
        Optional[:class:`.ApplicationCommand`]
            The command that was removed. If the command has not been added,
            ``None`` is returned instead.
        """
        if command.id:
            self._application_commands.pop(command.id, None)

        if command in self._pending_application_commands:
            self._pending_application_commands.remove(command)
            return command

    @property
    def get_command(self):
        """Shortcut for :meth:`.get_application_command`.

        .. note::
            Overridden in :class:`ext.commands.Bot`.

        .. versionadded:: 2.0
        """
        # TODO: Do something like we did in self.commands for this
        return self.get_application_command

    def get_application_command(
        self,
        name: str,
        guild_ids: list[int] | None = None,
        type: type[ApplicationCommand] = ApplicationCommand,
    ) -> ApplicationCommand | None:
        """Get an :class:`.ApplicationCommand` from the internal list
        of commands.

        .. versionadded:: 2.0

        Parameters
        ----------
        name: :class:`str`
            The qualified name of the command to get.
        guild_ids: List[:class:`int`]
            The guild ids associated to the command to get.
        type: Type[:class:`.ApplicationCommand`]
            The type of the command to get. Defaults to :class:`.ApplicationCommand`.

        Returns
        -------
        Optional[:class:`.ApplicationCommand`]
            The command that was requested. If not found, returns ``None``.
        """
        commands = self._application_commands.values()
        for command in commands:
            if command.name == name and isinstance(command, type):
                if guild_ids is not None and command.guild_ids != guild_ids:
                    return
                return command
            elif (names := name.split())[0] == command.name and isinstance(
                command, SlashCommandGroup
            ):
                while len(names) > 1:
                    command = get(commands, name=names.pop(0))
                    if not isinstance(command, SlashCommandGroup) or (
                        guild_ids is not None and command.guild_ids != guild_ids
                    ):
                        return
                    commands = command.subcommands
                command = get(commands, name=names.pop())
                if not isinstance(command, type) or (
                    guild_ids is not None and command.guild_ids != guild_ids
                ):
                    return
                return command

    async def get_desynced_commands(
        self,
        guild_id: int | None = None,
        prefetched: list[interactions.ApplicationCommand] | None = None,
    ) -> list[dict[str, Any]]:
        """|coro|

        Gets the list of commands that are desynced from discord. If ``guild_id`` is specified, it will only return
        guild commands that are desynced from said guild, else it will return global commands.

        .. note::
            This function is meant to be used internally, and should only be used if you want to override the default
            command registration behavior.

        .. versionadded:: 2.0

        Parameters
        ----------
        guild_id: Optional[:class:`int`]
            The guild id to get the desynced commands for, else global commands if unspecified.
        prefetched: Optional[List[:class:`.ApplicationCommand`]]
            If you already fetched the commands, you can pass them here to be used. Not recommended for typical usage.

        Returns
        -------
        List[Dict[:class:`str`, Any]]
            A list of the desynced commands. Each will come with at least the ``cmd`` and ``action`` keys, which
            respectively contain the command and the action to perform. Other keys may also be present depending on
            the action, including ``id``.
        """

        # We can suggest the user to upsert, edit, delete, or bulk upsert the commands

        def _check_command(cmd: ApplicationCommand, match: Mapping[str, Any]) -> bool:
            if isinstance(cmd, SlashCommandGroup):
                if len(cmd.subcommands) != len(match.get("options", [])):
                    return True
                for i, subcommand in enumerate(cmd.subcommands):
                    match_ = next(
                        (
                            data
                            for data in match["options"]
                            if data["name"] == subcommand.name
                        ),
                        MISSING,
                    )
                    if match_ is not MISSING and _check_command(subcommand, match_):
                        return True
            else:
                as_dict = cmd.to_dict()
                to_check = {
                    "nsfw": None,
                    "default_member_permissions": None,
                    "name": None,
                    "description": None,
                    "name_localizations": None,
                    "description_localizations": None,
                    "options": [
                        "type",
                        "name",
                        "description",
                        "autocomplete",
                        "choices",
                        "name_localizations",
                        "description_localizations",
                    ],
                    "contexts": None,
                    "integration_types": None,
                }
                for check, value in to_check.items():
                    if type(to_check[check]) == list:
                        # We need to do some falsy conversion here
                        # The API considers False (autocomplete) and [] (choices) to be falsy values
                        falsy_vals = (False, [])
                        for opt in value:
                            cmd_vals = (
                                [val.get(opt, MISSING) for val in as_dict[check]]
                                if check in as_dict
                                else []
                            )
                            for i, val in enumerate(cmd_vals):
                                if val in falsy_vals:
                                    cmd_vals[i] = MISSING
                            if match.get(
                                check, MISSING
                            ) is not MISSING and cmd_vals != [
                                val.get(opt, MISSING) for val in match[check]
                            ]:
                                # We have a difference
                                return True
                    elif getattr(cmd, check, None) != match.get(check):
                        # We have a difference
                        if (
                            check == "default_permission"
                            and getattr(cmd, check) is True
                            and match.get(check) is None
                        ):
                            # This is a special case
                            # TODO: Remove for perms v2
                            continue
                        return True
            return False

        return_value = []
        cmds = self.pending_application_commands.copy()

        if guild_id is None:
            pending = [cmd for cmd in cmds if cmd.guild_ids is None]
        else:
            pending = [
                cmd
                for cmd in cmds
                if cmd.guild_ids is not None and guild_id in cmd.guild_ids
            ]

        registered_commands: list[interactions.ApplicationCommand] = []
        if prefetched is not None:
            registered_commands = prefetched
        elif self._bot.user:
            if guild_id is None:
                registered_commands = await self._bot.http.get_global_commands(
                    self._bot.user.id
                )
            else:
                registered_commands = await self._bot.http.get_guild_commands(
                    self._bot.user.id, guild_id
                )

        registered_commands_dict = {cmd["name"]: cmd for cmd in registered_commands}
        # First let's check if the commands we have locally are the same as the ones on discord
        for cmd in pending:
            match = registered_commands_dict.get(cmd.name)
            if match is None:
                # We don't have this command registered
                return_value.append({"command": cmd, "action": "upsert"})
            elif _check_command(cmd, match):
                return_value.append(
                    {
                        "command": cmd,
                        "action": "edit",
                        "id": int(registered_commands_dict[cmd.name]["id"]),
                    }
                )
            else:
                # We have this command registered but it's the same
                return_value.append(
                    {"command": cmd, "action": None, "id": int(match["id"])}
                )

        # Now let's see if there are any commands on discord that we need to delete
        for cmd, value_ in registered_commands_dict.items():
            match = get(pending, name=registered_commands_dict[cmd]["name"])
            if match is None:
                # We have this command registered but not in our list
                return_value.append(
                    {
                        "command": registered_commands_dict[cmd]["name"],
                        "id": int(value_["id"]),
                        "action": "delete",
                    }
                )

                continue

        return return_value

    async def register_command(
        self,
        command: ApplicationCommand,
        force: bool = True,
        guild_ids: list[int] | None = None,
    ) -> None:
        """|coro|

        Registers a command. If the command has ``guild_ids`` set, or if the ``guild_ids`` parameter is passed,
        the command will be registered as a guild command for those guilds.

        Parameters
        ----------
        command: :class:`~.ApplicationCommand`
            The command to register.
        force: :class:`bool`
            Whether to force the command to be registered. If this is set to False, the command will only be registered
            if it seems to already be registered and up to date with our internal cache. Defaults to True.
        guild_ids: :class:`list`
            A list of guild ids to register the command for. If this is not set, the command's
            :attr:`ApplicationCommand.guild_ids` attribute will be used.

        Returns
        -------
        :class:`~.ApplicationCommand`
            The command that was registered
        """
        # TODO: Write this
        raise NotImplementedError

    async def register_commands(
        self,
        commands: list[ApplicationCommand] | None = None,
        guild_id: int | None = None,
        method: Literal["individual", "bulk", "auto"] = "bulk",
        force: bool = False,
        delete_existing: bool = True,
    ) -> list[interactions.ApplicationCommand]:
        """|coro|

        Register a list of commands.

        .. versionadded:: 2.0

        Parameters
        ----------
        commands: Optional[List[:class:`~.ApplicationCommand`]]
            A list of commands to register. If this is not set (``None``), then all commands will be registered.
        guild_id: Optional[int]
            If this is set, the commands will be registered as a guild command for the respective guild. If it is not
            set, the commands will be registered according to their :attr:`ApplicationCommand.guild_ids` attribute.
        method: Literal['individual', 'bulk', 'auto']
            The method to use when registering the commands. If this is set to "individual", then each command will be
            registered individually. If this is set to "bulk", then all commands will be registered in bulk. If this is
            set to "auto", then the method will be determined automatically. Defaults to "bulk".
        force: :class:`bool`
            Registers the commands regardless of the state of the command on Discord. This uses one less API call, but
            can result in hitting rate limits more often. Defaults to False.
        delete_existing: :class:`bool`
            Whether to delete existing commands that are not in the list of commands to register. Defaults to True.
        """
        if commands is None:
            commands = self.pending_application_commands

        commands = [copy.copy(cmd) for cmd in commands]

        if guild_id is not None:
            for cmd in commands:
                to_rep_with = [guild_id]
                cmd.guild_ids = to_rep_with

        is_global = guild_id is None

        registered = []

        if is_global:
            pending = list(filter(lambda c: c.guild_ids is None, commands))
            registration_methods = {
                "bulk": self._bot.http.bulk_upsert_global_commands,
                "upsert": self._bot.http.upsert_global_command,
                "delete": self._bot.http.delete_global_command,
                "edit": self._bot.http.edit_global_command,
            }

            def _register(
                method: Literal["bulk", "upsert", "delete", "edit"], *args, **kwargs
            ):
                return registration_methods[method](
                    self._bot.user and self._bot.user.id, *args, **kwargs
                )

        else:
            pending = list(
                filter(
                    lambda c: c.guild_ids is not None and guild_id in c.guild_ids,
                    commands,
                )
            )
            registration_methods = {
                "bulk": self._bot.http.bulk_upsert_guild_commands,
                "upsert": self._bot.http.upsert_guild_command,
                "delete": self._bot.http.delete_guild_command,
                "edit": self._bot.http.edit_guild_command,
            }

            def _register(
                method: Literal["bulk", "upsert", "delete", "edit"], *args, **kwargs
            ):
                return registration_methods[method](
                    self._bot.user and self._bot.user.id, guild_id, *args, **kwargs
                )

        def register(
            method: Literal["bulk", "upsert", "delete", "edit"],
            *args,
            cmd_name: str = None,
            guild_id: int | None = None,
            **kwargs,
        ):
            if kwargs.pop("_log", True):
                if method == "bulk":
                    _log.debug(
                        f"Bulk updating commands {[c['name'] for c in args[0]]} for"
                        f" guild {guild_id}"
                    )
                elif method == "upsert":
                    _log.debug(f"Creating command {cmd_name} for guild {guild_id}")  # type: ignore
                elif method == "edit":
                    _log.debug(f"Editing command {cmd_name} for guild {guild_id}")  # type: ignore
                elif method == "delete":
                    _log.debug(f"Deleting command {cmd_name} for guild {guild_id}")  # type: ignore
            return _register(method, *args, **kwargs)

        pending_actions = []

        if not force:
            prefetched_commands: list[interactions.ApplicationCommand] = []
            if self._bot.user:
                if guild_id is None:
                    prefetched_commands = await self._bot.http.get_global_commands(
                        self._bot.user.id
                    )
                else:
                    prefetched_commands = await self._bot.http.get_guild_commands(
                        self._bot.user.id, guild_id
                    )
            desynced = await self.get_desynced_commands(
                guild_id=guild_id, prefetched=prefetched_commands
            )

            for cmd in desynced:
                if cmd["action"] == "delete":
                    pending_actions.append(
                        {
                            "action": "delete" if delete_existing else None,
                            "command": collections.namedtuple("Command", ["name"])(
                                name=cmd["command"]
                            ),
                            "id": cmd["id"],
                        }
                    )
                    continue
                # We can assume the command item is a command, since it's only a string if action is delete
                match = get(pending, name=cmd["command"].name, type=cmd["command"].type)
                if match is None:
                    continue
                if cmd["action"] == "edit":
                    pending_actions.append(
                        {
                            "action": "edit",
                            "command": match,
                            "id": cmd["id"],
                        }
                    )
                elif cmd["action"] == "upsert":
                    pending_actions.append(
                        {
                            "action": "upsert",
                            "command": match,
                        }
                    )
                elif cmd["action"] is None:
                    pending_actions.append(
                        {
                            "action": None,
                            "command": match,
                        }
                    )
                else:
                    raise ValueError(f"Unknown action: {cmd['action']}")
            filtered_no_action = list(
                filter(lambda c: c["action"] is not None, pending_actions)
            )
            filtered_deleted = list(
                filter(lambda a: a["action"] != "delete", pending_actions)
            )
            if method == "bulk" or (
                method == "auto" and len(filtered_deleted) == len(pending)
            ):
                # Either the method is bulk or all the commands need to be modified, so we can just do a bulk upsert
                data = [cmd["command"].to_dict() for cmd in filtered_deleted]
                # If there's nothing to update, don't bother
                if len(filtered_no_action) == 0:
                    _log.debug("Skipping bulk command update: Commands are up to date")
                    registered = prefetched_commands
                else:
                    _log.debug(
                        "Bulk updating commands %s for guild %s",
                        {c["command"].name: c["action"] for c in pending_actions},
                        guild_id,
                    )
                    registered = await register("bulk", data, _log=False)
            else:
                if not filtered_no_action:
                    registered = []
                for cmd in filtered_no_action:
                    if cmd["action"] == "delete":
                        await register(
                            "delete",
                            cmd["id"],
                            cmd_name=cmd["command"].name,
                            guild_id=guild_id,
                        )
                        continue
                    if cmd["action"] == "edit":
                        registered.append(
                            await register(
                                "edit",
                                cmd["id"],
                                cmd["command"].to_dict(),
                                cmd_name=cmd["command"].name,
                                guild_id=guild_id,
                            )
                        )
                    elif cmd["action"] == "upsert":
                        registered.append(
                            await register(
                                "upsert",
                                cmd["command"].to_dict(),
                                cmd_name=cmd["command"].name,
                                guild_id=guild_id,
                            )
                        )
                    else:
                        raise ValueError(f"Unknown action: {cmd['action']}")

            # TODO: Our lists dont work sometimes, see if that can be fixed so we can avoid this second API call
            if method != "bulk":
                if self._bot.user:
                    if guild_id is None:
                        registered = await self._bot.http.get_global_commands(
                            self._bot.user.id
                        )
                    else:
                        registered = await self._bot.http.get_guild_commands(
                            self._bot.user.id, guild_id
                        )
        else:
            data = [cmd.to_dict() for cmd in pending]
            registered = await register("bulk", data, guild_id=guild_id)

        for i in registered:
            cmd = get(
                self.pending_application_commands,
                name=i["name"],
                type=i.get("type"),
            )
            if not cmd:
                raise ValueError(
                    f"Registered command {i['name']}, type {i.get('type')} not found in"
                    " pending commands"
                )
            cmd.id = i["id"]
            self._application_commands[cmd.id] = cmd

        return registered

    async def sync_commands(
        self,
        commands: list[ApplicationCommand] | None = None,
        method: Literal["individual", "bulk", "auto"] = "bulk",
        force: bool = False,
        guild_ids: list[int] | None = None,
        register_guild_commands: bool = True,
        check_guilds: list[int] | None = [],
        delete_existing: bool = True,
    ) -> None:
        """|coro|

        Registers all commands that have been added through :meth:`.add_application_command`. This method cleans up all
        commands over the API and should sync them with the internal cache of commands. It attempts to register the
        commands in the most efficient way possible, unless ``force`` is set to ``True``, in which case it will always
        register all commands.

        By default, this coroutine is called inside the :func:`.on_connect` event. If you choose to override the
        :func:`.on_connect` event, then you should invoke this coroutine as well such as the following:

        .. code-block:: python

            @bot.event
            async def on_connect():
                if bot.auto_sync_commands:
                    await bot.sync_commands()
                print(f"{bot.user.name} connected.")

        .. note::
            If you remove all guild commands from a particular guild, the library may not be able to detect and update
            the commands accordingly, as it would have to individually check for each guild. To force the library to
            unregister a guild's commands, call this function with ``commands=[]`` and ``guild_ids=[guild_id]``.

        .. versionadded:: 2.0

        Parameters
        ----------
        commands: Optional[List[:class:`~.ApplicationCommand`]]
            A list of commands to register. If this is not set (None), then all commands will be registered.
        method: Literal['individual', 'bulk', 'auto']
            The method to use when registering the commands. If this is set to "individual", then each command will be
            registered individually. If this is set to "bulk", then all commands will be registered in bulk. If this is
            set to "auto", then the method will be determined automatically. Defaults to "bulk".
        force: :class:`bool`
            Registers the commands regardless of the state of the command on Discord. This uses one less API call, but
            can result in hitting rate limits more often. Defaults to False.
        guild_ids: Optional[List[:class:`int`]]
            A list of guild ids to register the commands for. If this is not set, the commands'
            :attr:`~.ApplicationCommand.guild_ids` attribute will be used.
        register_guild_commands: :class:`bool`
            Whether to register guild commands. Defaults to True.
        check_guilds: Optional[List[:class:`int`]]
            A list of guilds ids to check for commands to unregister, since the bot would otherwise have to check all
            guilds. Unlike ``guild_ids``, this does not alter the commands' :attr:`~.ApplicationCommand.guild_ids`
            attribute, instead it adds the guild ids to a list of guilds to sync commands for. If
            ``register_guild_commands`` is set to False, then this parameter is ignored.
        delete_existing: :class:`bool`
            Whether to delete existing commands that are not in the list of commands to register. Defaults to True.
        """

        check_guilds = list(set((check_guilds or []) + (self._bot.debug_guilds or [])))

        if commands is None:
            commands = self.pending_application_commands

        if guild_ids is not None:
            for cmd in commands:
                cmd.guild_ids = guild_ids

        global_commands = [cmd for cmd in commands if cmd.guild_ids is None]
        registered_commands = await self.register_commands(
            global_commands, method=method, force=force, delete_existing=delete_existing
        )

        registered_guild_commands: dict[int, list[interactions.ApplicationCommand]] = {}

        if register_guild_commands:
            cmd_guild_ids: list[int] = []
            for cmd in commands:
                if cmd.guild_ids is not None:
                    cmd_guild_ids.extend(cmd.guild_ids)
            if check_guilds is not None:
                cmd_guild_ids.extend(check_guilds)
            for guild_id in set(cmd_guild_ids):
                guild_commands = [
                    cmd
                    for cmd in commands
                    if cmd.guild_ids is not None and guild_id in cmd.guild_ids
                ]
                app_cmds = await self.register_commands(
                    guild_commands,
                    guild_id=guild_id,
                    method=method,
                    force=force,
                    delete_existing=delete_existing,
                )
                registered_guild_commands[guild_id] = app_cmds

        for i in registered_commands:
            cmd = get(
                self.pending_application_commands,
                name=i["name"],
                guild_ids=None,
                type=i.get("type"),
            )
            if cmd:
                cmd.id = i["id"]
                self._application_commands[cmd.id] = cmd

        if register_guild_commands and registered_guild_commands:
            for guild_id, guild_cmds in registered_guild_commands.items():
                for i in guild_cmds:
                    cmd = find(
                        lambda cmd: cmd.name == i["name"]
                        and cmd.type == i.get("type")
                        and cmd.guild_ids is not None
                        and (guild_id := i.get("guild_id"))
                        and guild_id in cmd.guild_ids,
                        self.pending_application_commands,
                    )
                    if not cmd:
                        # command has not been added yet
                        continue
                    cmd.id = i["id"]
                    self._application_commands[cmd.id] = cmd

    async def process_application_commands(
        self, interaction: Interaction, auto_sync: bool | None = None
    ) -> None:
        """|coro|

        This function processes the commands that have been registered
        to the bot and other groups. Without this coroutine, none of the
        commands will be triggered.

        By default, this coroutine is called inside the :func:`.on_interaction`
        event. If you choose to override the :func:`.on_interaction` event, then
        you should invoke this coroutine as well.

        This function finds a registered command matching the interaction id from
        application commands and invokes it. If no matching command was
        found, it replies to the interaction with a default message.

        .. versionadded:: 2.0

        Parameters
        ----------
        interaction: :class:`discord.Interaction`
            The interaction to process
        auto_sync: Optional[:class:`bool`]
            Whether to automatically sync and unregister the command if it is not found in the internal cache. This will
            invoke the :meth:`~.Bot.sync_commands` method on the context of the command, either globally or per-guild,
            based on the type of the command, respectively. Defaults to :attr:`.Bot.auto_sync_commands`.
        """
        if auto_sync is None:
            auto_sync = self._bot.auto_sync_commands
        # TODO: find out why the isinstance check below doesn't stop the type errors below
        if interaction.type not in (
            InteractionType.application_command,
            InteractionType.auto_complete,
        ):
            return

        command: ApplicationCommand | None = None
        try:
            if interaction.data:
                command = self._application_commands[interaction.data["id"]]  # type: ignore
        except KeyError:
            for cmd in self.application_commands + self.pending_application_commands:
                if interaction.data:
                    guild_id = interaction.data.get("guild_id")
                    if guild_id:
                        guild_id = int(guild_id)
                    if cmd.name == interaction.data["name"] and (  # type: ignore
                        guild_id == cmd.guild_ids
                        or (
                            isinstance(cmd.guild_ids, list)
                            and guild_id in cmd.guild_ids
                        )
                    ):
                        command = cmd
                        break
            else:
                if auto_sync and interaction.data:
                    guild_id = interaction.data.get("guild_id")
                    if guild_id is None:
                        await self.sync_commands()
                    else:
                        await self.sync_commands(check_guilds=[guild_id])
                return self._bot.dispatch("unknown_application_command", interaction)

        if interaction.type is InteractionType.auto_complete:
            return self._bot.dispatch(
                "application_command_auto_complete", interaction, command
            )

        ctx = await self.get_application_context(interaction)
        if command:
            ctx.command = command
        await self.invoke_application_command(ctx)

    async def on_application_command_auto_complete(
        self, interaction: Interaction, command: ApplicationCommand
    ) -> None:
        async def callback() -> None:
            ctx = await self.get_autocomplete_context(interaction)
            ctx.command = command
            return await command.invoke_autocomplete_callback(ctx)

        autocomplete_task = self._bot.loop.create_task(callback())
        try:
            await self._bot.wait_for(
                "application_command_auto_complete",
                check=lambda i, c: c == command,
                timeout=3,
            )
        except asyncio.TimeoutError:
            return
        else:
            if not autocomplete_task.done():
                autocomplete_task.cancel()

    def slash_command(self, **kwargs):
        """A shortcut decorator that invokes :func:`command` and adds it to
        the internal command list via :meth:`add_application_command`.
        This shortcut is made specifically for :class:`.SlashCommand`.

        .. versionadded:: 2.0

        Returns
        -------
        Callable[..., :class:`SlashCommand`]
            A decorator that converts the provided method into a :class:`.SlashCommand`, adds it to the bot,
            then returns it.
        """
        return self.application_command(cls=SlashCommand, **kwargs)

    def user_command(self, **kwargs):
        """A shortcut decorator that invokes :func:`command` and adds it to
        the internal command list via :meth:`add_application_command`.
        This shortcut is made specifically for :class:`.UserCommand`.

        .. versionadded:: 2.0

        Returns
        -------
        Callable[..., :class:`UserCommand`]
            A decorator that converts the provided method into a :class:`.UserCommand`, adds it to the bot,
            then returns it.
        """
        return self.application_command(cls=UserCommand, **kwargs)

    def message_command(self, **kwargs):
        """A shortcut decorator that invokes :func:`command` and adds it to
        the internal command list via :meth:`add_application_command`.
        This shortcut is made specifically for :class:`.MessageCommand`.

        .. versionadded:: 2.0

        Returns
        -------
        Callable[..., :class:`MessageCommand`]
            A decorator that converts the provided method into a :class:`.MessageCommand`, adds it to the bot,
            then returns it.
        """
        return self.application_command(cls=MessageCommand, **kwargs)

    def application_command(self, **kwargs):
        """A shortcut decorator that invokes :func:`command` and adds it to
        the internal command list via :meth:`~.Bot.add_application_command`.

        .. versionadded:: 2.0

        Returns
        -------
        Callable[..., :class:`ApplicationCommand`]
            A decorator that converts the provided method into an :class:`.ApplicationCommand`, adds it to the bot,
            then returns it.
        """

        def decorator(func) -> ApplicationCommand:
            result = command(**kwargs)(func)
            self.add_application_command(result)
            return result

        return decorator

    def command(self, **kwargs):
        """An alias for :meth:`application_command`.

        .. note::

            This decorator is overridden by :class:`discord.ext.commands.Bot`.

        .. versionadded:: 2.0

        Returns
        -------
        Callable[..., :class:`ApplicationCommand`]
            A decorator that converts the provided method into an :class:`.ApplicationCommand`, adds it to the bot,
            then returns it.
        """
        return self.application_command(**kwargs)

    def create_group(
        self,
        name: str,
        description: str | None = None,
        guild_ids: list[int] | None = None,
        **kwargs,
    ) -> SlashCommandGroup:
        """A shortcut method that creates a slash command group with no subcommands and adds it to the internal
        command list via :meth:`add_application_command`.

        .. versionadded:: 2.0

        Parameters
        ----------
        name: :class:`str`
            The name of the group to create.
        description: Optional[:class:`str`]
            The description of the group to create.
        guild_ids: Optional[List[:class:`int`]]
            A list of the IDs of each guild this group should be added to, making it a guild command.
            This will be a global command if ``None`` is passed.
        kwargs:
            Any additional keyword arguments to pass to :class:`.SlashCommandGroup`.

        Returns
        -------
        SlashCommandGroup
            The slash command group that was created.
        """
        description = description or "No description provided."
        group = SlashCommandGroup(name, description, guild_ids, **kwargs)
        self.add_application_command(group)
        return group

    def group(
        self,
        name: str | None = None,
        description: str | None = None,
        guild_ids: list[int] | None = None,
    ) -> Callable[[type[SlashCommandGroup]], SlashCommandGroup]:
        """A shortcut decorator that initializes the provided subclass of :class:`.SlashCommandGroup`
        and adds it to the internal command list via :meth:`add_application_command`.

        .. versionadded:: 2.0

        Parameters
        ----------
        name: Optional[:class:`str`]
            The name of the group to create. This will resolve to the name of the decorated class if ``None`` is passed.
        description: Optional[:class:`str`]
            The description of the group to create.
        guild_ids: Optional[List[:class:`int`]]
            A list of the IDs of each guild this group should be added to, making it a guild command.
            This will be a global command if ``None`` is passed.

        Returns
        -------
        Callable[[Type[SlashCommandGroup]], SlashCommandGroup]
            The slash command group that was created.
        """

        def inner(cls: type[SlashCommandGroup]) -> SlashCommandGroup:
            group = cls(
                name or cls.__name__,
                (
                    description or inspect.cleandoc(cls.__doc__).splitlines()[0]
                    if cls.__doc__ is not None
                    else "No description provided"
                ),
                guild_ids=guild_ids,
            )
            self.add_application_command(group)
            return group

        return inner

    slash_group = group

    def walk_application_commands(self) -> Generator[ApplicationCommand]:
        """An iterator that recursively walks through all application commands and subcommands.

        Yields
        ------
        :class:`.ApplicationCommand`
            An application command from the internal list of application commands.
        """
        for command in self.application_commands:
            if isinstance(command, SlashCommandGroup):
                yield from command.walk_commands()
            yield command

    async def get_application_context(
        self, interaction: Interaction, cls: Any = ApplicationContext
    ) -> ApplicationContext:
        r"""|coro|

        Returns the invocation context from the interaction.

        This is a more low-level counter-part for :meth:`.process_application_commands`
        to allow users more fine-grained control over the processing.

        Parameters
        -----------
        interaction: :class:`discord.Interaction`
            The interaction to get the invocation context from.
        cls
            The factory class that will be used to create the context.
            By default, this is :class:`.ApplicationContext`. Should a custom
            class be provided, it must be similar enough to
            :class:`.ApplicationContext`\'s interface.

        Returns
        --------
        :class:`.ApplicationContext`
            The invocation context. The type of this can change via the
            ``cls`` parameter.
        """
        return cls(self, interaction)

    async def get_autocomplete_context(
        self, interaction: Interaction, cls: Any = AutocompleteContext
    ) -> AutocompleteContext:
        r"""|coro|

        Returns the autocomplete context from the interaction.

        This is a more low-level counter-part for :meth:`.process_application_commands`
        to allow users more fine-grained control over the processing.

        Parameters
        -----------
        interaction: :class:`discord.Interaction`
            The interaction to get the invocation context from.
        cls
            The factory class that will be used to create the context.
            By default, this is :class:`.AutocompleteContext`. Should a custom
            class be provided, it must be similar enough to
            :class:`.AutocompleteContext`\'s interface.

        Returns
        --------
        :class:`.AutocompleteContext`
            The autocomplete context. The type of this can change via the
            ``cls`` parameter.
        """
        return cls(self, interaction)

    async def invoke_application_command(self, ctx: ApplicationContext) -> None:
        """|coro|

        Invokes the application command given under the invocation
        context and handles all the internal event dispatch mechanisms.

        Parameters
        ----------
        ctx: :class:`.ApplicationCommand`
            The invocation context to invoke.
        """
        self._bot.dispatch("application_command", ctx)
        try:
            if await self._bot.can_run(ctx, call_once=True):
                await ctx.command.invoke(ctx)
            else:
                raise CheckFailure("The global check once functions failed.")
        except DiscordException as exc:
            await ctx.command.dispatch_error(ctx, exc)
        else:
            self._bot.dispatch("application_command_completion", ctx)

    @property
    @abstractmethod
    def _bot(self) -> Bot | AutoShardedBot: ...


class BotBase(ApplicationCommandMixin, CogMixin, ABC):
    _supports_prefixed_commands = False

    def __init__(self, description=None, *args, **options):
        super().__init__(*args, **options)
        self.__cogs = {}  # TYPE: Dict[str, Cog]
        self.__extensions = {}  # TYPE: Dict[str, types.ModuleType]
        self._checks = []  # TYPE: List[Check]
        self._check_once = []
        self._before_invoke = None
        self._after_invoke = None
        self.description = inspect.cleandoc(description) if description else ""
        self.owner_id = options.get("owner_id")
        self.owner_ids = options.get("owner_ids", set())
        self.auto_sync_commands = options.get("auto_sync_commands", True)

        self.debug_guilds = options.pop("debug_guilds", None)
        self.default_command_contexts = options.pop(
            "default_command_contexts",
            {
                InteractionContextType.guild,
                InteractionContextType.bot_dm,
                InteractionContextType.private_channel,
            },
        )

        self.default_command_integration_types = options.pop(
            "default_command_integration_types",
            {
                IntegrationType.guild_install,
            },
        )

        if self.owner_id and self.owner_ids:
            raise TypeError("Both owner_id and owner_ids are set.")

        if self.owner_ids and not isinstance(
            self.owner_ids, collections.abc.Collection
        ):
            raise TypeError(
                f"owner_ids must be a collection not {self.owner_ids.__class__!r}"
            )
        if not isinstance(self.default_command_contexts, collections.abc.Collection):
            raise TypeError(
                f"default_command_contexts must be a collection not {self.default_command_contexts.__class__!r}"
            )
        if not isinstance(
            self.default_command_integration_types, collections.abc.Collection
        ):
            raise TypeError(
                f"default_command_integration_types must be a collection not {self.default_command_integration_types.__class__!r}"
            )
        self.default_command_contexts = set(self.default_command_contexts)
        self.default_command_integration_types = set(
            self.default_command_integration_types
        )

        self._checks = []
        self._check_once = []
        self._before_invoke = None
        self._after_invoke = None

    async def on_connect(self):
        if self.auto_sync_commands:
            await self.sync_commands()

    async def on_interaction(self, interaction):
        await self.process_application_commands(interaction)

    async def on_application_command_error(
        self, context: ApplicationContext, exception: DiscordException
    ) -> None:
        """|coro|

        The default command error handler provided by the bot.

        By default, this prints to :data:`sys.stderr` however it could be
        overridden to have a different implementation.

        This only fires if you do not specify any listeners for command error.
        """
        if self._event_handlers.get("on_application_command_error", None):
            return
        command = context.command
        if command and command.has_error_handler():
            return

        cog = context.cog
        if cog and cog.has_error_handler():
            return

        print(f"Ignoring exception in command {context.command}:", file=sys.stderr)
        traceback.print_exception(
            type(exception), exception, exception.__traceback__, file=sys.stderr
        )

    # global check registration
    # TODO: Remove these from commands.Bot

    def check(self, func):
        """A decorator that adds a global check to the bot. A global check is similar to a :func:`.check` that is
        applied on a per-command basis except it is run before any command checks have been verified and applies to
        every command the bot has.

        .. note::

           This function can either be a regular function or a coroutine. Similar to a command :func:`.check`, this
           takes a single parameter of type :class:`.Context` and can only raise exceptions inherited from
           :exc:`.ApplicationCommandError`.

        Example
        -------
        .. code-block:: python3

            @bot.check
            def check_commands(ctx):
                return ctx.command.qualified_name in allowed_commands
        """
        # T was used instead of Check to ensure the type matches on return
        self.add_check(func)  # type: ignore
        return func

    def add_check(self, func, *, call_once: bool = False) -> None:
        """Adds a global check to the bot. This is the non-decorator interface to :meth:`.check` and
        :meth:`.check_once`.

        Parameters
        ----------
        func
            The function that was used as a global check.
        call_once: :class:`bool`
            If the function should only be called once per :meth:`.Bot.invoke` call.
        """

        if call_once:
            self._check_once.append(func)
        else:
            self._checks.append(func)

    def remove_check(self, func, *, call_once: bool = False) -> None:
        """Removes a global check from the bot.
        This function is idempotent and will not raise an exception
        if the function is not in the global checks.

        Parameters
        ----------
        func
            The function to remove from the global checks.
        call_once: :class:`bool`
            If the function was added with ``call_once=True`` in
            the :meth:`.Bot.add_check` call or using :meth:`.check_once`.
        """
        checks = self._check_once if call_once else self._checks

        try:
            checks.remove(func)
        except ValueError:
            pass

    def check_once(self, func):
        """A decorator that adds a "call once" global check to the bot. Unlike regular global checks, this one is called
        only once per :meth:`.Bot.invoke` call. Regular global checks are called whenever a command is called or
        :meth:`.Command.can_run` is called. This type of check bypasses that and ensures that it's called only once,
        even inside the default help command.

        .. note::

           When using this function the :class:`.Context` sent to a group subcommand may only parse the parent command
           and not the subcommands due to it being invoked once per :meth:`.Bot.invoke` call.

        .. note::

           This function can either be a regular function or a coroutine. Similar to a command :func:`.check`,
           this takes a single parameter of type :class:`.Context` and can only raise exceptions inherited from
           :exc:`.ApplicationCommandError`.

        Example
        -------
        .. code-block:: python3

            @bot.check_once
            def whitelist(ctx):
                return ctx.message.author.id in my_whitelist
        """
        self.add_check(func, call_once=True)
        return func

    async def can_run(
        self, ctx: ApplicationContext, *, call_once: bool = False
    ) -> bool:
        data = self._check_once if call_once else self._checks

        if not data:
            return True

        # type-checker doesn't distinguish between functions and methods
        return await async_all(f(ctx) for f in data)  # type: ignore

    def before_invoke(self, coro):
        """A decorator that registers a coroutine as a pre-invoke hook.
        A pre-invoke hook is called directly before the command is
        called. This makes it a useful function to set up database
        connections or any type of set up required.
        This pre-invoke hook takes a sole parameter, a :class:`.Context`.

        .. note::

            The :meth:`~.Bot.before_invoke` and :meth:`~.Bot.after_invoke` hooks are
            only called if all checks and argument parsing procedures pass
            without error. If any check or argument parsing procedures fail
            then the hooks are not called.

        Parameters
        ----------
        coro: :ref:`coroutine <coroutine>`
            The coroutine to register as the pre-invoke hook.

        Raises
        ------
        TypeError
            The coroutine passed is not actually a coroutine.
        """
        if not asyncio.iscoroutinefunction(coro):
            raise TypeError("The pre-invoke hook must be a coroutine.")

        self._before_invoke = coro
        return coro

    def after_invoke(self, coro):
        r"""A decorator that registers a coroutine as a post-invoke hook.
        A post-invoke hook is called directly after the command is
        called. This makes it a useful function to clean-up database
        connections or any type of clean up required.
        This post-invoke hook takes a sole parameter, a :class:`.Context`.

        .. note::

            Similar to :meth:`~.Bot.before_invoke`\, this is not called unless
            checks and argument parsing procedures succeed. This hook is,
            however, **always** called regardless of the internal command
            callback raising an error (i.e. :exc:`.CommandInvokeError`\).
            This makes it ideal for clean-up scenarios.

        Parameters
        -----------
        coro: :ref:`coroutine <coroutine>`
            The coroutine to register as the post-invoke hook.

        Raises
        -------
        TypeError
            The coroutine passed is not actually a coroutine.

        """
        if not asyncio.iscoroutinefunction(coro):
            raise TypeError("The post-invoke hook must be a coroutine.")

        self._after_invoke = coro
        return coro

    async def is_owner(self, user: User | Member) -> bool:
        """|coro|

        Checks if a :class:`~discord.User` or :class:`~discord.Member` is the owner of
        this bot.

        If an :attr:`owner_id` is not set, it is fetched automatically
        through the use of :meth:`~.Bot.application_info`.

        .. versionchanged:: 1.3
            The function also checks if the application is team-owned if
            :attr:`owner_ids` is not set.

        Parameters
        ----------
        user: Union[:class:`.abc.User`, :class:`.member.Member`]
            The user to check for.

        Returns
        -------
        :class:`bool`
            Whether the user is the owner.
        """

        if self.owner_id:
            return user.id == self.owner_id
        elif self.owner_ids:
            return user.id in self.owner_ids
        else:
            app = await self.application_info()  # type: ignore
            if app.team:
                self.owner_ids = ids = {m.id for m in app.team.members}
                return user.id in ids
            else:
                self.owner_id = owner_id = app.owner.id
                return user.id == owner_id


class Bot(BotBase, Client):
    """Represents a discord bot.

    This class is a subclass of :class:`discord.Client` and as a result
    anything that you can do with a :class:`discord.Client` you can do with
    this bot.

    This class also subclasses ``ApplicationCommandMixin`` to provide the functionality
    to manage commands.

    .. versionadded:: 2.0

    Attributes
    ----------
    description: :class:`str`
        The content prefixed into the default help message.
    owner_id: Optional[:class:`int`]
        The user ID that owns the bot. If this is not set and is then queried via
        :meth:`.is_owner` then it is fetched automatically using
        :meth:`~.Bot.application_info`.
    owner_ids: Optional[Collection[:class:`int`]]
        The user IDs that owns the bot. This is similar to :attr:`owner_id`.
        If this is not set and the application is team based, then it is
        fetched automatically using :meth:`~.Bot.application_info`.
        For performance reasons it is recommended to use a :class:`set`
        for the collection. You cannot set both ``owner_id`` and ``owner_ids``.

        .. versionadded:: 1.3
    debug_guilds: Optional[List[:class:`int`]]
        Guild IDs of guilds to use for testing commands.
        The bot will not create any global commands if debug guild IDs are passed.

        .. versionadded:: 2.0
    auto_sync_commands: :class:`bool`
        Whether to automatically sync slash commands. This will call :meth:`~.Bot.sync_commands` in :func:`discord.on_connect`, and in
        :attr:`.process_application_commands` if the command is not found. Defaults to ``True``.

        .. versionadded:: 2.0
    default_command_contexts: Collection[:class:`InteractionContextType`]
        The default context types that the bot will use for commands.
        Defaults to a set containing :attr:`InteractionContextType.guild`, :attr:`InteractionContextType.bot_dm`, and
        :attr:`InteractionContextType.private_channel`.

        .. versionadded:: 2.6
    default_command_integration_types: Collection[:class:`IntegrationType`]]
        The default integration types that the bot will use for commands.
        Defaults to a set containing :attr:`IntegrationType.guild_install`.

        .. versionadded:: 2.6
    """

    @property
    def _bot(self) -> Bot:
        return self


class AutoShardedBot(BotBase, AutoShardedClient):
    """This is similar to :class:`.Bot` except that it is inherited from
    :class:`discord.AutoShardedClient` instead.

    .. versionadded:: 2.0
    """

    @property
    def _bot(self) -> AutoShardedBot:
        return self