Automod improvements (#41)

* Add normalization to `message_content_contains` condition

* Adjust `message_content_matches` condition

* Adjust docstrings for some conditions

* Implement ValueFormatter for custom event field formatting

* Adjust docstring for AutomodLogOptions

* Add extra fields to member join/leave events

* Implement `log_message` action

* Implement role-based (per-guild) permissions

* Account for unresolved roles

* Disable pings by default in ResponsiveException

* Adjust AutomodEvent member fields

* Implement custom AllowedMentions for actions

* Clarify docstring for `log_message` action

* Update changelog

Author: Arcensoth

CHANGELOG.md

@@ -4,6 +4,17 @@ All notable changes to this project will be documented in this file.
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Changed
+
+- `automod` improvements (#41):
+  - Implemented role-based (per-guild) permissions
+  - Added a new `log_message` action that suppresses pings by default
+  - Pings are now suppressed by default in error messages
+  - Added normalization to the `message_content_contains` condition
+  - Added more fields to some events for string formatting
+
 ## [0.13.0] - 2021-08-22
 
 ### Added

commanderbot_ext/ext/automod/actions/abc/add_roles_to_target_base.py

@@ -18,5 +18,7 @@ def get_target(self, event: AutomodEvent) -> Optional[Member]:
     async def apply(self, event: AutomodEvent):
         if member := self.get_target(event):
             guild: Guild = member.guild
+            # TODO Warn about unresolved roles. #logging
             roles = [guild.get_role(role_id) for role_id in self.roles]
+            roles = [role for role in roles if role]
             await member.add_roles(roles)

commanderbot_ext/ext/automod/actions/abc/remove_roles_from_target_base.py

@@ -18,5 +18,7 @@ def get_target(self, event: AutomodEvent) -> Optional[Member]:
     async def apply(self, event: AutomodEvent):
         if member := self.get_target(event):
             guild: Guild = member.guild
+            # TODO Warn about unresolved roles. #logging
             roles = [guild.get_role(role_id) for role_id in self.roles]
+            roles = [role for role in roles if role]
             await member.remove_roles(roles)

commanderbot_ext/ext/automod/actions/log_message.py

@@ -0,0 +1,90 @@
+from dataclasses import dataclass, field
+from typing import Dict, Optional, Type, TypeVar
+
+from discord import Color
+from discord.abc import Messageable
+
+from commanderbot_ext.ext.automod.automod_action import AutomodAction, AutomodActionBase
+from commanderbot_ext.ext.automod.automod_event import AutomodEvent
+from commanderbot_ext.lib import AllowedMentions, ChannelID, JsonObject, ValueFormatter
+from commanderbot_ext.lib.utils import color_from_field_optional
+
+ST = TypeVar("ST")
+
+
+@dataclass
+class LogMessage(AutomodActionBase):
+    """
+    Send a log message, with pings disabled by default.
+
+    Attributes
+    ----------
+    content
+        The content of the message to send.
+    channel
+        The channel to send the message in. Defaults to the channel in context.
+    emoji
+        The emoji used to represent the type of message.
+    color
+        The emoji used to represent the type of message.
+    fields
+        A custom set of fields to display as part of the message. The key should
+        correspond to an event field, and the value is the title to use for it.
+    allowed_mentions
+        The types of mentions allowed in the message. Unless otherwise specified, all
+        mentions will be suppressed.
+    """
+
+    content: Optional[str] = None
+    channel: Optional[ChannelID] = None
+    emoji: Optional[str] = None
+    color: Optional[Color] = None
+    fields: Optional[Dict[str, str]] = None
+    allowed_mentions: Optional[AllowedMentions] = None
+
+    @classmethod
+    def from_data(cls: Type[ST], data: JsonObject) -> ST:
+        color = color_from_field_optional(data, "color")
+        allowed_mentions = AllowedMentions.from_field_optional(data, "allowed_mentions")
+        return cls(
+            description=data.get("description"),
+            content=data.get("content"),
+            channel=data.get("channel"),
+            emoji=data.get("emoji"),
+            color=color,
+            fields=data.get("fields"),
+            allowed_mentions=allowed_mentions,
+        )
+
+    async def resolve_channel(self, event: AutomodEvent) -> Optional[Messageable]:
+        if self.channel is not None:
+            return event.bot.get_channel(self.channel)
+        return event.channel
+
+    async def apply(self, event: AutomodEvent):
+        if channel := await self.resolve_channel(event):
+            parts = []
+            if self.emoji:
+                parts.append(self.emoji)
+            if self.content:
+                formatted_content = event.format_content(self.content)
+                parts.append(formatted_content)
+            if self.fields:
+                event_fields = event.get_fields()
+                field_parts = []
+                for field_key, field_title in self.fields.items():
+                    field_value = event_fields.get(field_key)
+                    if isinstance(field_value, ValueFormatter):
+                        field_value_str = str(field_value)
+                    else:
+                        field_value_str = f"`{field_value}`"
+                    field_parts.append(f"{field_title} {field_value_str}")
+                fields_str = ", ".join(field_parts)
+                parts.append(fields_str)
+            content = " ".join(parts)
+            allowed_mentions = self.allowed_mentions or AllowedMentions.none()
+            await channel.send(content, allowed_mentions=allowed_mentions)
+
+
+def create_action(data: JsonObject) -> AutomodAction:
+    return LogMessage.from_data(data)

commanderbot_ext/ext/automod/actions/send_message.py

@@ -1,11 +1,13 @@
-from dataclasses import dataclass
-from typing import Optional
+from dataclasses import dataclass, field
+from typing import Optional, Type, TypeVar
 
 from discord.abc import Messageable
 
 from commanderbot_ext.ext.automod.automod_action import AutomodAction, AutomodActionBase
 from commanderbot_ext.ext.automod.automod_event import AutomodEvent
-from commanderbot_ext.lib import ChannelID, JsonObject
+from commanderbot_ext.lib import AllowedMentions, ChannelID, JsonObject
+
+ST = TypeVar("ST")
 
 
 @dataclass
@@ -19,11 +21,24 @@ class SendMessage(AutomodActionBase):
         The content of the message to send.
     channel
         The channel to send the message in. Defaults to the channel in context.
+    allowed_mentions
+        The types of mentions allowed in the message. Unless otherwise specified, only
+        "everyone" mentions will be suppressed.
     """
 
     content: str
-
     channel: Optional[ChannelID] = None
+    allowed_mentions: Optional[AllowedMentions] = None
+
+    @classmethod
+    def from_data(cls: Type[ST], data: JsonObject) -> ST:
+        allowed_mentions = AllowedMentions.from_field_optional(data, "allowed_mentions")
+        return cls(
+            description=data.get("description"),
+            content=data.get("content"),
+            channel=data.get("channel"),
+            allowed_mentions=allowed_mentions,
+        )
 
     async def resolve_channel(self, event: AutomodEvent) -> Optional[Messageable]:
         if self.channel is not None:
@@ -33,7 +48,11 @@ async def resolve_channel(self, event: AutomodEvent) -> Optional[Messageable]:
     async def apply(self, event: AutomodEvent):
         if channel := await self.resolve_channel(event):
             content = event.format_content(self.content)
-            await channel.send(content)
+            allowed_mentions = self.allowed_mentions or AllowedMentions.not_everyone()
+            await channel.send(
+                content,
+                allowed_mentions=allowed_mentions,
+            )
 
 
 def create_action(data: JsonObject) -> AutomodAction:

commanderbot_ext/ext/automod/automod_cog.py

@@ -10,10 +10,12 @@
     RawMessageUpdateEvent,
     RawReactionActionEvent,
     Reaction,
+    Role,
     TextChannel,
     User,
 )
 from discord.abc import Messageable
+from discord.ext import commands
 from discord.ext.commands import Bot, Cog, group
 
 from commanderbot_ext.ext.automod.automod_data import AutomodData
@@ -53,6 +55,17 @@ def make_automod_store(bot: Bot, cog: Cog, options: AutomodOptions) -> AutomodSt
     raise UnsupportedDatabaseOptions(db_options)
 
 
+def check_can_run_automod():
+    async def predicate(ctx: GuildContext):
+        cog = ctx.cog
+        actor = ctx.author
+        if isinstance(cog, AutomodCog) and isinstance(actor, Member):
+            return await cog.state[ctx.guild].member_has_permission(actor)
+        return False
+
+    return commands.check(predicate)
+
+
 class AutomodCog(Cog, name="commanderbot_ext.ext.automod"):
     """
     Automate a variety of moderation tasks.
@@ -254,7 +267,11 @@ async def on_raw_reaction_remove(self, payload: RawReactionActionEvent):
         aliases=["am"],
     )
     @checks.guild_only()
-    @checks.is_administrator()
+    @commands.check_any(
+        checks.is_administrator(),
+        check_can_run_automod(),
+        commands.is_owner(),
+    )
     async def cmd_automod(self, ctx: GuildContext):
         if not ctx.invoked_subcommand:
             await ctx.send_help(self.cmd_automod)
@@ -269,17 +286,26 @@ async def cmd_automod_options(self, ctx: GuildContext):
         if not ctx.invoked_subcommand:
             await ctx.send_help(self.cmd_automod_options)
 
+    # @@ automod options log
+
     @cmd_automod_options.group(
         name="log",
         brief="Configure the default logging behaviour.",
     )
     async def cmd_automod_options_log(self, ctx: GuildContext):
         if not ctx.invoked_subcommand:
             if ctx.subcommand_passed:
-                await ctx.send_help(self.cmd_automod_options)
+                await ctx.send_help(self.cmd_automod_options_log)
             else:
                 await self.state[ctx.guild].show_default_log_options(ctx)
 
+    @cmd_automod_options_log.command(
+        name="show",
+        brief="Show the default logging behaviour.",
+    )
+    async def cmd_automod_options_log_show(self, ctx: GuildContext):
+        await self.state[ctx.guild].show_default_log_options(ctx)
+
     @cmd_automod_options_log.command(
         name="set",
         brief="Set the default logging behaviour.",
@@ -307,6 +333,43 @@ async def cmd_automod_options_log_set(
     async def cmd_automod_options_log_remove(self, ctx: GuildContext):
         await self.state[ctx.guild].remove_default_log_options(ctx)
 
+    # @@ automod options permit
+
+    # NOTE Only guild admins and bot owners can manage permitted roles.
+
+    @cmd_automod_options.group(
+        name="permit",
+        brief="Configure the set of roles permitted to manage automod.",
+    )
+    @checks.is_guild_admin_or_bot_owner()
+    async def cmd_automod_options_permit(self, ctx: GuildContext):
+        if not ctx.invoked_subcommand:
+            if ctx.subcommand_passed:
+                await ctx.send_help(self.cmd_automod_options_permit)
+            else:
+                await self.state[ctx.guild].show_permitted_roles(ctx)
+
+    @cmd_automod_options_permit.command(
+        name="show",
+        brief="Show the roles permitted to manage automod.",
+    )
+    async def cmd_automod_options_permit_show(self, ctx: GuildContext):
+        await self.state[ctx.guild].show_permitted_roles(ctx)
+
+    @cmd_automod_options_permit.command(
+        name="set",
+        brief="Set the roles permitted to manage automod.",
+    )
+    async def cmd_automod_options_permit_set(self, ctx: GuildContext, *roles: Role):
+        await self.state[ctx.guild].set_permitted_roles(ctx, *roles)
+
+    @cmd_automod_options_permit.command(
+        name="clear",
+        brief="Clear all roles permitted to manage automod.",
+    )
+    async def cmd_automod_options_permit_clear(self, ctx: GuildContext):
+        await self.state[ctx.guild].clear_permitted_roles(ctx)
+
     # @@ automod rules
 
     @cmd_automod.group(

commanderbot_ext/ext/automod/automod_data.py

@@ -5,12 +5,12 @@
 from datetime import datetime
 from typing import AsyncIterable, DefaultDict, Dict, Iterable, Optional, Set, Type
 
-from discord import Guild
+from discord import Guild, Member
 
 from commanderbot_ext.ext.automod.automod_event import AutomodEvent
 from commanderbot_ext.ext.automod.automod_log_options import AutomodLogOptions
 from commanderbot_ext.ext.automod.automod_rule import AutomodRule
-from commanderbot_ext.lib import GuildID, JsonObject, ResponsiveException
+from commanderbot_ext.lib import GuildID, JsonObject, ResponsiveException, RoleSet
 from commanderbot_ext.lib.json import to_data
 from commanderbot_ext.lib.utils import dict_without_ellipsis
 
@@ -51,20 +51,27 @@ def __init__(self, names: Set[str]):
 
 @dataclass
 class AutomodGuildData:
+    # Default logging configuration for this guild.
     default_log_options: Optional[AutomodLogOptions] = None
 
-    # Index rules by name for fast look-up in commands.
+    # Roles that are permitted to manage the extension within this guild.
+    permitted_roles: Optional[RoleSet] = None
+
+    # Index rules by name for faster look-up in commands.
     rules: Dict[str, AutomodRule] = field(init=False, default_factory=dict)
 
-    # Index rules by event type for faster initial access.
+    # Group rules by event type for faster look-up during event dispatch.
     rules_by_event_type: RulesByEventType = field(
         init=False, default_factory=lambda: defaultdict(lambda: set())
     )
 
     @staticmethod
     def from_data(data: JsonObject) -> AutomodGuildData:
+        default_log_options = AutomodLogOptions.from_field_optional(data, "log")
+        permitted_roles = RoleSet.from_field_optional(data, "permitted_roles")
         guild_data = AutomodGuildData(
-            default_log_options=AutomodLogOptions.from_field_optional(data, "log"),
+            default_log_options=default_log_options,
+            permitted_roles=permitted_roles,
         )
         for rule_data in data.get("rules", []):
             rule = AutomodRule.from_data(rule_data)
@@ -74,15 +81,29 @@ def from_data(data: JsonObject) -> AutomodGuildData:
     def to_data(self) -> JsonObject:
         return dict_without_ellipsis(
             log=self.default_log_options or ...,
+            permitted_roles=self.permitted_roles or ...,
             rules=list(self.rules.values()) or ...,
         )
 
     def set_default_log_options(
         self, log_options: Optional[AutomodLogOptions]
     ) -> Optional[AutomodLogOptions]:
-        old_log_options = self.default_log_options
+        old_value = self.default_log_options
         self.default_log_options = log_options
-        return old_log_options
+        return old_value
+
+    def set_permitted_roles(
+        self, permitted_roles: Optional[RoleSet]
+    ) -> Optional[RoleSet]:
+        old_value = self.permitted_roles
+        self.permitted_roles = permitted_roles
+        return old_value
+
+    def member_has_permission(self, member: Member) -> bool:
+        if self.permitted_roles is None:
+            return False
+        has_permission = self.permitted_roles.member_has_some(member)
+        return has_permission
 
     def all_rules(self) -> Iterable[AutomodRule]:
         yield from self.rules.values()
@@ -224,6 +245,20 @@ async def set_default_log_options(
     ) -> Optional[AutomodLogOptions]:
         return self.guilds[guild.id].set_default_log_options(log_options)
 
+    # @implements AutomodStore
+    async def get_permitted_roles(self, guild: Guild) -> Optional[RoleSet]:
+        return self.guilds[guild.id].permitted_roles
+
+    # @implements AutomodStore
+    async def set_permitted_roles(
+        self, guild: Guild, permitted_roles: Optional[RoleSet]
+    ) -> Optional[RoleSet]:
+        return self.guilds[guild.id].set_permitted_roles(permitted_roles)
+
+    # @implements AutomodStore
+    async def member_has_permission(self, guild: Guild, member: Member) -> bool:
+        return self.guilds[guild.id].member_has_permission(member)
+
     # @implements AutomodStore
     async def all_rules(self, guild: Guild) -> AsyncIterable[AutomodRule]:
         for rule in self.guilds[guild.id].all_rules():