Implement voiceloger

plugin/__init__.py

@@ -1 +1,5 @@
+import logging
+
+logger = logging.getLogger(__name__)
+
 from .discord import setup

plugin/data.py

@@ -0,0 +1,16 @@
+from data import Registry, RowModel, Table
+from data.columns import String, Timestamp, Integer, Bool
+
+
+class VoiceLogGuild(RowModel):
+    _tablename_ = "voicelog_guilds"
+    _cache_ = {}
+
+    guildid = Integer(primary=True)
+    webhook_url = String()
+    created_at = Timestamp()
+    _timestamp = Timestamp()
+
+
+class VoiceLogData(Registry):
+    voicelog_guilds = VoiceLogGuild.table

plugin/discord/__init__.py

@@ -0,0 +1,7 @@
+from .. import logger
+
+
+async def setup(bot):
+    from .cog import VoiceLogCog
+
+    await bot.add_cog(VoiceLogCog(bot))

plugin/discord/cog.py

@@ -0,0 +1,106 @@
+import discord
+from discord.ext import commands as cmds
+from discord import app_commands as appcmds
+
+from meta import LionBot, LionCog, LionContext
+from meta.logger import log_wrap
+from utils.lib import utc_now
+
+from . import logger
+from ..data import VoiceLogData, VoiceLogGuild
+from .lib import ThreadedWebhook
+
+
+class VoiceLogCog(LionCog):
+    def __init__(self, bot: LionBot):
+        self.bot = bot
+        self.data = bot.db.load_registry(VoiceLogData())
+
+    async def cog_load(self):
+        await self.data.init()
+
+    @LionCog.listener("on_voice_state_update")
+    @log_wrap(action="Voice Log Update")
+    async def voicelog_update(
+        self,
+        member: discord.Member,
+        before: discord.VoiceState,
+        after: discord.VoiceState,
+    ):
+        if member.bot:
+            return
+
+        after_channel = after.channel
+        before_channel = before.channel
+        if after_channel == before.channel:
+            return
+
+        row = await VoiceLogGuild.fetch(member.guild.id)
+        if row is None or row.webhook_url is None:
+            return
+
+        hook = ThreadedWebhook.from_url(logger.webhook_url, client=self.bot)
+
+        embed = discord.Embed(timestamp=utc_now())
+
+        if before_channel is None:
+            embed.title = "User Joined Voice Channel"
+            embed.description = f"{member.mention} joined {after_channel.mention}"
+        elif after_channel is None:
+            embed.title = "User Left Voice Channel"
+            embed.description = f"{member.mention} left {before_channel.mention}"
+        else:
+            embed.title = "User Switched Voice Channel"
+            embed.description = f"{member.mention} moved from {before_channel.mention} to {after_channel.mention}"
+
+        await hook.send(embed=embed)
+
+    @cmds.hybrid_group(
+        name="voicelog", description="Base command group for the voice logging system"
+    )
+    @appcmds.default_permissions(manage_guild=True)
+    async def voicelog_group(self, ctx: LionContext): ...
+
+    @voicelog_group.command(
+        name="enable",
+        description="Enable voice activity logging and set the webhook url to use.",
+    )
+    async def voicelog_enable(self, ctx: LionContext, webhook_url: str):
+        if not ctx.guild:
+            return
+        if not ctx.author.guild_permissions.manage_guild:
+            return
+
+        webhook = ThreadedWebhook.from_url(webhook_url, client=self.bot)
+        try:
+            embed = discord.Embed(
+                title="Testing",
+                description="Testing logging webhook, feel free to delete.",
+            )
+            await webhook.send(embed=embed, wait=True)
+            existing = await VoiceLogGuild.fetch(ctx.guild.id)
+            if existing:
+                await existing.update(webhook_url=webhook_url)
+            else:
+                await VoiceLogGuild.create(
+                    guildid=ctx.guild.id, webhook_url=webhook_url
+                )
+
+            await ctx.reply("Voice logging enabled!")
+        except discord.HTTPException:
+            await ctx.error_reply("Could not post to the given webhook!")
+
+    @voicelog_group.command(
+        name="disable", description="Disable voice activity logging in this server."
+    )
+    async def voicelog_disable(self, ctx: LionContext):
+        if not ctx.guild:
+            return
+        if not ctx.author.guild_permissions.manage_guild:
+            return
+
+        await self.data.voicelog_guilds.update_where(guildid=ctx.guild.id).set(
+            webhook_url=None
+        )
+
+        await ctx.reply("Voice Activity logging disabled.")

plugin/discord/lib.py

@@ -0,0 +1,25 @@
+import discord
+from urllib.parse import urlparse, parse_qs
+
+
+class ThreadedWebhook(discord.Webhook):
+    __slots__ = ("thread_id",)
+
+    def __init__(self, *args, thread_id=None, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.thread_id = thread_id
+
+    @classmethod
+    def from_url(cls, url: str, *args, **kwargs):
+        self = super().from_url(url, *args, **kwargs)
+        parse = urlparse(url)
+        if parse.query:
+            args = parse_qs(parse.query)
+            if "thread_id" in args:
+                self.thread_id = int(args["thread_id"][0])
+        return self
+
+    async def send(self, *args, **kwargs):
+        if self.thread_id is not None:
+            kwargs.setdefault("thread", discord.Object(self.thread_id))
+        return await super().send(*args, **kwargs)