115 lines
3.8 KiB
Python
115 lines
3.8 KiB
Python
import datetime
|
|
import re
|
|
import string
|
|
from typing import Optional
|
|
|
|
from apscheduler.triggers.date import DateTrigger
|
|
|
|
from edmond.plugin import Plugin
|
|
|
|
|
|
class ReminderPlugin(Plugin):
|
|
"""Reminders using an async scheduler."""
|
|
|
|
REQUIRED_CONFIGS = [
|
|
"commands", "at_word", "in_word", "day_letter", "hour_letter",
|
|
"minute_letter", "second_letter", "reminder_format", "done"
|
|
]
|
|
|
|
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
|
|
# self.bot.scheduler.add_job(, 'cron', hour='1', minute='52')
|
|
|
|
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],
|
|
reminder: str,
|
|
sender: str,
|
|
target: str
|
|
) -> None:
|
|
"""Remind something at a given time."""
|
|
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.bot.scheduler.add_job(
|
|
self.remind,
|
|
trigger=DateTrigger(when),
|
|
args=(reminder, sender, target)
|
|
)
|
|
self.bot.log_d(f"Scheduled for {when}, time was {time}.")
|
|
self.bot.say(target, self.config["done"])
|
|
|
|
async def remind(self, reminder: str, username: str, target: str) -> None:
|
|
reminder_format = self.config["reminder_format"]
|
|
message = reminder_format.format(username=username, reminder=reminder)
|
|
self.bot.say(target, message)
|