Edm0nd/edmond/plugins/reminder.py

import datetime
import re
import string
import uuid
from typing import Optional

from apscheduler.triggers.date import DateTrigger

from edmond.plugin import Plugin


class ReminderPlugin(Plugin):
    """Reminders using an async scheduler.

    Reminders can be set in a given amount of time or to a given hour. They are
    kept between restarts.
    """

    REQUIRED_CONFIGS = [
        "commands", "at_word", "in_word", "day_letter", "hour_letter",
        "minute_letter", "second_letter", "reminder_format", "done"
    ]

    def on_join(self, event):
        """Reschedule undelivered reminders."""
        nick = event.source.nick
        if nick != self.bot.nick:
            return

        joined_channel = event.target
        reminder_format = self.config["reminder_format"]
        saved_reminders = self.get_storage_value("reminders", [])
        self.set_storage_value("reminders", [])

        for reminder in saved_reminders:
            # Ignore reminders for other targets than this channel.
            if reminder["target"] != joined_channel:
                continue

            when = datetime.datetime.fromisoformat(reminder["when"])
            # If the timer passed, deliver it now and forget it.
            if when <= datetime.datetime.now():
                message = reminder_format.format(
                    username=reminder["sender"],
                    reminder=reminder["message"]
                )
                self.bot.say(reminder["target"], message)
                continue

            # Else schedule it properly (but skip storage backup, we're on it).
            self.schedule(
                when,
                reminder["message"],
                reminder["sender"],
                reminder["target"],
            )


    def on_pubmsg(self, event):
        if not self.should_handle_command(event.arguments[0]):
            return False

        target = event.target
        words = self.command.content.split()
        if len(words) < 3:
            self.bot.log_e("Not enough words in command.")
            self.signal_failure(event.target)
            return True

        mode = words[0]
        if mode not in (self.config["at_word"], self.config["in_word"]):
            self.bot.log_e("Invalid reminder mode.")
            self.signal_failure(event.target)
            return True

        time = self.parse_time(words[1])
        if not time:
            self.bot.log_e("Invalid time format.")
            self.signal_failure(event.target)
            return True

        sender = event.source.nick
        reminder = " ".join(words[2:])
        self.setup_reminder(mode, time, reminder, sender, target)
        return True

    def parse_time(self, time: str) -> Optional[dict[str, int]]:
        """Parse a time request string.

        Return a dict with day/hour/minute/second set as integers if specified.
        """
        time_re = re.compile(
            r"((?P<day>\d+)" + self.config["day_letter"] + ")?"
            r"((?P<hour>\d+)" + self.config["hour_letter"] + ")?"
            r"((?P<minute>\d+)" + self.config["minute_letter"] + ")?"
            r"((?P<second>\d+)" + self.config["second_letter"] + ")?"
        )

        # People tend to use formats such as "1h30", omitting the "m" for
        # minutes. If the time ends without a marker, add the minute letter.
        if time[-1] in string.digits:
            time += self.config["minute_letter"]

        if (match := time_re.match(time)):
            values = match.groupdict()
            return { k: int(v) for k, v in values.items() if v is not None }

        return None

    def setup_reminder(
        self,
        mode: str,
        time: dict[str, int],
        message: str,
        sender: str,
        target: str
    ) -> None:
        """Remind something at a given time.

        There are two possible modes, "at" for a given time and "in" for a
        given delay. In both cases, it should result to a datetime in the
        future, or it will be discarded.

        Reminders are saved in the bot's storage to prevent losing reminders
        between restarts. Only when a reminder is delivered it is removed from
        the storage.
        """
        now = datetime.datetime.now()
        if mode == self.config["at_word"]:
            if "day" in time: # "day" is not supported in at mode.
                del time["day"]
            when = datetime.datetime.today().replace(
                hour=time.get("hour", 0),
                minute=time.get("minute", 0),
                second=time.get("second", 0),
                microsecond=0
            )
        elif mode == self.config["in_word"]:
            time = { k + "s": v for k, v in time.items() }  # Use plural names.
            when = now + datetime.timedelta(**time)
        else:
            self.bot.log_e("Invalid reminder mode.")
            self.signal_failure(target)
            return

        if when <= now:
            self.bot.log_e(f"Trigger datetime is in the past: {when}")
            self.signal_failure(target)
            return

        self.schedule(when, message, sender, target)
        self.bot.say(target, self.config["done"])

    def schedule(
        self,
        when: datetime.datetime,
        message: str,
        sender: str,
        target: str,
    ) -> None:
        """Schedule the reminder to be delivered at the given datetime."""
        identifier = str(uuid.uuid4())
        # Store the reminder in case the bot shuts down before delivery.
        reminder = {
            "id": identifier,
            "when": when.isoformat(),
            "message": message,
            "sender": sender,
            "target": target,
        }
        self.append_storage_list_value("reminders", reminder)

        self.bot.scheduler.add_job(
            self.remind,
            trigger=DateTrigger(when),
            args=(identifier, message, sender, target)
        )
        self.bot.log_d(f"Scheduled {identifier} for {when}.")

    async def remind(
        self,
        identifier: str,
        reminder: str,
        username: str,
        target: str
    ) -> None:
        self.bot.log_d(
            f"Delivering reminder {identifier} in {target} for {username}…"
        )
        reminder_format = self.config["reminder_format"]
        message = reminder_format.format(username=username, reminder=reminder)
        self.bot.say(target, message)

        # Remove the reminder from the saved reminder list.
        saved_reminders = self.get_storage_value("reminders", [])
        for reminder in saved_reminders.copy():
            if reminder["id"] == identifier:
                saved_reminders.remove(reminder)
                break
        self.set_storage_value("reminders", saved_reminders)