Compare commits

...

2 Commits

@ -164,6 +164,17 @@
"separator": "or",
"not_enough": "Not enough choices!"
},
"reminder": {
"commands": ["reminder"],
"at_word": "at",
"in_word": "in",
"day_letter": "d",
"hour_letter": "h",
"minute_letter": "m",
"second_letter": "s",
"reminder_format": "⏰ {username}! {reminder}",
"done": "…✍️"
},
"sleep": {
"commands": ["sleep", "wake up"],
"aliases": {

@ -33,6 +33,7 @@ class Bot(irc.client_aio.AioSimpleIRCClient, Logger):
self.tasks: list[asyncio.Task] = []
self.done: bool = False
self.scheduler = AsyncIOScheduler()
self.scheduler.start()
@property
def nick(self) -> str:
@ -94,9 +95,6 @@ class Bot(irc.client_aio.AioSimpleIRCClient, Logger):
if event.source.nick == self.nick:
self.log_i(f"Joined {event.target}.")
self.channels.append(event.target)
self.scheduler.add_job(self.flubiz, 'cron', hour='1',
minute='52')
self.scheduler.start()
self.run_plugin_callbacks(event)
async def flubiz(self):

@ -90,11 +90,12 @@ class PlaylistOfTheDayPlugin(Plugin):
today.isoformat(),
)
def post_playlist(self, target):
def post_playlist(self, target: str) -> None:
playlist: list[str] = self.get_storage_value(self.PLAYLIST_KEY, [])
if not playlist:
self.bot.log_e("Playlist empty.")
self.signal_failure(target)
return
linkified_items = map(PlaylistOfTheDayPlugin.linkify, playlist)
html_items = map(lambda item: f"<li>{item}</li>", linkified_items)
@ -104,6 +105,7 @@ class PlaylistOfTheDayPlugin(Plugin):
if not url:
self.bot.log_e("Shrlok returned None.")
self.signal_failure(target)
return
date = self.get_storage_value(self.DATE_KEY, "")
if date != datetime.date.today().isoformat():

@ -0,0 +1,107 @@
import datetime
import re
import string
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) -> dict[str, int] | None:
"""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 = now.replace(**time)
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.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)

@ -0,0 +1,22 @@
import unittest
from edmond.tests.test_plugin import get_plugin_patcher
from ..reminder import ReminderPlugin
class TestReminderPlugin(unittest.TestCase):
def test_parse_time(self):
with get_plugin_patcher(ReminderPlugin):
plugin = ReminderPlugin()
plugin.config = {
"day_letter": "d",
"hour_letter": "h",
"minute_letter": "m",
"second_letter": "s",
}
result = plugin.parse_time("1d")
self.assertDictEqual(result, {"day": 1})
result = plugin.parse_time("1h30m")
self.assertDictEqual(result, {"hour": 1, "minute": 30})
result = plugin.parse_time("99d99m99s")
self.assertDictEqual(result, {"day": 99, "minute": 99, "second": 99})

@ -1,6 +1,5 @@
import unittest
from datetime import datetime, timedelta
from unittest.mock import Mock, patch
from edmond.tests.test_plugin import get_plugin_patcher
from ..sleep import SleepPlugin

@ -1,14 +0,0 @@
import unittest
from ..wikipedia import WikipediaPlugin
class TestWikipediaPlugin(unittest.TestCase):
def test_limit_text_length(self):
text = "lorem ipsum blah blah."
result = WikipediaPlugin.limit_text_length(text, max_length=10)
self.assertEqual(result, "lorem…")
result = WikipediaPlugin.limit_text_length(text, max_length=15)
self.assertEqual(result, "lorem ipsum…")
result = WikipediaPlugin.limit_text_length(text, max_length=30)
self.assertEqual(result, "lorem ipsum blah blah.")

@ -1,4 +1,5 @@
import time
from typing import cast
import wikipedia

Loading…
Cancel
Save