Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/croniter/croniter.py: 89%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1#!/usr/bin/env python
2# -*- coding: utf-8 -*-
4from __future__ import absolute_import, print_function, division
6import traceback as _traceback
7import copy
8import math
9import re
10import sys
11import inspect
12from time import time
13import datetime
14from dateutil.relativedelta import relativedelta
15from dateutil.tz import tzutc
16import calendar
17import binascii
18import random
20# as pytz is optional in thirdparty libs but we need it for good support under
21# python2, just test that it's well installed
22import pytz # noqa
24try:
25 from collections import OrderedDict
26except ImportError:
27 OrderedDict = dict # py26 degraded mode, expanders order will not be immutable
30M_ALPHAS = {'jan': 1, 'feb': 2, 'mar': 3, 'apr': 4, 'may': 5, 'jun': 6,
31 'jul': 7, 'aug': 8, 'sep': 9, 'oct': 10, 'nov': 11, 'dec': 12}
32DOW_ALPHAS = {'sun': 0, 'mon': 1, 'tue': 2, 'wed': 3, 'thu': 4, 'fri': 5, 'sat': 6}
33ALPHAS = {}
34for i in M_ALPHAS, DOW_ALPHAS:
35 ALPHAS.update(i)
36del i
37step_search_re = re.compile(r'^([^-]+)-([^-/]+)(/(\d+))?$')
38only_int_re = re.compile(r'^\d+$')
40WEEKDAYS = '|'.join(DOW_ALPHAS.keys())
41MONTHS = '|'.join(M_ALPHAS.keys())
42star_or_int_re = re.compile(r'^(\d+|\*)$')
43special_dow_re = re.compile(
44 (r'^(?P<pre>((?P<he>(({WEEKDAYS})(-({WEEKDAYS}))?)').format(WEEKDAYS=WEEKDAYS) +
45 (r'|(({MONTHS})(-({MONTHS}))?)|\w+)#)|l)(?P<last>\d+)$').format(MONTHS=MONTHS)
46)
47re_star = re.compile('[*]')
48hash_expression_re = re.compile(
49 r'^(?P<hash_type>h|r)(\((?P<range_begin>\d+)-(?P<range_end>\d+)\))?(\/(?P<divisor>\d+))?$'
50)
51MINUTE_FIELD = 0
52HOUR_FIELD = 1
53DAY_FIELD = 2
54MONTH_FIELD = 3
55DOW_FIELD = 4
56SECOND_FIELD = 5
57YEAR_FIELD = 6
58UNIX_FIELDS = (MINUTE_FIELD, HOUR_FIELD, DAY_FIELD, MONTH_FIELD, DOW_FIELD)
59SECOND_FIELDS = (MINUTE_FIELD, HOUR_FIELD, DAY_FIELD, MONTH_FIELD, DOW_FIELD, SECOND_FIELD)
60YEAR_FIELDS = (MINUTE_FIELD, HOUR_FIELD, DAY_FIELD, MONTH_FIELD, DOW_FIELD, SECOND_FIELD, YEAR_FIELD)
61CRON_FIELDS = {
62 'unix': UNIX_FIELDS,
63 'second': SECOND_FIELDS,
64 'year': YEAR_FIELDS,
65 len(UNIX_FIELDS): UNIX_FIELDS,
66 len(SECOND_FIELDS): SECOND_FIELDS,
67 len(YEAR_FIELDS): YEAR_FIELDS,
68}
69UNIX_CRON_LEN = len(UNIX_FIELDS)
70SECOND_CRON_LEN = len(SECOND_FIELDS)
71YEAR_CRON_LEN = len(YEAR_FIELDS)
72# retrocompat
73VALID_LEN_EXPRESSION = set([a for a in CRON_FIELDS if isinstance(a, int)])
74EXPRESSIONS = {}
77def timedelta_to_seconds(td):
78 return (td.microseconds + (td.seconds + td.days * 24 * 3600) * 10**6) \
79 / 10**6
82def datetime_to_timestamp(d):
83 if d.tzinfo is not None:
84 d = d.replace(tzinfo=None) - d.utcoffset()
86 return timedelta_to_seconds(d - datetime.datetime(1970, 1, 1))
89def _get_caller_globals_and_locals():
90 """
91 Returns the globals and locals of the calling frame.
93 Is there an alternative to frame hacking here?
94 """
95 caller_frame = inspect.stack()[2]
96 myglobals = caller_frame[0].f_globals
97 mylocals = caller_frame[0].f_locals
98 return myglobals, mylocals
101class CroniterError(ValueError):
102 """ General top-level Croniter base exception """
103 pass
106class CroniterBadTypeRangeError(TypeError):
107 """."""
110class CroniterBadCronError(CroniterError):
111 """ Syntax, unknown value, or range error within a cron expression """
112 pass
115class CroniterUnsupportedSyntaxError(CroniterBadCronError):
116 """ Valid cron syntax, but likely to produce inaccurate results """
117 # Extending CroniterBadCronError, which may be contridatory, but this allows
118 # catching both errors with a single exception. From a user perspective
119 # these will likely be handled the same way.
120 pass
123class CroniterBadDateError(CroniterError):
124 """ Unable to find next/prev timestamp match """
125 pass
128class CroniterNotAlphaError(CroniterBadCronError):
129 """ Cron syntax contains an invalid day or month abbreviation """
130 pass
133class croniter(object):
134 MONTHS_IN_YEAR = 12
135 RANGES = (
136 (0, 59),
137 (0, 23),
138 (1, 31),
139 (1, 12),
140 (0, 7),
141 (0, 59),
142 (1970, 2099)
143 )
144 DAYS = (
145 31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31
146 )
148 ALPHACONV = (
149 {}, # 0: min
150 {}, # 1: hour
151 {"l": "l"}, # 2: dom
152 # 3: mon
153 copy.deepcopy(M_ALPHAS),
154 # 4: dow
155 copy.deepcopy(DOW_ALPHAS),
156 # 5: second
157 {},
158 # 6: year
159 {}
160 )
162 LOWMAP = (
163 {},
164 {},
165 {0: 1},
166 {0: 1},
167 {7: 0},
168 {},
169 {}
170 )
172 LEN_MEANS_ALL = (
173 60,
174 24,
175 31,
176 12,
177 7,
178 60,
179 130
180 )
182 bad_length = 'Exactly 5, 6 or 7 columns has to be specified for iterator ' \
183 'expression.'
185 def __init__(self, expr_format, start_time=None, ret_type=float,
186 day_or=True, max_years_between_matches=None, is_prev=False,
187 hash_id=None, implement_cron_bug=False, second_at_beginning=None,
188 expand_from_start_time=False):
189 self._ret_type = ret_type
190 self._day_or = day_or
191 self._implement_cron_bug = implement_cron_bug
192 self.second_at_beginning = bool(second_at_beginning)
193 self._expand_from_start_time = expand_from_start_time
195 if hash_id:
196 if not isinstance(hash_id, (bytes, str)):
197 raise TypeError('hash_id must be bytes or UTF-8 string')
198 if not isinstance(hash_id, bytes):
199 hash_id = hash_id.encode('UTF-8')
201 self._max_years_btw_matches_explicitly_set = (
202 max_years_between_matches is not None)
203 if not self._max_years_btw_matches_explicitly_set:
204 max_years_between_matches = 50
205 self._max_years_between_matches = max(int(max_years_between_matches), 1)
207 if start_time is None:
208 start_time = time()
210 self.tzinfo = None
212 self.start_time = None
213 self.dst_start_time = None
214 self.cur = None
215 self.set_current(start_time, force=False)
217 self.expanded, self.nth_weekday_of_month = self.expand(
218 expr_format,
219 hash_id=hash_id,
220 from_timestamp=self.dst_start_time if self._expand_from_start_time else None,
221 second_at_beginning=second_at_beginning
222 )
223 self.fields = CRON_FIELDS[len(self.expanded)]
224 self.expressions = EXPRESSIONS[(expr_format, hash_id, second_at_beginning)]
225 self._is_prev = is_prev
227 @classmethod
228 def _alphaconv(cls, index, key, expressions):
229 try:
230 return cls.ALPHACONV[index][key]
231 except KeyError:
232 raise CroniterNotAlphaError(
233 "[{0}] is not acceptable".format(" ".join(expressions)))
235 def get_next(self, ret_type=None, start_time=None, update_current=True):
236 if start_time and self._expand_from_start_time:
237 raise ValueError("start_time is not supported when using expand_from_start_time = True.")
238 return self._get_next(ret_type or self._ret_type,
239 start_time=start_time,
240 is_prev=False,
241 update_current=update_current)
243 def get_prev(self, ret_type=None, start_time=None, update_current=True):
244 return self._get_next(ret_type or self._ret_type,
245 start_time=start_time,
246 is_prev=True,
247 update_current=update_current)
249 def get_current(self, ret_type=None):
250 ret_type = ret_type or self._ret_type
251 if issubclass(ret_type, datetime.datetime):
252 return self._timestamp_to_datetime(self.cur)
253 return self.cur
255 def set_current(self, start_time, force=True):
256 if (force or (self.cur is None)) and start_time is not None:
257 if isinstance(start_time, datetime.datetime):
258 self.tzinfo = start_time.tzinfo
259 start_time = self._datetime_to_timestamp(start_time)
261 self.start_time = start_time
262 self.dst_start_time = start_time
263 self.cur = start_time
264 return self.cur
266 @classmethod
267 def _datetime_to_timestamp(cls, d):
268 """
269 Converts a `datetime` object `d` into a UNIX timestamp.
270 """
271 return datetime_to_timestamp(d)
273 def _timestamp_to_datetime(self, timestamp):
274 """
275 Converts a UNIX timestamp `timestamp` into a `datetime` object.
276 """
277 result = datetime.datetime.fromtimestamp(timestamp, tz=tzutc()).replace(tzinfo=None)
278 if self.tzinfo:
279 result = result.replace(tzinfo=tzutc()).astimezone(self.tzinfo)
281 return result
283 @classmethod
284 def _timedelta_to_seconds(cls, td):
285 """
286 Converts a 'datetime.timedelta' object `td` into seconds contained in
287 the duration.
288 Note: We cannot use `timedelta.total_seconds()` because this is not
289 supported by Python 2.6.
290 """
291 return timedelta_to_seconds(td)
293 def _get_next(self, ret_type=None, start_time=None, is_prev=None, update_current=None):
294 if update_current is None:
295 update_current = True
296 self.set_current(start_time, force=True)
297 if is_prev is None:
298 is_prev = self._is_prev
299 self._is_prev = is_prev
300 expanded = self.expanded[:]
301 nth_weekday_of_month = self.nth_weekday_of_month.copy()
303 ret_type = ret_type or self._ret_type
305 if not issubclass(ret_type, (float, datetime.datetime)):
306 raise TypeError("Invalid ret_type, only 'float' or 'datetime' "
307 "is acceptable.")
309 # exception to support day of month and day of week as defined in cron
310 dom_dow_exception_processed = False
311 if (expanded[DAY_FIELD][0] != '*' and expanded[DOW_FIELD][0] != '*') and self._day_or:
312 # If requested, handle a bug in vixie cron/ISC cron where day_of_month and day_of_week form
313 # an intersection (AND) instead of a union (OR) if either field is an asterisk or starts with an asterisk
314 # (https://crontab.guru/cron-bug.html)
315 if (
316 self._implement_cron_bug and
317 (re_star.match(self.expressions[DAY_FIELD]) or re_star.match(self.expressions[DOW_FIELD]))
318 ):
319 # To produce a schedule identical to the cron bug, we'll bypass the code that
320 # makes a union of DOM and DOW, and instead skip to the code that does an intersect instead
321 pass
322 else:
323 bak = expanded[DOW_FIELD]
324 expanded[DOW_FIELD] = ['*']
325 t1 = self._calc(self.cur, expanded, nth_weekday_of_month, is_prev)
326 expanded[DOW_FIELD] = bak
327 expanded[DAY_FIELD] = ['*']
329 t2 = self._calc(self.cur, expanded, nth_weekday_of_month, is_prev)
330 if not is_prev:
331 result = t1 if t1 < t2 else t2
332 else:
333 result = t1 if t1 > t2 else t2
334 dom_dow_exception_processed = True
336 if not dom_dow_exception_processed:
337 result = self._calc(self.cur, expanded,
338 nth_weekday_of_month, is_prev)
340 # DST Handling for cron job spanning across days
341 dtstarttime = self._timestamp_to_datetime(self.dst_start_time)
342 dtstarttime_utcoffset = (
343 dtstarttime.utcoffset() or datetime.timedelta(0))
344 dtresult = self._timestamp_to_datetime(result)
345 lag = lag_hours = 0
346 # do we trigger DST on next crontab (handle backward changes)
347 dtresult_utcoffset = dtstarttime_utcoffset
348 if dtresult and self.tzinfo:
349 dtresult_utcoffset = dtresult.utcoffset()
350 lag_hours = (
351 self._timedelta_to_seconds(dtresult - dtstarttime) / (60 * 60)
352 )
353 lag = self._timedelta_to_seconds(
354 dtresult_utcoffset - dtstarttime_utcoffset
355 )
356 hours_before_midnight = 24 - dtstarttime.hour
357 if dtresult_utcoffset != dtstarttime_utcoffset:
358 if (
359 (lag > 0 and abs(lag_hours) >= hours_before_midnight)
360 or (lag < 0 and
361 ((3600 * abs(lag_hours) + abs(lag)) >= hours_before_midnight * 3600))
362 ):
363 dtresult_adjusted = dtresult - datetime.timedelta(seconds=lag)
364 result_adjusted = self._datetime_to_timestamp(dtresult_adjusted)
365 # Do the actual adjust only if the result time actually exists
366 if self._timestamp_to_datetime(result_adjusted).tzinfo == dtresult_adjusted.tzinfo:
367 dtresult = dtresult_adjusted
368 result = result_adjusted
369 self.dst_start_time = result
370 if update_current:
371 self.cur = result
372 if issubclass(ret_type, datetime.datetime):
373 result = dtresult
374 return result
376 # iterator protocol, to enable direct use of croniter
377 # objects in a loop, like "for dt in croniter('5 0 * * *'): ..."
378 # or for combining multiple croniters into single
379 # dates feed using 'itertools' module
380 def all_next(self, ret_type=None, start_time=None, update_current=None):
381 '''Generator of all consecutive dates. Can be used instead of
382 implicit call to __iter__, whenever non-default
383 'ret_type' has to be specified.
384 '''
385 # In a Python 3.7+ world: contextlib.suppress and contextlib.nullcontext could be used instead
386 try:
387 while True:
388 self._is_prev = False
389 yield self._get_next(ret_type or self._ret_type,
390 start_time=start_time, update_current=update_current)
391 start_time = None
392 except CroniterBadDateError:
393 if self._max_years_btw_matches_explicitly_set:
394 return
395 else:
396 raise
398 def all_prev(self, ret_type=None, start_time=None, update_current=None):
399 '''Generator of all previous dates.'''
400 try:
401 while True:
402 self._is_prev = True
403 yield self._get_next(ret_type or self._ret_type,
404 start_time=start_time, update_current=update_current)
405 start_time = None
406 except CroniterBadDateError:
407 if self._max_years_btw_matches_explicitly_set:
408 return
409 else:
410 raise
412 def iter(self, *args, **kwargs):
413 return (self._is_prev and self.all_prev or self.all_next)
415 def __iter__(self):
416 return self
417 __next__ = next = _get_next
419 def _calc(self, now, expanded, nth_weekday_of_month, is_prev):
420 if is_prev:
421 now = math.ceil(now)
422 nearest_diff_method = self._get_prev_nearest_diff
423 sign = -1
424 offset = (len(expanded) > UNIX_CRON_LEN or now % 60 > 0) and 1 or 60
425 else:
426 now = math.floor(now)
427 nearest_diff_method = self._get_next_nearest_diff
428 sign = 1
429 offset = (len(expanded) > UNIX_CRON_LEN) and 1 or 60
431 dst = now = self._timestamp_to_datetime(now + sign * offset)
433 month, year = dst.month, dst.year
434 current_year = now.year
435 DAYS = self.DAYS
437 def proc_year(d):
438 if len(expanded) == YEAR_CRON_LEN:
439 try:
440 expanded[YEAR_FIELD].index("*")
441 except ValueError:
442 # use None as range_val to indicate no loop
443 diff_year = nearest_diff_method(d.year, expanded[YEAR_FIELD], None)
444 if diff_year is None:
445 return None, d
446 elif diff_year != 0:
447 if is_prev:
448 d += relativedelta(years=diff_year, month=12, day=31,
449 hour=23, minute=59, second=59)
450 else:
451 d += relativedelta(years=diff_year, month=1, day=1,
452 hour=0, minute=0, second=0)
453 return True, d
454 return False, d
456 def proc_month(d):
457 try:
458 expanded[MONTH_FIELD].index('*')
459 except ValueError:
460 diff_month = nearest_diff_method(
461 d.month, expanded[MONTH_FIELD], self.MONTHS_IN_YEAR)
462 days = DAYS[month - 1]
463 if month == 2 and self.is_leap(year) is True:
464 days += 1
466 reset_day = 1
468 if diff_month is not None and diff_month != 0:
469 if is_prev:
470 d += relativedelta(months=diff_month)
471 reset_day = DAYS[d.month - 1]
472 if d.month == 2 and self.is_leap(d.year) is True:
473 reset_day += 1
474 d += relativedelta(
475 day=reset_day, hour=23, minute=59, second=59)
476 else:
477 d += relativedelta(months=diff_month, day=reset_day,
478 hour=0, minute=0, second=0)
479 return True, d
480 return False, d
482 def proc_day_of_month(d):
483 try:
484 expanded[DAY_FIELD].index('*')
485 except ValueError:
486 days = DAYS[month - 1]
487 if month == 2 and self.is_leap(year) is True:
488 days += 1
489 if 'l' in expanded[DAY_FIELD] and days == d.day:
490 return False, d
492 if is_prev:
493 days_in_prev_month = DAYS[
494 (month - 2) % self.MONTHS_IN_YEAR]
495 diff_day = nearest_diff_method(
496 d.day, expanded[DAY_FIELD], days_in_prev_month)
497 else:
498 diff_day = nearest_diff_method(d.day, expanded[DAY_FIELD], days)
500 if diff_day is not None and diff_day != 0:
501 if is_prev:
502 d += relativedelta(
503 days=diff_day, hour=23, minute=59, second=59)
504 else:
505 d += relativedelta(
506 days=diff_day, hour=0, minute=0, second=0)
507 return True, d
508 return False, d
510 def proc_day_of_week(d):
511 try:
512 expanded[DOW_FIELD].index('*')
513 except ValueError:
514 diff_day_of_week = nearest_diff_method(
515 d.isoweekday() % 7, expanded[DOW_FIELD], 7)
516 if diff_day_of_week is not None and diff_day_of_week != 0:
517 if is_prev:
518 d += relativedelta(days=diff_day_of_week,
519 hour=23, minute=59, second=59)
520 else:
521 d += relativedelta(days=diff_day_of_week,
522 hour=0, minute=0, second=0)
523 return True, d
524 return False, d
526 def proc_day_of_week_nth(d):
527 if '*' in nth_weekday_of_month:
528 s = nth_weekday_of_month['*']
529 for i in range(0, 7):
530 if i in nth_weekday_of_month:
531 nth_weekday_of_month[i].update(s)
532 else:
533 nth_weekday_of_month[i] = s
534 del nth_weekday_of_month['*']
536 candidates = []
537 for wday, nth in nth_weekday_of_month.items():
538 c = self._get_nth_weekday_of_month(d.year, d.month, wday)
539 for n in nth:
540 if n == "l":
541 candidate = c[-1]
542 elif len(c) < n:
543 continue
544 else:
545 candidate = c[n - 1]
546 if (
547 (is_prev and candidate <= d.day) or
548 (not is_prev and d.day <= candidate)
549 ):
550 candidates.append(candidate)
552 if not candidates:
553 if is_prev:
554 d += relativedelta(days=-d.day,
555 hour=23, minute=59, second=59)
556 else:
557 days = DAYS[month - 1]
558 if month == 2 and self.is_leap(year) is True:
559 days += 1
560 d += relativedelta(days=(days - d.day + 1),
561 hour=0, minute=0, second=0)
562 return True, d
564 candidates.sort()
565 diff_day = (candidates[-1] if is_prev else candidates[0]) - d.day
566 if diff_day != 0:
567 if is_prev:
568 d += relativedelta(days=diff_day,
569 hour=23, minute=59, second=59)
570 else:
571 d += relativedelta(days=diff_day,
572 hour=0, minute=0, second=0)
573 return True, d
574 return False, d
576 def proc_hour(d):
577 try:
578 expanded[HOUR_FIELD].index('*')
579 except ValueError:
580 diff_hour = nearest_diff_method(d.hour, expanded[HOUR_FIELD], 24)
581 if diff_hour is not None and diff_hour != 0:
582 if is_prev:
583 d += relativedelta(
584 hours=diff_hour, minute=59, second=59)
585 else:
586 d += relativedelta(hours=diff_hour, minute=0, second=0)
587 return True, d
588 return False, d
590 def proc_minute(d):
591 try:
592 expanded[MINUTE_FIELD].index('*')
593 except ValueError:
594 diff_min = nearest_diff_method(d.minute, expanded[MINUTE_FIELD], 60)
595 if diff_min is not None and diff_min != 0:
596 if is_prev:
597 d += relativedelta(minutes=diff_min, second=59)
598 else:
599 d += relativedelta(minutes=diff_min, second=0)
600 return True, d
601 return False, d
603 def proc_second(d):
604 if len(expanded) > UNIX_CRON_LEN:
605 try:
606 expanded[SECOND_FIELD].index('*')
607 except ValueError:
608 diff_sec = nearest_diff_method(d.second, expanded[SECOND_FIELD], 60)
609 if diff_sec is not None and diff_sec != 0:
610 d += relativedelta(seconds=diff_sec)
611 return True, d
612 else:
613 d += relativedelta(second=0)
614 return False, d
616 procs = [proc_year,
617 proc_month,
618 proc_day_of_month,
619 (proc_day_of_week_nth if nth_weekday_of_month
620 else proc_day_of_week),
621 proc_hour,
622 proc_minute,
623 proc_second]
625 while abs(year - current_year) <= self._max_years_between_matches:
626 next = False
627 stop = False
628 for proc in procs:
629 (changed, dst) = proc(dst)
630 # `None` can be set mostly for year processing
631 # so please see proc_year / _get_prev_nearest_diff / _get_next_nearest_diff
632 if changed is None:
633 stop = True
634 break
635 if changed:
636 month, year = dst.month, dst.year
637 next = True
638 break
639 if stop:
640 break
641 if next:
642 continue
643 return self._datetime_to_timestamp(dst.replace(microsecond=0))
645 if is_prev:
646 raise CroniterBadDateError("failed to find prev date")
647 raise CroniterBadDateError("failed to find next date")
649 def _get_next_nearest(self, x, to_check):
650 small = [item for item in to_check if item < x]
651 large = [item for item in to_check if item >= x]
652 large.extend(small)
653 return large[0]
655 def _get_prev_nearest(self, x, to_check):
656 small = [item for item in to_check if item <= x]
657 large = [item for item in to_check if item > x]
658 small.reverse()
659 large.reverse()
660 small.extend(large)
661 return small[0]
663 def _get_next_nearest_diff(self, x, to_check, range_val):
664 """
665 `range_val` is the range of a field.
666 If no available time, we can move to next loop(like next month).
667 `range_val` can also be set to `None` to indicate that there is no loop.
668 ( Currently, should only used for `year` field )
669 """
670 for i, d in enumerate(to_check):
671 if d == "l" and range_val is not None:
672 # if 'l' then it is the last day of month
673 # => its value of range_val
674 d = range_val
675 if d >= x:
676 return d - x
677 # When range_val is None and x not exists in to_check,
678 # `None` will be returned to suggest no more available time
679 if range_val is None:
680 return None
681 return to_check[0] - x + range_val
683 def _get_prev_nearest_diff(self, x, to_check, range_val):
684 """
685 `range_val` is the range of a field.
686 If no available time, we can move to previous loop(like previous month).
687 Range_val can also be set to `None` to indicate that there is no loop.
688 ( Currently should only used for `year` field )
689 """
690 candidates = to_check[:]
691 candidates.reverse()
692 for d in candidates:
693 if d != 'l' and d <= x:
694 return d - x
695 if 'l' in candidates:
696 return -x
697 # When range_val is None and x not exists in to_check,
698 # `None` will be returned to suggest no more available time
699 if range_val is None:
700 return None
701 candidate = candidates[0]
702 for c in candidates:
703 # fixed: c < range_val
704 # this code will reject all 31 day of month, 12 month, 59 second,
705 # 23 hour and so on.
706 # if candidates has just a element, this will not harmful.
707 # but candidates have multiple elements, then values equal to
708 # range_val will rejected.
709 if c <= range_val:
710 candidate = c
711 break
712 if candidate > range_val:
713 # fix crontab "0 6 30 3 *" condidates only a element,
714 # then get_prev error return 2021-03-02 06:00:00
715 return - x
716 return (candidate - x - range_val)
718 @staticmethod
719 def _get_nth_weekday_of_month(year, month, day_of_week):
720 """ For a given year/month return a list of days in nth-day-of-month order.
721 The last weekday of the month is always [-1].
722 """
723 w = (day_of_week + 6) % 7
724 c = calendar.Calendar(w).monthdayscalendar(year, month)
725 if c[0][0] == 0:
726 c.pop(0)
727 return tuple(i[0] for i in c)
729 def is_leap(self, year):
730 if year % 400 == 0 or (year % 4 == 0 and year % 100 != 0):
731 return True
732 else:
733 return False
735 @classmethod
736 def _expand(cls, expr_format, hash_id=None, second_at_beginning=False, from_timestamp=None):
737 # Split the expression in components, and normalize L -> l, MON -> mon,
738 # etc. Keep expr_format untouched so we can use it in the exception
739 # messages.
740 expr_aliases = {
741 '@midnight': ('0 0 * * *', 'h h(0-2) * * * h'),
742 '@hourly': ('0 * * * *', 'h * * * * h'),
743 '@daily': ('0 0 * * *', 'h h * * * h'),
744 '@weekly': ('0 0 * * 0', 'h h * * h h'),
745 '@monthly': ('0 0 1 * *', 'h h h * * h'),
746 '@yearly': ('0 0 1 1 *', 'h h h h * h'),
747 '@annually': ('0 0 1 1 *', 'h h h h * h'),
748 }
750 efl = expr_format.lower()
751 hash_id_expr = hash_id is not None and 1 or 0
752 try:
753 efl = expr_aliases[efl][hash_id_expr]
754 except KeyError:
755 pass
757 expressions = efl.split()
759 if len(expressions) not in VALID_LEN_EXPRESSION:
760 raise CroniterBadCronError(cls.bad_length)
762 if len(expressions) > UNIX_CRON_LEN and second_at_beginning:
763 # move second to it's own(6th) field to process by same logical
764 expressions.insert(SECOND_FIELD, expressions.pop(0))
766 expanded = []
767 nth_weekday_of_month = {}
769 for i, expr in enumerate(expressions):
770 for expanderid, expander in EXPANDERS.items():
771 expr = expander(cls).expand(efl, i, expr, hash_id=hash_id, from_timestamp=from_timestamp)
773 if "?" in expr:
774 if expr != "?":
775 raise CroniterBadCronError(
776 "[{0}] is not acceptable. Question mark can not "
777 "used with other characters".format(expr_format))
778 if i not in [DAY_FIELD, DOW_FIELD]:
779 raise CroniterBadCronError(
780 "[{0}] is not acceptable. Question mark can only used "
781 "in day_of_month or day_of_week".format(expr_format))
782 # currently just trade `?` as `*`
783 expr = "*"
785 e_list = expr.split(',')
786 res = []
788 while len(e_list) > 0:
789 e = e_list.pop()
790 nth = None
792 if i == DOW_FIELD:
793 # Handle special case in the dow expression: 2#3, l3
794 special_dow_rem = special_dow_re.match(str(e))
795 if special_dow_rem:
796 g = special_dow_rem.groupdict()
797 he, last = g.get('he', ''), g.get('last', '')
798 if he:
799 e = he
800 try:
801 nth = int(last)
802 assert (nth >= 1 and nth <= 5)
803 except (KeyError, ValueError, AssertionError):
804 raise CroniterBadCronError(
805 "[{0}] is not acceptable. Invalid day_of_week "
806 "value: '{1}'".format(expr_format, nth))
807 elif last:
808 e = last
809 nth = g['pre'] # 'l'
811 # Before matching step_search_re, normalize "*" to "{min}-{max}".
812 # Example: in the minute field, "*/5" normalizes to "0-59/5"
813 t = re.sub(r'^\*(\/.+)$', r'%d-%d\1' % (
814 cls.RANGES[i][0],
815 cls.RANGES[i][1]),
816 str(e))
817 m = step_search_re.search(t)
819 if not m:
820 # Before matching step_search_re,
821 # normalize "{start}/{step}" to "{start}-{max}/{step}".
822 # Example: in the minute field, "10/5" normalizes to "10-59/5"
823 t = re.sub(r'^(.+)\/(.+)$', r'\1-%d/\2' % (
824 cls.RANGES[i][1]),
825 str(e))
826 m = step_search_re.search(t)
828 if m:
829 # early abort if low/high are out of bounds
831 (low, high, step) = m.group(1), m.group(2), m.group(4) or 1
832 if i == DAY_FIELD and high == 'l':
833 high = '31'
835 if not only_int_re.search(low):
836 low = "{0}".format(cls._alphaconv(i, low, expressions))
838 if not only_int_re.search(high):
839 high = "{0}".format(cls._alphaconv(i, high, expressions))
841 if (
842 not low or not high or int(low) > int(high)
843 or not only_int_re.search(str(step))
844 ):
845 if i == DOW_FIELD and high == '0':
846 # handle -Sun notation -> 7
847 high = '7'
848 else:
849 raise CroniterBadCronError(
850 "[{0}] is not acceptable".format(expr_format))
852 low, high, step = map(int, [low, high, step])
853 if (
854 max(low, high) > max(cls.RANGES[i][0], cls.RANGES[i][1])
855 ):
856 raise CroniterBadCronError(
857 "{0} is out of bands".format(expr_format))
859 if from_timestamp:
860 low = cls._get_low_from_current_date_number(i, int(step), int(from_timestamp))
862 try:
863 rng = range(low, high + 1, step)
864 except ValueError as exc:
865 raise CroniterBadCronError(
866 'invalid range: {0}'.format(exc))
867 e_list += (["{0}#{1}".format(item, nth) for item in rng]
868 if i == DOW_FIELD and nth and nth != "l" else rng)
869 else:
870 if t.startswith('-'):
871 raise CroniterBadCronError((
872 "[{0}] is not acceptable,"
873 "negative numbers not allowed"
874 ).format(expr_format))
875 if not star_or_int_re.search(t):
876 t = cls._alphaconv(i, t, expressions)
878 try:
879 t = int(t)
880 except ValueError:
881 pass
883 if t in cls.LOWMAP[i] and not (
884 # do not support 0 as a month either for classical 5 fields cron,
885 # 6fields second repeat form or 7 fields year form
886 # but still let conversion happen if day field is shifted
887 (i in [DAY_FIELD, MONTH_FIELD] and len(expressions) == UNIX_CRON_LEN) or
888 (i in [MONTH_FIELD, DOW_FIELD] and len(expressions) == SECOND_CRON_LEN) or
889 (i in [DAY_FIELD, MONTH_FIELD, DOW_FIELD] and len(expressions) == YEAR_CRON_LEN)
890 ):
891 t = cls.LOWMAP[i][t]
893 if (
894 t not in ["*", "l"]
895 and (int(t) < cls.RANGES[i][0] or
896 int(t) > cls.RANGES[i][1])
897 ):
898 raise CroniterBadCronError(
899 "[{0}] is not acceptable, out of range".format(
900 expr_format))
902 res.append(t)
904 if i == DOW_FIELD and nth:
905 if t not in nth_weekday_of_month:
906 nth_weekday_of_month[t] = set()
907 nth_weekday_of_month[t].add(nth)
909 res = set(res)
910 res = sorted(res, key=lambda i: "{:02}".format(i) if isinstance(i, int) else i)
911 if len(res) == cls.LEN_MEANS_ALL[i]:
912 # Make sure the wildcard is used in the correct way (avoid over-optimization)
913 if (
914 (i == DAY_FIELD and '*' not in expressions[DOW_FIELD]) or
915 (i == DOW_FIELD and '*' not in expressions[DAY_FIELD])
916 ):
917 pass
918 else:
919 res = ['*']
921 expanded.append(['*'] if (len(res) == 1
922 and res[0] == '*')
923 else res)
925 # Check to make sure the dow combo in use is supported
926 if nth_weekday_of_month:
927 dow_expanded_set = set(expanded[DOW_FIELD])
928 dow_expanded_set = dow_expanded_set.difference(nth_weekday_of_month.keys())
929 dow_expanded_set.discard("*")
930 # Skip: if it's all weeks instead of wildcard
931 if dow_expanded_set and len(set(expanded[DOW_FIELD])) != cls.LEN_MEANS_ALL[DOW_FIELD]:
932 raise CroniterUnsupportedSyntaxError(
933 "day-of-week field does not support mixing literal values and nth day of week syntax. "
934 "Cron: '{}' dow={} vs nth={}".format(expr_format, dow_expanded_set, nth_weekday_of_month))
936 EXPRESSIONS[(expr_format, hash_id, second_at_beginning)] = expressions
937 return expanded, nth_weekday_of_month
939 @classmethod
940 def expand(cls, expr_format, hash_id=None, second_at_beginning=False, from_timestamp=None):
941 """Shallow non Croniter ValueError inside a nice CroniterBadCronError"""
942 try:
943 return cls._expand(expr_format, hash_id=hash_id,
944 second_at_beginning=second_at_beginning,
945 from_timestamp=from_timestamp)
946 except (ValueError,) as exc:
947 error_type, error_instance, traceback = sys.exc_info()
948 if isinstance(exc, CroniterError):
949 raise
950 if int(sys.version[0]) >= 3:
951 trace = _traceback.format_exc()
952 globs, locs = _get_caller_globals_and_locals()
953 raise CroniterBadCronError(trace)
954 else:
955 raise CroniterBadCronError("{0}".format(exc))
957 @classmethod
958 def _get_low_from_current_date_number(cls, i, step, from_timestamp):
959 dt = datetime.datetime.fromtimestamp(from_timestamp, tz=datetime.timezone.utc)
960 if i == MINUTE_FIELD:
961 return dt.minute % step
962 if i == HOUR_FIELD:
963 return dt.hour % step
964 if i == DAY_FIELD:
965 return ((dt.day - 1) % step) + 1
966 if i == MONTH_FIELD:
967 return dt.month % step
968 if i == DOW_FIELD:
969 return (dt.weekday() + 1) % step
971 raise ValueError("Can't get current date number for index larger than 4")
973 @classmethod
974 def is_valid(cls, expression, hash_id=None, encoding='UTF-8',
975 second_at_beginning=False):
976 if hash_id:
977 if not isinstance(hash_id, (bytes, str)):
978 raise TypeError('hash_id must be bytes or UTF-8 string')
979 if not isinstance(hash_id, bytes):
980 hash_id = hash_id.encode(encoding)
981 try:
982 cls.expand(expression, hash_id=hash_id,
983 second_at_beginning=second_at_beginning)
984 except CroniterError:
985 return False
986 else:
987 return True
989 @classmethod
990 def match(cls, cron_expression, testdate, day_or=True, second_at_beginning=False):
991 return cls.match_range(cron_expression, testdate, testdate, day_or, second_at_beginning)
993 @classmethod
994 def match_range(cls, cron_expression, from_datetime, to_datetime,
995 day_or=True, second_at_beginning=False):
996 cron = cls(cron_expression, to_datetime, ret_type=datetime.datetime,
997 day_or=day_or, second_at_beginning=second_at_beginning)
998 tdp = cron.get_current(datetime.datetime)
999 if not tdp.microsecond:
1000 tdp += relativedelta(microseconds=1)
1001 cron.set_current(tdp, force=True)
1002 try:
1003 tdt = cron.get_prev()
1004 except CroniterBadDateError:
1005 return False
1006 precision_in_seconds = 1 if len(cron.expanded) > UNIX_CRON_LEN else 60
1007 duration_in_second = (to_datetime - from_datetime).total_seconds() + precision_in_seconds
1008 return (max(tdp, tdt) - min(tdp, tdt)).total_seconds() < duration_in_second
1011def croniter_range(start, stop, expr_format, ret_type=None, day_or=True, exclude_ends=False,
1012 _croniter=None,
1013 second_at_beginning=False,
1014 expand_from_start_time=False):
1015 """
1016 Generator that provides all times from start to stop matching the given cron expression.
1017 If the cron expression matches either 'start' and/or 'stop', those times will be returned as
1018 well unless 'exclude_ends=True' is passed.
1020 You can think of this function as sibling to the builtin range function for datetime objects.
1021 Like range(start,stop,step), except that here 'step' is a cron expression.
1022 """
1023 _croniter = _croniter or croniter
1024 auto_rt = datetime.datetime
1025 # type is used in first if branch for perfs reasons
1026 if (
1027 type(start) is not type(stop) and not (
1028 isinstance(start, type(stop)) or
1029 isinstance(stop, type(start)))
1030 ):
1031 raise CroniterBadTypeRangeError(
1032 "The start and stop must be same type. {0} != {1}".
1033 format(type(start), type(stop)))
1034 if isinstance(start, (float, int)):
1035 start, stop = (datetime.datetime.fromtimestamp(t, tzutc()).replace(tzinfo=None) for t in (start, stop))
1036 auto_rt = float
1037 if ret_type is None:
1038 ret_type = auto_rt
1039 if not exclude_ends:
1040 ms1 = relativedelta(microseconds=1)
1041 if start < stop: # Forward (normal) time order
1042 start -= ms1
1043 stop += ms1
1044 else: # Reverse time order
1045 start += ms1
1046 stop -= ms1
1047 year_span = math.floor(abs(stop.year - start.year)) + 1
1048 ic = _croniter(expr_format, start, ret_type=datetime.datetime, day_or=day_or,
1049 max_years_between_matches=year_span,
1050 second_at_beginning=second_at_beginning,
1051 expand_from_start_time=expand_from_start_time)
1052 # define a continue (cont) condition function and step function for the main while loop
1053 if start < stop: # Forward
1054 def cont(v):
1055 return v < stop
1056 step = ic.get_next
1057 else: # Reverse
1058 def cont(v):
1059 return v > stop
1060 step = ic.get_prev
1061 try:
1062 dt = step()
1063 while cont(dt):
1064 if ret_type is float:
1065 yield ic.get_current(float)
1066 else:
1067 yield dt
1068 dt = step()
1069 except CroniterBadDateError:
1070 # Stop iteration when this exception is raised; no match found within the given year range
1071 return
1074class HashExpander:
1076 def __init__(self, cronit):
1077 self.cron = cronit
1079 def do(self, idx, hash_type="h", hash_id=None, range_end=None, range_begin=None):
1080 """Return a hashed/random integer given range/hash information"""
1081 if range_end is None:
1082 range_end = self.cron.RANGES[idx][1]
1083 if range_begin is None:
1084 range_begin = self.cron.RANGES[idx][0]
1085 if hash_type == 'r':
1086 crc = random.randint(0, 0xFFFFFFFF)
1087 else:
1088 crc = binascii.crc32(hash_id) & 0xFFFFFFFF
1089 return ((crc >> idx) % (range_end - range_begin + 1)) + range_begin
1091 def match(self, efl, idx, expr, hash_id=None, **kw):
1092 return hash_expression_re.match(expr)
1094 def expand(self, efl, idx, expr, hash_id=None, match='', **kw):
1095 """Expand a hashed/random expression to its normal representation"""
1096 if match == '':
1097 match = self.match(efl, idx, expr, hash_id, **kw)
1098 if not match:
1099 return expr
1100 m = match.groupdict()
1102 if m['hash_type'] == 'h' and hash_id is None:
1103 raise CroniterBadCronError('Hashed definitions must include hash_id')
1105 if m['range_begin'] and m['range_end']:
1106 if int(m['range_begin']) >= int(m['range_end']):
1107 raise CroniterBadCronError('Range end must be greater than range begin')
1109 if m['range_begin'] and m['range_end'] and m['divisor']:
1110 # Example: H(30-59)/10 -> 34-59/10 (i.e. 34,44,54)
1111 if int(m["divisor"]) == 0:
1112 raise CroniterBadCronError("Bad expression: {0}".format(expr))
1114 return '{0}-{1}/{2}'.format(
1115 self.do(
1116 idx,
1117 hash_type=m['hash_type'],
1118 hash_id=hash_id,
1119 range_begin=int(m['range_begin']),
1120 range_end=int(m['divisor']) - 1 + int(m['range_begin']),
1121 ),
1122 int(m['range_end']),
1123 int(m['divisor']),
1124 )
1125 elif m['range_begin'] and m['range_end']:
1126 # Example: H(0-29) -> 12
1127 return str(
1128 self.do(
1129 idx,
1130 hash_type=m['hash_type'],
1131 hash_id=hash_id,
1132 range_end=int(m['range_end']),
1133 range_begin=int(m['range_begin']),
1134 )
1135 )
1136 elif m['divisor']:
1137 # Example: H/15 -> 7-59/15 (i.e. 7,22,37,52)
1138 if int(m["divisor"]) == 0:
1139 raise CroniterBadCronError("Bad expression: {0}".format(expr))
1141 return '{0}-{1}/{2}'.format(
1142 self.do(
1143 idx,
1144 hash_type=m['hash_type'],
1145 hash_id=hash_id,
1146 range_begin=self.cron.RANGES[idx][0],
1147 range_end=int(m['divisor']) - 1 + self.cron.RANGES[idx][0],
1148 ),
1149 self.cron.RANGES[idx][1],
1150 int(m['divisor']),
1151 )
1152 else:
1153 # Example: H -> 32
1154 return str(
1155 self.do(
1156 idx,
1157 hash_type=m['hash_type'],
1158 hash_id=hash_id,
1159 )
1160 )
1163EXPANDERS = OrderedDict([
1164 ('hash', HashExpander),
1165])