diff options
| author | 2019-09-23 16:51:22 +1000 | |
|---|---|---|
| committer | 2019-09-23 16:51:22 +1000 | |
| commit | d49befb1800135000f324588c593acdb2d1bebcb (patch) | |
| tree | 38d3475bcd71b55264acf90a23c49bca93774e7a /bot/cogs/moderation.py | |
| parent | Fix date formatting bug in infraction search (diff) | |
| parent | Apply suggestions from code review (diff) | |
Update linting (#406)
Update linting
Diffstat (limited to '')
| -rw-r--r-- | bot/cogs/moderation.py | 251 |
1 files changed, 75 insertions, 176 deletions
diff --git a/bot/cogs/moderation.py b/bot/cogs/moderation.py index fea86c33e..81b3864a7 100644 --- a/bot/cogs/moderation.py +++ b/bot/cogs/moderation.py @@ -33,6 +33,7 @@ APPEALABLE_INFRACTIONS = ("Ban", "Mute") def proxy_user(user_id: str) -> Object: + """Create a proxy user for the provided user_id for situations where a Member or User object cannot be resolved.""" try: user_id = int(user_id) except ValueError: @@ -47,9 +48,7 @@ UserTypes = Union[Member, User, proxy_user] class Moderation(Scheduler, Cog): - """ - Server moderation tools. - """ + """Server moderation tools.""" def __init__(self, bot: Bot): self.bot = bot @@ -58,10 +57,12 @@ class Moderation(Scheduler, Cog): @property def mod_log(self) -> ModLog: + """Get currently loaded ModLog cog instance.""" return self.bot.get_cog("ModLog") @Cog.listener() - async def on_ready(self): + async def on_ready(self) -> None: + """Schedule expiration for previous infractions.""" # Schedule expiration for previous infractions infractions = await self.bot.api_client.get( 'bot/infractions', params={'active': 'true'} @@ -74,14 +75,8 @@ class Moderation(Scheduler, Cog): @with_role(*MODERATION_ROLES) @command() - async def warn(self, ctx: Context, user: UserTypes, *, reason: str = None): - """ - Create a warning infraction in the database for a user. - - **`user`:** Accepts user mention, ID, etc. - **`reason`:** The reason for the warning. - """ - + async def warn(self, ctx: Context, user: UserTypes, *, reason: str = None) -> None: + """Create a warning infraction in the database for a user.""" infraction = await post_infraction(ctx, user, type="warning", reason=reason) if infraction is None: return @@ -120,14 +115,8 @@ class Moderation(Scheduler, Cog): @with_role(*MODERATION_ROLES) @command() - async def kick(self, ctx: Context, user: Member, *, reason: str = None): - """ - Kicks a user. - - **`user`:** Accepts user mention, ID, etc. - **`reason`:** The reason for the kick. - """ - + async def kick(self, ctx: Context, user: Member, *, reason: str = None) -> None: + """Kicks a user with the provided reason.""" if not await self.respect_role_hierarchy(ctx, user, 'kick'): # Ensure ctx author has a higher top role than the target user # Warning is sent to ctx by the helper method @@ -176,14 +165,8 @@ class Moderation(Scheduler, Cog): @with_role(*MODERATION_ROLES) @command() - async def ban(self, ctx: Context, user: UserTypes, *, reason: str = None): - """ - Create a permanent ban infraction in the database for a user. - - **`user`:** Accepts user mention, ID, etc. - **`reason`:** The reason for the ban. - """ - + async def ban(self, ctx: Context, user: UserTypes, *, reason: str = None) -> None: + """Create a permanent ban infraction for a user with the provided reason.""" if not await self.respect_role_hierarchy(ctx, user, 'ban'): # Ensure ctx author has a higher top role than the target user # Warning is sent to ctx by the helper method @@ -242,14 +225,8 @@ class Moderation(Scheduler, Cog): @with_role(*MODERATION_ROLES) @command() - async def mute(self, ctx: Context, user: Member, *, reason: str = None): - """ - Create a permanent mute infraction in the database for a user. - - **`user`:** Accepts user mention, ID, etc. - **`reason`:** The reason for the mute. - """ - + async def mute(self, ctx: Context, user: Member, *, reason: str = None) -> None: + """Create a permanent mute infraction for a user with the provided reason.""" if await already_has_active_infraction(ctx=ctx, user=user, type="mute"): return @@ -304,11 +281,9 @@ class Moderation(Scheduler, Cog): @command() 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. + Create a temporary mute infraction for a user with the provided expiration and reason. - **`user`:** Accepts user mention, ID, etc. - **`duration`:** The duration for the temporary mute infraction - **`reason`:** The reason for the temporary mute. + Duration strings are parsed per: http://strftime.org/ """ expiration = duration @@ -372,11 +347,9 @@ class Moderation(Scheduler, Cog): @command() 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. + Create a temporary ban infraction for a user with the provided expiration and reason. - **`user`:** Accepts user mention, ID, etc. - **`duration`:** The duration for the temporary ban infraction - **`reason`:** The reason for the temporary ban. + Duration strings are parsed per: http://strftime.org/ """ expiration = duration @@ -453,12 +426,10 @@ class Moderation(Scheduler, Cog): @command(hidden=True, aliases=['shadowwarn', 'swarn', 'shadow_warn']) async def note(self, ctx: Context, user: UserTypes, *, reason: str = None) -> None: """ - Create a private infraction note in the database for a user. + Create a private infraction note in the database for a user with the provided reason. - **`user`:** accepts user mention, ID, etc. - **`reason`:** The reason for the warning. + This does not send the user a notification """ - infraction = await post_infraction(ctx, user, type="warning", reason=reason, hidden=True) if infraction is None: return @@ -485,12 +456,10 @@ class Moderation(Scheduler, Cog): @command(hidden=True, aliases=['shadowkick', 'skick']) async def shadow_kick(self, ctx: Context, user: Member, *, reason: str = None) -> None: """ - Kicks a user. + Kick a user for the provided reason. - **`user`:** accepts user mention, ID, etc. - **`reason`:** The reason for the kick. + This does not send the user a notification. """ - if not await self.respect_role_hierarchy(ctx, user, 'shadowkick'): # Ensure ctx author has a higher top role than the target user # Warning is sent to ctx by the helper method @@ -538,12 +507,10 @@ class Moderation(Scheduler, Cog): @command(hidden=True, aliases=['shadowban', 'sban']) async def shadow_ban(self, ctx: Context, user: UserTypes, *, reason: str = None) -> None: """ - Create a permanent ban infraction in the database for a user. + Create a permanent ban infraction for a user with the provided reason. - **`user`:** Accepts user mention, ID, etc. - **`reason`:** The reason for the ban. + This does not send the user a notification. """ - if not await self.respect_role_hierarchy(ctx, user, 'shadowban'): # Ensure ctx author has a higher top role than the target user # Warning is sent to ctx by the helper method @@ -595,12 +562,10 @@ class Moderation(Scheduler, Cog): @command(hidden=True, aliases=['shadowmute', 'smute']) async def shadow_mute(self, ctx: Context, user: Member, *, reason: str = None) -> None: """ - Create a permanent mute infraction in the database for a user. + Create a permanent mute infraction for a user with the provided reason. - **`user`:** Accepts user mention, ID, etc. - **`reason`:** The reason for the mute. + This does not send the user a notification. """ - if await already_has_active_infraction(ctx=ctx, user=user, type="mute"): return @@ -635,19 +600,14 @@ class Moderation(Scheduler, Cog): @with_role(*MODERATION_ROLES) @command(hidden=True, aliases=["shadowtempmute, stempmute"]) async def shadow_tempmute( - self, - ctx: Context, - user: Member, - duration: ExpirationDate, - *, - 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. + Create a temporary mute infraction for a user with the provided reason. - **`user`:** Accepts user mention, ID, etc. - **`duration`:** The duration for the temporary mute infraction - **`reason`:** The reason for the temporary mute. + Duration strings are parsed per: http://strftime.org/ + + This does not send the user a notification. """ expiration = duration @@ -693,19 +653,14 @@ class Moderation(Scheduler, Cog): @with_role(*MODERATION_ROLES) @command(hidden=True, aliases=["shadowtempban, stempban"]) async def shadow_tempban( - self, - ctx: Context, - user: UserTypes, - duration: ExpirationDate, - *, - 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. + Create a temporary ban infraction for a user with the provided reason. - **`user`:** Accepts user mention, ID, etc. - **`duration`:** The duration for the temporary ban infraction - **`reason`:** The reason for the temporary ban. + Duration strings are parsed per: http://strftime.org/ + + This does not send the user a notification. """ expiration = duration @@ -774,12 +729,7 @@ class Moderation(Scheduler, Cog): @with_role(*MODERATION_ROLES) @command() async def unmute(self, ctx: Context, user: UserTypes) -> None: - """ - Deactivates the active mute infraction for a user. - - **`user`:** Accepts user mention, ID, etc. - """ - + """Deactivates the active mute infraction for a user.""" try: # check the current active infraction response = await self.bot.api_client.get( @@ -857,12 +807,7 @@ class Moderation(Scheduler, Cog): @with_role(*MODERATION_ROLES) @command() async def unban(self, ctx: Context, user: UserTypes) -> None: - """ - Deactivates the active ban infraction for a user. - - **`user`:** Accepts user mention, ID, etc. - """ - + """Deactivates the active ban infraction for a user.""" try: # check the current active infraction response = await self.bot.api_client.get( @@ -925,16 +870,14 @@ class Moderation(Scheduler, Cog): @with_role(*MODERATION_ROLES) @group(name='infraction', aliases=('infr', 'infractions', 'inf'), invoke_without_command=True) - async def infraction_group(self, ctx: Context): + async def infraction_group(self, ctx: Context) -> None: """Infraction manipulation commands.""" - await ctx.invoke(self.bot.get_command("help"), "infraction") @with_role(*MODERATION_ROLES) @infraction_group.group(name='edit', invoke_without_command=True) - async def infraction_edit_group(self, ctx: Context): + async def infraction_edit_group(self, ctx: Context) -> None: """Infraction editing commands.""" - await ctx.invoke(self.bot.get_command("help"), "infraction", "edit") @with_role(*MODERATION_ROLES) @@ -942,15 +885,12 @@ class Moderation(Scheduler, Cog): async def edit_duration( self, ctx: Context, infraction_id: int, expires_at: Union[ExpirationDate, str] - ): + ) -> None: """ Sets the duration of the given infraction, relative to the time of updating. - **`infraction_id`:** the id of the infraction - **`expires_at`:** the new expiration date of the infraction. - Use "permanent" to mark the infraction as permanent. + Duration strings are parsed per: http://strftime.org/, use "permanent" to mark the infraction as permanent. """ - if isinstance(expires_at, str) and expires_at != 'permanent': raise BadArgument( "If `expires_at` is given as a non-datetime, " @@ -1031,12 +971,7 @@ class Moderation(Scheduler, Cog): @with_role(*MODERATION_ROLES) @infraction_edit_group.command(name="reason") 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 - **`reason`:** The new reason of the infraction - """ - + """Edit the reason of the given infraction.""" try: old_infraction = await self.bot.api_client.get( 'bot/infractions/' + str(infraction_id) @@ -1087,11 +1022,8 @@ class Moderation(Scheduler, Cog): @with_role(*MODERATION_ROLES) @infraction_group.group(name="search", invoke_without_command=True) - async def infraction_search_group(self, ctx: Context, query: InfractionSearchQuery): - """ - Searches for infractions in the database. - """ - + async def infraction_search_group(self, ctx: Context, query: InfractionSearchQuery) -> None: + """Searches for infractions in the database.""" if isinstance(query, User): await ctx.invoke(self.search_user, query) @@ -1100,11 +1032,8 @@ class Moderation(Scheduler, Cog): @with_role(*MODERATION_ROLES) @infraction_search_group.command(name="user", aliases=("member", "id")) - async def search_user(self, ctx: Context, user: Union[User, proxy_user]): - """ - Search for infractions by member. - """ - + async def search_user(self, ctx: Context, user: Union[User, proxy_user]) -> None: + """Search for infractions by member.""" infraction_list = await self.bot.api_client.get( 'bot/infractions', params={'user__id': str(user.id)} @@ -1117,11 +1046,8 @@ class Moderation(Scheduler, Cog): @with_role(*MODERATION_ROLES) @infraction_search_group.command(name="reason", aliases=("match", "regex", "re")) - async def search_reason(self, ctx: Context, reason: str): - """ - Search for infractions by their reason. Use Re2 for matching. - """ - + async def search_reason(self, ctx: Context, reason: str) -> None: + """Search for infractions by their reason. Use Re2 for matching.""" infraction_list = await self.bot.api_client.get( 'bot/infractions', params={'search': reason} ) @@ -1134,8 +1060,8 @@ class Moderation(Scheduler, Cog): # endregion # region: Utility functions - async def send_infraction_list(self, ctx: Context, embed: Embed, infractions: list): - + async def send_infraction_list(self, ctx: Context, embed: Embed, infractions: list) -> None: + """Send a paginated embed of infractions for the specified user.""" if not infractions: await ctx.send(f":warning: No infractions could be found for that query.") return @@ -1158,17 +1084,9 @@ class Moderation(Scheduler, Cog): # region: Utility functions def schedule_expiration( - self, - loop: asyncio.AbstractEventLoop, - infraction_object: Dict[str, Union[str, int, bool]] + self, loop: asyncio.AbstractEventLoop, infraction_object: Dict[str, Union[str, int, bool]] ) -> None: - """ - Schedules a task to expire a temporary infraction. - - :param loop: the asyncio event loop - :param infraction_object: the infraction object to expire at the end of the task - """ - + """Schedules a task to expire a temporary infraction.""" infraction_id = infraction_object["id"] if infraction_id in self.scheduled_tasks: return @@ -1177,12 +1095,8 @@ class Moderation(Scheduler, Cog): self.scheduled_tasks[infraction_id] = task - def cancel_expiration(self, infraction_id: str): - """ - Un-schedules a task set to expire a temporary infraction. - :param infraction_id: the ID of the infraction in question - """ - + def cancel_expiration(self, infraction_id: str) -> None: + """Un-schedules a task set to expire a temporary infraction.""" task = self.scheduled_tasks.get(infraction_id) if task is None: log.warning(f"Failed to unschedule {infraction_id}: no task found.") @@ -1193,13 +1107,11 @@ class Moderation(Scheduler, Cog): 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 - marked as inactive on the website, and the expiration task is cancelled. + Marks an infraction expired after the delay from time of scheduling to time of expiration. - :param infraction_object: the infraction in question + At the time of expiration, the infraction is marked as inactive on the website, and the + expiration task is cancelled. The user is then notified via DM. """ - infraction_id = infraction_object["id"] # transform expiration to delay in seconds @@ -1224,11 +1136,9 @@ class Moderation(Scheduler, Cog): 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. - :param infraction_object: the infraction in question + This co-routine does not cancel or un-schedule an expiration task. """ - guild: Guild = self.bot.get_guild(constants.Guild.id) user_id = infraction_object["user"] infraction_type = infraction_object["type"] @@ -1254,6 +1164,7 @@ class Moderation(Scheduler, Cog): ) def _infraction_to_string(self, infraction_object: Dict[str, Union[str, int, bool]]) -> str: + """Convert the infraction object to a string representation.""" actor_id = infraction_object["actor"] guild: Guild = self.bot.get_guild(constants.Guild.id) actor = guild.get_member(actor_id) @@ -1283,21 +1194,17 @@ class Moderation(Scheduler, Cog): 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 :) + Attempt to notify a user, via DM, of their fresh infraction. - :param user: The user to send the message to. - :param infr_type: The type of infraction, as a string. - :param duration: The duration of the infraction. - :param reason: The reason for the infraction. + Returns a boolean indicator of whether the DM was successful. """ - if isinstance(expires_at, datetime): expires_at = expires_at.strftime('%c') @@ -1328,14 +1235,10 @@ class Moderation(Scheduler, Cog): icon_url: str = Icons.user_verified ) -> bool: """ - Notify a user that an infraction has been lifted. + Attempt to notify a user, via DM, of their expired infraction. - :param user: The user to send the message to. - :param title: The title of the embed. - :param content: The content of the embed. - :param icon_url: URL for the title icon. + Optionally returns a boolean indicator of whether the DM was successful. """ - embed = Embed( description=content, colour=Colour(Colours.soft_green) @@ -1349,10 +1252,8 @@ class Moderation(Scheduler, Cog): """ A helper method for sending an embed to a user's DMs. - :param user: The user to send the embed to. - :param embed: The embed to send. + Returns a boolean indicator of DM success. """ - # sometimes `user` is a `discord.Object`, so let's make it a proper user. user = await self.bot.fetch_user(user.id) @@ -1366,7 +1267,8 @@ class Moderation(Scheduler, Cog): ) return False - async def log_notify_failure(self, target: str, actor: Member, infraction_type: str): + async def log_notify_failure(self, target: str, actor: Member, infraction_type: str) -> None: + """Send a mod log entry if an attempt to DM the target user has failed.""" await self.mod_log.send_log_message( icon_url=Icons.token_removed, content=actor.mention, @@ -1381,7 +1283,8 @@ class Moderation(Scheduler, Cog): # endregion @staticmethod - async def cog_command_error(ctx: Context, error) -> None: + async def cog_command_error(ctx: Context, error: Exception) -> None: + """Send a notification to the invoking context on a Union failure.""" if isinstance(error, BadUnionArgument): if User in error.converters: await ctx.send(str(error.errors[0])) @@ -1391,15 +1294,11 @@ class Moderation(Scheduler, Cog): async def respect_role_hierarchy(ctx: Context, target: UserTypes, infr_type: str) -> bool: """ Check if the highest role of the invoking member is greater than that of the target member. + If this check fails, a warning is sent to the invoking ctx. Returns True always if target is not a discord.Member instance. - - :param ctx: The command context when invoked. - :param target: The target of the infraction. - :param infr_type: The type of infraction. """ - if not isinstance(target, Member): return True @@ -1419,6 +1318,6 @@ class Moderation(Scheduler, Cog): def setup(bot: Bot) -> None: - """Sets up the Moderation cog.""" + """Moderation cog load.""" bot.add_cog(Moderation(bot)) log.info("Cog loaded: Moderation") |