aboutsummaryrefslogtreecommitdiffstats
path: root/bot/cogs/moderation.py
diff options
context:
space:
mode:
authorGravatar scragly <[email protected]>2019-09-23 16:51:22 +1000
committerGravatar GitHub <[email protected]>2019-09-23 16:51:22 +1000
commitd49befb1800135000f324588c593acdb2d1bebcb (patch)
tree38d3475bcd71b55264acf90a23c49bca93774e7a /bot/cogs/moderation.py
parentFix date formatting bug in infraction search (diff)
parentApply suggestions from code review (diff)
Update linting (#406)
Update linting
Diffstat (limited to '')
-rw-r--r--bot/cogs/moderation.py251
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")