457 lines
14 KiB
Python
457 lines
14 KiB
Python
|
|
'''
|
|
crontab.py
|
|
|
|
Written July 15, 2011 by Josiah Carlson
|
|
Released under the GNU LGPL v2.1 and v3
|
|
available:
|
|
http://www.gnu.org/licenses/old-licenses/lgpl-2.1.html
|
|
http://www.gnu.org/licenses/lgpl.html
|
|
|
|
Other licenses may be available upon request.
|
|
|
|
'''
|
|
|
|
from collections import namedtuple
|
|
from datetime import datetime, timedelta
|
|
import sys
|
|
import warnings
|
|
|
|
_ranges = [
|
|
(0, 59),
|
|
(0, 59),
|
|
(0, 23),
|
|
(1, 31),
|
|
(1, 12),
|
|
(0, 6),
|
|
(1970, 2099),
|
|
]
|
|
|
|
ENTRIES = len(_ranges)
|
|
SECOND_OFFSET, MINUTE_OFFSET, HOUR_OFFSET, DAY_OFFSET, MONTH_OFFSET, WEEK_OFFSET, YEAR_OFFSET = range(ENTRIES)
|
|
|
|
_attribute = [
|
|
'second',
|
|
'minute',
|
|
'hour',
|
|
'day',
|
|
'month',
|
|
'isoweekday',
|
|
'year'
|
|
]
|
|
_alternate = {
|
|
MONTH_OFFSET: {'jan': 1, 'feb': 2, 'mar': 3, 'apr': 4, 'may': 5, 'jun': 6,
|
|
'jul': 7, 'aug': 8, 'sep': 9, 'oct': 10, 'nov':11, 'dec':12},
|
|
WEEK_OFFSET: {'sun': 0, 'mon': 1, 'tue': 2, 'wed': 3, 'thu': 4, 'fri': 5,
|
|
'sat': 6},
|
|
}
|
|
_aliases = {
|
|
'@yearly': '0 0 1 1 *',
|
|
'@annually': '0 0 1 1 *',
|
|
'@monthly': '0 0 1 * *',
|
|
'@weekly': '0 0 * * 0',
|
|
'@daily': '0 0 * * *',
|
|
'@hourly': '0 * * * *',
|
|
}
|
|
|
|
WARNING_CHANGE_MESSAGE = '''\
|
|
Version 0.22.0+ of crontab will use datetime.utcnow() and
|
|
datetime.utcfromtimestamp() instead of datetime.now() and
|
|
datetime.fromtimestamp() as was previous. This had been a bug, which will be
|
|
remedied. If you would like to keep the *old* behavior:
|
|
`ct.next(..., default_utc=False)` . If you want to use the new behavior *now*:
|
|
`ct.next(..., default_utc=True)`. If you pass a datetime object with a tzinfo
|
|
attribute that is not None, timezones will *just work* to the best of their
|
|
ability. There are tests...'''
|
|
|
|
|
|
if sys.version_info >= (3, 0):
|
|
_number_types = (int, float)
|
|
xrange = range
|
|
else:
|
|
_number_types = (int, long, float)
|
|
|
|
SECOND = timedelta(seconds=1)
|
|
MINUTE = timedelta(minutes=1)
|
|
HOUR = timedelta(hours=1)
|
|
DAY = timedelta(days=1)
|
|
WEEK = timedelta(days=7)
|
|
MONTH = timedelta(days=28)
|
|
YEAR = timedelta(days=365)
|
|
|
|
WARN_CHANGE = object()
|
|
|
|
# find the next scheduled time
|
|
def _end_of_month(dt):
|
|
ndt = dt + DAY
|
|
while dt.month == ndt.month:
|
|
dt += DAY
|
|
return ndt.replace(day=1) - DAY
|
|
|
|
def _month_incr(dt, m):
|
|
odt = dt
|
|
dt += MONTH
|
|
while dt.month == odt.month:
|
|
dt += DAY
|
|
# get to the first of next month, let the backtracking handle it
|
|
dt = dt.replace(day=1)
|
|
return dt - odt
|
|
|
|
def _year_incr(dt, m):
|
|
# simple leapyear stuff works for 1970-2099 :)
|
|
mod = dt.year % 4
|
|
if mod == 0 and (dt.month, dt.day) < (2, 29):
|
|
return YEAR + DAY
|
|
if mod == 3 and (dt.month, dt.day) > (2, 29):
|
|
return YEAR + DAY
|
|
return YEAR
|
|
|
|
_increments = [
|
|
lambda *a: SECOND,
|
|
lambda *a: MINUTE,
|
|
lambda *a: HOUR,
|
|
lambda *a: DAY,
|
|
_month_incr,
|
|
lambda *a: DAY,
|
|
_year_incr,
|
|
lambda dt,x: dt.replace(second=0),
|
|
lambda dt,x: dt.replace(minute=0),
|
|
lambda dt,x: dt.replace(hour=0),
|
|
lambda dt,x: dt.replace(day=1) if x > DAY else dt,
|
|
lambda dt,x: dt.replace(month=1) if x > DAY else dt,
|
|
lambda dt,x: dt,
|
|
]
|
|
|
|
# find the previously scheduled time
|
|
def _day_decr(dt, m):
|
|
if m.day.input != 'l':
|
|
return -DAY
|
|
odt = dt
|
|
ndt = dt = dt - DAY
|
|
while dt.month == ndt.month:
|
|
dt -= DAY
|
|
return dt - odt
|
|
|
|
def _month_decr(dt, m):
|
|
odt = dt
|
|
# get to the last day of last month, let the backtracking handle it
|
|
dt = dt.replace(day=1) - DAY
|
|
return dt - odt
|
|
|
|
def _year_decr(dt, m):
|
|
# simple leapyear stuff works for 1970-2099 :)
|
|
mod = dt.year % 4
|
|
if mod == 0 and (dt.month, dt.day) > (2, 29):
|
|
return -(YEAR + DAY)
|
|
if mod == 1 and (dt.month, dt.day) < (2, 29):
|
|
return -(YEAR + DAY)
|
|
return -YEAR
|
|
|
|
def _day_decr_reset(dt, x):
|
|
if x >= -DAY:
|
|
return dt
|
|
cur = dt.month
|
|
while dt.month == cur:
|
|
dt += DAY
|
|
return dt - DAY
|
|
|
|
_decrements = [
|
|
lambda *a: -SECOND,
|
|
lambda *a: -MINUTE,
|
|
lambda *a: -HOUR,
|
|
_day_decr,
|
|
_month_decr,
|
|
lambda *a: -DAY,
|
|
_year_decr,
|
|
lambda dt,x: dt.replace(second=59),
|
|
lambda dt,x: dt.replace(minute=59),
|
|
lambda dt,x: dt.replace(hour=23),
|
|
_day_decr_reset,
|
|
lambda dt,x: dt.replace(month=12) if x < -DAY else dt,
|
|
lambda dt,x: dt,
|
|
_year_decr,
|
|
]
|
|
|
|
Matcher = namedtuple('Matcher', 'second, minute, hour, day, month, weekday, year')
|
|
|
|
def _assert(condition, message, *args):
|
|
if not condition:
|
|
raise ValueError(message%args)
|
|
|
|
class _Matcher(object):
|
|
__slots__ = 'allowed', 'end', 'any', 'input', 'which', 'split'
|
|
def __init__(self, which, entry):
|
|
_assert(0 <= which <= YEAR_OFFSET,
|
|
"improper number of cron entries specified")
|
|
self.input = entry.lower()
|
|
self.split = self.input.split(',')
|
|
self.which = which
|
|
self.allowed = set()
|
|
self.end = None
|
|
self.any = '*' in self.split or '?' in self.split
|
|
|
|
for it in self.split:
|
|
al, en = self._parse_crontab(which, it)
|
|
if al is not None:
|
|
self.allowed.update(al)
|
|
self.end = en
|
|
_assert(self.end is not None,
|
|
"improper item specification: %r", entry.lower()
|
|
)
|
|
self.allowed = frozenset(self.allowed)
|
|
|
|
def __call__(self, v, dt):
|
|
for i, x in enumerate(self.split):
|
|
if x == 'l':
|
|
if v == _end_of_month(dt).day:
|
|
return True
|
|
|
|
elif x.startswith('l'):
|
|
# We have to do this in here, otherwise we can end up, for
|
|
# example, accepting *any* Friday instead of the *last* Friday.
|
|
if dt.month == (dt + WEEK).month:
|
|
continue
|
|
|
|
x = x[1:]
|
|
if x.isdigit():
|
|
x = int(x) if x != '7' else 0
|
|
if v == x:
|
|
return True
|
|
continue
|
|
|
|
start, end = map(int, x.partition('-')[::2])
|
|
allowed = set(range(start, end+1))
|
|
if 7 in allowed:
|
|
allowed.add(0)
|
|
if v in allowed:
|
|
return True
|
|
|
|
return self.any or v in self.allowed
|
|
|
|
def __lt__(self, other):
|
|
if self.any:
|
|
return self.end < other
|
|
return all(item < other for item in self.allowed)
|
|
|
|
def __gt__(self, other):
|
|
if self.any:
|
|
return _ranges[self.which][0] > other
|
|
return all(item > other for item in self.allowed)
|
|
|
|
def __eq__(self, other):
|
|
if self.any:
|
|
return other.any
|
|
return self.allowed == other.allowed
|
|
|
|
def __hash__(self):
|
|
return hash((self.any, self.allowed))
|
|
|
|
def _parse_crontab(self, which, entry):
|
|
'''
|
|
This parses a single crontab field and returns the data necessary for
|
|
this matcher to accept the proper values.
|
|
|
|
See the README for information about what is accepted.
|
|
'''
|
|
|
|
# this handles day of week/month abbreviations
|
|
def _fix(it):
|
|
if which in _alternate and not it.isdigit():
|
|
if it in _alternate[which]:
|
|
return _alternate[which][it]
|
|
_assert(it.isdigit(),
|
|
"invalid range specifier: %r (%r)", it, entry)
|
|
it = int(it, 10)
|
|
_assert(_start <= it <= _end_limit,
|
|
"item value %r out of range [%r, %r]",
|
|
it, _start, _end_limit)
|
|
return it
|
|
|
|
# this handles individual items/ranges
|
|
def _parse_piece(it):
|
|
if '-' in it:
|
|
start, end = map(_fix, it.split('-'))
|
|
# Allow "sat-sun"
|
|
if which in (DAY_OFFSET, WEEK_OFFSET) and end == 0:
|
|
end = 7
|
|
elif it == '*':
|
|
start = _start
|
|
end = _end
|
|
else:
|
|
start = _fix(it)
|
|
end = _end
|
|
if increment is None:
|
|
return set([start])
|
|
|
|
_assert(_start <= start <= _end_limit,
|
|
"%s range start value %r out of range [%r, %r]",
|
|
_attribute[which], start, _start, _end_limit)
|
|
_assert(_start <= end <= _end_limit,
|
|
"%s range end value %r out of range [%r, %r]",
|
|
_attribute[which], end, _start, _end_limit)
|
|
_assert(start <= end,
|
|
"%s range start value %r > end value %r",
|
|
_attribute[which], start, end)
|
|
return set(range(start, end+1, increment or 1))
|
|
|
|
_start, _end = _ranges[which]
|
|
_end_limit = _end
|
|
# wildcards
|
|
if entry in ('*', '?'):
|
|
if entry == '?':
|
|
_assert(which in (DAY_OFFSET, WEEK_OFFSET),
|
|
"cannot use '?' in the %r field", _attribute[which])
|
|
return None, _end
|
|
|
|
# last day of the month
|
|
if entry == 'l':
|
|
_assert(which == DAY_OFFSET,
|
|
"you can only specify a bare 'L' in the 'day' field")
|
|
return None, _end
|
|
|
|
# for the last 'friday' of the month, for example
|
|
elif entry.startswith('l'):
|
|
_assert(which == WEEK_OFFSET,
|
|
"you can only specify a leading 'L' in the 'weekday' field")
|
|
es, _, ee = entry[1:].partition('-')
|
|
_assert((entry[1:].isdigit() and 0 <= int(es) <= 7) or
|
|
(_ and es.isdigit() and ee.isdigit() and 0 <= int(es) <= 7 and 0 <= int(ee) <= 7),
|
|
"last <day> specifier must include a day number or range in the 'weekday' field, you entered %r", entry)
|
|
return None, _end
|
|
|
|
increment = None
|
|
# increments
|
|
if '/' in entry:
|
|
entry, increment = entry.split('/')
|
|
increment = int(increment, 10)
|
|
_assert(increment > 0,
|
|
"you can only use positive increment values, you provided %r",
|
|
increment)
|
|
|
|
# allow Sunday to be specified as weekday 7
|
|
if which == WEEK_OFFSET:
|
|
_end_limit = 7
|
|
|
|
# handle singles and ranges
|
|
good = _parse_piece(entry)
|
|
|
|
# change Sunday to weekday 0
|
|
if which == WEEK_OFFSET and 7 in good:
|
|
good.discard(7)
|
|
good.add(0)
|
|
|
|
return good, _end
|
|
|
|
|
|
class CronTab(object):
|
|
__slots__ = 'matchers',
|
|
def __init__(self, crontab):
|
|
self.matchers = self._make_matchers(crontab)
|
|
|
|
def _make_matchers(self, crontab):
|
|
'''
|
|
This constructs the full matcher struct.
|
|
'''
|
|
crontab = _aliases.get(crontab, crontab)
|
|
ct = crontab.split()
|
|
if len(ct) == 5:
|
|
ct.insert(0, '0')
|
|
ct.append('*')
|
|
elif len(ct) == 6:
|
|
ct.insert(0, '0')
|
|
_assert(len(ct) == 7,
|
|
"improper number of cron entries specified; got %i need 5 to 7"%(len(ct,)))
|
|
|
|
matchers = [_Matcher(which, entry) for which, entry in enumerate(ct)]
|
|
|
|
return Matcher(*matchers)
|
|
|
|
def _test_match(self, index, dt):
|
|
'''
|
|
This tests the given field for whether it matches with the current
|
|
datetime object passed.
|
|
'''
|
|
at = _attribute[index]
|
|
attr = getattr(dt, at)
|
|
if index == WEEK_OFFSET:
|
|
attr = attr() % 7
|
|
return self.matchers[index](attr, dt)
|
|
|
|
def next(self, now=None, increments=_increments, delta=True, default_utc=WARN_CHANGE):
|
|
'''
|
|
How long to wait in seconds before this crontab entry can next be
|
|
executed.
|
|
'''
|
|
if default_utc is WARN_CHANGE and (isinstance(now, _number_types) or (now and not now.tzinfo) or now is None):
|
|
warnings.warn(WARNING_CHANGE_MESSAGE, FutureWarning, 2)
|
|
default_utc = False
|
|
|
|
now = now or (datetime.utcnow() if default_utc and default_utc is not WARN_CHANGE else datetime.now())
|
|
if isinstance(now, _number_types):
|
|
now = datetime.utcfromtimestamp(now) if default_utc else datetime.fromtimestamp(now)
|
|
|
|
# handle timezones if the datetime object has a timezone and get a
|
|
# reasonable future/past start time
|
|
onow, now = now, now.replace(tzinfo=None)
|
|
tz = onow.tzinfo
|
|
future = now.replace(microsecond=0) + increments[0]()
|
|
if future < now:
|
|
# we are going backwards...
|
|
_test = lambda: future.year < self.matchers.year
|
|
if now.microsecond:
|
|
future = now.replace(microsecond=0)
|
|
else:
|
|
# we are going forwards
|
|
_test = lambda: self.matchers.year < future.year
|
|
|
|
# Start from the year and work our way down. Any time we increment a
|
|
# higher-magnitude value, we reset all lower-magnitude values. This
|
|
# gets us performance without sacrificing correctness. Still more
|
|
# complicated than a brute-force approach, but also orders of
|
|
# magnitude faster in basically all cases.
|
|
to_test = ENTRIES - 1
|
|
while to_test >= 0:
|
|
if not self._test_match(to_test, future):
|
|
inc = increments[to_test](future, self.matchers)
|
|
future += inc
|
|
for i in xrange(0, to_test):
|
|
future = increments[ENTRIES+i](future, inc)
|
|
try:
|
|
if _test():
|
|
return None
|
|
except:
|
|
print(future, type(future), type(inc))
|
|
raise
|
|
to_test = ENTRIES-1
|
|
continue
|
|
to_test -= 1
|
|
|
|
# verify the match
|
|
match = [self._test_match(i, future) for i in xrange(ENTRIES)]
|
|
_assert(all(match),
|
|
"\nYou have discovered a bug with crontab, please notify the\n" \
|
|
"author with the following information:\n" \
|
|
"crontab: %r\n" \
|
|
"now: %r", ' '.join(m.input for m in self.matchers), now)
|
|
delay = future - now
|
|
if tz:
|
|
delay += tz.utcoffset(now)
|
|
delay -= tz.utcoffset(future)
|
|
|
|
if not delta:
|
|
begin = datetime(1970, 1, 1)
|
|
delay = future - begin
|
|
if tz:
|
|
delay -= tz.utcoffset(future)
|
|
return delay.days * 86400 + delay.seconds + delay.microseconds / 1000000.
|
|
|
|
def previous(self, now=None, delta=True, default_utc=WARN_CHANGE):
|
|
return self.next(now, _decrements, delta, default_utc)
|
|
|
|
def test(self, entry):
|
|
if isinstance(entry, _number_types):
|
|
entry = datetime.utcfromtimestamp(entry)
|
|
for index in xrange(ENTRIES):
|
|
if not self._test_match(index, entry):
|
|
return False
|
|
return True
|