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\d+)" + self.config["day_letter"] + ")?" r"((?P\d+)" + self.config["hour_letter"] + ")?" r"((?P\d+)" + self.config["minute_letter"] + ")?" r"((?P\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)