Implement Moderation logging and replication with the discord.py server (#38)

* add moderation logging capability from discord.py server

* apply feedback

Author: AbstractUmbra

constants/constants.py

@@ -59,6 +59,7 @@ class Channels(CONSTANTS):
     MYSTBIN_DEV = 698366338774728714
 
     FORUM_LOGS = 1114743569786347550
+    DPY_MOD_LOGS = 955843398001127434
 
 
 class Colours(CONSTANTS):

core/__init__.py

@@ -27,4 +27,5 @@
 from .context import *
 from .converters import *
 from .core import *
+from .enums import *
 from .errors import *

core/bot.py

@@ -135,6 +135,38 @@ async def on_command_error(self, ctx: Context, error: commands.CommandError) ->
 
         self.log_handler.error("Command Error", extra={"embed": embed})
 
+    async def get_or_fetch_user(
+        self,
+        target_id: int,
+        /,
+        *,
+        guild: discord.Guild | None = None,
+        cache: dict[int, discord.User | discord.Member] | None = None,
+    ) -> discord.User | discord.Member | None:
+        if guild:
+            user = guild.get_member(target_id)
+            if not user:
+                try:
+                    user = await guild.fetch_member(target_id)
+                except discord.HTTPException:
+                    return
+
+            if cache:
+                cache[target_id] = user
+            return user
+
+        user = self.get_user(target_id)
+        if not user:
+            try:
+                user = await self.fetch_user(target_id)
+            except discord.HTTPException:
+                return
+
+        if cache:
+            cache[target_id] = user
+
+        return user
+
     async def start(self, token: str, *, reconnect: bool = True) -> None:
         try:
             await super().start(token=token, reconnect=reconnect)

core/enums.py

@@ -0,0 +1,12 @@
+from discord.enums import Enum
+
+
+__all__ = ("DiscordPyModerationEvent",)
+
+
+class DiscordPyModerationEvent(Enum):
+    ban = 1
+    kick = 2
+    mute = 3
+    unban = 4
+    helpblock = 5

core/utils/formatters.py

@@ -20,10 +20,16 @@
 OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 SOFTWARE.
 """
+import random
+
+from discord import Colour
 from discord.utils import escape_markdown
 
 
-__all__ = ("to_codeblock",)
+__all__ = (
+    "to_codeblock",
+    "random_pastel_colour",
+)
 
 
 def to_codeblock(
@@ -41,3 +47,7 @@ def to_codeblock(
     if escape_md:
         content = escape_markdown(content)
     return f"```{language}\n{content}\n```"
+
+
+def random_pastel_colour() -> Colour:
+    return Colour.from_hsv(random.random(), 0.28, 0.97)

modules/moderation.py

@@ -25,17 +25,34 @@
 import asyncio
 import base64
 import binascii
+import datetime
 import re
-from typing import Any
+from textwrap import shorten
+from typing import TYPE_CHECKING, Any, Self, TypeAlias
 
 import discord
 import yarl
 from discord.ext import commands
 
 import core
+from constants import Channels
+from core.context import Interaction
+from core.utils import random_pastel_colour
 
 
+if TYPE_CHECKING:
+    from types_.papi import ModLogPayload, PythonistaAPIWebsocketPayload
+
+    ModLogType: TypeAlias = PythonistaAPIWebsocketPayload[ModLogPayload]
+
 TOKEN_RE = re.compile(r"[a-zA-Z0-9_-]{23,28}\.[a-zA-Z0-9_-]{6,7}\.[a-zA-Z0-9_-]{27}")
+PROSE_LOOKUP = {
+    1: "banned",
+    2: "kicked",
+    3: "muted",
+    4: "unbanned",
+    5: "helpblocked",
+}
 
 
 def validate_token(token: str) -> bool:
@@ -45,17 +62,47 @@ def validate_token(token: str) -> bool:
         user_id = int(base64.b64decode(user_id + "==", validate=True))
     except (ValueError, binascii.Error):
         return False
-    else:
-        return True
+    return True
 
 
 class GithubError(commands.CommandError):
     pass
 
 
+class ModerationRespostView(discord.ui.View):
+    message: discord.Message | discord.WebhookMessage
+
+    def __init__(self, *, timeout: float | None = 180, target_id: int, target_reason: str) -> None:
+        super().__init__(timeout=timeout)
+        self.target: discord.Object = discord.Object(id=target_id, type=discord.Member)
+        self.target_reason: str = target_reason
+
+    def _disable_all_buttons(self) -> None:
+        for item in self.children:
+            if isinstance(item, (discord.ui.Button, discord.ui.Select)):
+                item.disabled = True
+
+    async def on_timeout(self) -> None:
+        self._disable_all_buttons()
+        await self.message.edit(view=self)
+
+    @discord.ui.button(label="Ban", emoji="\U0001f528")
+    async def ban_button(self, interaction: Interaction, button: discord.ui.Button[Self]) -> None:
+        assert interaction.guild
+        await interaction.response.defer(ephemeral=False)
+
+        reason = f"Banned due to grievances in discord.py: {self.target_reason!r}"
+        await interaction.guild.ban(
+            self.target,
+            reason=shorten(reason, width=128, placeholder="..."),
+        )
+        await interaction.followup.send("Banned.")
+
+
 class Moderation(commands.Cog):
     def __init__(self, bot: core.Bot, /) -> None:
         self.bot = bot
+        self.dpy_mod_cache: dict[int, discord.User | discord.Member] = {}
         self._req_lock = asyncio.Lock()
 
     async def github_request(
@@ -73,7 +120,7 @@ async def github_request(
 
         hdrs = {
             "Accept": "application/vnd.github.inertia-preview+json",
-            "User-Agent": "RoboDanny DPYExclusive Cog",
+            "User-Agent": "PythonistaBot Moderation Cog",
             "Authorization": f"token {api_key}",
         }
 
@@ -148,6 +195,56 @@ async def find_discord_tokens(self, message: discord.Message) -> None:
         )
         await message.reply(msg)
 
+    @commands.Cog.listener()
+    async def on_papi_dpy_modlog(self, payload: ModLogType, /) -> None:
+        moderation_payload = payload["payload"]
+        moderation_event = core.DiscordPyModerationEvent(moderation_payload["moderation_event_type"])
+
+        embed = discord.Embed(
+            title=f"Discord.py Moderation Event: {moderation_event.name.title()}",
+            colour=random_pastel_colour(),
+        )
+
+        target_id = moderation_payload["target_id"]
+        target = await self.bot.get_or_fetch_user(target_id)
+
+        moderation_reason = moderation_payload["reason"]
+
+        moderator_id = moderation_payload["author_id"]
+        moderator = self.dpy_mod_cache.get(moderator_id) or await self.bot.get_or_fetch_user(
+            moderator_id, cache=self.dpy_mod_cache
+        )
+
+        if moderator:
+            self.dpy_mod_cache[moderator.id] = moderator
+            moderator_format = f"{moderator.name} {PROSE_LOOKUP[moderation_event.value]} "
+            embed.set_author(name=moderator.name, icon_url=moderator.display_avatar.url)
+        else:
+            moderator_format = f"Unknown Moderator with ID: {moderator_id} {PROSE_LOOKUP[moderation_event.value]} "
+            embed.set_author(name=f"Unknown Moderator.")
+
+        if target:
+            target_format = target.name
+            embed.set_footer(text=f"{target.name} | {target_id}", icon_url=target.display_avatar.url)
+        else:
+            target_format = f"An unknown user with ID {target_id}"
+            embed.set_footer(text=f"Not Found | {target_id}")
+        embed.add_field(name="Reason", value=moderation_reason or "No reason given.")
+
+        embed.description = moderator_format + target_format
+
+        when = datetime.datetime.fromisoformat(moderation_payload["event_time"])
+        embed.timestamp = when
+
+        guild = self.bot.get_guild(490948346773635102)
+        assert guild
+
+        channel = guild.get_channel(Channels.DPY_MOD_LOGS)
+        assert isinstance(channel, discord.TextChannel)  # This is static
+
+        view = ModerationRespostView(timeout=900, target_id=target_id, target_reason=moderation_reason)
+        view.message = await channel.send(embed=embed, view=view)
+
 
 async def setup(bot: core.Bot) -> None:
     await bot.add_cog(Moderation(bot))

types_/papi.py

@@ -0,0 +1,30 @@
+from typing import Generic, Literal, TypedDict, TypeVar
+
+
+PayloadT = TypeVar("PayloadT", bound=TypedDict)
+
+
+class ModLogPayload(TypedDict):
+    # event types:
+    """
+    1. Ban
+    2. Kick
+    3. Mute
+    4. Unban
+    5. Helpblock?
+    """
+    moderation_event_type: Literal[1, 2, 3, 4, 5]
+    guild_id: int
+    target_id: int
+    author_id: int
+    reason: str
+    event_time: str  # isoformatted datetime
+
+
+class PythonistaAPIWebsocketPayload(TypedDict, Generic[PayloadT]):
+    op: int
+    subscription: str
+    application: int
+    application_name: str
+    payload: PayloadT
+    user_id: int