aboutsummaryrefslogtreecommitdiffstats
path: root/bot/cogs/moderation.py
diff options
context:
space:
mode:
authorGravatar sco1 <[email protected]>2019-09-10 17:57:42 -0400
committerGravatar sco1 <[email protected]>2019-09-10 17:57:42 -0400
commitd20905ef77c86cd348cb8a86014b68012861d3b0 (patch)
treebc4e9cfca3363fc2c3332f20247ce2f37a0f0b0a /bot/cogs/moderation.py
parentDocstring linting chunk 2 (diff)
Docstring linting chunk 3
Diffstat (limited to '')
-rw-r--r--bot/cogs/moderation.py258
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")