feat(silence): allow to silence threads

bot/exts/moderation/silence.py

@@ -2,9 +2,11 @@
 from collections import OrderedDict
 from contextlib import suppress
 from datetime import UTC, datetime, timedelta
+from typing import Any
 
 from async_rediscache import RedisCache
 from discord import Guild, PermissionOverwrite, TextChannel, Thread, VoiceChannel
+from discord.abc import Messageable
 from discord.ext import commands, tasks
 from discord.ext.commands import Context
 from discord.utils import MISSING
@@ -33,6 +35,7 @@
 MSG_UNSILENCE_SUCCESS = f"{constants.Emojis.check_mark} unsilenced {{channel}}."
 
 TextOrVoiceChannel = TextChannel | VoiceChannel
+SilenceableChannel = TextChannel | VoiceChannel | Thread
 
 VOICE_CHANNELS = {
     constants.Channels.code_help_voice_0: constants.Channels.code_help_chat_0,
@@ -57,17 +60,17 @@ def __init__(self, alert_channel: TextChannel):
             time=MISSING,
             name=None
         )
-        self._silenced_channels = {}
+        self._silenced_channels: dict[SilenceableChannel, int] = {}
         self._alert_channel = alert_channel
 
-    def add_channel(self, channel: TextOrVoiceChannel) -> None:
+    def add_channel(self, channel: SilenceableChannel) -> None:
         """Add channel to `_silenced_channels` and start loop if not launched."""
         if not self._silenced_channels:
             self.start()
             log.info("Starting notifier loop.")
         self._silenced_channels[channel] = self._current_loop
 
-    def remove_channel(self, channel: TextChannel) -> None:
+    def remove_channel(self, channel: SilenceableChannel) -> None:
         """Remove channel from `_silenced_channels` and stop loop if no channels remain."""
         with suppress(KeyError):
             del self._silenced_channels[channel]
@@ -92,7 +95,7 @@ async def _notifier(self) -> None:
             )
 
 
-async def _select_lock_channel(args: OrderedDict[str, any]) -> TextOrVoiceChannel:
+async def _select_lock_channel(args: OrderedDict[str, Any]) -> SilenceableChannel:
     """Passes the channel to be silenced to the resource lock."""
     channel, _ = Silence.parse_silence_args(args["ctx"], args["duration_or_channel"], args["duration"])
     return channel
@@ -130,8 +133,8 @@ async def cog_load(self) -> None:
     async def send_message(
         self,
         message: str,
-        source_channel: TextChannel,
-        target_channel: TextOrVoiceChannel,
+        source_channel: Messageable,
+        target_channel: SilenceableChannel,
         *,
         alert_target: bool = False
     ) -> None:
@@ -159,7 +162,7 @@ async def send_message(
     async def silence(
         self,
         ctx: Context,
-        duration_or_channel: TextOrVoiceChannel | HushDurationConverter = None,
+        duration_or_channel: SilenceableChannel | HushDurationConverter | None = None,
         duration: HushDurationConverter = 10,
         *,
         kick: bool = False
@@ -178,12 +181,6 @@ async def silence(
         channel_info = f"#{channel} ({channel.id})"
         log.debug(f"{ctx.author} is silencing channel {channel_info}.")
 
-        # Since threads don't have specific overrides, we cannot silence them individually.
-        # The parent channel has to be muted or the thread should be archived.
-        if isinstance(channel, Thread):
-            await ctx.send(":x: Threads cannot be silenced.")
-            return
-
         if not await self._set_silence_overwrites(channel, kick=kick):
             log.info(f"Tried to silence channel {channel_info} but the channel was already silenced.")
             await self.send_message(MSG_SILENCE_FAIL, ctx.channel, channel, alert_target=False)
@@ -210,12 +207,12 @@ async def silence(
     @staticmethod
     def parse_silence_args(
         ctx: Context,
-        duration_or_channel: TextOrVoiceChannel | int,
+        duration_or_channel: SilenceableChannel | int | None,
         duration: HushDurationConverter
-    ) -> tuple[TextOrVoiceChannel, int | None]:
+    ) -> tuple[SilenceableChannel, int | None]:
         """Helper method to parse the arguments of the silence command."""
         if duration_or_channel:
-            if isinstance(duration_or_channel, TextChannel | VoiceChannel):
+            if isinstance(duration_or_channel, TextChannel | VoiceChannel | Thread):
                 channel = duration_or_channel
             else:
                 channel = ctx.channel
@@ -228,8 +225,11 @@ def parse_silence_args(
 
         return channel, duration
 
-    async def _set_silence_overwrites(self, channel: TextOrVoiceChannel, *, kick: bool = False) -> bool:
+    async def _set_silence_overwrites(self, channel: SilenceableChannel, *, kick: bool = False) -> bool:
         """Set silence permission overwrites for `channel` and return True if successful."""
+        if channel.id in self.scheduler:
+            return False
+
         # Get the original channel overwrites
         if isinstance(channel, TextChannel):
             role = self._everyone_role
@@ -242,6 +242,11 @@ async def _set_silence_overwrites(self, channel: TextOrVoiceChannel, *, kick: bo
                 send_messages_in_threads=overwrite.send_messages_in_threads
             )
 
+        elif isinstance(channel, Thread):
+            log.info(f"Silencing thread #{channel.parent}/{channel} ({channel.parent.id}/{channel.id}).")
+            await channel.edit(locked=True)
+            return True
+
         else:
             role = self._verified_voice_role
             overwrite = channel.overwrites_for(role)
@@ -250,7 +255,7 @@ async def _set_silence_overwrites(self, channel: TextOrVoiceChannel, *, kick: bo
                 prev_overwrites.update(connect=overwrite.connect)
 
         # Stop if channel was already silenced
-        if channel.id in self.scheduler or all(val is False for val in prev_overwrites.values()):
+        if all(val is False for val in prev_overwrites.values()):
             return False
 
         # Set new permissions, store
@@ -260,7 +265,7 @@ async def _set_silence_overwrites(self, channel: TextOrVoiceChannel, *, kick: bo
 
         return True
 
-    async def _schedule_unsilence(self, ctx: Context, channel: TextOrVoiceChannel, duration: int | None) -> None:
+    async def _schedule_unsilence(self, ctx: Context, channel: SilenceableChannel, duration: int | None) -> None:
         """Schedule `ctx.channel` to be unsilenced if `duration` is not None."""
         if duration is None:
             await self.unsilence_timestamps.set(channel.id, -1)
@@ -270,7 +275,7 @@ async def _schedule_unsilence(self, ctx: Context, channel: TextOrVoiceChannel, d
             await self.unsilence_timestamps.set(channel.id, unsilence_time.timestamp())
 
     @commands.command(aliases=("unhush",))
-    async def unsilence(self, ctx: Context, *, channel: TextOrVoiceChannel = None) -> None:
+    async def unsilence(self, ctx: Context, *, channel: SilenceableChannel | None = None) -> None:
         """
         Unsilence the given channel if given, else the current one.
 
@@ -282,7 +287,7 @@ async def unsilence(self, ctx: Context, *, channel: TextOrVoiceChannel = None) -
         await self._unsilence_wrapper(channel, ctx)
 
     @lock_arg(LOCK_NAMESPACE, "channel", raise_error=True)
-    async def _unsilence_wrapper(self, channel: TextOrVoiceChannel, ctx: Context | None = None) -> None:
+    async def _unsilence_wrapper(self, channel: SilenceableChannel, ctx: Context | None = None) -> None:
         """
         Unsilence `channel` and send a success/failure message to ctx.channel.
 
@@ -297,6 +302,10 @@ async def _unsilence_wrapper(self, channel: TextOrVoiceChannel, ctx: Context | N
             if isinstance(channel, VoiceChannel):
                 overwrite = channel.overwrites_for(self._verified_voice_role)
                 has_channel_overwrites = overwrite.speak is False
+
+            elif isinstance(channel, Thread):
+                has_channel_overwrites = False
+
             else:
                 overwrite = channel.overwrites_for(self._everyone_role)
                 has_channel_overwrites = overwrite.send_messages is False or overwrite.add_reactions is False
@@ -310,15 +319,15 @@ async def _unsilence_wrapper(self, channel: TextOrVoiceChannel, ctx: Context | N
         else:
             await self.send_message(MSG_UNSILENCE_SUCCESS, msg_channel, channel, alert_target=True)
 
-    async def _unsilence(self, channel: TextOrVoiceChannel) -> bool:
+    async def _unsilence(self, channel: SilenceableChannel) -> bool:
         """
         Unsilence `channel`.
 
         If `channel` has a silence task scheduled or has its previous overwrites cached, unsilence
         it, cancel the task, and remove it from the notifier. Notify admins if it has a task but
         not cached overwrites.
 
-        Return `True` if channel permissions were changed, `False` otherwise.
+        Return `True` if channel was unsilenced, `False` otherwise.
         """
         # Get stored overwrites, and return if channel is unsilenced
         prev_overwrites = await self.previous_overwrites.get(channel.id)
@@ -331,6 +340,12 @@ async def _unsilence(self, channel: TextOrVoiceChannel) -> bool:
             role = self._everyone_role
             overwrite = channel.overwrites_for(role)
             permissions = "`Send Messages` and `Add Reactions`"
+        elif isinstance(channel, Thread):
+            await channel.edit(locked=False)
+            log.info(f"Unsilenced thread #{channel.parent}/{channel} ({channel.parent_id}/{channel.id}).")
+            self.scheduler.cancel(channel.id)
+            self.notifier.remove_channel(channel)
+            return True
         else:
             role = self._verified_voice_role
             overwrite = channel.overwrites_for(role)