parent
f9c23cd5f5
commit
2d84c6b696
2 changed files with 21 additions and 4 deletions
|
@ -1,7 +1,7 @@
|
|||
import logging
|
||||
|
||||
from flask_wtf import Form
|
||||
from wtforms import StringField, validators, SubmitField, ValidationError
|
||||
from wtforms import StringField, TextAreaField, validators, SubmitField, ValidationError
|
||||
from croniter import croniter
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
@ -19,6 +19,10 @@ class CronAlarmForm(Form):
|
|||
description="the frequency specification, as in the <tt>cron</tt> command; "
|
||||
'see <a href="https://crontab.guru/">crontab.guru</a> for a hepl with cron format',
|
||||
)
|
||||
exclude = TextAreaField(
|
||||
"cron-like format; any matching time will be excluded",
|
||||
description="Another cron-like thing to _exclude_ events",
|
||||
)
|
||||
submit = SubmitField("Submit")
|
||||
|
||||
def populate_from_timespec(self, timespec):
|
||||
|
@ -26,15 +30,28 @@ class CronAlarmForm(Form):
|
|||
self.nick.data = timespec["nick"]
|
||||
if "cron_format" in timespec:
|
||||
self.cron_format.data = timespec["cron_format"]
|
||||
if "exclude" in timespec:
|
||||
if type(timespec["exclude"]) is str:
|
||||
self.exclude.data = timespec["exclude"]
|
||||
else:
|
||||
self.exclude.data = "\n".join(timespec["exclude"])
|
||||
|
||||
def validate_cron_format(self, field):
|
||||
if not croniter.is_valid(field.data):
|
||||
raise ValidationError("formato di cron non valido")
|
||||
|
||||
def validate_exclude(self, field):
|
||||
for line in field.data.split("\n"):
|
||||
if line.strip() and not croniter.is_valid(line):
|
||||
raise ValidationError("formato di exclude non valido: %s" % line)
|
||||
|
||||
|
||||
def cronalarm_receive(form):
|
||||
return {
|
||||
"kind": "cron",
|
||||
"nick": form.nick.data,
|
||||
"cron_format": form.cron_format.data,
|
||||
"exclude": [
|
||||
line.strip() for line in form.exclude.data.split("\n") if line.strip()
|
||||
],
|
||||
}
|
||||
|
|
|
@ -22,14 +22,14 @@ class CronAlarm(Alarm):
|
|||
line.strip() for line in obj["exclude"].split("\n") if line.strip()
|
||||
]
|
||||
else:
|
||||
self.exclude = obj["exclude"]
|
||||
self.exclude = [excl for excl in obj["exclude"] if excl.strip()]
|
||||
else:
|
||||
self.exclude = []
|
||||
if not croniter.is_valid(self.cron_format):
|
||||
raise ValueError("Invalid cron_format: %s" % self.cron_format)
|
||||
raise ValueError("Invalid cron_format: `%s`" % self.cron_format)
|
||||
for exclude in self.exclude:
|
||||
if not croniter.is_valid(exclude):
|
||||
raise ValueError("Invalid exclude: %s" % exclude)
|
||||
raise ValueError("Invalid exclude: `%s`" % exclude)
|
||||
|
||||
def is_excluded(self, dt):
|
||||
base = dt - timedelta(seconds=1)
|
||||
|
|
Loading…
Reference in a new issue