jarvis-bot/jarvis/utils/cogs.py

107 lines
4.5 KiB
Python

"""Cog wrapper for command caching."""
import logging
from jarvis_core.db import q
from jarvis_core.db.models import Ban, Kick, Mute, Setting, Warning
from naff import Client, Cog, InteractionContext
from naff.models.discord.components import ActionRow, Button, ButtonStyles
from naff.models.discord.embed import EmbedField
from jarvis.utils import build_embed
MODLOG_LOOKUP = {"Ban": Ban, "Kick": Kick, "Mute": Mute, "Warning": Warning}
IGNORE_COMMANDS = {"Ban": ["bans"], "Kick": [], "Mute": ["unmute"], "Warning": ["warnings"]}
class ModcaseCog(Cog):
"""Cog wrapper for moderation case logging."""
def __init__(self, bot: Client):
self.bot = bot
self.logger = logging.getLogger(__name__)
self.add_cog_postrun(self.log)
async def log(self, ctx: InteractionContext, *_args: list, **kwargs: dict) -> None:
"""
Log a moderation activity in a moderation case.
Args:
ctx: Command context
"""
name = self.__name__.replace("Cog", "")
if name in MODLOG_LOOKUP and ctx.command not in IGNORE_COMMANDS[name]:
user = kwargs.pop("user", None)
if not user and not ctx.target_id:
self.logger.warning("Admin action %s missing user, exiting", name)
return
if ctx.target_id:
user = ctx.target
coll = MODLOG_LOOKUP.get(name, None)
if not coll:
self.logger.warning("Unsupported action %s, exiting", name)
return
action = await coll.find_one(q(user=user.id, guild=ctx.guild_id, active=True))
if not action:
self.logger.warning("Missing action %s, exiting", name)
return
notify = await Setting.find_one(q(guild=ctx.guild.id, setting="notify", value=True))
if notify and name not in ("Kick", "Ban"): # Ignore Kick and Ban, as these are unique
fields = (
EmbedField(name="Action Type", value=name, inline=False),
EmbedField(
name="Reason", value=kwargs.get("reason", None) or "N/A", inline=False
),
)
embed = build_embed(
title="Admin action taken",
description=f"Admin action has been taken against you in {ctx.guild.name}",
fields=fields,
)
guild_url = f"https://discord.com/channels/{ctx.guild.id}"
embed.set_author(name=ctx.guild.name, icon_url=ctx.guild.icon.url, url=guild_url)
embed.set_thumbnail(url=ctx.guild.icon.url)
try:
await user.send(embed=embed)
except Exception:
self.logger.debug("User not warned of action due to closed DMs")
lookup_key = f"{user.id}|{ctx.guild.id}"
async with self.bot.redis.lock("lock|" + lookup_key):
if await self.bot.redis.get(lookup_key):
self.logger.debug(f"User {user.id} in {ctx.guild.id} already has pending case")
return
modlog = await Setting.find_one(q(guild=ctx.guild.id, setting="modlog"))
if not modlog:
return
channel = await ctx.guild.fetch_channel(modlog.value)
if not channel:
self.logger.warn(
f"Guild {ctx.guild.id} modlog channel no longer exists, deleting"
)
await modlog.delete()
return
embed = build_embed(
title="Recent Action Taken",
description="Would you like to open a moderation case for {user.mention}?",
fields=[],
)
embed.set_author(
name=user.username + "#" + user.discriminator, icon_url=user.display_avatar.url
)
components = [
ActionRow(
Button(style=ButtonStyles.RED, emoji="✖️", custom_id="modcase|no"),
Button(style=ButtonStyles.GREEN, emoji="✔️", custom_id="modcase|yes"),
)
]
message = await ctx.send(embed=embed, components=components)
await self.bot.redis.set(lookup_key, f"{name.lower()}|{action.id}")
await self.bot.redis.set(f"msg|{message.id}", user.id)