Logging Start (#19)

* feat: Error And Guild Join/Leave Logging

* chore: Run Black

Author: Icebluewolf

bot/cogs/developer.py

@@ -0,0 +1,70 @@
+import discord
+from discord import ApplicationContext
+
+from utils.bot import SurveyWolf
+from main import bot as survey_wolf_bot
+
+
+class Developer(discord.Cog, guild_ids=survey_wolf_bot.config["dev_guilds"]):
+    def __init__(self, bot):
+        self.bot: SurveyWolf = bot
+
+    log_text = {
+        "error_logging_webhook": "Errors",
+        "server_join_leave_webhook": "Guild Join/Leave",
+    }
+    logs = [discord.OptionChoice(x[1], x[0]) for x in log_text.items()]
+    logging = discord.SlashCommandGroup("logging", "Actions For The Discord Facing Logging")
+
+    async def cog_before_invoke(self, ctx: ApplicationContext) -> None:
+        if ctx.guild_id not in self.bot.config["dev_guilds"]:
+            raise Exception("Guild Not Authorized To Run Developer Command")
+
+        if not await self.bot.is_owner(ctx.user):
+            raise Exception("User Not Authorized To Run Developer Command")
+
+    @discord.Cog.listener()
+    async def on_guild_join(self, guild):
+        await self.bot.config["server_join_leave_webhook"].send(
+            f"Joined A New Server: {guild.id} Total: {len(self.bot.guilds)}"
+        )
+
+    @discord.Cog.listener()
+    async def on_guild_remove(self, guild):
+        await self.bot.config["server_join_leave_webhook"].send(
+            f"Left A Server: {guild.id} Total: {len(self.bot.guilds)}"
+        )
+
+    async def _remove_webhook(self, w: discord.Webhook | None):
+        if w is not None:
+            try:
+                if w.is_partial():
+                    w = await self.bot.fetch_webhook(w.id)
+                await w.delete(reason="Logging Moved Or Disabled")
+            except discord.NotFound:
+                pass
+
+    @logging.command(description="Creates A Webhook In The Current Channel For The Specified Log")
+    async def set(
+        self, ctx: discord.ApplicationContext, log: discord.Option(str, description="The Log To Set", choices=logs)
+    ):
+        w = await ctx.channel.create_webhook(
+            name=f"{self.bot.user.name} {self.log_text[log]} Log",
+            avatar=await self.bot.user.avatar.read() if self.bot.user.avatar else None,
+            reason="Logging Enabled",
+        )
+        await self._remove_webhook(self.bot.config[log])
+        self.bot.update_config(log, w, w.url)
+        await ctx.respond("Logging Set", ephemeral=True)
+
+    @logging.command(description="Removes The Webhook For The Specified Log")
+    async def unset(
+        self, ctx: discord.ApplicationContext, log: discord.Option(str, description="The Log To Remove", choices=logs)
+    ):
+        await self._remove_webhook(self.bot.config[log])
+        self.bot.update_config(log, None, "None")
+        await ctx.respond("Logging Unset", ephemeral=True)
+
+
+def setup(bot):
+    bot.add_cog(Developer(bot))

bot/main.py

@@ -5,7 +5,7 @@
 
 load_dotenv()
 
-COGS = ["utility", "survey.creation", "survey.active", "survey.results"]
+COGS = ["utility", "survey.creation", "survey.active", "survey.results", "developer"]
 
 intents = discord.Intents.default()
 

bot/utils/bot.py

@@ -1,9 +1,13 @@
+import traceback
 from abc import ABC
+from typing import Any
+
 import yaml
 
 import discord
 from utils.database import database
-from discord import Interaction
+from . import embed_factory as ef
+from discord import Interaction, ApplicationContext, DiscordException
 
 
 class AdvContext(discord.ApplicationContext):
@@ -29,12 +33,85 @@ class AdvContext(discord.ApplicationContext):
 class SurveyWolf(discord.Bot, ABC):
     def __init__(self, description=None, *args, **options):
         super().__init__(description=description, *args, **options)
+        self._did_on_ready = False
 
         with open("config.yaml") as stream:
             config = yaml.safe_load(stream)
 
-        self.ERROR_LOGGING_WEBHOOK = config["error_logging_webhook"]
-        self.SERVER_JOIN_LEAVE_WEBHOOK = config["server_join_leave_webhook"]
+        self._raw_config = config
+        self.config = self._raw_config.copy()
+
+    async def on_ready(self):
+        if self._did_on_ready:
+            return
+        self._did_on_ready = True
+        # Do Some Additional Processing On Some Config Items
+        self.config.update(
+            {
+                "error_logging_webhook": await self._create_webhook(self.config["error_logging_webhook"]),
+                "server_join_leave_webhook": await self._create_webhook(self.config["server_join_leave_webhook"]),
+            }
+        )
+
+    async def _create_webhook(self, url: str) -> discord.Webhook | None:
+        if url == "None":
+            return None
+
+        try:
+            return await self.fetch_webhook(discord.Webhook.from_url(url, session=self.http._HTTPClient__session).id)
+        except discord.NotFound:
+            return None
+
+    def update_config(self, key: str, value, raw=None) -> None:
+        """
+        Updates the config with the key and value. Updates the config file with the raw value
+        :param key: The Config Key
+        :param value: The Value The Bot Should Retrieve
+        :param raw: The Value That Should Be Stored In The Config File. Defaults To `value`
+        """
+        if raw is None:
+            raw = value
+
+        self._raw_config[key] = raw
+        with open("config.yaml", "w") as stream:
+            yaml.safe_dump(self._raw_config, stream)
+        self.config[key] = value
 
     async def get_application_context(self, interaction: Interaction, cls=None) -> discord.ApplicationContext:
         return await super().get_application_context(interaction, cls=cls or AdvContext)
+
+    @staticmethod
+    def _split_text(text: str, max_length: int, newline: bool = False) -> list[str]:
+        texts = []
+        while len(text) > max_length:
+            if newline:
+                try:
+                    ind = text[:max_length].rindex("\n")
+                except ValueError:
+                    # No newline was found so fall back to character count
+                    ind = max_length
+            else:
+                ind = max_length
+
+            texts.append(text[:ind])
+            # The +1 is to remove the newline character
+            text = text[ind + 1:]
+        texts.append(text)
+        return texts
+
+    async def on_application_command_error(self, ctx: ApplicationContext, exception: DiscordException) -> None:
+        if (w := self.config["error_logging_webhook"]) is not None:
+            text = f"Error In {ctx.command.qualified_name} Guild ID: {ctx.guild_id} Channel ID: {ctx.channel_id}\n"
+            text += "".join(traceback.format_exception(type(exception), exception, exception.__traceback__))
+            for i in self._split_text(text, 1990, newline=True):
+                await w.send(f"```py\n{i}\n```")
+        await ctx.respond(embed=await ef.error("An Error Occurred"))
+        raise exception
+
+    async def on_error(self, event_method: str, *args: Any, **kwargs: Any) -> None:
+        if (w := self.config["error_logging_webhook"]) is not None:
+            text = "Error In " + event_method + "\n"
+            text += traceback.format_exc()
+            for i in self._split_text(text, 1990, newline=True):
+                await w.send(f"```py\n{i}\n```")
+        traceback.print_exc()