Compare commits
2 commits
7c030b771d
...
6f41e5f407
Author | SHA1 | Date | |
---|---|---|---|
dece | 6f41e5f407 | ||
dece | 85a1c7ddce |
|
@ -164,6 +164,17 @@
|
||||||
"separator": "or",
|
"separator": "or",
|
||||||
"not_enough": "Not enough choices!"
|
"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": {
|
"sleep": {
|
||||||
"commands": ["sleep", "wake up"],
|
"commands": ["sleep", "wake up"],
|
||||||
"aliases": {
|
"aliases": {
|
||||||
|
|
|
@ -33,6 +33,7 @@ class Bot(irc.client_aio.AioSimpleIRCClient, Logger):
|
||||||
self.tasks: list[asyncio.Task] = []
|
self.tasks: list[asyncio.Task] = []
|
||||||
self.done: bool = False
|
self.done: bool = False
|
||||||
self.scheduler = AsyncIOScheduler()
|
self.scheduler = AsyncIOScheduler()
|
||||||
|
self.scheduler.start()
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def nick(self) -> str:
|
def nick(self) -> str:
|
||||||
|
@ -94,9 +95,6 @@ class Bot(irc.client_aio.AioSimpleIRCClient, Logger):
|
||||||
if event.source.nick == self.nick:
|
if event.source.nick == self.nick:
|
||||||
self.log_i(f"Joined {event.target}.")
|
self.log_i(f"Joined {event.target}.")
|
||||||
self.channels.append(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)
|
self.run_plugin_callbacks(event)
|
||||||
|
|
||||||
async def flubiz(self):
|
async def flubiz(self):
|
||||||
|
|
|
@ -90,11 +90,12 @@ class PlaylistOfTheDayPlugin(Plugin):
|
||||||
today.isoformat(),
|
today.isoformat(),
|
||||||
)
|
)
|
||||||
|
|
||||||
def post_playlist(self, target):
|
def post_playlist(self, target: str) -> None:
|
||||||
playlist: list[str] = self.get_storage_value(self.PLAYLIST_KEY, [])
|
playlist: list[str] = self.get_storage_value(self.PLAYLIST_KEY, [])
|
||||||
if not playlist:
|
if not playlist:
|
||||||
self.bot.log_e("Playlist empty.")
|
self.bot.log_e("Playlist empty.")
|
||||||
self.signal_failure(target)
|
self.signal_failure(target)
|
||||||
|
return
|
||||||
|
|
||||||
linkified_items = map(PlaylistOfTheDayPlugin.linkify, playlist)
|
linkified_items = map(PlaylistOfTheDayPlugin.linkify, playlist)
|
||||||
html_items = map(lambda item: f"<li>{item}</li>", linkified_items)
|
html_items = map(lambda item: f"<li>{item}</li>", linkified_items)
|
||||||
|
@ -104,6 +105,7 @@ class PlaylistOfTheDayPlugin(Plugin):
|
||||||
if not url:
|
if not url:
|
||||||
self.bot.log_e("Shrlok returned None.")
|
self.bot.log_e("Shrlok returned None.")
|
||||||
self.signal_failure(target)
|
self.signal_failure(target)
|
||||||
|
return
|
||||||
|
|
||||||
date = self.get_storage_value(self.DATE_KEY, "")
|
date = self.get_storage_value(self.DATE_KEY, "")
|
||||||
if date != datetime.date.today().isoformat():
|
if date != datetime.date.today().isoformat():
|
||||||
|
|
107
edmond/plugins/reminder.py
Normal file
107
edmond/plugins/reminder.py
Normal file
|
@ -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)
|
22
edmond/plugins/tests/test_reminder.py
Normal file
22
edmond/plugins/tests/test_reminder.py
Normal file
|
@ -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
|
import unittest
|
||||||
from datetime import datetime, timedelta
|
from datetime import datetime, timedelta
|
||||||
from unittest.mock import Mock, patch
|
|
||||||
|
|
||||||
from edmond.tests.test_plugin import get_plugin_patcher
|
from edmond.tests.test_plugin import get_plugin_patcher
|
||||||
from ..sleep import SleepPlugin
|
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
|
import time
|
||||||
|
|
||||||
from typing import cast
|
from typing import cast
|
||||||
|
|
||||||
import wikipedia
|
import wikipedia
|
||||||
|
|
Loading…
Reference in a new issue