Edm0nd/edmond/plugins/reminder.py

201 lines
6.6 KiB
Python
Raw Normal View History

import datetime
import re
import string
2023-10-10 15:14:48 +02:00
import uuid
2023-08-29 14:28:01 +02:00
from typing import Optional
from apscheduler.triggers.date import DateTrigger
from edmond.plugin import Plugin
class ReminderPlugin(Plugin):
2023-10-10 15:14:48 +02:00
"""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"
]
2023-10-10 15:14:48 +02:00
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
2023-08-29 14:28:01 +02:00
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],
2023-10-10 15:14:48 +02:00
message: str,
sender: str,
target: str
) -> None:
2023-10-10 15:14:48 +02:00
"""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"]
2023-08-29 14:28:01 +02:00
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
2023-10-10 15:14:48 +02:00
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),
2023-10-10 15:14:48 +02:00
args=(identifier, message, sender, target)
)
2023-10-10 15:14:48 +02:00
self.bot.log_d(f"Scheduled {identifier} for {when}.")
2023-10-10 15:14:48 +02:00
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)
2023-10-10 15:14:48 +02:00
# 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)