diff options
| -rw-r--r-- | bot/cogs/moderation.py | 206 | ||||
| -rw-r--r-- | bot/cogs/superstarify/__init__.py | 11 | ||||
| -rw-r--r-- | bot/cogs/verification.py | 4 | ||||
| -rw-r--r-- | bot/cogs/watchchannels/talentpool.py | 5 | ||||
| -rw-r--r-- | bot/cogs/watchchannels/watchchannel.py | 15 | ||||
| -rw-r--r-- | bot/utils/time.py | 7 |
6 files changed, 91 insertions, 157 deletions
diff --git a/bot/cogs/moderation.py b/bot/cogs/moderation.py index 46009ffd2..b596f36e6 100644 --- a/bot/cogs/moderation.py +++ b/bot/cogs/moderation.py @@ -19,7 +19,7 @@ from bot.decorators import with_role from bot.pagination import LinePaginator 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 +from bot.utils.time import INFRACTION_FORMAT, format_infraction, wait_until log = logging.getLogger(__name__) @@ -44,6 +44,15 @@ def proxy_user(user_id: str) -> Object: return user +def permanent_duration(expires_at: str) -> str: + """Only allow an expiration to be 'permanent' if it is a string.""" + expires_at = expires_at.lower() + if expires_at != "permanent": + raise BadArgument + else: + return expires_at + + UserTypes = Union[Member, User, proxy_user] @@ -241,11 +250,7 @@ class Moderation(Scheduler, Cog): reason=reason ) - infraction_expiration = ( - datetime - .fromisoformat(infraction["expires_at"][:-1]) - .strftime('%c') - ) + infraction_expiration = format_infraction(infraction["expires_at"]) self.schedule_task(ctx.bot.loop, infraction["id"], infraction) @@ -314,11 +319,7 @@ class Moderation(Scheduler, Cog): except Forbidden: action_result = False - infraction_expiration = ( - datetime - .fromisoformat(infraction["expires_at"][:-1]) - .strftime('%c') - ) + infraction_expiration = format_infraction(infraction["expires_at"]) self.schedule_task(ctx.bot.loop, infraction["id"], infraction) @@ -505,11 +506,7 @@ class Moderation(Scheduler, Cog): self.mod_log.ignore(Event.member_update, user.id) await user.add_roles(self._muted_role, reason=reason) - infraction_expiration = ( - datetime - .fromisoformat(infraction["expires_at"][:-1]) - .strftime('%c') - ) + infraction_expiration = format_infraction(infraction["expires_at"]) self.schedule_task(ctx.bot.loop, infraction["id"], infraction) await ctx.send(f":ok_hand: muted {user.mention} until {infraction_expiration}.") @@ -562,11 +559,7 @@ class Moderation(Scheduler, Cog): except Forbidden: action_result = False - infraction_expiration = ( - datetime - .fromisoformat(infraction["expires_at"][:-1]) - .strftime('%c') - ) + infraction_expiration = format_infraction(infraction["expires_at"]) self.schedule_task(ctx.bot.loop, infraction["id"], infraction) await ctx.send(f":ok_hand: banned {user.mention} until {infraction_expiration}.") @@ -745,121 +738,72 @@ class Moderation(Scheduler, Cog): 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) -> None: - """Infraction editing commands.""" - await ctx.invoke(self.bot.get_command("help"), "infraction", "edit") - - @with_role(*MODERATION_ROLES) - @infraction_edit_group.command(name="duration") - async def edit_duration( - self, ctx: Context, - infraction_id: int, expires_at: Union[Duration, str] + @infraction_group.command(name='edit') + async def infraction_edit( + self, + ctx: Context, + infraction_id: int, + expires_at: Union[Duration, permanent_duration, None], + *, + reason: str = None ) -> None: """ - Sets the duration of the given infraction, relative to the time of updating. + Edit the duration and/or the reason of an infraction. - Duration strings are parsed per: http://strftime.org/, use "permanent" to mark the infraction as permanent. + Durations are relative to the time of updating. + 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, " - "it must be `permanent`." - ) - if expires_at == 'permanent': - expires_at = None - - try: - previous_infraction = await self.bot.api_client.get( - 'bot/infractions/' + str(infraction_id) - ) - - # check the current active infraction - infraction = await self.bot.api_client.patch( - 'bot/infractions/' + str(infraction_id), - json={ - 'expires_at': ( - expires_at.isoformat() - if expires_at is not None - else None - ) - } - ) - - # Re-schedule - self.cancel_task(infraction['id']) - loop = asyncio.get_event_loop() - self.schedule_task(loop, infraction['id'], infraction) - - if expires_at is None: - await ctx.send(f":ok_hand: Updated infraction: marked as permanent.") - else: - human_expiry = ( - datetime - .fromisoformat(infraction['expires_at'][:-1]) - .strftime('%c') - ) - await ctx.send( - ":ok_hand: Updated infraction: set to expire on " - f"{human_expiry}." - ) - - except Exception: - log.exception("There was an error updating an infraction.") - await ctx.send(":x: There was an error updating the infraction.") - return - - # Get information about the infraction's user - user_id = infraction["user"] - user = ctx.guild.get_member(user_id) - - if user: - member_text = f"{user.mention} (`{user.id}`)" - thumbnail = user.avatar_url_as(static_format="png") + if expires_at is None and reason is None: + # Unlike UserInputError, the error handler will show a specified message for BadArgument + raise BadArgument("Neither a new expiry nor a new reason was specified.") + + # Retrieve the previous infraction for its information. + old_infraction = await self.bot.api_client.get(f'bot/infractions/{infraction_id}') + + request_data = {} + confirm_messages = [] + log_text = "" + + if expires_at == "permanent": + request_data['expires_at'] = None + confirm_messages.append("marked as permanent") + elif expires_at is not None: + request_data['expires_at'] = expires_at.isoformat() + confirm_messages.append(f"set to expire on {expires_at.strftime(INFRACTION_FORMAT)}") else: - member_text = f"`{user_id}`" - thumbnail = None + confirm_messages.append("expiry unchanged") - # The infraction's actor - actor_id = infraction["actor"] - actor = ctx.guild.get_member(actor_id) or f"`{actor_id}`" + if reason: + request_data['reason'] = reason + confirm_messages.append("set a new reason") + log_text += f""" + Previous reason: {old_infraction['reason']} + New reason: {reason} + """.rstrip() + else: + confirm_messages.append("reason unchanged") - await self.mod_log.send_log_message( - icon_url=Icons.pencil, - colour=Colour.blurple(), - title="Infraction edited", - thumbnail=thumbnail, - text=textwrap.dedent(f""" - Member: {member_text} - Actor: {actor} - Edited by: {ctx.message.author} - Previous expiry: {previous_infraction['expires_at']} - New expiry: {infraction['expires_at']} - """) + # Update the infraction + new_infraction = await self.bot.api_client.patch( + f'bot/infractions/{infraction_id}', + json=request_data, ) - @with_role(*MODERATION_ROLES) - @infraction_edit_group.command(name="reason") - 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) - ) + # Re-schedule infraction if the expiration has been updated + if 'expires_at' in request_data: + self.cancel_task(new_infraction['id']) + loop = asyncio.get_event_loop() + self.schedule_task(loop, new_infraction['id'], new_infraction) - updated_infraction = await self.bot.api_client.patch( - 'bot/infractions/' + str(infraction_id), - json={'reason': reason} - ) - await ctx.send(f":ok_hand: Updated infraction: set reason to \"{reason}\".") + log_text += f""" + Previous expiry: {old_infraction['expires_at'] or "Permanent"} + New expiry: {new_infraction['expires_at'] or "Permanent"} + """.rstrip() - except Exception: - log.exception("There was an error updating an infraction.") - await ctx.send(":x: There was an error updating the infraction.") - return + await ctx.send(f":ok_hand: Updated infraction: {' & '.join(confirm_messages)}") # Get information about the infraction's user - user_id = updated_infraction['user'] + user_id = new_infraction['user'] user = ctx.guild.get_member(user_id) if user: @@ -870,7 +814,7 @@ class Moderation(Scheduler, Cog): thumbnail = None # The infraction's actor - actor_id = updated_infraction['actor'] + actor_id = new_infraction['actor'] actor = ctx.guild.get_member(actor_id) or f"`{actor_id}`" await self.mod_log.send_log_message( @@ -881,9 +825,7 @@ class Moderation(Scheduler, Cog): text=textwrap.dedent(f""" Member: {user_text} Actor: {actor} - Edited by: {ctx.message.author} - Previous reason: {old_infraction['reason']} - New reason: {updated_infraction['reason']} + Edited by: {ctx.message.author}{log_text} """) ) @@ -1041,11 +983,11 @@ class Moderation(Scheduler, Cog): active = infraction_object["active"] user_id = infraction_object["user"] hidden = infraction_object["hidden"] - created = datetime.fromisoformat(infraction_object["inserted_at"][:-1]).strftime("%Y-%m-%d %H:%M") + created = format_infraction(infraction_object["inserted_at"]) if infraction_object["expires_at"] is None: expires = "*Permanent*" else: - expires = datetime.fromisoformat(infraction_object["expires_at"][:-1]).strftime("%Y-%m-%d %H:%M") + expires = format_infraction(infraction_object["expires_at"]) lines = textwrap.dedent(f""" {"**===============**" if active else "==============="} @@ -1076,7 +1018,7 @@ class Moderation(Scheduler, Cog): Returns a boolean indicator of whether the DM was successful. """ if isinstance(expires_at, datetime): - expires_at = expires_at.strftime('%c') + expires_at = expires_at.strftime(INFRACTION_FORMAT) embed = Embed( description=textwrap.dedent(f""" @@ -1152,8 +1094,8 @@ class Moderation(Scheduler, Cog): # endregion - @staticmethod - async def cog_command_error(ctx: Context, error: Exception) -> None: + # This cannot be static (must have a __func__ attribute). + async def cog_command_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: diff --git a/bot/cogs/superstarify/__init__.py b/bot/cogs/superstarify/__init__.py index b1936ef3a..87021eded 100644 --- a/bot/cogs/superstarify/__init__.py +++ b/bot/cogs/superstarify/__init__.py @@ -1,6 +1,5 @@ import logging import random -from datetime import datetime from discord import Colour, Embed, Member from discord.errors import Forbidden @@ -13,6 +12,7 @@ from bot.constants import Icons, MODERATION_ROLES, POSITIVE_REPLIES from bot.converters import Duration from bot.decorators import with_role from bot.utils.moderation import post_infraction +from bot.utils.time import format_infraction log = logging.getLogger(__name__) NICKNAME_POLICY_URL = "https://pythondiscord.com/pages/rules/#wiki-toc-nickname-policy" @@ -71,10 +71,7 @@ class Superstarify(Cog): f"Changing the nick back to {before.display_name}." ) await after.edit(nick=forced_nick) - end_timestamp_human = ( - datetime.fromisoformat(infraction['expires_at'][:-1]) - .strftime('%c') - ) + end_timestamp_human = format_infraction(infraction['expires_at']) try: await after.send( @@ -113,9 +110,7 @@ class Superstarify(Cog): [infraction] = active_superstarifies forced_nick = get_nick(infraction['id'], member.id) await member.edit(nick=forced_nick) - end_timestamp_human = ( - datetime.fromisoformat(infraction['expires_at'][:-1]).strftime('%c') - ) + end_timestamp_human = format_infraction(infraction['expires_at']) try: await member.send( diff --git a/bot/cogs/verification.py b/bot/cogs/verification.py index b0c250603..f0a099f27 100644 --- a/bot/cogs/verification.py +++ b/bot/cogs/verification.py @@ -141,8 +141,8 @@ class Verification(Cog): f"{ctx.author.mention} Unsubscribed from <#{Channels.announcements}> notifications." ) - @staticmethod - async def cog_command_error(ctx: Context, error: Exception) -> None: + # This cannot be static (must have a __func__ attribute). + async def cog_command_error(self, ctx: Context, error: Exception) -> None: """Check for & ignore any InChannelCheckFailure.""" if isinstance(error, InChannelCheckFailure): error.handled = True diff --git a/bot/cogs/watchchannels/talentpool.py b/bot/cogs/watchchannels/talentpool.py index ffe7693a9..4a23902d5 100644 --- a/bot/cogs/watchchannels/talentpool.py +++ b/bot/cogs/watchchannels/talentpool.py @@ -10,6 +10,7 @@ from bot.api import ResponseCodeError from bot.constants import Channels, Guild, Roles, Webhooks from bot.decorators import with_role from bot.pagination import LinePaginator +from bot.utils import time from .watchchannel import WatchChannel, proxy_user log = logging.getLogger(__name__) @@ -198,7 +199,7 @@ class TalentPool(WatchChannel, Cog, name="Talentpool"): log.debug(active) log.debug(type(nomination_object["inserted_at"])) - start_date = self._get_human_readable(nomination_object["inserted_at"]) + start_date = time.format_infraction(nomination_object["inserted_at"]) if active: lines = textwrap.dedent( f""" @@ -212,7 +213,7 @@ class TalentPool(WatchChannel, Cog, name="Talentpool"): """ ) else: - end_date = self._get_human_readable(nomination_object["ended_at"]) + end_date = time.format_infraction(nomination_object["ended_at"]) lines = textwrap.dedent( f""" =============== diff --git a/bot/cogs/watchchannels/watchchannel.py b/bot/cogs/watchchannels/watchchannel.py index e78282900..122e3dfe8 100644 --- a/bot/cogs/watchchannels/watchchannel.py +++ b/bot/cogs/watchchannels/watchchannel.py @@ -1,5 +1,4 @@ import asyncio -import datetime import logging import re import textwrap @@ -8,6 +7,7 @@ from collections import defaultdict, deque from dataclasses import dataclass from typing import Optional +import dateutil.parser import discord from discord import Color, Embed, HTTPException, Message, Object, errors from discord.ext.commands import BadArgument, Bot, Cog, Context @@ -321,22 +321,11 @@ class WatchChannel(metaclass=CogABCMeta): @staticmethod def _get_time_delta(time_string: str) -> str: """Returns the time in human-readable time delta format.""" - date_time = datetime.datetime.strptime( - time_string, - "%Y-%m-%dT%H:%M:%S.%fZ" - ).replace(tzinfo=None) + date_time = dateutil.parser.isoparse(time_string).replace(tzinfo=None) time_delta = time_since(date_time, precision="minutes", max_units=1) return time_delta - @staticmethod - def _get_human_readable(time_string: str, output_format: str = "%Y-%m-%d %H:%M:%S") -> str: - date_time = datetime.datetime.strptime( - time_string, - "%Y-%m-%dT%H:%M:%S.%fZ" - ).replace(tzinfo=None) - return date_time.strftime(output_format) - def _remove_user(self, user_id: int) -> None: """Removes a user from a watch channel.""" self.watched_users.pop(user_id, None) diff --git a/bot/utils/time.py b/bot/utils/time.py index c529ccc2b..da28f2c76 100644 --- a/bot/utils/time.py +++ b/bot/utils/time.py @@ -1,9 +1,11 @@ import asyncio import datetime +import dateutil.parser from dateutil.relativedelta import relativedelta RFC1123_FORMAT = "%a, %d %b %Y %H:%M:%S GMT" +INFRACTION_FORMAT = "%Y-%m-%d %H:%M" def _stringify_time_unit(value: int, unit: str) -> str: @@ -95,3 +97,8 @@ async def wait_until(time: datetime.datetime) -> None: # Incorporate a small delay so we don't rapid-fire the event due to time precision errors if delay_seconds > 1.0: await asyncio.sleep(delay_seconds) + + +def format_infraction(timestamp: str) -> str: + """Format an infraction timestamp to a more readable ISO 8601 format.""" + return dateutil.parser.isoparse(timestamp).strftime(INFRACTION_FORMAT) |