Warnsystem version 1.4 (#117)

I am merging this PR now, as big updates simply do not fit my schedule anymore. I will now do more frequent updates with less features. This also mean minor version bumps will not contain as many features as you had in versions 1.1 to 1.3

The liked issues that were not completed for this release will be kept open, and may be done in the future.
As an engineer student, free time is starting to become less present. If anyone wants to help maintaining the repo, you're welcome to do so and seek my help if needed.

* Bump version number

* Add end user data functions

* Whitelist for antispam (not tested yet)

* Toggle for checking edited messages with automod

Not tested yet
Resolves #111

docs/warnsystem.rst

@@ -2,7 +2,7 @@
 WarnSystem
 ==========
 
-.. note:: These docs refers to the version **1.2.1**.
+.. note:: These docs refers to the version **1.4.0**.
     Make sure you're under the good version by typing ``[p]cog update``.
 
 This is the guide for the ``warnsystem`` cog. Everything you need is here.

warnsystem/api.py

@@ -1303,20 +1303,25 @@ def disable_automod(self):
         if hasattr(self, "automod_warn_task"):
             self.automod_warn_task.cancel()
 
-    async def automod_on_message(self, message: discord.Message):
+    async def _check_if_automod_valid(self, message: discord.Message):
         guild = message.guild
         member = message.author
         if not guild:
-            return
+            return False
         if member.bot:
-            return
+            return False
         if guild.owner_id == member.id:
-            return
+            return False
         if not self.cache.is_automod_enabled(guild):
-            return
+            return False
         if await self.bot.is_automod_immune(message):
-            return
+            return False
         if await self.bot.is_mod(member):
+            return False
+        return True
+
+    async def automod_on_message(self, message: discord.Message):
+        if not await self._check_if_automod_valid(message):
             return
         # we run all tasks concurrently
         # results are returned in the same order (either None or an exception)
@@ -1336,6 +1341,18 @@ async def automod_on_message(self, message: discord.Message):
                 exc_info=antispam_exception,
             )
 
+    async def automod_on_message_edit(self, before: discord.Message, after: discord.Message):
+        if not await self._check_if_automod_valid(after):
+            return
+        try:
+            await self.automod_process_regex(after)
+        except Exception as e:
+            log.error(
+                f"[Guild {after.guild.id}] Error while "
+                "processing edited message for regex automod.",
+                exc_info=e,
+            )
+
     async def _safe_regex_search(self, regex: re.Pattern, message: discord.Message):
         """
         Mostly safe regex search to prevent reDOS from user defined regex patterns
@@ -1421,6 +1438,9 @@ async def automod_process_antispam(self, message: discord.Message):
         antispam_data = await self.cache.get_automod_antispam(guild)
         if antispam_data is False:
             return
+        for word in antispam_data["whitelist"]:
+            if word in message.content:
+                return
 
         # we slowly go across each key, if it doesn't exist, data is created then the
         # function ends since there's no data to check

warnsystem/automod.py

@@ -247,6 +247,33 @@ async def automod_regex_show(self, ctx: commands.Context, name: str):
         )
         await ctx.send(embed=embed)
 
+    @automod_regex.command(name="edited")
+    async def automod_regex_edited(self, ctx: commands.Context, enable: bool = None):
+        """
+        Defines if the bot should check edited messages.
+        """
+        guild = ctx.guild
+        if enable is not None:
+            if enable is True:
+                await self.cache.set_automod_regex_edited(guild, True)
+                await ctx.send(_("The bot will now check edited messages."))
+            else:
+                await self.cache.set_automod_regex_edited(guild, False)
+                await ctx.send(_("The bot will no longer check edited messages."))
+        else:
+            current = await self.data.guild(guild).automod.regex_edited_messages()
+            await ctx.send(
+                _(
+                    "Edited message check is currently {state}.\n"
+                    "Type `{prefix}automod regex edited {arg}` to {action} it."
+                ).format(
+                    state=_("enabled") if current else _("disabled"),
+                    prefix=ctx.clean_prefix,
+                    arg=not current,
+                    action=_("enable") if not current else _("disable"),
+                )
+            )
+
     @automod.group(name="warn")
     async def automod_warn(self, ctx: commands.Context):
         """
@@ -651,6 +678,85 @@ async def automod_antispam_warn(
             )
         )
 
+    @automod_antispam.group(name="whitelist")
+    async def automod_antispam_whitelist(self, ctx: commands.Context):
+        """
+        Manage word whitelist ignored for antispam.
+        """
+        pass
+
+    @automod_antispam_whitelist.command(name="add")
+    async def automod_antispam_whitelist_add(self, ctx: commands.Context, *words: str):
+        """
+        Add multiple words for the whitelist.
+
+        If you want to add words with spaces, use quotes.
+        """
+        guild = ctx.guild
+        if not words:
+            await ctx.send_help()
+            return
+        async with self.data.guild(guild).automod.antispam.whitelist() as whitelist:
+            for word in words:
+                if word in whitelist:
+                    await ctx.send(_("`{word}` is already in the whitelist.").format(word=word))
+                    return
+            whitelist.extend(words)
+        await self.cache.update_automod_antispam()
+        if len(words) == 1:
+            await ctx.send(_("Added one word to the whitelist."))
+        else:
+            await ctx.send(_("Added {num} words to the whitelist.").format(num=len(words)))
+
+    @automod_antispam_whitelist.command(name="delete", aliases=["del", "remove"])
+    async def automod_antispam_whitelist_delete(self, ctx: commands.Context, *words: str):
+        """
+        Remove multiple words for the whitelist.
+
+        If you want to remove words with spaces, use quotes.
+        """
+        guild = ctx.guild
+        if not words:
+            await ctx.send_help()
+            return
+        async with self.data.guild(guild).automod.antispam.whitelist() as whitelist:
+            for word in words:
+                if word not in whitelist:
+                    await ctx.send(_("`{word}` isn't in the whitelist.").format(word=word))
+                    return
+            whitelist = [x for x in whitelist if x not in words]
+        await self.cache.update_automod_antispam()
+        if len(words) == 1:
+            await ctx.send(_("Removed one word from the whitelist."))
+        else:
+            await ctx.send(_("Removed {num} words from the whitelist.").format(num=len(words)))
+
+    @automod_antispam_whitelist.command(name="list")
+    async def automod_antispam_whitelist_list(self, ctx: commands.Context):
+        """
+        List words in the whitelist.
+        """
+        guild = ctx.guild
+        async with self.data.guild(guild).automod.antispam.whitelist() as whitelist:
+            if not whitelist:
+                await ctx.send(_("Whitelist is empty."))
+                return
+            text = _("__{num} words registered in the whitelist__\n").format(
+                num=len(whitelist)
+            ) + ", ".join(whitelist)
+        for page in pagify(text, delims=[", ", "\n"]):
+            await ctx.send(page)
+
+    @automod_antispam_whitelist.command(name="clear")
+    async def automod_antispam_whitelist_clear(self, ctx: commands.Context):
+        """
+        Clear the whitelist.
+        """
+        guild = ctx.guild
+        await self.data.guild(guild).automod.antispam.whitelist.set([])
+        await self.cache.update_automod_antispam()
+        await ctx.tick()
+
     @automod_antispam.command(name="info")
     async def automod_antispam_info(self, ctx: commands.Context):
         """
@@ -679,12 +785,14 @@ async def automod_antispam_info(self, ctx: commands.Context):
                 "Max messages allowed within the threshold: **{max_messages}**\n"
                 "Threshold: **{delay} seconds**\n"
                 "Delay before reset: **{reset_delay} seconds**  "
-                "*(see `{prefix}automod antispam delay` for details about this)*"
+                "*(see `{prefix}automod antispam delay` for details about this)*\n"
+                "Number of whitelisted words: {whitelist}"
             ).format(
                 max_messages=antispam_settings["max_messages"],
                 delay=antispam_settings["delay"],
                 reset_delay=antispam_settings["delay_before_action"],
                 prefix=ctx.clean_prefix,
+                whitelist=len(antispam_settings["whitelist"]),
             ),
             inline=False,
         )

warnsystem/cache.py

@@ -26,12 +26,15 @@ def __init__(self, bot: Red, config: Config):
         self.automod_enabled = []
         self.automod_antispam = {}
         self.automod_regex = {}
+        self.automod_regex_edited = []
 
     async def init_automod_enabled(self):
         for guild_id, data in (await self.data.all_guilds()).items():
             try:
                 if data["automod"]["enabled"] is True:
                     self.automod_enabled.append(guild_id)
+                if data["automod"]["regex_edited_messages"] is True:
+                    self.automod_regex_edited.append(guild_id)
             except KeyError:
                 pass
 
@@ -164,3 +167,13 @@ async def remove_automod_regex(self, guild: discord.Guild, name: str):
             del self.automod_regex[guild.id][name]
         except KeyError:
             pass
+
+    async def set_automod_regex_edited(self, guild: discord.Guild, enable: bool):
+        await self.data.guild(guild).automod.regex_edited_messages.set(enable)
+        if enable is False and guild.id in self.automod_regex_edited:
+            self.automod_regex_edited.remove(guild.id)
+        elif enable is True and guild.id not in self.automod_regex_edited:
+            self.automod_regex_edited.append(guild.id)
+
+    def is_automod_regex_edited_enabled(self, guild: discord.Guild):
+        return guild.id in self.automod_regex_edited

warnsystem/warnsystem.py

@@ -4,6 +4,7 @@
 import asyncio
 import re
 
+from io import BytesIO
 from typing import Optional
 from asyncio import TimeoutError as AsyncTimeoutError
 from abc import ABC
@@ -200,7 +201,9 @@ class WarnSystem(SettingsMixin, AutomodMixin, BaseCog, metaclass=CompositeMetaCl
                     "reason": "Sending messages too fast!",
                     "time": None,
                 },
+                "whitelist": [],
             },
+            "regex_edited_messages": False,  # if the bot should check message edits
             "regex": {},  # all regex expressions
             "warnings": [],  # all automatic warns
         },
@@ -224,7 +227,7 @@ def __init__(self, bot):
 
         self.task: asyncio.Task
 
-    __version__ = "1.3.22"
+    __version__ = "1.4.0"
     __author__ = ["retke (El Laggron)"]
 
     # helpers
@@ -1482,6 +1485,105 @@ async def on_command_error(self, ctx, error):
                 exc_info=error.original,
             )
 
+    async def _red_get_data_for_user(self, *, user_id: int):
+        readme = (
+            "--- WarnSystem user data ---\n\n\n"
+            "This cog is a tool for moderators and administrators for taking actions against "
+            "members of their server and log it. You can read more about this cog here:\n"
+            "https://github.com/retke/Laggrons-Dumb-Cogs/tree/v3/warnsystem#warnsystem\n\n"
+            "As soon as a member is warned, the cog will store the following data:\n"
+            "- User ID\n"
+            "- Warn level (from 1 to 5: warn, mute, kick, softban, ban)\n"
+            "- Warn reason\n"
+            "- Warn author (responsible moderator. can be the bot in case of automated warns)\n"
+            "- Date and time of the warn\n"
+            "- Duration of the warn (only in case of a temporary mute/ban)\n"
+            "- List of the roles the member had when he was muted "
+            "(only for mutes since version 1.2)\n\n"
+            "A list of files is provided, one for each server. The ID of the server is the name "
+            "of the file. Servers without registered warnings are not included.\n\n"
+            "Additonal notes:\n"
+            "- The timezone for date and time is UTC.\n"
+            "- For durations, the raw number of seconds is included.\n"
+            "- The end date of a temp warn is obtained by adding the duration to the date.\n"
+            "- The responsible moderator of a warn is not included, as this is private data.\n\n\n"
+            "Author of WarnSystem: retke (El Laggron)\n"
+            "Repo: https://github.com/retke/Laggrons-Dumb-Cogs\n"
+            "Contact info is in the README of that repo.\n"
+        )
+        file = BytesIO()
+        file.write(readme.encode("utf-8"))
+        files = {"README": file}
+        all_modlogs = await self.data.custom("MODLOGS").all()
+        for guild_id, modlogs in all_modlogs.items():
+            if str(user_id) not in modlogs:
+                files[guild_id] = BytesIO()
+            guild = self.bot.get_guild(int(guild_id))
+            text = "Modlogs registered for server {guild}\n".format(
+                guild=guild.name if guild else f"{guild_id} (not found)"
+            )
+            for i, modlog in enumerate(modlogs[str(user_id)]["x"]):
+                text += (
+                    "\n\n\n--- Case {number} ---\nLevel:     {level}\nReason:    {reason}\n"
+                ).format(number=i + 1, **modlog)
+                text += "Date:      {date}\n".format(
+                    date=self.api._format_datetime(self.api._get_datetime(modlog["time"]))
+                )
+                if modlog["duration"]:
+                    duration = self.api._get_timedelta(modlog["duration"])
+                    text += "Duration:  {duration} (raw: {raw}s)\n".format(
+                        duration=self.api._format_timedelta(duration),
+                        raw=modlog["duration"],
+                    )
+                if modlog["roles"]:
+                    text += "Roles:     {roles}\n".format(roles=", ".join(modlog["roles"]))
+            file = BytesIO()
+            file.write(text.encode("utf-8"))
+            files[guild_id] = file
+        return files
+
+    async def red_get_data_for_user(self, *, user_id: int):
+        try:
+            data = await self._red_get_data_for_user(user_id=user_id)
+        except Exception as e:
+            log.error(
+                f"User {user_id} has requested end user data but an exception occured!", exc_info=e
+            )
+            raise
+        else:
+            log.info(
+                f"User {user_id} has requested end user data, which was successfully provided."
+            )
+            return data
+
+    async def _red_delete_data_for_user(self, *, requester: str, user_id: int):
+        allowed_requesters = ("discord_deleted_user",)
+        if requester not in allowed_requesters:
+            return False
+        async with self.data.custom("MODLOGS").all() as all_modlogs:
+            for guild_id, modlogs in all_modlogs.items():
+                try:
+                    del all_modlogs[guild_id][str(user_id)]
+                except KeyError:
+                    pass
+        return True
+
+    async def red_delete_data_for_user(self, *, requester: str, user_id: int):
+        try:
+            result = await self._red_delete_data_for_user(requester=requester, user_id=user_id)
+        except Exception as e:
+            log.error(
+                f"User {user_id} has requested end user data deletion but an exception occured!",
+                exc_info=e,
+            )
+            raise
+        else:
+            if result is True:
+                log.info(
+                    f"User {user_id} has requested end user data "
+                    "deletion, which was successfully done."
+                )
+
     # correctly unload the cog
     def __unload(self):
         self.cog_unload()