diff options
| author | 2022-02-09 07:58:04 +0200 | |
|---|---|---|
| committer | 2022-02-09 07:58:04 +0200 | |
| commit | f037942d3b2ba77a568a12fc4fa703fee3274d90 (patch) | |
| tree | 06cd59b9931ee8dfabda0b55c6fa3ff3e8f12bf0 | |
| parent | Merge pull request #2073 from minalike/incident-unfurl-embed-pfp (diff) | |
Disable Reminders Cog (#2074)
| -rw-r--r-- | bot/exts/utils/reminders.py | 492 | 
1 files changed, 0 insertions, 492 deletions
| diff --git a/bot/exts/utils/reminders.py b/bot/exts/utils/reminders.py deleted file mode 100644 index 289d00356..000000000 --- a/bot/exts/utils/reminders.py +++ /dev/null @@ -1,492 +0,0 @@ -import random -import textwrap -import typing as t -from datetime import datetime, timezone -from operator import itemgetter - -import discord -from dateutil.parser import isoparse -from discord.ext.commands import Cog, Context, Greedy, group - -from bot.bot import Bot -from bot.constants import Guild, Icons, MODERATION_ROLES, POSITIVE_REPLIES, Roles, STAFF_PARTNERS_COMMUNITY_ROLES -from bot.converters import Duration, UnambiguousUser -from bot.log import get_logger -from bot.pagination import LinePaginator -from bot.utils import scheduling, time -from bot.utils.checks import has_any_role_check, has_no_roles_check -from bot.utils.lock import lock_arg -from bot.utils.members import get_or_fetch_member -from bot.utils.messages import send_denial -from bot.utils.scheduling import Scheduler - -log = get_logger(__name__) - -LOCK_NAMESPACE = "reminder" -WHITELISTED_CHANNELS = Guild.reminder_whitelist -MAXIMUM_REMINDERS = 5 - -Mentionable = t.Union[discord.Member, discord.Role] -ReminderMention = t.Union[UnambiguousUser, discord.Role] - - -class Reminders(Cog): -    """Provide in-channel reminder functionality.""" - -    def __init__(self, bot: Bot): -        self.bot = bot -        self.scheduler = Scheduler(self.__class__.__name__) - -        scheduling.create_task(self.reschedule_reminders(), event_loop=self.bot.loop) - -    def cog_unload(self) -> None: -        """Cancel scheduled tasks.""" -        self.scheduler.cancel_all() - -    async def reschedule_reminders(self) -> None: -        """Get all current reminders from the API and reschedule them.""" -        await self.bot.wait_until_guild_available() -        response = await self.bot.api_client.get( -            'bot/reminders', -            params={'active': 'true'} -        ) - -        now = datetime.now(timezone.utc) - -        for reminder in response: -            is_valid, *_ = self.ensure_valid_reminder(reminder) -            if not is_valid: -                continue - -            remind_at = isoparse(reminder['expiration']) - -            # If the reminder is already overdue ... -            if remind_at < now: -                await self.send_reminder(reminder, remind_at) -            else: -                self.schedule_reminder(reminder) - -    def ensure_valid_reminder(self, reminder: dict) -> t.Tuple[bool, discord.User, discord.TextChannel]: -        """Ensure reminder author and channel can be fetched otherwise delete the reminder.""" -        user = self.bot.get_user(reminder['author']) -        channel = self.bot.get_channel(reminder['channel_id']) -        is_valid = True -        if not user or not channel: -            is_valid = False -            log.info( -                f"Reminder {reminder['id']} invalid: " -                f"User {reminder['author']}={user}, Channel {reminder['channel_id']}={channel}." -            ) -            scheduling.create_task(self.bot.api_client.delete(f"bot/reminders/{reminder['id']}")) - -        return is_valid, user, channel - -    @staticmethod -    async def _send_confirmation( -        ctx: Context, -        on_success: str, -        reminder_id: t.Union[str, int] -    ) -> None: -        """Send an embed confirming the reminder change was made successfully.""" -        embed = discord.Embed( -            description=on_success, -            colour=discord.Colour.green(), -            title=random.choice(POSITIVE_REPLIES) -        ) - -        footer_str = f"ID: {reminder_id}" - -        embed.set_footer(text=footer_str) - -        await ctx.send(embed=embed) - -    @staticmethod -    async def _check_mentions(ctx: Context, mentions: t.Iterable[Mentionable]) -> t.Tuple[bool, str]: -        """ -        Returns whether or not the list of mentions is allowed. - -        Conditions: -        - Role reminders are Mods+ -        - Reminders for other users are Helpers+ - -        If mentions aren't allowed, also return the type of mention(s) disallowed. -        """ -        if await has_no_roles_check(ctx, *STAFF_PARTNERS_COMMUNITY_ROLES): -            return False, "members/roles" -        elif await has_no_roles_check(ctx, *MODERATION_ROLES): -            return all(isinstance(mention, (discord.User, discord.Member)) for mention in mentions), "roles" -        else: -            return True, "" - -    @staticmethod -    async def validate_mentions(ctx: Context, mentions: t.Iterable[Mentionable]) -> bool: -        """ -        Filter mentions to see if the user can mention, and sends a denial if not allowed. - -        Returns whether or not the validation is successful. -        """ -        mentions_allowed, disallowed_mentions = await Reminders._check_mentions(ctx, mentions) - -        if not mentions or mentions_allowed: -            return True -        else: -            await send_denial(ctx, f"You can't mention other {disallowed_mentions} in your reminder!") -            return False - -    async def get_mentionables(self, mention_ids: t.List[int]) -> t.Iterator[Mentionable]: -        """Converts Role and Member ids to their corresponding objects if possible.""" -        guild = self.bot.get_guild(Guild.id) -        for mention_id in mention_ids: -            member = await get_or_fetch_member(guild, mention_id) -            if mentionable := (member or guild.get_role(mention_id)): -                yield mentionable - -    def schedule_reminder(self, reminder: dict) -> None: -        """A coroutine which sends the reminder once the time is reached, and cancels the running task.""" -        reminder_datetime = isoparse(reminder['expiration']) -        self.scheduler.schedule_at(reminder_datetime, reminder["id"], self.send_reminder(reminder)) - -    async def _edit_reminder(self, reminder_id: int, payload: dict) -> dict: -        """ -        Edits a reminder in the database given the ID and payload. - -        Returns the edited reminder. -        """ -        # Send the request to update the reminder in the database -        reminder = await self.bot.api_client.patch( -            'bot/reminders/' + str(reminder_id), -            json=payload -        ) -        return reminder - -    async def _reschedule_reminder(self, reminder: dict) -> None: -        """Reschedule a reminder object.""" -        log.trace(f"Cancelling old task #{reminder['id']}") -        self.scheduler.cancel(reminder["id"]) - -        log.trace(f"Scheduling new task #{reminder['id']}") -        self.schedule_reminder(reminder) - -    @lock_arg(LOCK_NAMESPACE, "reminder", itemgetter("id"), raise_error=True) -    async def send_reminder(self, reminder: dict, expected_time: t.Optional[time.Timestamp] = None) -> None: -        """Send the reminder.""" -        is_valid, user, channel = self.ensure_valid_reminder(reminder) -        if not is_valid: -            # No need to cancel the task too; it'll simply be done once this coroutine returns. -            return -        embed = discord.Embed() -        if expected_time: -            embed.colour = discord.Colour.red() -            embed.set_author( -                icon_url=Icons.remind_red, -                name="Sorry, your reminder should have arrived earlier!" -            ) -        else: -            embed.colour = discord.Colour.og_blurple() -            embed.set_author( -                icon_url=Icons.remind_blurple, -                name="It has arrived!" -            ) - -        # Let's not use a codeblock to keep emojis and mentions working. Embeds are safe anyway. -        embed.description = f"Here's your reminder: {reminder['content']}" - -        # Here the jump URL is in the format of base_url/guild_id/channel_id/message_id -        additional_mentions = ' '.join([ -            mentionable.mention async for mentionable in self.get_mentionables(reminder["mentions"]) -        ]) - -        jump_url = reminder.get("jump_url") -        embed.description += f"\n[Jump back to when you created the reminder]({jump_url})" -        partial_message = channel.get_partial_message(int(jump_url.split("/")[-1])) -        try: -            await partial_message.reply(content=f"{additional_mentions}", embed=embed) -        except discord.HTTPException as e: -            log.info( -                f"There was an error when trying to reply to a reminder invocation message, {e}, " -                "fall back to using jump_url" -            ) -            await channel.send(content=f"{user.mention} {additional_mentions}", embed=embed) - -        log.debug(f"Deleting reminder #{reminder['id']} (the user has been reminded).") -        await self.bot.api_client.delete(f"bot/reminders/{reminder['id']}") - -    @group(name="remind", aliases=("reminder", "reminders", "remindme"), invoke_without_command=True) -    async def remind_group( -        self, ctx: Context, mentions: Greedy[ReminderMention], expiration: Duration, *, content: t.Optional[str] = None -    ) -> None: -        """ -        Commands for managing your reminders. - -        The `expiration` duration of `!remind new` supports the following symbols for each unit of time: -        - years: `Y`, `y`, `year`, `years` -        - months: `m`, `month`, `months` -        - weeks: `w`, `W`, `week`, `weeks` -        - days: `d`, `D`, `day`, `days` -        - hours: `H`, `h`, `hour`, `hours` -        - minutes: `M`, `minute`, `minutes` -        - seconds: `S`, `s`, `second`, `seconds` - -        For example, to set a reminder that expires in 3 days and 1 minute, you can do `!remind new 3d1M Do something`. -        """ -        await self.new_reminder(ctx, mentions=mentions, expiration=expiration, content=content) - -    @remind_group.command(name="new", aliases=("add", "create")) -    async def new_reminder( -        self, ctx: Context, mentions: Greedy[ReminderMention], expiration: Duration, *, content: t.Optional[str] = None -    ) -> None: -        """ -        Set yourself a simple reminder. - -        The `expiration` duration supports the following symbols for each unit of time: -        - years: `Y`, `y`, `year`, `years` -        - months: `m`, `month`, `months` -        - weeks: `w`, `W`, `week`, `weeks` -        - days: `d`, `D`, `day`, `days` -        - hours: `H`, `h`, `hour`, `hours` -        - minutes: `M`, `minute`, `minutes` -        - seconds: `S`, `s`, `second`, `seconds` - -        For example, to set a reminder that expires in 3 days and 1 minute, you can do `!remind new 3d1M Do something`. -        """ -        # If the user is not staff, partner or part of the python community, -        # we need to verify whether or not to make a reminder at all. -        if await has_no_roles_check(ctx, *STAFF_PARTNERS_COMMUNITY_ROLES): - -            # If they don't have permission to set a reminder in this channel -            if ctx.channel.id not in WHITELISTED_CHANNELS: -                await send_denial(ctx, "Sorry, you can't do that here!") -                return - -            # Get their current active reminders -            active_reminders = await self.bot.api_client.get( -                'bot/reminders', -                params={ -                    'author__id': str(ctx.author.id) -                } -            ) - -            # Let's limit this, so we don't get 10 000 -            # reminders from kip or something like that :P -            if len(active_reminders) > MAXIMUM_REMINDERS: -                await send_denial(ctx, "You have too many active reminders!") -                return - -        # Remove duplicate mentions -        mentions = set(mentions) -        mentions.discard(ctx.author) - -        # Filter mentions to see if the user can mention members/roles -        if not await self.validate_mentions(ctx, mentions): -            return - -        mention_ids = [mention.id for mention in mentions] - -        # If `content` isn't provided then we try to get message content of a replied message -        if not content: -            if reference := ctx.message.reference: -                if isinstance((resolved_message := reference.resolved), discord.Message): -                    content = resolved_message.content -            # If we weren't able to get the content of a replied message -            if content is None: -                await send_denial(ctx, "Your reminder must have a content and/or reply to a message.") -                return - -            # If the replied message has no content (e.g. only attachments/embeds) -            if content == "": -                content = "See referenced message." - -        # Now we can attempt to actually set the reminder. -        reminder = await self.bot.api_client.post( -            'bot/reminders', -            json={ -                'author': ctx.author.id, -                'channel_id': ctx.message.channel.id, -                'jump_url': ctx.message.jump_url, -                'content': content, -                'expiration': expiration.isoformat(), -                'mentions': mention_ids, -            } -        ) - -        formatted_time = time.discord_timestamp(expiration, time.TimestampFormats.DAY_TIME) -        mention_string = f"Your reminder will arrive on {formatted_time}" - -        if mentions: -            mention_string += f" and will mention {len(mentions)} other(s)" -        mention_string += "!" - -        # Confirm to the user that it worked. -        await self._send_confirmation( -            ctx, -            on_success=mention_string, -            reminder_id=reminder["id"] -        ) - -        self.schedule_reminder(reminder) - -    @remind_group.command(name="list") -    async def list_reminders(self, ctx: Context) -> None: -        """View a paginated embed of all reminders for your user.""" -        # Get all the user's reminders from the database. -        data = await self.bot.api_client.get( -            'bot/reminders', -            params={'author__id': str(ctx.author.id)} -        ) - -        # Make a list of tuples so it can be sorted by time. -        reminders = sorted( -            ( -                (rem['content'], rem['expiration'], rem['id'], rem['mentions']) -                for rem in data -            ), -            key=itemgetter(1) -        ) - -        lines = [] - -        for content, remind_at, id_, mentions in reminders: -            # Parse and humanize the time, make it pretty :D -            expiry = time.format_relative(remind_at) - -            mentions = ", ".join([ -                # Both Role and User objects have the `name` attribute -                mention.name async for mention in self.get_mentionables(mentions) -            ]) -            mention_string = f"\n**Mentions:** {mentions}" if mentions else "" - -            text = textwrap.dedent(f""" -            **Reminder #{id_}:** *expires {expiry}* (ID: {id_}){mention_string} -            {content} -            """).strip() - -            lines.append(text) - -        embed = discord.Embed() -        embed.colour = discord.Colour.og_blurple() -        embed.title = f"Reminders for {ctx.author}" - -        # Remind the user that they have no reminders :^) -        if not lines: -            embed.description = "No active reminders could be found." -            await ctx.send(embed=embed) -            return - -        # Construct the embed and paginate it. -        embed.colour = discord.Colour.og_blurple() - -        await LinePaginator.paginate( -            lines, -            ctx, embed, -            max_lines=3, -            empty=True -        ) - -    @remind_group.group(name="edit", aliases=("change", "modify"), invoke_without_command=True) -    async def edit_reminder_group(self, ctx: Context) -> None: -        """ -        Commands for modifying your current reminders. - -        The `expiration` duration supports the following symbols for each unit of time: -        - years: `Y`, `y`, `year`, `years` -        - months: `m`, `month`, `months` -        - weeks: `w`, `W`, `week`, `weeks` -        - days: `d`, `D`, `day`, `days` -        - hours: `H`, `h`, `hour`, `hours` -        - minutes: `M`, `minute`, `minutes` -        - seconds: `S`, `s`, `second`, `seconds` - -        For example, to edit a reminder to expire in 3 days and 1 minute, you can do `!remind edit duration 1234 3d1M`. -        """ -        await ctx.send_help(ctx.command) - -    @edit_reminder_group.command(name="duration", aliases=("time",)) -    async def edit_reminder_duration(self, ctx: Context, id_: int, expiration: Duration) -> None: -        """ -        Edit one of your reminder's expiration. - -        The `expiration` duration supports the following symbols for each unit of time: -        - years: `Y`, `y`, `year`, `years` -        - months: `m`, `month`, `months` -        - weeks: `w`, `W`, `week`, `weeks` -        - days: `d`, `D`, `day`, `days` -        - hours: `H`, `h`, `hour`, `hours` -        - minutes: `M`, `minute`, `minutes` -        - seconds: `S`, `s`, `second`, `seconds` - -        For example, to edit a reminder to expire in 3 days and 1 minute, you can do `!remind edit duration 1234 3d1M`. -        """ -        await self.edit_reminder(ctx, id_, {'expiration': expiration.isoformat()}) - -    @edit_reminder_group.command(name="content", aliases=("reason",)) -    async def edit_reminder_content(self, ctx: Context, id_: int, *, content: str) -> None: -        """Edit one of your reminder's content.""" -        await self.edit_reminder(ctx, id_, {"content": content}) - -    @edit_reminder_group.command(name="mentions", aliases=("pings",)) -    async def edit_reminder_mentions(self, ctx: Context, id_: int, mentions: Greedy[ReminderMention]) -> None: -        """Edit one of your reminder's mentions.""" -        # Remove duplicate mentions -        mentions = set(mentions) -        mentions.discard(ctx.author) - -        # Filter mentions to see if the user can mention members/roles -        if not await self.validate_mentions(ctx, mentions): -            return - -        mention_ids = [mention.id for mention in mentions] -        await self.edit_reminder(ctx, id_, {"mentions": mention_ids}) - -    @lock_arg(LOCK_NAMESPACE, "id_", raise_error=True) -    async def edit_reminder(self, ctx: Context, id_: int, payload: dict) -> None: -        """Edits a reminder with the given payload, then sends a confirmation message.""" -        if not await self._can_modify(ctx, id_): -            return -        reminder = await self._edit_reminder(id_, payload) - -        # Send a confirmation message to the channel -        await self._send_confirmation( -            ctx, -            on_success="That reminder has been edited successfully!", -            reminder_id=id_, -        ) -        await self._reschedule_reminder(reminder) - -    @remind_group.command("delete", aliases=("remove", "cancel")) -    @lock_arg(LOCK_NAMESPACE, "id_", raise_error=True) -    async def delete_reminder(self, ctx: Context, id_: int) -> None: -        """Delete one of your active reminders.""" -        if not await self._can_modify(ctx, id_): -            return - -        await self.bot.api_client.delete(f"bot/reminders/{id_}") -        self.scheduler.cancel(id_) - -        await self._send_confirmation( -            ctx, -            on_success="That reminder has been deleted successfully!", -            reminder_id=id_ -        ) - -    async def _can_modify(self, ctx: Context, reminder_id: t.Union[str, int]) -> bool: -        """ -        Check whether the reminder can be modified by the ctx author. - -        The check passes when the user is an admin, or if they created the reminder. -        """ -        if await has_any_role_check(ctx, Roles.admins): -            return True - -        api_response = await self.bot.api_client.get(f"bot/reminders/{reminder_id}") -        if not api_response["author"] == ctx.author.id: -            log.debug(f"{ctx.author} is not the reminder author and does not pass the check.") -            await send_denial(ctx, "You can't modify reminders of other users!") -            return False - -        log.debug(f"{ctx.author} is the reminder author and passes the check.") -        return True - - -def setup(bot: Bot) -> None: -    """Load the Reminders cog.""" -    bot.add_cog(Reminders(bot)) | 
