diff options
| author | 2019-09-10 17:57:42 -0400 | |
|---|---|---|
| committer | 2019-09-10 17:57:42 -0400 | |
| commit | d20905ef77c86cd348cb8a86014b68012861d3b0 (patch) | |
| tree | bc4e9cfca3363fc2c3332f20247ce2f37a0f0b0a /bot/cogs/moderation.py | |
| parent | Docstring linting chunk 2 (diff) | |
Docstring linting chunk 3
Diffstat (limited to '')
| -rw-r--r-- | bot/cogs/moderation.py | 258 |
1 files changed, 85 insertions, 173 deletions
diff --git a/bot/cogs/moderation.py b/bot/cogs/moderation.py index 1dc2c70d6..28956e636 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): - """ - Server moderation tools. - """ + """Server moderation tools.""" def __init__(self, bot: Bot): self.bot = bot @@ -58,10 +57,11 @@ class Moderation(Scheduler): @property def mod_log(self) -> ModLog: + """Get currently loaded ModLog cog instance.""" return self.bot.get_cog("ModLog") - async def on_ready(self): - # Schedule expiration for previous infractions + async def on_ready(self) -> None: + """Schedule expiration for previous infractions.""" infractions = await self.bot.api_client.get( 'bot/infractions', params={'active': 'true'} ) @@ -73,14 +73,13 @@ class Moderation(Scheduler): @with_role(*MODERATION_ROLES) @command() - async def warn(self, ctx: Context, user: UserTypes, *, reason: str = None): + async def warn(self, ctx: Context, user: UserTypes, *, reason: str = None) -> None: """ Create a warning infraction in the database for a user. **`user`:** Accepts user mention, ID, etc. **`reason`:** The reason for the warning. """ - response_object = await post_infraction(ctx, user, type="warning", reason=reason) if response_object is None: return @@ -123,14 +122,8 @@ class Moderation(Scheduler): @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 @@ -183,14 +176,8 @@ class Moderation(Scheduler): @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 @@ -260,14 +247,8 @@ class Moderation(Scheduler): @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.""" active_mutes = await self.bot.api_client.get( 'bot/infractions', params={ @@ -334,15 +315,12 @@ class Moderation(Scheduler): async def tempmute( self, ctx: Context, user: Member, expiration: 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/ """ - active_mutes = await self.bot.api_client.get( 'bot/infractions', params={ @@ -416,15 +394,12 @@ class Moderation(Scheduler): @command() async def tempban( self, ctx: Context, user: UserTypes, expiry: 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. - **`expiry`:** The duration for the temporary ban infraction - **`reason`:** The reason for the temporary ban. + Duration strings are parsed per: http://strftime.org/ """ - 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 @@ -510,14 +485,12 @@ 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. + 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 """ - response_object = await post_infraction( ctx, user, type="warning", reason=reason, hidden=True ) @@ -545,14 +518,12 @@ class Moderation(Scheduler): @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. + 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 @@ -598,14 +569,12 @@ class Moderation(Scheduler): @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. + 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 @@ -652,14 +621,12 @@ class Moderation(Scheduler): @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. + 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. """ - response_object = await post_infraction(ctx, user, type="mute", reason=reason, hidden=True) if response_object is None: return @@ -692,15 +659,14 @@ class Moderation(Scheduler): @command(hidden=True, aliases=["shadowtempmute, stempmute"]) async def shadow_tempmute( self, ctx: Context, user: Member, duration: str, *, 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. + """ response_object = await post_infraction( ctx, user, type="mute", reason=reason, duration=duration, hidden=True ) @@ -741,15 +707,14 @@ class Moderation(Scheduler): @command(hidden=True, aliases=["shadowtempban, stempban"]) async def shadow_tempban( self, ctx: Context, user: UserTypes, duration: str, *, 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. + """ 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 @@ -811,13 +776,8 @@ class Moderation(Scheduler): @with_role(*MODERATION_ROLES) @command() - async def unmute(self, ctx: Context, user: Member): - """ - Deactivates the active mute infraction for a user. - - **`user`:** Accepts user mention, ID, etc. - """ - + async def unmute(self, ctx: Context, user: Member) -> None: + """Deactivates the active mute infraction for a user.""" try: # check the current active infraction response = await self.bot.api_client.get( @@ -881,13 +841,8 @@ class Moderation(Scheduler): @with_role(*MODERATION_ROLES) @command() - async def unban(self, ctx: Context, user: UserTypes): - """ - Deactivates the active ban infraction for a user. - - **`user`:** Accepts user mention, ID, etc. - """ - + async def unban(self, ctx: Context, user: UserTypes) -> None: + """Deactivates the active ban infraction for a user.""" try: # check the current active infraction response = await self.bot.api_client.get( @@ -938,16 +893,14 @@ class Moderation(Scheduler): @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) @@ -955,15 +908,12 @@ class Moderation(Scheduler): 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, " @@ -1043,13 +993,8 @@ 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): - """ - Sets the reason of the given infraction. - **`infraction_id`:** the id of the infraction - **`reason`:** The new reason of the infraction - """ - + async def edit_reason(self, ctx: Context, infraction_id: int, *, reason: str) -> None: + """Edit the reason of the given infraction.""" try: old_infraction = await self.bot.api_client.get( 'bot/infractions/' + str(infraction_id) @@ -1099,11 +1044,8 @@ class Moderation(Scheduler): @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) @@ -1112,11 +1054,8 @@ class Moderation(Scheduler): @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)} @@ -1129,11 +1068,8 @@ class Moderation(Scheduler): @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} ) @@ -1146,8 +1082,8 @@ class Moderation(Scheduler): # 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 @@ -1169,14 +1105,8 @@ class Moderation(Scheduler): # endregion # region: Utility functions - def schedule_expiration(self, loop: asyncio.AbstractEventLoop, infraction_object: dict): - """ - 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 - """ - + def schedule_expiration(self, loop: asyncio.AbstractEventLoop, infraction_object: dict) -> None: + """Schedules a task to expire a temporary infraction.""" infraction_id = infraction_object["id"] if infraction_id in self.scheduled_tasks: return @@ -1185,12 +1115,8 @@ class Moderation(Scheduler): 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.") @@ -1199,15 +1125,13 @@ 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) -> 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 @@ -1229,14 +1153,12 @@ class Moderation(Scheduler): icon_url=Icons.user_unmute ) - async def _deactivate_infraction(self, infraction_object): + async def _deactivate_infraction(self, infraction_object: dict) -> 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"] @@ -1258,7 +1180,8 @@ class Moderation(Scheduler): json={"active": False} ) - def _infraction_to_string(self, infraction_object): + def _infraction_to_string(self, infraction_object: dict) -> 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) @@ -1285,16 +1208,12 @@ class Moderation(Scheduler): async def notify_infraction( 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. + Optionally returns a boolean indicator of whether the DM was successful. """ - if isinstance(expires_at, datetime): expires_at = expires_at.strftime('%c') @@ -1320,16 +1239,12 @@ class Moderation(Scheduler): async def notify_pardon( 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. + 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) @@ -1339,14 +1254,12 @@ 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. - :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.get_user_info(user.id) @@ -1360,7 +1273,8 @@ class Moderation(Scheduler): ) 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, @@ -1374,7 +1288,8 @@ class Moderation(Scheduler): # endregion - async def __error(self, ctx, error): + async def __error(self, 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])) @@ -1382,15 +1297,11 @@ class Moderation(Scheduler): async def respect_role_hierarchy(self, 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 @@ -1409,6 +1320,7 @@ class Moderation(Scheduler): return target_is_lower -def setup(bot): +def setup(bot: Bot) -> None: + """Moderation cog load.""" bot.add_cog(Moderation(bot)) log.info("Cog loaded: Moderation") |