diff options
| -rw-r--r-- | bot/cogs/moderation.py | 283 | ||||
| -rw-r--r-- | bot/utils/moderation.py | 20 |
2 files changed, 149 insertions, 154 deletions
diff --git a/bot/cogs/moderation.py b/bot/cogs/moderation.py index 1dc2c70d6..bec2f98c1 100644 --- a/bot/cogs/moderation.py +++ b/bot/cogs/moderation.py @@ -2,7 +2,7 @@ import asyncio import logging import textwrap from datetime import datetime -from typing import Union +from typing import Dict, Union from discord import ( Colour, Embed, Forbidden, Guild, HTTPException, Member, Object, User @@ -17,7 +17,7 @@ from bot.constants import Colours, Event, Icons, MODERATION_ROLES from bot.converters import ExpirationDate, InfractionSearchQuery from bot.decorators import with_role from bot.pagination import LinePaginator -from bot.utils.moderation import post_infraction +from bot.utils.moderation import already_has_active_infraction, post_infraction from bot.utils.scheduling import Scheduler, create_task from bot.utils.time import wait_until @@ -81,15 +81,11 @@ class Moderation(Scheduler): **`reason`:** The reason for the warning. """ - response_object = await post_infraction(ctx, user, type="warning", reason=reason) - if response_object is None: + infraction = await post_infraction(ctx, user, type="warning", reason=reason) + if infraction is None: return - notified = await self.notify_infraction( - user=user, - infr_type="Warning", - reason=reason - ) + notified = await self.notify_infraction(user=user, infr_type="Warning", reason=reason) dm_result = ":incoming_envelope: " if notified else "" action = f"{dm_result}:ok_hand: warned {user.mention}" @@ -118,7 +114,7 @@ class Moderation(Scheduler): Reason: {reason} """), content=log_content, - footer=f"ID {response_object['infraction']['id']}" + footer=f"ID {infraction['id']}" ) @with_role(*MODERATION_ROLES) @@ -136,15 +132,11 @@ class Moderation(Scheduler): # Warning is sent to ctx by the helper method return - response_object = await post_infraction(ctx, user, type="kick", reason=reason) - if response_object is None: + infraction = await post_infraction(ctx, user, type="kick", reason=reason) + if infraction is None: return - notified = await self.notify_infraction( - user=user, - infr_type="Kick", - reason=reason - ) + notified = await self.notify_infraction(user=user, infr_type="Kick", reason=reason) self.mod_log.ignore(Event.member_remove, user.id) @@ -178,7 +170,7 @@ class Moderation(Scheduler): Reason: {reason} """), content=log_content, - footer=f"ID {response_object['infraction']['id']}" + footer=f"ID {infraction['id']}" ) @with_role(*MODERATION_ROLES) @@ -196,22 +188,11 @@ class Moderation(Scheduler): # Warning is sent to ctx by the helper method return - active_bans = await self.bot.api_client.get( - 'bot/infractions', - params={ - 'active': 'true', - 'type': 'ban', - 'user__id': str(user.id) - } - ) - if active_bans: - return await ctx.send( - ":x: According to my records, this user is already banned. " - f"See infraction **#{active_bans[0]['id']}**." - ) + if await already_has_active_infraction(ctx=ctx, user=user, type="ban"): + return - response_object = await post_infraction(ctx, user, type="ban", reason=reason) - if response_object is None: + infraction = await post_infraction(ctx, user, type="ban", reason=reason) + if infraction is None: return notified = await self.notify_infraction( @@ -255,7 +236,7 @@ class Moderation(Scheduler): Reason: {reason} """), content=log_content, - footer=f"ID {response_object['infraction']['id']}" + footer=f"ID {infraction['id']}" ) @with_role(*MODERATION_ROLES) @@ -268,22 +249,11 @@ class Moderation(Scheduler): **`reason`:** The reason for the mute. """ - active_mutes = await self.bot.api_client.get( - 'bot/infractions', - params={ - 'active': 'true', - 'type': 'mute', - 'user__id': str(user.id) - } - ) - if active_mutes: - return await ctx.send( - ":x: According to my records, this user is already muted. " - f"See infraction **#{active_mutes[0]['id']}**." - ) + if await already_has_active_infraction(ctx=ctx, user=user, type="mute"): + return - response_object = await post_infraction(ctx, user, type="mute", reason=reason) - if response_object is None: + infraction = await post_infraction(ctx, user, type="mute", reason=reason) + if infraction is None: return self.mod_log.ignore(Event.member_update, user.id) @@ -323,7 +293,7 @@ class Moderation(Scheduler): Reason: {reason} """), content=log_content, - footer=f"ID {response_object['infraction']['id']}" + footer=f"ID {infraction['id']}" ) # endregion @@ -331,10 +301,7 @@ class Moderation(Scheduler): @with_role(*MODERATION_ROLES) @command() - async def tempmute( - self, ctx: Context, user: Member, expiration: ExpirationDate, - *, reason: str = None - ): + async def tempmute(self, ctx: Context, user: Member, duration: ExpirationDate, *, reason: str = None) -> None: """ Create a temporary mute infraction in the database for a user. @@ -342,26 +309,14 @@ class Moderation(Scheduler): **`duration`:** The duration for the temporary mute infraction **`reason`:** The reason for the temporary mute. """ + expiration = duration - active_mutes = await self.bot.api_client.get( - 'bot/infractions', - params={ - 'active': 'true', - 'type': 'mute', - 'user__id': str(user.id) - } - ) - if active_mutes: - return await ctx.send( - ":x: According to my records, this user is already muted. " - f"See infraction **#{active_mutes[0]['id']}**." - ) + if await already_has_active_infraction(ctx=ctx, user=user, type="mute"): + return - infraction = await post_infraction( - ctx, user, - type="mute", reason=reason, - expires_at=expiration - ) + infraction = await post_infraction(ctx, user, type="mute", reason=reason, expires_at=expiration) + if infraction is None: + return self.mod_log.ignore(Event.member_update, user.id) await user.add_roles(self._muted_role, reason=reason) @@ -414,47 +369,32 @@ class Moderation(Scheduler): @with_role(*MODERATION_ROLES) @command() - async def tempban( - self, ctx: Context, user: UserTypes, expiry: ExpirationDate, *, reason: str = None - ): + async def tempban(self, ctx: Context, user: UserTypes, duration: ExpirationDate, *, reason: str = None) -> None: """ Create a temporary ban infraction in the database for a user. **`user`:** Accepts user mention, ID, etc. - **`expiry`:** The duration for the temporary ban infraction + **`duration`:** The duration for the temporary ban infraction **`reason`:** The reason for the temporary ban. """ + expiration = duration if not await self.respect_role_hierarchy(ctx, user, 'tempban'): # Ensure ctx author has a higher top role than the target user # Warning is sent to ctx by the helper method return - active_bans = await self.bot.api_client.get( - 'bot/infractions', - params={ - 'active': 'true', - 'type': 'ban', - 'user__id': str(user.id) - } - ) - if active_bans: - return await ctx.send( - ":x: According to my records, this user is already banned. " - f"See infraction **#{active_bans[0]['id']}**." - ) + if await already_has_active_infraction(ctx=ctx, user=user, type="ban"): + return - infraction = await post_infraction( - ctx, user, type="ban", - reason=reason, expires_at=expiry - ) + infraction = await post_infraction(ctx, user, type="ban", reason=reason, expires_at=expiration) if infraction is None: return notified = await self.notify_infraction( user=user, infr_type="Ban", - expires_at=expiry, + expires_at=expiration, reason=reason ) @@ -510,7 +450,7 @@ class Moderation(Scheduler): @with_role(*MODERATION_ROLES) @command(hidden=True, aliases=['shadowwarn', 'swarn', 'shadow_warn']) - async def note(self, ctx: Context, user: UserTypes, *, reason: str = None): + async def note(self, ctx: Context, user: UserTypes, *, reason: str = None) -> None: """ Create a private infraction note in the database for a user. @@ -518,11 +458,8 @@ class Moderation(Scheduler): **`reason`:** The reason for the warning. """ - response_object = await post_infraction( - ctx, user, type="warning", reason=reason, hidden=True - ) - - if response_object is None: + infraction = await post_infraction(ctx, user, type="warning", reason=reason, hidden=True) + if infraction is None: return if reason is None: @@ -540,12 +477,12 @@ class Moderation(Scheduler): Actor: {ctx.message.author} Reason: {reason} """), - footer=f"ID {response_object['infraction']['id']}" + footer=f"ID {infraction['id']}" ) @with_role(*MODERATION_ROLES) @command(hidden=True, aliases=['shadowkick', 'skick']) - async def shadow_kick(self, ctx: Context, user: Member, *, reason: str = None): + async def shadow_kick(self, ctx: Context, user: Member, *, reason: str = None) -> None: """ Kicks a user. @@ -558,8 +495,8 @@ class Moderation(Scheduler): # Warning is sent to ctx by the helper method return - response_object = await post_infraction(ctx, user, type="kick", reason=reason, hidden=True) - if response_object is None: + infraction = await post_infraction(ctx, user, type="kick", reason=reason, hidden=True) + if infraction is None: return self.mod_log.ignore(Event.member_remove, user.id) @@ -593,12 +530,12 @@ class Moderation(Scheduler): Reason: {reason} """), content=log_content, - footer=f"ID {response_object['infraction']['id']}" + footer=f"ID {infraction['id']}" ) @with_role(*MODERATION_ROLES) @command(hidden=True, aliases=['shadowban', 'sban']) - async def shadow_ban(self, ctx: Context, user: UserTypes, *, reason: str = None): + async def shadow_ban(self, ctx: Context, user: UserTypes, *, reason: str = None) -> None: """ Create a permanent ban infraction in the database for a user. @@ -611,8 +548,11 @@ class Moderation(Scheduler): # Warning is sent to ctx by the helper method return - response_object = await post_infraction(ctx, user, type="ban", reason=reason, hidden=True) - if response_object is None: + if await already_has_active_infraction(ctx=ctx, user=user, type="ban"): + return + + infraction = await post_infraction(ctx, user, type="ban", reason=reason, hidden=True) + if infraction is None: return self.mod_log.ignore(Event.member_ban, user.id) @@ -647,12 +587,12 @@ class Moderation(Scheduler): Reason: {reason} """), content=log_content, - footer=f"ID {response_object['infraction']['id']}" + footer=f"ID {infraction['id']}" ) @with_role(*MODERATION_ROLES) @command(hidden=True, aliases=['shadowmute', 'smute']) - async def shadow_mute(self, ctx: Context, user: Member, *, reason: str = None): + async def shadow_mute(self, ctx: Context, user: Member, *, reason: str = None) -> None: """ Create a permanent mute infraction in the database for a user. @@ -660,8 +600,11 @@ class Moderation(Scheduler): **`reason`:** The reason for the mute. """ - response_object = await post_infraction(ctx, user, type="mute", reason=reason, hidden=True) - if response_object is None: + if await already_has_active_infraction(ctx=ctx, user=user, type="mute"): + return + + infraction = await post_infraction(ctx, user, type="mute", reason=reason, hidden=True) + if infraction is None: return self.mod_log.ignore(Event.member_update, user.id) @@ -682,7 +625,7 @@ class Moderation(Scheduler): Actor: {ctx.message.author} Reason: {reason} """), - footer=f"ID {response_object['infraction']['id']}" + footer=f"ID {infraction['id']}" ) # endregion @@ -691,8 +634,13 @@ class Moderation(Scheduler): @with_role(*MODERATION_ROLES) @command(hidden=True, aliases=["shadowtempmute, stempmute"]) async def shadow_tempmute( - self, ctx: Context, user: Member, duration: str, *, reason: str = None - ): + self, + ctx: Context, + user: Member, + duration: ExpirationDate, + *, + reason: str = None + ) -> None: """ Create a temporary mute infraction in the database for a user. @@ -700,20 +648,25 @@ class Moderation(Scheduler): **`duration`:** The duration for the temporary mute infraction **`reason`:** The reason for the temporary mute. """ + expiration = duration - response_object = await post_infraction( - ctx, user, type="mute", reason=reason, duration=duration, hidden=True - ) - if response_object is None: + if await already_has_active_infraction(ctx=ctx, user=user, type="mute"): + return + + infraction = await post_infraction(ctx, user, type="mute", reason=reason, expires_at=expiration, hidden=True) + if infraction is None: return self.mod_log.ignore(Event.member_update, user.id) await user.add_roles(self._muted_role, reason=reason) - infraction_object = response_object["infraction"] - infraction_expiration = infraction_object["expires_at"] + infraction_expiration = ( + datetime + .fromisoformat(infraction["expires_at"][:-1]) + .strftime('%c') + ) - self.schedule_expiration(ctx.bot.loop, infraction_object) + self.schedule_task(ctx.bot.loop, infraction["id"], infraction) if reason is None: await ctx.send(f":ok_hand: muted {user.mention} until {infraction_expiration}.") @@ -731,17 +684,21 @@ class Moderation(Scheduler): Member: {user.mention} (`{user.id}`) Actor: {ctx.message.author} Reason: {reason} - Duration: {duration} Expires: {infraction_expiration} """), - footer=f"ID {response_object['infraction']['id']}" + footer=f"ID {infraction['id']}" ) @with_role(*MODERATION_ROLES) @command(hidden=True, aliases=["shadowtempban, stempban"]) async def shadow_tempban( - self, ctx: Context, user: UserTypes, duration: str, *, reason: str = None - ): + self, + ctx: Context, + user: UserTypes, + duration: ExpirationDate, + *, + reason: str = None + ) -> None: """ Create a temporary ban infraction in the database for a user. @@ -749,16 +706,18 @@ class Moderation(Scheduler): **`duration`:** The duration for the temporary ban infraction **`reason`:** The reason for the temporary ban. """ + expiration = duration if not await self.respect_role_hierarchy(ctx, user, 'shadowtempban'): # Ensure ctx author has a higher top role than the target user # Warning is sent to ctx by the helper method return - response_object = await post_infraction( - ctx, user, type="ban", reason=reason, duration=duration, hidden=True - ) - if response_object is None: + if await already_has_active_infraction(ctx=ctx, user=user, type="ban"): + return + + infraction = await post_infraction(ctx, user, type="ban", reason=reason, expires_at=expiration, hidden=True) + if infraction is None: return self.mod_log.ignore(Event.member_ban, user.id) @@ -770,10 +729,13 @@ class Moderation(Scheduler): except Forbidden: action_result = False - infraction_object = response_object["infraction"] - infraction_expiration = infraction_object["expires_at"] + infraction_expiration = ( + datetime + .fromisoformat(infraction["expires_at"][:-1]) + .strftime('%c') + ) - self.schedule_expiration(ctx.bot.loop, infraction_object) + self.schedule_task(ctx.bot.loop, infraction["id"], infraction) if reason is None: await ctx.send(f":ok_hand: banned {user.mention} until {infraction_expiration}.") @@ -799,11 +761,10 @@ class Moderation(Scheduler): Member: {user.mention} (`{user.id}`) Actor: {ctx.message.author} Reason: {reason} - Duration: {duration} Expires: {infraction_expiration} """), content=log_content, - footer=f"ID {response_object['infraction']['id']}" + footer=f"ID {infraction['id']}" ) # endregion @@ -811,7 +772,7 @@ class Moderation(Scheduler): @with_role(*MODERATION_ROLES) @command() - async def unmute(self, ctx: Context, user: Member): + async def unmute(self, ctx: Context, user: Member) -> None: """ Deactivates the active mute infraction for a user. @@ -833,9 +794,10 @@ class Moderation(Scheduler): if not response: # no active infraction - return await ctx.send( + await ctx.send( f":x: There is no active mute infraction for user {user.mention}." ) + return infraction = response[0] await self._deactivate_infraction(infraction) @@ -881,7 +843,7 @@ class Moderation(Scheduler): @with_role(*MODERATION_ROLES) @command() - async def unban(self, ctx: Context, user: UserTypes): + async def unban(self, ctx: Context, user: UserTypes) -> None: """ Deactivates the active ban infraction for a user. @@ -906,9 +868,10 @@ class Moderation(Scheduler): if not response: # no active infraction - return await ctx.send( + await ctx.send( f":x: There is no active ban infraction for user {user.mention}." ) + return infraction = response[0] await self._deactivate_infraction(infraction) @@ -1043,7 +1006,7 @@ class Moderation(Scheduler): @with_role(*MODERATION_ROLES) @infraction_edit_group.command(name="reason") - async def edit_reason(self, ctx: Context, infraction_id: int, *, reason: str): + async def edit_reason(self, ctx: Context, infraction_id: int, *, reason: str) -> None: """ Sets the reason of the given infraction. **`infraction_id`:** the id of the infraction @@ -1063,7 +1026,8 @@ class Moderation(Scheduler): except Exception: log.exception("There was an error updating an infraction.") - return await ctx.send(":x: There was an error updating the infraction.") + await ctx.send(":x: There was an error updating the infraction.") + return # Get information about the infraction's user user_id = updated_infraction['user'] @@ -1169,7 +1133,11 @@ class Moderation(Scheduler): # endregion # region: Utility functions - def schedule_expiration(self, loop: asyncio.AbstractEventLoop, infraction_object: dict): + def schedule_expiration( + self, + loop: asyncio.AbstractEventLoop, + infraction_object: Dict[str, Union[str, int, bool]] + ) -> None: """ Schedules a task to expire a temporary infraction. @@ -1199,7 +1167,7 @@ class Moderation(Scheduler): log.debug(f"Unscheduled {infraction_id}.") del self.scheduled_tasks[infraction_id] - async def _scheduled_task(self, infraction_object: dict): + async def _scheduled_task(self, infraction_object: Dict[str, Union[str, int, bool]]) -> None: """ A co-routine which marks an infraction as expired after the delay from the time of scheduling to the time of expiration. At the time of expiration, the infraction is @@ -1229,7 +1197,7 @@ class Moderation(Scheduler): icon_url=Icons.user_unmute ) - async def _deactivate_infraction(self, infraction_object): + async def _deactivate_infraction(self, infraction_object: Dict[str, Union[str, int, bool]]) -> None: """ A co-routine which marks an infraction as inactive on the website. This co-routine does not cancel or un-schedule an expiration task. @@ -1258,7 +1226,7 @@ class Moderation(Scheduler): json={"active": False} ) - def _infraction_to_string(self, infraction_object): + def _infraction_to_string(self, infraction_object: Dict[str, Union[str, int, bool]]) -> str: actor_id = infraction_object["actor"] guild: Guild = self.bot.get_guild(constants.Guild.id) actor = guild.get_member(actor_id) @@ -1283,9 +1251,12 @@ class Moderation(Scheduler): return lines.strip() async def notify_infraction( - self, user: Union[User, Member], infr_type: str, - expires_at: Union[datetime, str] = 'N/A', reason: str = "No reason provided." - ): + self, + user: Union[User, Member], + infr_type: str, + expires_at: Union[datetime, str] = 'N/A', + reason: str = "No reason provided." + ) -> bool: """ Notify a user of their fresh infraction :) @@ -1318,9 +1289,12 @@ class Moderation(Scheduler): return await self.send_private_embed(user, embed) async def notify_pardon( - self, user: Union[User, Member], title: str, content: str, - icon_url: str = Icons.user_verified - ): + self, + user: Union[User, Member], + title: str, + content: str, + icon_url: str = Icons.user_verified + ) -> bool: """ Notify a user that an infraction has been lifted. @@ -1339,7 +1313,7 @@ class Moderation(Scheduler): return await self.send_private_embed(user, embed) - async def send_private_embed(self, user: Union[User, Member], embed: Embed): + async def send_private_embed(self, user: Union[User, Member], embed: Embed) -> bool: """ A helper method for sending an embed to a user's DMs. @@ -1374,7 +1348,7 @@ class Moderation(Scheduler): # endregion - async def __error(self, ctx, error): + async def __error(self, ctx: Context, error) -> None: if isinstance(error, BadUnionArgument): if User in error.converters: await ctx.send(str(error.errors[0])) @@ -1409,6 +1383,7 @@ class Moderation(Scheduler): return target_is_lower -def setup(bot): +def setup(bot: Bot) -> None: + """Sets up the Moderation cog.""" bot.add_cog(Moderation(bot)) log.info("Cog loaded: Moderation") diff --git a/bot/utils/moderation.py b/bot/utils/moderation.py index c1eb98dd6..152f9d538 100644 --- a/bot/utils/moderation.py +++ b/bot/utils/moderation.py @@ -44,3 +44,23 @@ async def post_infraction( return return response + + +async def already_has_active_infraction(ctx: Context, user: Union[Member, Object, User], type: str) -> bool: + """Checks if a user already has an active infraction of the given type.""" + active_infractions = await ctx.bot.api_client.get( + 'bot/infractions', + params={ + 'active': 'true', + 'type': type, + 'user__id': str(user.id) + } + ) + if active_infractions: + await ctx.send( + f":x: According to my records, this user already has a {type} infraction. " + f"See infraction **#{active_infractions[0]['id']}**." + ) + return True + else: + return False |