-
Notifications
You must be signed in to change notification settings - Fork 0
/
reminders.py
147 lines (126 loc) · 9.49 KB
/
reminders.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
"""Module containing reminders functionality for the guild bot."""
import datetime
from typing import Dict, List
from pytz import utc
import apscheduler
import apscheduler.triggers.date
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
class Reminders:
"""Reminders functionality for the guild bot"""
def __init__(self, reminders_database_path: str):
self.reminders_database_path = reminders_database_path
self.scheduler : AsyncIOScheduler = None
def start(self, event_loop):
"""Start the reminder service."""
self.scheduler = AsyncIOScheduler(event_loop=event_loop, timezone='UTC')
self.scheduler.add_jobstore(SQLAlchemyJobStore(url='sqlite:///' + self.reminders_database_path))
self.scheduler.start()
def stop(self):
"""Stop the reminder service immediately."""
self.scheduler.shutdown(wait=False)
del self.scheduler
def createOrResetPeriodicStatusUpdateCallback(self, callback: callable):
"""Create or reset the status update callback for the entire bot."""
self.scheduler.add_job(callback, trigger='interval', minutes=1, id='periodic-status', name='periodic-status',
coalesce=True, max_instances=1, replace_existing=True)
def addDailyReminder(self, owner_name: str, owner_id: str, callback: callable, callback_args: List[str]):
"""Add a daily reminder that is fired within a short time window after daily reset based on WotV world time (midnight at UTC -8).
args:
owner_name name of the owner, used in the description of the task.
ower_id unique ID of the owner, used to construct IDs for the reminders.
callback the callback function (must be a callable) to be invoked for the reminder.
callback_args positional arguments to be passed to the callback.
The reminder will have the name "<owner_name>#daily", i.e. if the owner_name is "bob" then the ID of the reminder is "bob#daily"
"""
job_id = owner_id + '#daily'
job_desc = '#daily reminder for ' + owner_name + ' (id=' + owner_id + ')'
self.scheduler.add_job(callback, id=job_id, name=job_desc, coalesce=True, max_instances=1, replace_existing=True, args=callback_args,
trigger='cron', # Cron scheduler to get easy daily scheduling
hour=8, # World time is at UTC-8 so execute at midnight World Time
minute=5, # ... but actually spread it out a bit over 5 minutes by anchoring at 5 minutes past the hour and...
jitter=300) # Jittering the time by as much as 300 seconds (5 minutes) in any direciton.
def getDailyReminder(self, owner_id: str) -> apscheduler.job.Job:
"""Return the specified ownwer's daily reminder job."""
return self.scheduler.get_job(owner_id + '#daily')
def hasDailyReminder(self, owner_id: str) -> bool:
"""Return true if the specified user has a daily reminder configured."""
return self.getDailyReminder(owner_id) is not None
def cancelDailyReminder(self, owner_id: str):
"""Cancel any daily reminder configured for the specified owner."""
job: apscheduler.job.Job = self.scheduler.get_job(owner_id + '#daily')
if job:
job.remove()
def addWhimsyReminder(self, owner_name: str, owner_id: str, nrg_reminder_callback: callable, nrg_reminder_args: List[str], spawn_reminder_callback: callable,
spawn_reminder_args: List[str], nrg_time_ms_override: int = None, spawn_time_ms_override: int = None):
"""Add a whimsy shop reminder. Actually a pair of reminders, one for NRG spending and one for whimsy spawning.
The first reminder is set for 30 minutes after now, and reminds the user that they can now start spending NRG.
The second reminder is set for 60 minutes after now, and reminds the user that they can now spawn a new Whimsy shop.
args:
owner_name name of the owner, used in the description of the task.
ower_id unique ID of the owner, used to construct IDs for the reminders.
nrg_reminder_callback the callback function (must be a callable) to be invoked for the nrg reminder.
nrg_reminder_args positional arguments to be passed to the nrg_reminder_callback.
spawn_reminder_callback the callback function (must be a callable) to be invoked for the whimsy spawn reminder.
spawn_reminder_args positional arguments to be passed to the spawn_reminder_callback.
nrg_time_ms_override if specified, overrides the amount of time before the nrg reminder fires from 30 minutes to the specified number of ms
spawn_time_ms_override if specified, overrides the amount of time before the spawn reminder fires from 60 minutes to the specified number of ms
The nrg reminder will have the name "<owner_name>#whimsy-nrg", i.e. if the owner_name is "bob" then the ID of the reminder is "bob#whimsy-nrg"
The spawn reminder will have the name "<owner_name>#whimsy-spawn", i.e. if the owner_name is "bob" then the ID of the reminder is "bob#whimsy-spawn"
"""
nrg_job_id = owner_id + '#whimsy-nrg'
nrg_job_desc = '#whimsy-nrg reminder for ' + owner_name + ' (id=' + owner_id + ')'
now = datetime.datetime.now(tz=utc)
nrg_execute_at = now + datetime.timedelta(minutes=30)
if nrg_time_ms_override:
nrg_execute_at = now + datetime.timedelta(milliseconds=nrg_time_ms_override)
spawn_job_id = owner_id + '#whimsy-spawn'
spawn_job_desc = '#whimsy-spawn reminder for ' + owner_name + ' (id=' + owner_id + ')'
spawn_execute_at = now + datetime.timedelta(hours=1)
if spawn_time_ms_override:
spawn_execute_at = now + datetime.timedelta(milliseconds=spawn_time_ms_override)
self.scheduler.add_job(nrg_reminder_callback, trigger='date', run_date=nrg_execute_at, args=nrg_reminder_args, kwargs=None,
id=nrg_job_id, name=nrg_job_desc, misfire_grace_time=30*60, coalesce=True, max_instances=1, replace_existing=True)
self.scheduler.add_job(spawn_reminder_callback, trigger='date', run_date=spawn_execute_at, args=spawn_reminder_args, kwargs=None,
id=spawn_job_id, name=spawn_job_desc, misfire_grace_time=30*60, coalesce=True, max_instances=1, replace_existing=True)
def getWhimsyReminders(self, owner_id: str) -> Dict[str, apscheduler.job.Job]:
"""Fetch any whimsy reminders outstanding for the specified owner id.
The returned dictionary contains 2 entries:
'nrg': <the NRG reminder, or None if there is no such reminder or the reminder has expired.>
'spawn': <the spawn reminder, or None if there is no such reminder or the reminder has expired.>
"""
return {
'nrg': self.scheduler.get_job(owner_id + '#whimsy-nrg'),
'spawn': self.scheduler.get_job(owner_id + '#whimsy-spawn'),
}
def hasPendingWhimsyNrgReminder(self, owner_id: str) -> bool:
"""Return true if the specified user has a pending whimsy shop NRG reminder."""
scheduled: Dict[str, apscheduler.job.Job] = self.getWhimsyReminders(owner_id)
return scheduled and 'nrg' in scheduled and scheduled['nrg'] and scheduled['nrg'].next_run_time and scheduled['nrg'].next_run_time > datetime.datetime.now(tz=utc)
def hasPendingWhimsySpawnReminder(self, owner_id: str) -> bool:
"""Return true if the specified user has a pending whimsy shop spawn reminder."""
scheduled: Dict[str, apscheduler.job.Job] = self.getWhimsyReminders(owner_id)
return scheduled and 'spawn' in scheduled and scheduled['spawn'] and scheduled['spawn'].next_run_time and scheduled['spawn'].next_run_time > datetime.datetime.now(tz=utc)
def timeTillWhimsyNrgReminder(self, owner_id: str) -> int:
"""If the specified user has a whimsy-shop NRG reminder in the future, return the number of seconds until that reminder fires; else return None."""
scheduled: Dict[str, apscheduler.job.Job] = self.getWhimsyReminders(owner_id)
if scheduled and 'nrg' in scheduled and scheduled['nrg'] and scheduled['nrg'].next_run_time and scheduled['nrg'].next_run_time > datetime.datetime.now(tz=utc):
next_run_time: datetime.datetime = scheduled['nrg'].next_run_time
return (next_run_time - datetime.datetime.now(tz=utc)).total_seconds()
return None
def timeTillWhimsySpawnReminder(self, owner_id: str) -> int:
"""If the specified user has a whimsy-shop spawn reminder in the future, return the number of seconds until that reminder fires; else return None."""
scheduled: Dict[str, apscheduler.job.Job] = self.getWhimsyReminders(owner_id)
if scheduled and 'spawn' in scheduled and scheduled['spawn'] and scheduled['spawn'].next_run_time and scheduled['spawn'].next_run_time > datetime.datetime.now(tz=utc):
next_run_time: datetime.datetime = scheduled['spawn'].next_run_time
return (next_run_time - datetime.datetime.now(tz=utc)).total_seconds()
return None
def cancelWhimsyReminders(self, owner_id: str):
"""Cancels any and all oustanding whimsy reminders for the specified owner."""
job: apscheduler.job.Job = None
job = self.scheduler.get_job(owner_id + '#whimsy-nrg')
if job:
job.remove()
job = self.scheduler.get_job(owner_id + '#whimsy-spawn')
if job:
job.remove()