Error handling

dynamo/bot.py

@@ -17,6 +17,7 @@
 log = logging.getLogger(__name__)
 
 initial_extensions = (
+    get_cog("errors"),
     get_cog("help"),
     get_cog("events"),
     get_cog("general"),

dynamo/extensions/cogs/dev.py

@@ -1,30 +1,28 @@
 import importlib
-import logging
 import sys
 from pathlib import Path
 
 import discord
 from discord.ext import commands
 
 from dynamo.bot import Dynamo
+from dynamo.utils.base_cog import DynamoCog
 from dynamo.utils.cache import cached_functions
 from dynamo.utils.context import Context, Status
 from dynamo.utils.converter import GuildConverter
 from dynamo.utils.helper import ROOT, get_cog
 
-log = logging.getLogger(__name__)
-
 # Don't unload these
 BLACKLIST_UTILS: set[str] = {
     "dynamo.utils.cache",
 }
 
 
-class Dev(commands.GroupCog, group_name="dev"):
+class Dev(DynamoCog):
     """Dev-only commands"""
 
     def __init__(self, bot: Dynamo) -> None:
-        self.bot: Dynamo = bot
+        super().__init__(bot)
 
     async def cog_check(self, ctx: commands.Context) -> bool:  # type: ignore[override]
         return await self.bot.is_owner(ctx.author)
@@ -88,7 +86,7 @@ async def load(self, ctx: commands.Context, *, module: str) -> None:
         try:
             await self.bot.load_extension(m := get_cog(module))
         except commands.ExtensionError:
-            log.exception("Failed to load %s", m)
+            self.log.exception("Failed to load %s", m)
         else:
             await ctx.send(Status.OK)
 
@@ -104,7 +102,7 @@ async def unload(self, ctx: commands.Context, *, module: str) -> None:
         try:
             await self.bot.unload_extension(m := get_cog(module))
         except commands.ExtensionError:
-            log.exception("Failed to unload %s", m)
+            self.log.exception("Failed to unload %s", m)
         else:
             await ctx.send(Status.OK)
 
@@ -120,7 +118,7 @@ async def _reload(self, ctx: commands.Context, *, module: str) -> None:
         try:
             await self.bot.reload_extension(m := get_cog(module))
         except commands.ExtensionError:
-            log.exception("Failed to reload %s", m)
+            self.log.exception("Failed to reload %s", m)
         else:
             await ctx.send(Status.OK)
 
@@ -146,22 +144,22 @@ async def _reload_all(self, ctx: Context) -> None:
             try:
                 importlib.reload(sys.modules[module])
             except (KeyError, ModuleNotFoundError):
-                log.exception("Failed to reload %s", module)
-        log.debug("Reloaded %d/%d utilities", len(utils_modules), len(all_utils))
+                self.log.exception("Failed to reload %s", module)
+        self.log.debug("Reloaded %d/%d utilities", len(utils_modules), len(all_utils))
 
         extensions = set(self.bot.extensions)
         statuses: set[tuple[Status, str]] = set()
         for ext in extensions:
             try:
                 await self.reload_or_load_extension(ext)
             except commands.ExtensionError:
-                log.exception("Failed to reload extension %s", ext)
+                self.log.exception("Failed to reload extension %s", ext)
                 statuses.add((Status.FAILURE, ext))
             else:
                 statuses.add((Status.SUCCESS, ext))
 
         success_count = sum(1 for status, _ in statuses if status == Status.SUCCESS)
-        log.debug("Reloaded %d/%d extensions", success_count, len(extensions))
+        self.log.debug("Reloaded %d/%d extensions", success_count, len(extensions))
         await ctx.send("\n".join(f"{status} `{ext}`" for status, ext in statuses))
 
     @commands.hybrid_group(name="cache", aliases=("c",))
@@ -178,3 +176,7 @@ async def shutdown(self, ctx: commands.Context) -> None:
 
 async def setup(bot: Dynamo) -> None:
     await bot.add_cog(Dev(bot))
+
+
+async def teardown(bot: Dynamo) -> None:
+    await bot.remove_cog(Dev.__name__)

dynamo/extensions/cogs/errors.py

@@ -0,0 +1,156 @@
+from typing import Any, Callable, Coroutine, Mapping
+
+import discord
+from discord import Interaction, app_commands
+from discord.ext import commands
+
+from dynamo.bot import Dynamo
+from dynamo.utils.base_cog import DynamoCog
+
+
+class Errors(DynamoCog):
+    """Handles errors for the bot."""
+
+    command_error_messages: Mapping[type[commands.CommandError], str] = {
+        commands.CommandNotFound: "Command not found: `{}`.",
+        commands.MissingRequiredArgument: "Missing required argument: `{}`.",
+        commands.BadArgument: "Bad argument.",
+        commands.CommandOnCooldown: "You are on cooldown. Try again in `{:.2f}` seconds.",
+        commands.TooManyArguments: "Too many arguments.",
+        commands.MissingPermissions: "You are not allowed to use this command.",
+        commands.BotMissingPermissions: "I am not allowed to use this command.",
+        commands.NoPrivateMessage: "This command can only be used in a server.",
+        commands.NotOwner: "You are not the owner of this bot.",
+        commands.DisabledCommand: "This command is disabled.",
+        commands.CheckFailure: "You do not have permission to use this command.",
+    }
+
+    app_command_error_messages: Mapping[type[app_commands.AppCommandError], str] = {
+        app_commands.CommandNotFound: "Command not found: `{}`.",
+        app_commands.CommandOnCooldown: "You are on cooldown. Try again in `{:.2f}` seconds.",
+        app_commands.MissingPermissions: "You are not allowed to use this command.",
+        app_commands.BotMissingPermissions: "I am not allowed to use this command.",
+        app_commands.NoPrivateMessage: "This command can only be used in a server.",
+        app_commands.CheckFailure: "You do not have permission to use this command.",
+    }
+
+    def __init__(self, bot: Dynamo) -> None:
+        super().__init__(bot)
+
+        self.old_tree_error: Callable[
+            [Interaction[Dynamo], app_commands.AppCommandError], Coroutine[Any, Any, None]
+        ] = self.bot.tree.on_error
+        self.bot.tree.on_error = self.on_app_command_error
+
+    async def cog_unload(self) -> None:
+        """Restores the old tree error handler on unload."""
+        self.bot.tree.on_error = self.old_tree_error
+        await super().cog_unload()
+
+    def get_command_error_message(self, error: commands.CommandError) -> str:
+        """Get the error message for the given error.
+
+        Parameters
+        ----------
+        error : commands.CommandError
+            The error.
+
+        Returns
+        -------
+        str
+            The error message.
+        """
+        return self.command_error_messages.get(type(error), "An unknown error occurred.")
+
+    def get_app_command_error_message(self, error: app_commands.AppCommandError) -> str:
+        """
+        Get the error message for the given error.
+
+        Parameters
+        ----------
+        error : app_commands.AppCommandError
+            The error.
+
+        Returns
+        -------
+        str
+            The error message.
+        """
+        return self.app_command_error_messages.get(type(error), "An unknown error occurred.")
+
+    @commands.Cog.listener()
+    async def on_command_error(self, ctx: commands.Context, error: commands.CommandError) -> None:
+        """
+        Event that triggers when a command fails.
+
+        Parameters
+        ----------
+        ctx : commands.Context
+            The context.
+        error : commands.CommandError
+            The error.
+        """
+        self.log.error("%s called by %s raised an exception: %s. (%s)", ctx.command, ctx.author, error, ctx.message)
+
+        error_message = self.get_command_error_message(error)
+
+        if isinstance(error, commands.CommandNotFound):
+            error_message = error_message.format(ctx.invoked_with)
+
+        elif isinstance(error, commands.MissingRequiredArgument):
+            error_message = error_message.format(error.param.name)
+
+        elif isinstance(error, commands.CommandOnCooldown):
+            error_message = error_message.format(error.retry_after)
+
+        await ctx.reply(error_message)
+
+        if await self.bot.is_owner(ctx.author):
+            command_name = ctx.command.name if ctx.command else "Unknown Command"
+            await ctx.author.send(f"An error occurred while running the command `{command_name}`: {error}")
+
+    @commands.Cog.listener()
+    async def on_app_command_error(self, interaction: Interaction, error: app_commands.AppCommandError) -> None:
+        """
+        Event that triggers when a command fails.
+
+        Parameters
+        ----------
+        interaction : Interaction
+            The interaction object.
+        error : app_commands.AppCommandError
+            The exception.
+        """
+        if interaction.command is None:
+            self.log.error("Command not found: %s.", interaction.data)
+            command_name = interaction.data.get("name", "")
+            await interaction.response.send_message(f"Command not found: `{command_name}`.", ephemeral=True)
+            return
+
+        self.log.error("%s called by %s raised an exception: %s.", interaction.command.name, interaction.user, error)
+
+        error_message = self.get_app_command_error_message(error)
+
+        if isinstance(error, app_commands.CommandNotFound):
+            error_message = error_message.format(interaction.command.name)
+
+        elif isinstance(error, app_commands.CommandOnCooldown):
+            error_message = error_message.format(error.retry_after)
+
+        try:
+            await interaction.response.send_message(error_message, ephemeral=True)
+        except (discord.HTTPException, discord.InteractionResponded, TypeError, ValueError):
+            await interaction.followup.send(error_message, ephemeral=True)
+
+        if await self.bot.is_owner(interaction.user):
+            await interaction.user.send(
+                f"An error occurred while running the command `{interaction.command.name}`: {error}"
+            )
+
+
+async def setup(bot: Dynamo) -> None:
+    await bot.add_cog(Errors(bot))
+
+
+async def teardown(bot: Dynamo) -> None:
+    await bot.remove_cog(Errors.__name__)

dynamo/extensions/cogs/events.py

@@ -5,6 +5,7 @@
 from discord.ext import commands
 
 from dynamo.bot import Dynamo
+from dynamo.utils.base_cog import DynamoCog
 from dynamo.utils.cache import async_lru_cache
 
 log = logging.getLogger(__name__)
@@ -47,11 +48,11 @@ async def get_interested(event: discord.ScheduledEvent) -> str:
     return f"`[{event.name}]({event.url}) {' '.join(f'<@{u.id}>' for u in users) or "No users found"}`"
 
 
-class Events(commands.Cog, name="Events"):
+class Events(DynamoCog):
     """Scheduled event related commands"""
 
     def __init__(self, bot: Dynamo) -> None:
-        self.bot: Dynamo = bot
+        super().__init__(bot)
 
     def cog_check(self, ctx: commands.Context) -> bool:
         return ctx.guild is not None
@@ -84,3 +85,7 @@ async def event(self, ctx: commands.Context, event: int | None) -> None:
 
 async def setup(bot: Dynamo) -> None:
     await bot.add_cog(Events(bot))
+
+
+async def teardown(bot: Dynamo) -> None:
+    await bot.remove_cog(Events.__name__)

dynamo/extensions/cogs/general.py

@@ -7,6 +7,7 @@
 from discord.ext import commands
 
 from dynamo.bot import Dynamo
+from dynamo.utils.base_cog import DynamoCog
 from dynamo.utils.context import Context
 from dynamo.utils.converter import MemberTransformer
 from dynamo.utils.format import human_join
@@ -31,11 +32,11 @@ def embed_from_user(user: discord.Member | discord.User | discord.ClientUser) ->
     return e
 
 
-class General(commands.GroupCog, group_name="general"):
+class General(DynamoCog):
     """Generic commands"""
 
     def __init__(self, bot: Dynamo) -> None:
-        self.bot: Dynamo = bot
+        super().__init__(bot)
 
     @commands.hybrid_command(name="ping")
     async def ping(self, ctx: commands.Context) -> None:
@@ -99,7 +100,7 @@ async def identicon(
         file = discord.File(BytesIO(idt_bytes), filename=f"{fname}.png")
 
         cmd_mention = await self.bot.tree.find_mention_for("general identicon", guild=ctx.guild)  # type: ignore[attr-defined]
-        prefix = self.bot.prefixes.get(ctx.guild.id, ["d!", "d?"])[0]  # type: ignore[union-attr]
+        prefix = self.bot.prefixes.get(ctx.guild.id, ["d!", "d?"])[0]
         description = f"**Command:**\n{cmd_mention} {display_name}\n{prefix}identicon {display_name}"
 
         e = discord.Embed(title=display_name, description=description, color=discord.Color.from_rgb(*fg.as_tuple()))
@@ -155,3 +156,7 @@ async def spotify(self, ctx: Context, user: discord.Member | discord.User | None
 
 async def setup(bot: Dynamo) -> None:
     await bot.add_cog(General(bot))
+
+
+async def teardown(bot: Dynamo) -> None:
+    await bot.remove_cog(General.__name__)

dynamo/extensions/cogs/help.py

@@ -5,6 +5,7 @@
 from discord.ext import commands
 
 from dynamo.bot import Dynamo
+from dynamo.utils.base_cog import DynamoCog
 from dynamo.utils.format import human_join
 
 CogT = TypeVar("CogT", bound=commands.Cog)
@@ -36,7 +37,7 @@ def __init__(self) -> None:
                 "aliases": ["commands", "h"],
             }
         )
-        self.blacklisted = ["help_command"]
+        self.blacklisted = [Help.__name__]
 
     async def send(self, **kwargs: Any) -> None:
         await self.get_destination().send(**kwargs)
@@ -118,18 +119,24 @@ async def send_cog_help(self, cog: commands.Cog) -> None:
         )
 
 
-class Help(commands.Cog, name="help_command"):
+class Help(DynamoCog):
     def __init__(self, bot: Dynamo) -> None:
-        self.bot: Dynamo = bot
-        self._original_help_command = bot.help_command
+        super().__init__(bot)
+        self._original_help_command = self.bot.help_command
         help_command = DynamoHelp()
         help_command.cog = self
-        bot.help_command = help_command
+        self.bot.help_command = help_command
+        self.log.debug("Using custom help command")
 
     async def cog_unload(self) -> None:
         self.bot.help_command = self._original_help_command
+        self.log.debug("Restoring original help command")
         await super().cog_unload()
 
 
 async def setup(bot: Dynamo) -> None:
     await bot.add_cog(Help(bot))
+
+
+async def teardown(bot: Dynamo) -> None:
+    await bot.remove_cog(Help.__name__)