diff options
Diffstat (limited to 'bot/exts/moderation/infraction/infractions.py')
| -rw-r--r-- | bot/exts/moderation/infraction/infractions.py | 114 |
1 files changed, 70 insertions, 44 deletions
diff --git a/bot/exts/moderation/infraction/infractions.py b/bot/exts/moderation/infraction/infractions.py index f19323c7c..eaba97703 100644 --- a/bot/exts/moderation/infraction/infractions.py +++ b/bot/exts/moderation/infraction/infractions.py @@ -10,11 +10,10 @@ from discord.ext.commands import Context, command from bot import constants from bot.bot import Bot from bot.constants import Event -from bot.converters import Duration, Expiry, FetchedMember +from bot.converters import Duration, Expiry, MemberOrUser, UnambiguousMemberOrUser from bot.decorators import respect_role_hierarchy from bot.exts.moderation.infraction import _utils from bot.exts.moderation.infraction._scheduler import InfractionScheduler -from bot.exts.moderation.infraction._utils import UserSnowflake from bot.utils.messages import format_user log = logging.getLogger(__name__) @@ -54,7 +53,7 @@ class Infractions(InfractionScheduler, commands.Cog): # region: Permanent infractions @command() - async def warn(self, ctx: Context, user: FetchedMember, *, reason: t.Optional[str] = None) -> None: + async def warn(self, ctx: Context, user: UnambiguousMemberOrUser, *, reason: t.Optional[str] = None) -> None: """Warn a user for the given reason.""" if not isinstance(user, Member): await ctx.send(":x: The user doesn't appear to be on the server.") @@ -67,7 +66,7 @@ class Infractions(InfractionScheduler, commands.Cog): await self.apply_infraction(ctx, infraction, user) @command() - async def kick(self, ctx: Context, user: FetchedMember, *, reason: t.Optional[str] = None) -> None: + async def kick(self, ctx: Context, user: UnambiguousMemberOrUser, *, reason: t.Optional[str] = None) -> None: """Kick a user for the given reason.""" if not isinstance(user, Member): await ctx.send(":x: The user doesn't appear to be on the server.") @@ -79,7 +78,7 @@ class Infractions(InfractionScheduler, commands.Cog): async def ban( self, ctx: Context, - user: FetchedMember, + user: UnambiguousMemberOrUser, duration: t.Optional[Expiry] = None, *, reason: t.Optional[str] = None @@ -95,7 +94,7 @@ class Infractions(InfractionScheduler, commands.Cog): async def purgeban( self, ctx: Context, - user: FetchedMember, + user: UnambiguousMemberOrUser, duration: t.Optional[Expiry] = None, *, reason: t.Optional[str] = None @@ -111,7 +110,7 @@ class Infractions(InfractionScheduler, commands.Cog): async def voiceban( self, ctx: Context, - user: FetchedMember, + user: UnambiguousMemberOrUser, duration: t.Optional[Expiry] = None, *, reason: t.Optional[str] @@ -129,7 +128,7 @@ class Infractions(InfractionScheduler, commands.Cog): @command(aliases=["mute"]) async def tempmute( self, ctx: Context, - user: FetchedMember, + user: UnambiguousMemberOrUser, duration: t.Optional[Expiry] = None, *, reason: t.Optional[str] = None @@ -163,7 +162,7 @@ class Infractions(InfractionScheduler, commands.Cog): async def tempban( self, ctx: Context, - user: FetchedMember, + user: UnambiguousMemberOrUser, duration: Expiry, *, reason: t.Optional[str] = None @@ -189,7 +188,7 @@ class Infractions(InfractionScheduler, commands.Cog): async def tempvoiceban( self, ctx: Context, - user: FetchedMember, + user: UnambiguousMemberOrUser, duration: Expiry, *, reason: t.Optional[str] @@ -215,7 +214,7 @@ class Infractions(InfractionScheduler, commands.Cog): # region: Permanent shadow infractions @command(hidden=True) - async def note(self, ctx: Context, user: FetchedMember, *, reason: t.Optional[str] = None) -> None: + async def note(self, ctx: Context, user: UnambiguousMemberOrUser, *, reason: t.Optional[str] = None) -> None: """Create a private note for a user with the given reason without notifying the user.""" infraction = await _utils.post_infraction(ctx, user, "note", reason, hidden=True, active=False) if infraction is None: @@ -224,7 +223,7 @@ class Infractions(InfractionScheduler, commands.Cog): await self.apply_infraction(ctx, infraction, user) @command(hidden=True, aliases=['shadowban', 'sban']) - async def shadow_ban(self, ctx: Context, user: FetchedMember, *, reason: t.Optional[str] = None) -> None: + async def shadow_ban(self, ctx: Context, user: UnambiguousMemberOrUser, *, reason: t.Optional[str] = None) -> None: """Permanently ban a user for the given reason without notifying the user.""" await self.apply_ban(ctx, user, reason, hidden=True) @@ -235,7 +234,7 @@ class Infractions(InfractionScheduler, commands.Cog): async def shadow_tempban( self, ctx: Context, - user: FetchedMember, + user: UnambiguousMemberOrUser, duration: Expiry, *, reason: t.Optional[str] = None @@ -261,17 +260,17 @@ class Infractions(InfractionScheduler, commands.Cog): # region: Remove infractions (un- commands) @command() - async def unmute(self, ctx: Context, user: FetchedMember) -> None: + async def unmute(self, ctx: Context, user: UnambiguousMemberOrUser) -> None: """Prematurely end the active mute infraction for the user.""" await self.pardon_infraction(ctx, "mute", user) @command() - async def unban(self, ctx: Context, user: FetchedMember) -> None: + async def unban(self, ctx: Context, user: UnambiguousMemberOrUser) -> None: """Prematurely end the active ban infraction for the user.""" await self.pardon_infraction(ctx, "ban", user) @command(aliases=("uvban",)) - async def unvoiceban(self, ctx: Context, user: FetchedMember) -> None: + async def unvoiceban(self, ctx: Context, user: UnambiguousMemberOrUser) -> None: """Prematurely end the active voice ban infraction for the user.""" await self.pardon_infraction(ctx, "voice_ban", user) @@ -280,8 +279,19 @@ class Infractions(InfractionScheduler, commands.Cog): async def apply_mute(self, ctx: Context, user: Member, reason: t.Optional[str], **kwargs) -> None: """Apply a mute infraction with kwargs passed to `post_infraction`.""" - if await _utils.get_active_infraction(ctx, user, "mute"): - return + if active := await _utils.get_active_infraction(ctx, user, "mute", send_msg=False): + if active["actor"] != self.bot.user.id: + await _utils.send_active_infraction_message(ctx, active) + return + + # Allow the current mute attempt to override an automatically triggered mute. + log_text = await self.deactivate_infraction(active, notify=False) + if "Failure" in log_text: + await ctx.send( + f":x: can't override infraction **mute** for {user.mention}: " + f"failed to deactivate. {log_text['Failure']}" + ) + return infraction = await _utils.post_infraction(ctx, user, "mute", reason, active=True, **kwargs) if infraction is None: @@ -320,7 +330,7 @@ class Infractions(InfractionScheduler, commands.Cog): async def apply_ban( self, ctx: Context, - user: UserSnowflake, + user: MemberOrUser, reason: t.Optional[str], purge_days: t.Optional[int] = 0, **kwargs @@ -345,7 +355,7 @@ class Infractions(InfractionScheduler, commands.Cog): return log.trace("Old tempban is being replaced by new permaban.") - await self.pardon_infraction(ctx, "ban", user, is_temporary) + await self.pardon_infraction(ctx, "ban", user, send_msg=is_temporary) infraction = await _utils.post_infraction(ctx, user, "ban", reason, active=True, **kwargs) if infraction is None: @@ -376,7 +386,7 @@ class Infractions(InfractionScheduler, commands.Cog): await bb_cog.apply_unwatch(ctx, user, bb_reason, send_message=False) @respect_role_hierarchy(member_arg=2) - async def apply_voice_ban(self, ctx: Context, user: UserSnowflake, reason: t.Optional[str], **kwargs) -> None: + async def apply_voice_ban(self, ctx: Context, user: MemberOrUser, reason: t.Optional[str], **kwargs) -> None: """Apply a voice ban infraction with kwargs passed to `post_infraction`.""" if await _utils.get_active_infraction(ctx, user, "voice_ban"): return @@ -403,8 +413,15 @@ class Infractions(InfractionScheduler, commands.Cog): # endregion # region: Base pardon functions - async def pardon_mute(self, user_id: int, guild: discord.Guild, reason: t.Optional[str]) -> t.Dict[str, str]: - """Remove a user's muted role, DM them a notification, and return a log dict.""" + async def pardon_mute( + self, + user_id: int, + guild: discord.Guild, + reason: t.Optional[str], + *, + notify: bool = True + ) -> t.Dict[str, str]: + """Remove a user's muted role, optionally DM them a notification, and return a log dict.""" user = guild.get_member(user_id) log_text = {} @@ -413,16 +430,17 @@ class Infractions(InfractionScheduler, commands.Cog): self.mod_log.ignore(Event.member_update, user.id) await user.remove_roles(self._muted_role, reason=reason) - # DM the user about the expiration. - notified = await _utils.notify_pardon( - user=user, - title="You have been unmuted", - content="You may now send messages in the server.", - icon_url=_utils.INFRACTION_ICONS["mute"][1] - ) + if notify: + # DM the user about the expiration. + notified = await _utils.notify_pardon( + user=user, + title="You have been unmuted", + content="You may now send messages in the server.", + icon_url=_utils.INFRACTION_ICONS["mute"][1] + ) + log_text["DM"] = "Sent" if notified else "**Failed**" log_text["Member"] = format_user(user) - log_text["DM"] = "Sent" if notified else "**Failed**" else: log.info(f"Failed to unmute user {user_id}: user not found") log_text["Failure"] = "User was not found in the guild." @@ -444,31 +462,39 @@ class Infractions(InfractionScheduler, commands.Cog): return log_text - async def pardon_voice_ban(self, user_id: int, guild: discord.Guild, reason: t.Optional[str]) -> t.Dict[str, str]: - """Add Voice Verified role back to user, DM them a notification, and return a log dict.""" + async def pardon_voice_ban( + self, + user_id: int, + guild: discord.Guild, + *, + notify: bool = True + ) -> t.Dict[str, str]: + """Optionally DM the user a pardon notification and return a log dict.""" user = guild.get_member(user_id) log_text = {} if user: - # DM user about infraction expiration - notified = await _utils.notify_pardon( - user=user, - title="Voice ban ended", - content="You have been unbanned and can verify yourself again in the server.", - icon_url=_utils.INFRACTION_ICONS["voice_ban"][1] - ) + if notify: + # DM user about infraction expiration + notified = await _utils.notify_pardon( + user=user, + title="Voice ban ended", + content="You have been unbanned and can verify yourself again in the server.", + icon_url=_utils.INFRACTION_ICONS["voice_ban"][1] + ) + log_text["DM"] = "Sent" if notified else "**Failed**" log_text["Member"] = format_user(user) - log_text["DM"] = "Sent" if notified else "**Failed**" else: log_text["Info"] = "User was not found in the guild." return log_text - async def _pardon_action(self, infraction: _utils.Infraction) -> t.Optional[t.Dict[str, str]]: + async def _pardon_action(self, infraction: _utils.Infraction, notify: bool) -> t.Optional[t.Dict[str, str]]: """ Execute deactivation steps specific to the infraction's type and return a log dict. + If `notify` is True, notify the user of the pardon via DM where applicable. If an infraction type is unsupported, return None instead. """ guild = self.bot.get_guild(constants.Guild.id) @@ -476,11 +502,11 @@ class Infractions(InfractionScheduler, commands.Cog): reason = f"Infraction #{infraction['id']} expired or was pardoned." if infraction["type"] == "mute": - return await self.pardon_mute(user_id, guild, reason) + return await self.pardon_mute(user_id, guild, reason, notify=notify) elif infraction["type"] == "ban": return await self.pardon_ban(user_id, guild, reason) elif infraction["type"] == "voice_ban": - return await self.pardon_voice_ban(user_id, guild, reason) + return await self.pardon_voice_ban(user_id, guild, notify=notify) # endregion |