Merge pull request #31 from Puh00/rework/rpgchat

Rework/rpgchat

cogs/chat.py

@@ -0,0 +1,63 @@
+from util.toru_rpg import ToruRpg
+from util.error_handler import async_ignore_multiple_errors
+
+from discord import Member, Message
+from discord.ext import commands
+from discord.ext.commands import Context, errors
+
+from pymongo.errors import ConnectionFailure, ServerSelectionTimeoutError
+
+
+class RpgChat(commands.Cog):
+    def __init__(self, client) -> None:
+        self.client = client
+        self.handler = ToruRpg()
+
+    @commands.Cog.listener()
+    @async_ignore_multiple_errors([ServerSelectionTimeoutError, ConnectionFailure])
+    async def on_member_join(self, member: Member):
+        self.handler.register(member.id, member.guild.id)
+
+    @commands.Cog.listener()
+    @async_ignore_multiple_errors([ServerSelectionTimeoutError, ConnectionFailure])
+    async def on_member_remove(self, member: Member):
+        self.handler.unregister(member.id, member.guild.id)
+
+    @commands.Cog.listener()
+    @async_ignore_multiple_errors([ServerSelectionTimeoutError, ConnectionFailure])
+    async def on_message(self, message: Message):
+        ctx = await self.client.get_context(message)
+        # check if the message is a command
+        if not ctx.valid:
+            self.handler.add_exp(
+                message.author.id, message.guild.id, len(message.content)
+            )
+
+    @commands.command()
+    async def exp(self, ctx: Context):
+        exp = self.handler.get_exp(ctx.author.id, ctx.guild.id)
+        await ctx.send(
+            f"{ctx.author.mention} has: `{exp['current_exp']}/{exp['required_exp']}` xp"
+        )
+
+    @commands.command()
+    async def level(self, ctx: Context):
+        level = self.handler.get_level(ctx.author.id, ctx.guild.id)
+        await ctx.send(f"{ctx.author.mention}'s level is: `{level}`")
+
+    @exp.error
+    @level.error
+    async def chat_error(self, ctx: Context, error):
+        message = ":x: "
+        if isinstance(error, errors.CommandInvokeError):
+            cause = error.__cause__
+            if isinstance(cause, (ServerSelectionTimeoutError, ConnectionFailure)):
+                message += "The database is currently taking a nap, try again later!"
+        else:
+            message += f"Command failed to execute due to: ```\n{error}\n```"
+
+        await ctx.send(message)
+
+
+def setup(client):
+    client.add_cog(RpgChat(client))

util/error_handler.py

@@ -0,0 +1,98 @@
+"""Error Handler
+
+!!WORK IN PROGRESS!!
+
+This script provides useful decorators/methods for handling errors.
+"""
+
+import logging
+import functools
+from typing import Type
+
+
+def async_ignore_an_error(error_to_ignore: Type[Exception]):
+    """Decorator used to ignore a specific error for an async function
+
+    Parameters
+    ----------
+    error_to_ignore : Type[Exception]
+        The type of the error to be ignored
+    """
+
+    def decorator(func):
+        @functools.wraps(func)
+        async def wrapper(*args, **kwargs):
+            try:
+                await func(*args, **kwargs)
+            except error_to_ignore:
+                logging.info(f"Ignored an error: {error_to_ignore}")
+
+        return wrapper
+
+    return decorator
+
+
+def ignore_an_error(error_to_ignore: Type[Exception]):
+    """Decorator used to ignore a specific error for a blocking function
+
+    Parameters
+    ----------
+    error_to_ignore : Type[Exception]
+        The type of the error to be ignored
+    """
+
+    def decorator(func):
+        @functools.wraps(func)
+        def wrapper(*args, **kwargs):
+            try:
+                func(*args, **kwargs)
+            except error_to_ignore:
+                logging.info(f"Ignored an error: {error_to_ignore}")
+
+        return wrapper
+
+    return decorator
+
+
+def async_ignore_multiple_errors(errors_to_ignore: list[Type[Exception]]):
+    """Decorator used to ignore a list of errors for an async function
+
+    Parameters
+    ----------
+    errors_to_ignore : list[Type[Exception]]
+        The list of error types to ignore
+    """
+
+    def decorator(func):
+        @functools.wraps(func)
+        async def wrapper(*args, **kwargs):
+            try:
+                await func(*args, **kwargs)
+            except tuple(errors_to_ignore) as e:
+                logging.info(f"Ignored an error: {e}")
+
+        return wrapper
+
+    return decorator
+
+
+def ignore_multiple_errors(errors_to_ignore: list[Type[Exception]]):
+    """Decorator used to ignore a list of errors for a blocking function
+
+    Parameters
+    ----------
+    errors_to_ignore : list[Type[Exception]]
+        The list of error types to ignore
+    """
+
+    def decorator(func):
+        @functools.wraps(func)
+        def wrapper(*args, **kwargs):
+            try:
+                func(*args, **kwargs)
+            except tuple(errors_to_ignore) as e:
+                logging.info(f"Ignored an error: {e}")
+
+        return wrapper
+
+    return decorator