Merge pull request #3 from matthewflegg/patch-1

Patched AI chatbot errors and created a generic error handler

Author: v0idzdev

_chatbot/__init__.py

@@ -4,28 +4,25 @@
 
 from discord.ext import commands
 from dotenv import load_dotenv
-
 load_dotenv(".env")
 
 intents = discord.Intents.all()
 intents.members = True
 
 bot = commands.Bot(command_prefix=".", intents=intents)
+extensions = ["greetings", "miscellaneous", "admin", "music",
+              "help", "chatbot", "error_handler"]
 
-extensions = ["greetings", "miscellaneous", "admin", "music", "help"]
-_ = [bot.load_extension(f"extensions.{ext}") for ext in extensions]
-
-optional_extensions = ["chatbot"]
-_ = [bot.load_extension(f"optional_extensions.{opt_ext}") for opt_ext in optional_extensions]
-
+for ext in extensions:
+    bot.load_extension(f"extensions.{ext}")
 
 @bot.event
 async def on_ready():
     """Displays a series of logs to the screen when the bot has loaded"""
 
-    log.info(f"Username: {bot.user.name}")
-    log.info(f"Bot ID: {bot.user.id}")
-    log.info(f"Bot loaded successfully")
+    print(f"Username: {bot.user.name}")
+    print(f"Bot ID: {bot.user.id}")
+    print(f"Bot loaded successfully")
 
 
 TOKEN = os.getenv("TOKEN")

app.py

@@ -0,0 +1 @@
+from .model import init_chatbot

chatbot/__init__.py

@@ -10,7 +10,7 @@ def init_chatbot():
     chatbot = ChatBot_(
         "Beep Boop Bot",
         storage_adapter="chatterbot.storage.SQLStorageAdapter",
-        database_uri="sqlite:///_chatbot_db/db.sqlite3"
+        database_uri="sqlite:///chatbot\chatbot_db\db.sqlite3"
     )
 
     trainer = ChatterBotCorpusTrainer(chatbot)

_chatbot/_model.py renamed to chatbot/model.py

@@ -3,6 +3,10 @@
 from discord.ext import commands
 
 
+class AdminCommandError(commands.CommandError):
+    """Exception class for admin commands"""
+
+
 class Admin(commands.Cog):
     """Command category for administrator only commands"""
     def __init__(self, bot):
@@ -24,10 +28,10 @@ async def ban_(self, ctx: commands.Context, user: discord.Member, *, reason=None
 
         await ctx.send(f"**{user.name}** has successfully been banned")
 
-    @commands.command(name="kick", aliases=["k"], description="Kicks a user from a server")
+    @commands.command(name="kick", aliases=["k"], description="Kicks a user from a server.")
     @commands.has_permissions(kick_members=True)
     async def kick_(self, ctx: commands.Context, user: discord.Member, *, reason=None):
-        """Kicks a user from a server,
+        """Kicks a user from a server.
         Requires the 'Kick members' permission
         """
 

extensions/admin.py

@@ -0,0 +1,86 @@
+from datetime import datetime
+import discord
+import asyncio
+import chatbot
+from discord.ext import commands
+
+
+class ChatBot(commands.Cog):
+    """Command/event category for the AI chatbot"""
+    def __init__(self, bot: commands.Bot):
+        self.chatbot = chatbot.init_chatbot() # Initialise the chatbot model from the _chatbot module
+        self.bot = bot
+        self.enabled_channels = [] # A list of all the channels AI chat is enabled in (!!! THIS MIGHT BE DODGY !!!)
+        self.time_enabled = None
+
+    @commands.Cog.listener()
+    async def on_message(self, message: discord.Message):
+        """Gets a response from the AI chatbot when someone sends a message"""
+
+        if message.author.bot or message.channel not in self.enabled_channels:
+            return # Do nothing if the bot sent the message or if this isn't an enabled channel
+        if message.created_at < self.time_enabled: # Ignore all messages before the chatbot was enabled
+            return
+        if message.content.startswith("."): # If the message is a command, ignore it
+            return
+
+        # Show 'bot is typing' for 2 seconds
+        ctx = await self.bot.get_context(message)
+
+        async with ctx.typing():
+            await asyncio.sleep(2)
+
+        # Get and send an AI response
+        response_ = self.chatbot.get_response(message.content)
+        await message.channel.send(f"{message.author.mention}: {response_}")
+
+    @commands.command(name="aienable", aliases=["ai", "enableaichat"],
+                      description="Enables AI chatbot responses to all messages within a channel."
+                                + "Requires 'Manage channels' permissions")
+    @commands.has_permissions(manage_channels=True)
+    async def enable_ai_chat(self, ctx: commands.Context):
+        """Enables AI chatbot responses in a text channel"""
+
+        # Add the channel to the list. Check in on_message if this is an enabled channel
+        if ctx.channel not in self.enabled_channels: # Check if not already enabled
+            self.enabled_channels.append(ctx.channel)
+            self.time_enabled = datetime.now()
+
+            await ctx.send("✅ Done! I will now use AI to respond to all messages in this channel\n" +
+                        "You can use .aidisable or .disableaichat to disable chatbot responses")
+        else:
+            await ctx.send("❌ You've already enabled AI chat in this channel.")
+
+    @enable_ai_chat.error
+    async def enable_ai_chat_error(self, ctx: commands.Context, error):
+        """Called when the enable AI chat command throws an error"""
+
+        if isinstance(error, commands.MissingPermissions):
+            await ctx.send("❌ You don't have permission to enable AI chat in this channel")
+
+    @commands.command(name="aidisable", aliases=["disableaichat"],
+                      description="Disables AI chatbot responses to all messages within a channel."
+                                + "Requires 'Manage channels' permissions")
+    @commands.has_permissions(manage_channels=True)
+    async def disable_ai_chat(self, ctx: commands.Context):
+        """Disables AI chatbot responses in a text channel"""
+
+        if ctx.channel in self.enabled_channels: # If the channel is enabled, disable it
+            self.enabled_channels.remove(ctx.channel)
+
+            await ctx.send("✅ Done! I will no longer use AI to respond to all messages in this channel\n" +
+                        "You can use .aienable, .ai, or .enableaichat to enable chatbot responses")
+        else:
+            await ctx.send("❌ You've already disabled AI chat in this channel.")
+
+    @disable_ai_chat.error
+    async def disable_ai_chat_error(self, ctx: commands.Context, error):
+        """Called when the disable AI chat command throws an error"""
+
+        if isinstance(error, commands.MissingPermissions):
+            await ctx.send("❌ You don't have permission to disable AI chat in this channel")
+
+
+def setup(bot: commands.Bot):
+    """Adds the 'ChatBot' cog to the bot"""
+    bot.add_cog(ChatBot(bot))

extensions/chatbot.py

@@ -0,0 +1,75 @@
+import traceback
+import discord
+import math
+import sys
+from discord.ext import commands
+
+
+class GenericErrorHandler(commands.Cog):
+    """Generic error handler.
+    Used if local (cog-specific) error handlers fail to catch exceptions.
+    """
+    def __init__(self, bot: commands.Bot):
+        self.bot = bot
+
+    @commands.Cog.listener()
+    async def on_command_error(self, ctx: commands.Context, error):
+        """Generic command error handler."""
+        if hasattr(ctx.command, "on_error"): # Don't override local errors
+            return
+
+        error = getattr(error, "original", error)
+
+        if isinstance(error, commands.CommandNotFound):
+            return
+
+        if isinstance(error, commands.BotMissingPermissions): # Error handling for missing bot perms
+            missing = [perm.replace("_", " ").replace("guild", "server").title() for perm in error.missing_perms]
+
+            if len(missing) > 2:
+                format_ = f"{'**, **'.join(missing[: -1])}, and {missing[-1]}"
+            else:
+                format_ = " and ".join(missing)
+
+            message_ = f"I need the **{format_}** permission(s) to use this command."
+            await ctx.send(message_)
+            return
+
+        if isinstance(error, commands.DisabledCommand): # Command is disabled
+            await ctx.send("This command has been disabled.")
+            return
+
+        if isinstance(error, commands.CommandOnCooldown): # Command is currently on cooldown
+            await ctx.send(f"This command is currently on cooldown. Try again in {math.ceil(error.retry_after)}s")
+            return
+
+        if isinstance(error, commands.MissingPermissions): # Error handling for missing user perms
+            missing = [perm.replace("_", " ").replace("guild", "server").title() for perm in error.missing_perms]
+
+            if len(missing) > 2:
+                format_ = f"{'**, **'.join(missing[: -1])}, and {missing[-1]}"
+            else:
+                format_ = " and ".join(missing)
+
+            message_ = f"You need the **{format_}** permission(s) to use this command."
+            await ctx.send(message_)
+            return
+
+        if isinstance(error, commands.UserInputError): # If the bot receives bad input
+            await ctx.send("Sorry, that input is invalid.")
+            return
+
+        if isinstance(error, commands.NoPrivateMessage): # If the user can't use the command in a DM
+            try:
+                await ctx.author.send("You can't use this command in private messages.")
+            except discord.Forbidden:
+                pass
+            return
+
+        print(f"Ignoring exceptions in command {ctx.command}", file=sys.stderr) # Print errors to stderr
+        traceback.print_exception(type(error), error, error.__traceback__, file=sys.stderr)
+
+
+def setup(bot: commands.Bot):
+    """Adds the 'GenericErrorHandler' cog to the bot."""
+    bot.add_cog(GenericErrorHandler(bot))