Coverage for /usr/local/lib/python3.12/site-packages/prefect/_vendor/croniter/croniter.py: 14%
717 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 13:38 +0000
1#!/usr/bin/env python
2# -*- coding: utf-8 -*-
4from __future__ import absolute_import, division, print_function 1a
6import binascii 1a
7import calendar 1a
8import copy 1a
9import datetime 1a
10import math 1a
11import platform 1a
12import random 1a
13import re 1a
14import struct 1a
15import sys 1a
16import traceback as _traceback 1a
17from time import time 1a
19# as pytz is optional in thirdparty libs but we need it for good support under
20# python2, just test that it's well installed
21import pytz # noqa 1a
22from dateutil.relativedelta import relativedelta 1a
23from dateutil.tz import tzutc 1a
26def is_32bit(): 1a
27 """
28 Detect if Python is running in 32-bit mode.
29 Compatible with Python 2.6 and later versions.
30 Returns True if running on 32-bit Python, False for 64-bit.
31 """
32 # Method 1: Check pointer size
33 bits = struct.calcsize("P") * 8 1a
35 # Method 2: Check platform architecture string
36 try: 1a
37 architecture = platform.architecture()[0] 1a
38 except RuntimeError:
39 architecture = None
41 # Method 3: Check maxsize (sys.maxint in Python 2)
42 try: 1a
43 # Python 2
44 is_small_maxsize = sys.maxint <= 2**32 1a
45 except AttributeError: 1a
46 # Python 3
47 is_small_maxsize = sys.maxsize <= 2**32 1a
49 # Evaluate all available methods
50 is_32 = False 1a
52 if bits == 32: 52 ↛ 53line 52 didn't jump to line 53 because the condition on line 52 was never true1a
53 is_32 = True
54 elif architecture and "32" in architecture: 54 ↛ 55line 54 didn't jump to line 55 because the condition on line 54 was never true1a
55 is_32 = True
56 elif is_small_maxsize: 56 ↛ 57line 56 didn't jump to line 57 because the condition on line 56 was never true1a
57 is_32 = True
59 return is_32 1a
62try: 1a
63 # https://github.com/python/cpython/issues/101069 detection
64 if is_32bit(): 64 ↛ 65line 64 didn't jump to line 65 because the condition on line 64 was never true1a
65 datetime.datetime.fromtimestamp(3999999999)
66 OVERFLOW32B_MODE = False 1a
67except OverflowError:
68 OVERFLOW32B_MODE = True
70try: 1a
71 from collections import OrderedDict 1a
72except ImportError:
73 OrderedDict = dict # py26 degraded mode, expanders order will not be immutable
76try: 1a
77 # py3 recent
78 UTC_DT = datetime.timezone.utc 1a
79except AttributeError:
80 UTC_DT = pytz.utc
81EPOCH = datetime.datetime.fromtimestamp(0, UTC_DT) 1a
83# fmt: off
84M_ALPHAS = { 1a
85 "jan": 1, "feb": 2, "mar": 3, "apr": 4, # noqa: E241
86 "may": 5, "jun": 6, "jul": 7, "aug": 8, # noqa: E241
87 "sep": 9, "oct": 10, "nov": 11, "dec": 12,
88}
89DOW_ALPHAS = { 1a
90 "sun": 0, "mon": 1, "tue": 2, "wed": 3, "thu": 4, "fri": 5, "sat": 6
91}
93MINUTE_FIELD = 0 1a
94HOUR_FIELD = 1 1a
95DAY_FIELD = 2 1a
96MONTH_FIELD = 3 1a
97DOW_FIELD = 4 1a
98SECOND_FIELD = 5 1a
99YEAR_FIELD = 6 1a
101UNIX_FIELDS = (MINUTE_FIELD, HOUR_FIELD, DAY_FIELD, MONTH_FIELD, DOW_FIELD) # noqa: E222 1a
102SECOND_FIELDS = (MINUTE_FIELD, HOUR_FIELD, DAY_FIELD, MONTH_FIELD, DOW_FIELD, SECOND_FIELD) # noqa: E222 1a
103YEAR_FIELDS = (MINUTE_FIELD, HOUR_FIELD, DAY_FIELD, MONTH_FIELD, DOW_FIELD, SECOND_FIELD, YEAR_FIELD) # noqa: E222 1a
104# fmt: on
106step_search_re = re.compile(r"^([^-]+)-([^-/]+)(/(\d+))?$") 1a
107only_int_re = re.compile(r"^\d+$") 1a
109WEEKDAYS = "|".join(DOW_ALPHAS.keys()) 1a
110MONTHS = "|".join(M_ALPHAS.keys()) 1a
111star_or_int_re = re.compile(r"^(\d+|\*)$") 1a
112special_dow_re = re.compile( 1a
113 (r"^(?P<pre>((?P<he>(({WEEKDAYS})(-({WEEKDAYS}))?)").format(WEEKDAYS=WEEKDAYS)
114 + (r"|(({MONTHS})(-({MONTHS}))?)|\w+)#)|l)(?P<last>\d+)$").format(MONTHS=MONTHS)
115)
116re_star = re.compile("[*]") 1a
117hash_expression_re = re.compile( 1a
118 r"^(?P<hash_type>h|r)(\((?P<range_begin>\d+)-(?P<range_end>\d+)\))?(\/(?P<divisor>\d+))?$"
119)
121CRON_FIELDS = { 1a
122 "unix": UNIX_FIELDS,
123 "second": SECOND_FIELDS,
124 "year": YEAR_FIELDS,
125 len(UNIX_FIELDS): UNIX_FIELDS,
126 len(SECOND_FIELDS): SECOND_FIELDS,
127 len(YEAR_FIELDS): YEAR_FIELDS,
128}
129UNIX_CRON_LEN = len(UNIX_FIELDS) 1a
130SECOND_CRON_LEN = len(SECOND_FIELDS) 1a
131YEAR_CRON_LEN = len(YEAR_FIELDS) 1a
132# retrocompat
133VALID_LEN_EXPRESSION = set(a for a in CRON_FIELDS if isinstance(a, int)) 1a
134TIMESTAMP_TO_DT_CACHE = {} 1a
135EXPRESSIONS = {} 1a
136MARKER = object() 1a
139def timedelta_to_seconds(td): 1a
140 return (td.microseconds + (td.seconds + td.days * 24 * 3600) * 10**6) / 10**6
143def datetime_to_timestamp(d): 1a
144 if d.tzinfo is not None:
145 d = d.replace(tzinfo=None) - d.utcoffset()
147 return timedelta_to_seconds(d - datetime.datetime(1970, 1, 1))
150class CroniterError(ValueError): 1a
151 """General top-level Croniter base exception"""
154class CroniterBadTypeRangeError(TypeError): 1a
155 """."""
158class CroniterBadCronError(CroniterError): 1a
159 """Syntax, unknown value, or range error within a cron expression"""
162class CroniterUnsupportedSyntaxError(CroniterBadCronError): 1a
163 """Valid cron syntax, but likely to produce inaccurate results"""
165 # Extending CroniterBadCronError, which may be contridatory, but this allows
166 # catching both errors with a single exception. From a user perspective
167 # these will likely be handled the same way.
170class CroniterBadDateError(CroniterError): 1a
171 """Unable to find next/prev timestamp match"""
174class CroniterNotAlphaError(CroniterBadCronError): 1a
175 """Cron syntax contains an invalid day or month abbreviation"""
178class croniter(object): 1a
179 MONTHS_IN_YEAR = 12 1a
181 # This helps with expanding `*` fields into `lower-upper` ranges. Each item
182 # in this tuple maps to the corresponding field index
183 RANGES = ( 1a
184 (0, 59),
185 (0, 23),
186 (1, 31),
187 (1, 12),
188 (0, 6),
189 (0, 59),
190 (1970, 2099),
191 )
192 DAYS = (31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31) 1a
194 ALPHACONV = ( 1a
195 {}, # 0: min
196 {}, # 1: hour
197 {"l": "l"}, # 2: dom
198 # 3: mon
199 copy.deepcopy(M_ALPHAS),
200 # 4: dow
201 copy.deepcopy(DOW_ALPHAS),
202 # 5: second
203 {},
204 # 6: year
205 {},
206 )
208 LOWMAP = ( 1a
209 {},
210 {},
211 {0: 1},
212 {0: 1},
213 {7: 0},
214 {},
215 {},
216 )
218 LEN_MEANS_ALL = ( 1a
219 60,
220 24,
221 31,
222 12,
223 7,
224 60,
225 130,
226 )
228 def __init__( 1a
229 self,
230 expr_format,
231 start_time=None,
232 ret_type=float,
233 day_or=True,
234 max_years_between_matches=None,
235 is_prev=False,
236 hash_id=None,
237 implement_cron_bug=False,
238 second_at_beginning=None,
239 expand_from_start_time=False,
240 ):
241 self._ret_type = ret_type
242 self._day_or = day_or
243 self._implement_cron_bug = implement_cron_bug
244 self.second_at_beginning = bool(second_at_beginning)
245 self._expand_from_start_time = expand_from_start_time
247 if hash_id:
248 if not isinstance(hash_id, (bytes, str)):
249 raise TypeError("hash_id must be bytes or UTF-8 string")
250 if not isinstance(hash_id, bytes):
251 hash_id = hash_id.encode("UTF-8")
253 self._max_years_btw_matches_explicitly_set = (
254 max_years_between_matches is not None
255 )
256 if not self._max_years_btw_matches_explicitly_set:
257 max_years_between_matches = 50
258 self._max_years_between_matches = max(int(max_years_between_matches), 1)
260 if start_time is None:
261 start_time = time()
263 self.tzinfo = None
265 self.start_time = None
266 self.dst_start_time = None
267 self.cur = None
268 self.set_current(start_time, force=False)
270 self.expanded, self.nth_weekday_of_month = self.expand(
271 expr_format,
272 hash_id=hash_id,
273 from_timestamp=self.dst_start_time
274 if self._expand_from_start_time
275 else None,
276 second_at_beginning=second_at_beginning,
277 )
278 self.fields = CRON_FIELDS[len(self.expanded)]
279 self.expressions = EXPRESSIONS[(expr_format, hash_id, second_at_beginning)]
280 self._is_prev = is_prev
282 @classmethod 1a
283 def _alphaconv(cls, index, key, expressions): 1a
284 try:
285 return cls.ALPHACONV[index][key]
286 except KeyError:
287 raise CroniterNotAlphaError(
288 "[{0}] is not acceptable".format(" ".join(expressions))
289 )
291 def get_next(self, ret_type=None, start_time=None, update_current=True): 1a
292 if start_time and self._expand_from_start_time:
293 raise ValueError(
294 "start_time is not supported when using expand_from_start_time = True."
295 )
296 return self._get_next(
297 ret_type=ret_type,
298 start_time=start_time,
299 is_prev=False,
300 update_current=update_current,
301 )
303 def get_prev(self, ret_type=None, start_time=None, update_current=True): 1a
304 return self._get_next(
305 ret_type=ret_type,
306 start_time=start_time,
307 is_prev=True,
308 update_current=update_current,
309 )
311 def get_current(self, ret_type=None): 1a
312 ret_type = ret_type or self._ret_type
313 if issubclass(ret_type, datetime.datetime):
314 return self.timestamp_to_datetime(self.cur)
315 return self.cur
317 def set_current(self, start_time, force=True): 1a
318 if (force or (self.cur is None)) and start_time is not None:
319 if isinstance(start_time, datetime.datetime):
320 self.tzinfo = start_time.tzinfo
321 start_time = self.datetime_to_timestamp(start_time)
323 self.start_time = start_time
324 self.dst_start_time = start_time
325 self.cur = start_time
326 return self.cur
328 @staticmethod 1a
329 def datetime_to_timestamp(d): 1a
330 """
331 Converts a `datetime` object `d` into a UNIX timestamp.
332 """
333 return datetime_to_timestamp(d)
335 _datetime_to_timestamp = datetime_to_timestamp # retrocompat 1a
337 def timestamp_to_datetime(self, timestamp, tzinfo=MARKER): 1a
338 """
339 Converts a UNIX `timestamp` into a `datetime` object.
340 """
341 if tzinfo is MARKER: # allow to give tzinfo=None even if self.tzinfo is set
342 tzinfo = self.tzinfo
343 k = timestamp
344 if tzinfo:
345 k = (timestamp, repr(tzinfo))
346 try:
347 return TIMESTAMP_TO_DT_CACHE[k]
348 except KeyError:
349 pass
350 if OVERFLOW32B_MODE:
351 # degraded mode to workaround Y2038
352 # see https://github.com/python/cpython/issues/101069
353 result = EPOCH.replace(tzinfo=None) + datetime.timedelta(seconds=timestamp)
354 else:
355 result = datetime.datetime.fromtimestamp(timestamp, tz=tzutc()).replace(
356 tzinfo=None
357 )
358 if tzinfo:
359 result = result.replace(tzinfo=UTC_DT).astimezone(tzinfo)
360 TIMESTAMP_TO_DT_CACHE[(result, repr(result.tzinfo))] = result
361 return result
363 _timestamp_to_datetime = timestamp_to_datetime # retrocompat 1a
365 @staticmethod 1a
366 def timedelta_to_seconds(td): 1a
367 """
368 Converts a 'datetime.timedelta' object `td` into seconds contained in
369 the duration.
370 Note: We cannot use `timedelta.total_seconds()` because this is not
371 supported by Python 2.6.
372 """
373 return timedelta_to_seconds(td)
375 _timedelta_to_seconds = timedelta_to_seconds # retrocompat 1a
377 def _get_next( 1a
378 self,
379 ret_type=None,
380 start_time=None,
381 is_prev=None,
382 update_current=None,
383 ):
384 if update_current is None:
385 update_current = True
386 self.set_current(start_time, force=True)
387 if is_prev is None:
388 is_prev = self._is_prev
389 self._is_prev = is_prev
390 expanded = self.expanded[:]
391 nth_weekday_of_month = self.nth_weekday_of_month.copy()
393 ret_type = ret_type or self._ret_type
395 if not issubclass(ret_type, (float, datetime.datetime)):
396 raise TypeError(
397 "Invalid ret_type, only 'float' or 'datetime' is acceptable."
398 )
400 # exception to support day of month and day of week as defined in cron
401 dom_dow_exception_processed = False
402 if (
403 expanded[DAY_FIELD][0] != "*" and expanded[DOW_FIELD][0] != "*"
404 ) and self._day_or:
405 # If requested, handle a bug in vixie cron/ISC cron where day_of_month and day_of_week form
406 # an intersection (AND) instead of a union (OR) if either field is an asterisk or starts with an asterisk
407 # (https://crontab.guru/cron-bug.html)
408 if self._implement_cron_bug and (
409 re_star.match(self.expressions[DAY_FIELD])
410 or re_star.match(self.expressions[DOW_FIELD])
411 ):
412 # To produce a schedule identical to the cron bug, we'll bypass the code that
413 # makes a union of DOM and DOW, and instead skip to the code that does an intersect instead
414 pass
415 else:
416 bak = expanded[DOW_FIELD]
417 expanded[DOW_FIELD] = ["*"]
418 t1 = self._calc(self.cur, expanded, nth_weekday_of_month, is_prev)
419 expanded[DOW_FIELD] = bak
420 expanded[DAY_FIELD] = ["*"]
422 t2 = self._calc(self.cur, expanded, nth_weekday_of_month, is_prev)
423 if not is_prev:
424 result = t1 if t1 < t2 else t2
425 else:
426 result = t1 if t1 > t2 else t2
427 dom_dow_exception_processed = True
429 if not dom_dow_exception_processed:
430 result = self._calc(self.cur, expanded, nth_weekday_of_month, is_prev)
432 # DST Handling for cron job spanning across days
433 dtstarttime = self._timestamp_to_datetime(self.dst_start_time)
434 dtstarttime_utcoffset = dtstarttime.utcoffset() or datetime.timedelta(0)
435 dtresult = self.timestamp_to_datetime(result)
436 lag = lag_hours = 0
437 # do we trigger DST on next crontab (handle backward changes)
438 dtresult_utcoffset = dtstarttime_utcoffset
439 if dtresult and self.tzinfo:
440 dtresult_utcoffset = dtresult.utcoffset()
441 lag_hours = self._timedelta_to_seconds(dtresult - dtstarttime) / (60 * 60)
442 lag = self._timedelta_to_seconds(dtresult_utcoffset - dtstarttime_utcoffset)
443 hours_before_midnight = 24 - dtstarttime.hour
444 if dtresult_utcoffset != dtstarttime_utcoffset:
445 if (lag > 0 and abs(lag_hours) >= hours_before_midnight) or (
446 lag < 0
447 and ((3600 * abs(lag_hours) + abs(lag)) >= hours_before_midnight * 3600)
448 ):
449 dtresult_adjusted = dtresult - datetime.timedelta(seconds=lag)
450 result_adjusted = self._datetime_to_timestamp(dtresult_adjusted)
451 # Do the actual adjust only if the result time actually exists
452 if (
453 self._timestamp_to_datetime(result_adjusted).tzinfo
454 == dtresult_adjusted.tzinfo
455 ):
456 dtresult = dtresult_adjusted
457 result = result_adjusted
458 self.dst_start_time = result
459 if update_current:
460 self.cur = result
461 if issubclass(ret_type, datetime.datetime):
462 result = dtresult
463 return result
465 # iterator protocol, to enable direct use of croniter
466 # objects in a loop, like "for dt in croniter("5 0 * * *'): ..."
467 # or for combining multiple croniters into single
468 # dates feed using 'itertools' module
469 def all_next(self, ret_type=None, start_time=None, update_current=None): 1a
470 """
471 Returns a generator yielding consecutive dates.
473 May be used instead of an implicit call to __iter__ whenever a
474 non-default `ret_type` needs to be specified.
475 """
476 # In a Python 3.7+ world: contextlib.suppress and contextlib.nullcontext could be used instead
477 try:
478 while True:
479 self._is_prev = False
480 yield self._get_next(
481 ret_type=ret_type,
482 start_time=start_time,
483 update_current=update_current,
484 )
485 start_time = None
486 except CroniterBadDateError:
487 if self._max_years_btw_matches_explicitly_set:
488 return
489 raise
491 def all_prev(self, ret_type=None, start_time=None, update_current=None): 1a
492 """
493 Returns a generator yielding previous dates.
494 """
495 try:
496 while True:
497 self._is_prev = True
498 yield self._get_next(
499 ret_type=ret_type,
500 start_time=start_time,
501 update_current=update_current,
502 )
503 start_time = None
504 except CroniterBadDateError:
505 if self._max_years_btw_matches_explicitly_set:
506 return
507 raise
509 def iter(self, *args, **kwargs): 1a
510 return self.all_prev if self._is_prev else self.all_next
512 def __iter__(self): 1a
513 return self
515 __next__ = next = _get_next 1a
517 def _calc(self, now, expanded, nth_weekday_of_month, is_prev): 1a
518 if is_prev:
519 now = math.ceil(now)
520 nearest_diff_method = self._get_prev_nearest_diff
521 sign = -1
522 offset = 1 if (len(expanded) > UNIX_CRON_LEN or now % 60 > 0) else 60
523 else:
524 now = math.floor(now)
525 nearest_diff_method = self._get_next_nearest_diff
526 sign = 1
527 offset = 1 if (len(expanded) > UNIX_CRON_LEN) else 60
529 dst = now = self.timestamp_to_datetime(now + sign * offset)
531 month, year = dst.month, dst.year
532 current_year = now.year
533 DAYS = self.DAYS
535 def proc_year(d):
536 if len(expanded) == YEAR_CRON_LEN:
537 try:
538 expanded[YEAR_FIELD].index("*")
539 except ValueError:
540 # use None as range_val to indicate no loop
541 diff_year = nearest_diff_method(d.year, expanded[YEAR_FIELD], None)
542 if diff_year is None:
543 return None, d
544 if diff_year != 0:
545 if is_prev:
546 d += relativedelta(
547 years=diff_year,
548 month=12,
549 day=31,
550 hour=23,
551 minute=59,
552 second=59,
553 )
554 else:
555 d += relativedelta(
556 years=diff_year,
557 month=1,
558 day=1,
559 hour=0,
560 minute=0,
561 second=0,
562 )
563 return True, d
564 return False, d
566 def proc_month(d):
567 try:
568 expanded[MONTH_FIELD].index("*")
569 except ValueError:
570 diff_month = nearest_diff_method(
571 d.month, expanded[MONTH_FIELD], self.MONTHS_IN_YEAR
572 )
573 reset_day = 1
575 if diff_month is not None and diff_month != 0:
576 if is_prev:
577 d += relativedelta(months=diff_month)
578 reset_day = DAYS[d.month - 1]
579 if d.month == 2 and self.is_leap(d.year) is True:
580 reset_day += 1
581 d += relativedelta(day=reset_day, hour=23, minute=59, second=59)
582 else:
583 d += relativedelta(
584 months=diff_month, day=reset_day, hour=0, minute=0, second=0
585 )
586 return True, d
587 return False, d
589 def proc_day_of_month(d):
590 try:
591 expanded[DAY_FIELD].index("*")
592 except ValueError:
593 days = DAYS[month - 1]
594 if month == 2 and self.is_leap(year) is True:
595 days += 1
596 if "l" in expanded[DAY_FIELD] and days == d.day:
597 return False, d
599 if is_prev:
600 days_in_prev_month = DAYS[(month - 2) % self.MONTHS_IN_YEAR]
601 diff_day = nearest_diff_method(
602 d.day, expanded[DAY_FIELD], days_in_prev_month
603 )
604 else:
605 diff_day = nearest_diff_method(d.day, expanded[DAY_FIELD], days)
607 if diff_day is not None and diff_day != 0:
608 if is_prev:
609 d += relativedelta(days=diff_day, hour=23, minute=59, second=59)
610 else:
611 d += relativedelta(days=diff_day, hour=0, minute=0, second=0)
612 return True, d
613 return False, d
615 def proc_day_of_week(d):
616 try:
617 expanded[DOW_FIELD].index("*")
618 except ValueError:
619 diff_day_of_week = nearest_diff_method(
620 d.isoweekday() % 7, expanded[DOW_FIELD], 7
621 )
622 if diff_day_of_week is not None and diff_day_of_week != 0:
623 if is_prev:
624 d += relativedelta(
625 days=diff_day_of_week, hour=23, minute=59, second=59
626 )
627 else:
628 d += relativedelta(
629 days=diff_day_of_week, hour=0, minute=0, second=0
630 )
631 return True, d
632 return False, d
634 def proc_day_of_week_nth(d):
635 if "*" in nth_weekday_of_month:
636 s = nth_weekday_of_month["*"]
637 for i in range(0, 7):
638 if i in nth_weekday_of_month:
639 nth_weekday_of_month[i].update(s)
640 else:
641 nth_weekday_of_month[i] = s
642 del nth_weekday_of_month["*"]
644 candidates = []
645 for wday, nth in nth_weekday_of_month.items():
646 c = self._get_nth_weekday_of_month(d.year, d.month, wday)
647 for n in nth:
648 if n == "l":
649 candidate = c[-1]
650 elif len(c) < n:
651 continue
652 else:
653 candidate = c[n - 1]
654 if (is_prev and candidate <= d.day) or (
655 not is_prev and d.day <= candidate
656 ):
657 candidates.append(candidate)
659 if not candidates:
660 if is_prev:
661 d += relativedelta(days=-d.day, hour=23, minute=59, second=59)
662 else:
663 days = DAYS[month - 1]
664 if month == 2 and self.is_leap(year) is True:
665 days += 1
666 d += relativedelta(
667 days=(days - d.day + 1), hour=0, minute=0, second=0
668 )
669 return True, d
671 candidates.sort()
672 diff_day = (candidates[-1] if is_prev else candidates[0]) - d.day
673 if diff_day != 0:
674 if is_prev:
675 d += relativedelta(days=diff_day, hour=23, minute=59, second=59)
676 else:
677 d += relativedelta(days=diff_day, hour=0, minute=0, second=0)
678 return True, d
679 return False, d
681 def proc_hour(d):
682 try:
683 expanded[HOUR_FIELD].index("*")
684 except ValueError:
685 diff_hour = nearest_diff_method(d.hour, expanded[HOUR_FIELD], 24)
686 if diff_hour is not None and diff_hour != 0:
687 if is_prev:
688 d += relativedelta(hours=diff_hour, minute=59, second=59)
689 else:
690 d += relativedelta(hours=diff_hour, minute=0, second=0)
691 return True, d
692 return False, d
694 def proc_minute(d):
695 try:
696 expanded[MINUTE_FIELD].index("*")
697 except ValueError:
698 diff_min = nearest_diff_method(d.minute, expanded[MINUTE_FIELD], 60)
699 if diff_min is not None and diff_min != 0:
700 if is_prev:
701 d += relativedelta(minutes=diff_min, second=59)
702 else:
703 d += relativedelta(minutes=diff_min, second=0)
704 return True, d
705 return False, d
707 def proc_second(d):
708 if len(expanded) > UNIX_CRON_LEN:
709 try:
710 expanded[SECOND_FIELD].index("*")
711 except ValueError:
712 diff_sec = nearest_diff_method(d.second, expanded[SECOND_FIELD], 60)
713 if diff_sec is not None and diff_sec != 0:
714 d += relativedelta(seconds=diff_sec)
715 return True, d
716 else:
717 d += relativedelta(second=0)
718 return False, d
720 procs = [
721 proc_year,
722 proc_month,
723 proc_day_of_month,
724 (proc_day_of_week_nth if nth_weekday_of_month else proc_day_of_week),
725 proc_hour,
726 proc_minute,
727 proc_second,
728 ]
730 while abs(year - current_year) <= self._max_years_between_matches:
731 next = False
732 stop = False
733 for proc in procs:
734 (changed, dst) = proc(dst)
735 # `None` can be set mostly for year processing
736 # so please see proc_year / _get_prev_nearest_diff / _get_next_nearest_diff
737 if changed is None:
738 stop = True
739 break
740 if changed:
741 month, year = dst.month, dst.year
742 next = True
743 break
744 if stop:
745 break
746 if next:
747 continue
748 return self.datetime_to_timestamp(dst.replace(microsecond=0))
750 if is_prev:
751 raise CroniterBadDateError("failed to find prev date")
752 raise CroniterBadDateError("failed to find next date")
754 @staticmethod 1a
755 def _get_next_nearest(x, to_check): 1a
756 small = [item for item in to_check if item < x]
757 large = [item for item in to_check if item >= x]
758 large.extend(small)
759 return large[0]
761 @staticmethod 1a
762 def _get_prev_nearest(x, to_check): 1a
763 small = [item for item in to_check if item <= x]
764 large = [item for item in to_check if item > x]
765 small.reverse()
766 large.reverse()
767 small.extend(large)
768 return small[0]
770 @staticmethod 1a
771 def _get_next_nearest_diff(x, to_check, range_val): 1a
772 """
773 `range_val` is the range of a field.
774 If no available time, we can move to next loop(like next month).
775 `range_val` can also be set to `None` to indicate that there is no loop.
776 ( Currently, should only used for `year` field )
777 """
778 for i, d in enumerate(to_check):
779 if d == "l" and range_val is not None:
780 # if 'l' then it is the last day of month
781 # => its value of range_val
782 d = range_val
783 if d >= x:
784 return d - x
785 # When range_val is None and x not exists in to_check,
786 # `None` will be returned to suggest no more available time
787 if range_val is None:
788 return None
789 return to_check[0] - x + range_val
791 @staticmethod 1a
792 def _get_prev_nearest_diff(x, to_check, range_val): 1a
793 """
794 `range_val` is the range of a field.
795 If no available time, we can move to previous loop(like previous month).
796 Range_val can also be set to `None` to indicate that there is no loop.
797 ( Currently should only used for `year` field )
798 """
799 candidates = to_check[:]
800 candidates.reverse()
801 for d in candidates:
802 if d != "l" and d <= x:
803 return d - x
804 if "l" in candidates:
805 return -x
806 # When range_val is None and x not exists in to_check,
807 # `None` will be returned to suggest no more available time
808 if range_val is None:
809 return None
810 candidate = candidates[0]
811 for c in candidates:
812 # fixed: c < range_val
813 # this code will reject all 31 day of month, 12 month, 59 second,
814 # 23 hour and so on.
815 # if candidates has just a element, this will not harmful.
816 # but candidates have multiple elements, then values equal to
817 # range_val will rejected.
818 if c <= range_val:
819 candidate = c
820 break
821 # fix crontab "0 6 30 3 *" condidates only a element, then get_prev error return 2021-03-02 06:00:00
822 if candidate > range_val:
823 return -range_val
824 return candidate - x - range_val
826 @staticmethod 1a
827 def _get_nth_weekday_of_month(year, month, day_of_week): 1a
828 """For a given year/month return a list of days in nth-day-of-month order.
829 The last weekday of the month is always [-1].
830 """
831 w = (day_of_week + 6) % 7
832 c = calendar.Calendar(w).monthdayscalendar(year, month)
833 if c[0][0] == 0:
834 c.pop(0)
835 return tuple(i[0] for i in c)
837 @staticmethod 1a
838 def is_leap(year): 1a
839 return bool(year % 400 == 0 or (year % 4 == 0 and year % 100 != 0))
841 @classmethod 1a
842 def value_alias(cls, val, field_index, len_expressions=UNIX_CRON_LEN): 1a
843 if isinstance(len_expressions, (list, dict, tuple, set)):
844 len_expressions = len(len_expressions)
845 if val in cls.LOWMAP[field_index] and not (
846 # do not support 0 as a month either for classical 5 fields cron,
847 # 6fields second repeat form or 7 fields year form
848 # but still let conversion happen if day field is shifted
849 (
850 field_index in [DAY_FIELD, MONTH_FIELD]
851 and len_expressions == UNIX_CRON_LEN
852 )
853 or (
854 field_index in [MONTH_FIELD, DOW_FIELD]
855 and len_expressions == SECOND_CRON_LEN
856 )
857 or (
858 field_index in [DAY_FIELD, MONTH_FIELD, DOW_FIELD]
859 and len_expressions == YEAR_CRON_LEN
860 )
861 ):
862 val = cls.LOWMAP[field_index][val]
863 return val
865 @classmethod 1a
866 def _expand( 1a
867 cls,
868 expr_format,
869 hash_id=None,
870 second_at_beginning=False,
871 from_timestamp=None,
872 ):
873 # Split the expression in components, and normalize L -> l, MON -> mon,
874 # etc. Keep expr_format untouched so we can use it in the exception
875 # messages.
876 expr_aliases = {
877 "@midnight": ("0 0 * * *", "h h(0-2) * * * h"),
878 "@hourly": ("0 * * * *", "h * * * * h"),
879 "@daily": ("0 0 * * *", "h h * * * h"),
880 "@weekly": ("0 0 * * 0", "h h * * h h"),
881 "@monthly": ("0 0 1 * *", "h h h * * h"),
882 "@yearly": ("0 0 1 1 *", "h h h h * h"),
883 "@annually": ("0 0 1 1 *", "h h h h * h"),
884 }
886 efl = expr_format.lower()
887 hash_id_expr = 1 if hash_id is not None else 0
888 try:
889 efl = expr_aliases[efl][hash_id_expr]
890 except KeyError:
891 pass
893 expressions = efl.split()
895 if len(expressions) not in VALID_LEN_EXPRESSION:
896 raise CroniterBadCronError(
897 "Exactly 5, 6 or 7 columns has to be specified for iterator expression."
898 )
900 if len(expressions) > UNIX_CRON_LEN and second_at_beginning:
901 # move second to it's own(6th) field to process by same logical
902 expressions.insert(SECOND_FIELD, expressions.pop(0))
904 expanded = []
905 nth_weekday_of_month = {}
907 for field_index, expr in enumerate(expressions):
908 for expanderid, expander in EXPANDERS.items():
909 expr = expander(cls).expand(
910 efl,
911 field_index,
912 expr,
913 hash_id=hash_id,
914 from_timestamp=from_timestamp,
915 )
917 if "?" in expr:
918 if expr != "?":
919 raise CroniterBadCronError(
920 "[{0}] is not acceptable. Question mark can not used with other characters".format(
921 expr_format
922 )
923 )
924 if field_index not in [DAY_FIELD, DOW_FIELD]:
925 raise CroniterBadCronError(
926 "[{0}] is not acceptable. Question mark can only used in day_of_month or day_of_week".format(
927 expr_format
928 )
929 )
930 # currently just trade `?` as `*`
931 expr = "*"
933 e_list = expr.split(",")
934 res = []
936 while len(e_list) > 0:
937 e = e_list.pop()
938 nth = None
940 if field_index == DOW_FIELD:
941 # Handle special case in the dow expression: 2#3, l3
942 special_dow_rem = special_dow_re.match(str(e))
943 if special_dow_rem:
944 g = special_dow_rem.groupdict()
945 he, last = g.get("he", ""), g.get("last", "")
946 if he:
947 e = he
948 try:
949 nth = int(last)
950 assert 5 >= nth >= 1
951 except (KeyError, ValueError, AssertionError):
952 raise CroniterBadCronError(
953 "[{0}] is not acceptable. Invalid day_of_week value: '{1}'".format(
954 expr_format, nth
955 )
956 )
957 elif last:
958 e = last
959 nth = g["pre"] # 'l'
961 # Before matching step_search_re, normalize "*" to "{min}-{max}".
962 # Example: in the minute field, "*/5" normalizes to "0-59/5"
963 t = re.sub(
964 r"^\*(\/.+)$",
965 r"%d-%d\1"
966 % (cls.RANGES[field_index][0], cls.RANGES[field_index][1]),
967 str(e),
968 )
969 m = step_search_re.search(t)
971 if not m:
972 # Before matching step_search_re,
973 # normalize "{start}/{step}" to "{start}-{max}/{step}".
974 # Example: in the minute field, "10/5" normalizes to "10-59/5"
975 t = re.sub(
976 r"^(.+)\/(.+)$",
977 r"\1-%d/\2" % (cls.RANGES[field_index][1]),
978 str(e),
979 )
980 m = step_search_re.search(t)
982 if m:
983 # early abort if low/high are out of bounds
984 (low, high, step) = m.group(1), m.group(2), m.group(4) or 1
985 if field_index == DAY_FIELD and high == "l":
986 high = "31"
988 if not only_int_re.search(low):
989 low = "{0}".format(
990 cls._alphaconv(field_index, low, expressions)
991 )
993 if not only_int_re.search(high):
994 high = "{0}".format(
995 cls._alphaconv(field_index, high, expressions)
996 )
998 # normally, it's already guarded by the RE that should not accept not-int values.
999 if not only_int_re.search(str(step)):
1000 raise CroniterBadCronError(
1001 "[{0}] step '{2}' in field {1} is not acceptable".format(
1002 expr_format, field_index, step
1003 )
1004 )
1005 step = int(step)
1007 for band in low, high:
1008 if not only_int_re.search(str(band)):
1009 raise CroniterBadCronError(
1010 "[{0}] bands '{2}-{3}' in field {1} are not acceptable".format(
1011 expr_format, field_index, low, high
1012 )
1013 )
1015 low, high = [
1016 cls.value_alias(int(_val), field_index, expressions)
1017 for _val in (low, high)
1018 ]
1020 if max(low, high) > max(
1021 cls.RANGES[field_index][0], cls.RANGES[field_index][1]
1022 ):
1023 raise CroniterBadCronError(
1024 "{0} is out of bands".format(expr_format)
1025 )
1027 if from_timestamp:
1028 low = cls._get_low_from_current_date_number(
1029 field_index, int(step), int(from_timestamp)
1030 )
1032 # Handle when the second bound of the range is in backtracking order:
1033 # eg: X-Sun or X-7 (Sat-Sun) in DOW, or X-Jan (Apr-Jan) in MONTH
1034 if low > high:
1035 whole_field_range = list(
1036 range(
1037 cls.RANGES[field_index][0],
1038 cls.RANGES[field_index][1] + 1,
1039 1,
1040 )
1041 )
1042 # Add FirstBound -> ENDRANGE, respecting step
1043 rng = list(range(low, cls.RANGES[field_index][1] + 1, step))
1044 # Then 0 -> SecondBound, but skipping n first occurences according to step
1045 # EG to respect such expressions : Apr-Jan/3
1046 to_skip = 0
1047 if rng:
1048 already_skipped = list(reversed(whole_field_range)).index(
1049 rng[-1]
1050 )
1051 curpos = whole_field_range.index(rng[-1])
1052 if ((curpos + step) > len(whole_field_range)) and (
1053 already_skipped < step
1054 ):
1055 to_skip = step - already_skipped
1056 rng += list(
1057 range(cls.RANGES[field_index][0] + to_skip, high + 1, step)
1058 )
1059 # if we include a range type: Jan-Jan, or Sun-Sun,
1060 # it means the whole cycle (all days of week, # all monthes of year, etc)
1061 elif low == high:
1062 rng = list(
1063 range(
1064 cls.RANGES[field_index][0],
1065 cls.RANGES[field_index][1] + 1,
1066 step,
1067 )
1068 )
1069 else:
1070 try:
1071 rng = list(range(low, high + 1, step))
1072 except ValueError as exc:
1073 raise CroniterBadCronError("invalid range: {0}".format(exc))
1075 rng = (
1076 ["{0}#{1}".format(item, nth) for item in rng]
1077 if field_index == DOW_FIELD and nth and nth != "l"
1078 else rng
1079 )
1080 e_list += [a for a in rng if a not in e_list]
1081 else:
1082 if t.startswith("-"):
1083 raise CroniterBadCronError(
1084 "[{0}] is not acceptable,"
1085 "negative numbers not allowed".format(expr_format)
1086 )
1087 if not star_or_int_re.search(t):
1088 t = cls._alphaconv(field_index, t, expressions)
1090 try:
1091 t = int(t)
1092 except ValueError:
1093 pass
1095 t = cls.value_alias(t, field_index, expressions)
1097 if t not in ["*", "l"] and (
1098 int(t) < cls.RANGES[field_index][0]
1099 or int(t) > cls.RANGES[field_index][1]
1100 ):
1101 raise CroniterBadCronError(
1102 "[{0}] is not acceptable, out of range".format(expr_format)
1103 )
1105 res.append(t)
1107 if field_index == DOW_FIELD and nth:
1108 if t not in nth_weekday_of_month:
1109 nth_weekday_of_month[t] = set()
1110 nth_weekday_of_month[t].add(nth)
1112 res = set(res)
1113 res = sorted(
1114 res, key=lambda i: "{:02}".format(i) if isinstance(i, int) else i
1115 )
1116 if len(res) == cls.LEN_MEANS_ALL[field_index]:
1117 # Make sure the wildcard is used in the correct way (avoid over-optimization)
1118 if (field_index == DAY_FIELD and "*" not in expressions[DOW_FIELD]) or (
1119 field_index == DOW_FIELD and "*" not in expressions[DAY_FIELD]
1120 ):
1121 pass
1122 else:
1123 res = ["*"]
1125 expanded.append(["*"] if (len(res) == 1 and res[0] == "*") else res)
1127 # Check to make sure the dow combo in use is supported
1128 if nth_weekday_of_month:
1129 dow_expanded_set = set(expanded[DOW_FIELD])
1130 dow_expanded_set = dow_expanded_set.difference(nth_weekday_of_month.keys())
1131 dow_expanded_set.discard("*")
1132 # Skip: if it's all weeks instead of wildcard
1133 if (
1134 dow_expanded_set
1135 and len(set(expanded[DOW_FIELD])) != cls.LEN_MEANS_ALL[DOW_FIELD]
1136 ):
1137 raise CroniterUnsupportedSyntaxError(
1138 "day-of-week field does not support mixing literal values and nth day of week syntax. "
1139 "Cron: '{}' dow={} vs nth={}".format(
1140 expr_format, dow_expanded_set, nth_weekday_of_month
1141 )
1142 )
1144 EXPRESSIONS[(expr_format, hash_id, second_at_beginning)] = expressions
1145 return expanded, nth_weekday_of_month
1147 @classmethod 1a
1148 def expand( 1a
1149 cls,
1150 expr_format,
1151 hash_id=None,
1152 second_at_beginning=False,
1153 from_timestamp=None,
1154 ):
1155 """
1156 Expand a cron expression format into a noramlized format of
1157 list[list[int | 'l' | '*']]. The first list representing each element
1158 of the epxression, and each sub-list representing the allowed values
1159 for that expression component.
1161 A tuple is returned, the first value being the expanded epxression
1162 list, and the second being a `nth_weekday_of_month` mapping.
1164 Examples:
1166 # Every minute
1167 >>> croniter.expand('* * * * *')
1168 ([['*'], ['*'], ['*'], ['*'], ['*']], {})
1170 # On the hour
1171 >>> croniter.expand('0 0 * * *')
1172 ([[0], [0], ['*'], ['*'], ['*']], {})
1174 # Hours 0-5 and 10 monday through friday
1175 >>> croniter.expand('0-5,10 * * * mon-fri')
1176 ([[0, 1, 2, 3, 4, 5, 10], ['*'], ['*'], ['*'], [1, 2, 3, 4, 5]], {})
1178 Note that some special values such as nth day of week are expanded to a
1179 special mapping format for later processing:
1181 # Every minute on the 3rd tuesday of the month
1182 >>> croniter.expand('* * * * 2#3')
1183 ([['*'], ['*'], ['*'], ['*'], [2]], {2: {3}})
1185 # Every hour on the last day of the month
1186 >>> croniter.expand('0 * l * *')
1187 ([[0], ['*'], ['l'], ['*'], ['*']], {})
1189 # On the hour every 15 seconds
1190 >>> croniter.expand('0 0 * * * */15')
1191 ([[0], [0], ['*'], ['*'], ['*'], [0, 15, 30, 45]], {})
1192 """
1193 try:
1194 return cls._expand(
1195 expr_format,
1196 hash_id=hash_id,
1197 second_at_beginning=second_at_beginning,
1198 from_timestamp=from_timestamp,
1199 )
1200 except (ValueError,) as exc:
1201 if isinstance(exc, CroniterError):
1202 raise
1203 if int(sys.version[0]) >= 3:
1204 trace = _traceback.format_exc()
1205 raise CroniterBadCronError(trace)
1206 raise CroniterBadCronError("{0}".format(exc))
1208 @classmethod 1a
1209 def _get_low_from_current_date_number(cls, field_index, step, from_timestamp): 1a
1210 dt = datetime.datetime.fromtimestamp(from_timestamp, tz=UTC_DT)
1211 if field_index == MINUTE_FIELD:
1212 return dt.minute % step
1213 if field_index == HOUR_FIELD:
1214 return dt.hour % step
1215 if field_index == DAY_FIELD:
1216 return ((dt.day - 1) % step) + 1
1217 if field_index == MONTH_FIELD:
1218 return dt.month % step
1219 if field_index == DOW_FIELD:
1220 return (dt.weekday() + 1) % step
1222 raise ValueError("Can't get current date number for index larger than 4")
1224 @classmethod 1a
1225 def is_valid( 1a
1226 cls,
1227 expression,
1228 hash_id=None,
1229 encoding="UTF-8",
1230 second_at_beginning=False,
1231 ):
1232 if hash_id:
1233 if not isinstance(hash_id, (bytes, str)):
1234 raise TypeError("hash_id must be bytes or UTF-8 string")
1235 if not isinstance(hash_id, bytes):
1236 hash_id = hash_id.encode(encoding)
1237 try:
1238 cls.expand(
1239 expression, hash_id=hash_id, second_at_beginning=second_at_beginning
1240 )
1241 except CroniterError:
1242 return False
1243 return True
1245 @classmethod 1a
1246 def match(cls, cron_expression, testdate, day_or=True, second_at_beginning=False): 1a
1247 return cls.match_range(
1248 cron_expression, testdate, testdate, day_or, second_at_beginning
1249 )
1251 @classmethod 1a
1252 def match_range( 1a
1253 cls,
1254 cron_expression,
1255 from_datetime,
1256 to_datetime,
1257 day_or=True,
1258 second_at_beginning=False,
1259 ):
1260 cron = cls(
1261 cron_expression,
1262 to_datetime,
1263 ret_type=datetime.datetime,
1264 day_or=day_or,
1265 second_at_beginning=second_at_beginning,
1266 )
1267 tdp = cron.get_current(datetime.datetime)
1268 if not tdp.microsecond:
1269 tdp += relativedelta(microseconds=1)
1270 cron.set_current(tdp, force=True)
1271 try:
1272 tdt = cron.get_prev()
1273 except CroniterBadDateError:
1274 return False
1275 precision_in_seconds = 1 if len(cron.expanded) > UNIX_CRON_LEN else 60
1276 duration_in_second = (
1277 to_datetime - from_datetime
1278 ).total_seconds() + precision_in_seconds
1279 return (max(tdp, tdt) - min(tdp, tdt)).total_seconds() < duration_in_second
1282def croniter_range( 1a
1283 start,
1284 stop,
1285 expr_format,
1286 ret_type=None,
1287 day_or=True,
1288 exclude_ends=False,
1289 _croniter=None,
1290 second_at_beginning=False,
1291 expand_from_start_time=False,
1292):
1293 """
1294 Generator that provides all times from start to stop matching the given cron expression.
1295 If the cron expression matches either 'start' and/or 'stop', those times will be returned as
1296 well unless 'exclude_ends=True' is passed.
1298 You can think of this function as sibling to the builtin range function for datetime objects.
1299 Like range(start,stop,step), except that here 'step' is a cron expression.
1300 """
1301 _croniter = _croniter or croniter
1302 auto_rt = datetime.datetime
1303 # type is used in first if branch for perfs reasons
1304 if type(start) is not type(stop) and not (
1305 isinstance(start, type(stop)) or isinstance(stop, type(start))
1306 ):
1307 raise CroniterBadTypeRangeError(
1308 "The start and stop must be same type. {0} != {1}".format(
1309 type(start), type(stop)
1310 )
1311 )
1312 if isinstance(start, (float, int)):
1313 start, stop = (
1314 datetime.datetime.fromtimestamp(t, tzutc()).replace(tzinfo=None)
1315 for t in (start, stop)
1316 )
1317 auto_rt = float
1318 if ret_type is None:
1319 ret_type = auto_rt
1320 if not exclude_ends:
1321 ms1 = relativedelta(microseconds=1)
1322 if start < stop: # Forward (normal) time order
1323 start -= ms1
1324 stop += ms1
1325 else: # Reverse time order
1326 start += ms1
1327 stop -= ms1
1328 year_span = math.floor(abs(stop.year - start.year)) + 1
1329 ic = _croniter(
1330 expr_format,
1331 start,
1332 ret_type=datetime.datetime,
1333 day_or=day_or,
1334 max_years_between_matches=year_span,
1335 second_at_beginning=second_at_beginning,
1336 expand_from_start_time=expand_from_start_time,
1337 )
1338 # define a continue (cont) condition function and step function for the main while loop
1339 if start < stop: # Forward
1341 def cont(v):
1342 return v < stop
1344 step = ic.get_next
1345 else: # Reverse
1347 def cont(v):
1348 return v > stop
1350 step = ic.get_prev
1351 try:
1352 dt = step()
1353 while cont(dt):
1354 if ret_type is float:
1355 yield ic.get_current(float)
1356 else:
1357 yield dt
1358 dt = step()
1359 except CroniterBadDateError:
1360 # Stop iteration when this exception is raised; no match found within the given year range
1361 return
1364class HashExpander: 1a
1365 def __init__(self, cronit): 1a
1366 self.cron = cronit
1368 def do(self, idx, hash_type="h", hash_id=None, range_end=None, range_begin=None): 1a
1369 """Return a hashed/random integer given range/hash information"""
1370 if range_end is None:
1371 range_end = self.cron.RANGES[idx][1]
1372 if range_begin is None:
1373 range_begin = self.cron.RANGES[idx][0]
1374 if hash_type == "r":
1375 crc = random.randint(0, 0xFFFFFFFF)
1376 else:
1377 crc = binascii.crc32(hash_id) & 0xFFFFFFFF
1378 return ((crc >> idx) % (range_end - range_begin + 1)) + range_begin
1380 def match(self, efl, idx, expr, hash_id=None, **kw): 1a
1381 return hash_expression_re.match(expr)
1383 def expand(self, efl, idx, expr, hash_id=None, match="", **kw): 1a
1384 """Expand a hashed/random expression to its normal representation"""
1385 if match == "":
1386 match = self.match(efl, idx, expr, hash_id, **kw)
1387 if not match:
1388 return expr
1389 m = match.groupdict()
1391 if m["hash_type"] == "h" and hash_id is None:
1392 raise CroniterBadCronError("Hashed definitions must include hash_id")
1394 if m["range_begin"] and m["range_end"]:
1395 if int(m["range_begin"]) >= int(m["range_end"]):
1396 raise CroniterBadCronError("Range end must be greater than range begin")
1398 if m["range_begin"] and m["range_end"] and m["divisor"]:
1399 # Example: H(30-59)/10 -> 34-59/10 (i.e. 34,44,54)
1400 if int(m["divisor"]) == 0:
1401 raise CroniterBadCronError("Bad expression: {0}".format(expr))
1403 return "{0}-{1}/{2}".format(
1404 self.do(
1405 idx,
1406 hash_type=m["hash_type"],
1407 hash_id=hash_id,
1408 range_begin=int(m["range_begin"]),
1409 range_end=int(m["divisor"]) - 1 + int(m["range_begin"]),
1410 ),
1411 int(m["range_end"]),
1412 int(m["divisor"]),
1413 )
1414 elif m["range_begin"] and m["range_end"]:
1415 # Example: H(0-29) -> 12
1416 return str(
1417 self.do(
1418 idx,
1419 hash_type=m["hash_type"],
1420 hash_id=hash_id,
1421 range_end=int(m["range_end"]),
1422 range_begin=int(m["range_begin"]),
1423 )
1424 )
1425 elif m["divisor"]:
1426 # Example: H/15 -> 7-59/15 (i.e. 7,22,37,52)
1427 if int(m["divisor"]) == 0:
1428 raise CroniterBadCronError("Bad expression: {0}".format(expr))
1430 return "{0}-{1}/{2}".format(
1431 self.do(
1432 idx,
1433 hash_type=m["hash_type"],
1434 hash_id=hash_id,
1435 range_begin=self.cron.RANGES[idx][0],
1436 range_end=int(m["divisor"]) - 1 + self.cron.RANGES[idx][0],
1437 ),
1438 self.cron.RANGES[idx][1],
1439 int(m["divisor"]),
1440 )
1441 else:
1442 # Example: H -> 32
1443 return str(
1444 self.do(
1445 idx,
1446 hash_type=m["hash_type"],
1447 hash_id=hash_id,
1448 )
1449 )
1452EXPANDERS = OrderedDict( 1a
1453 [
1454 ("hash", HashExpander),
1455 ]
1456)