Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/croniter/croniter.py: 82%
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
2import binascii
3import calendar
4import copy
5import datetime
6import math
7import platform
8import random
9import re
10import struct
11import sys
12import traceback as _traceback
13from time import time
14from typing import Any, Literal, Optional, Union
16from dateutil.relativedelta import relativedelta
17from dateutil.tz import datetime_exists, tzutc
19ExpandedExpression = list[Union[int, Literal["*", "l"]]]
22def is_32bit() -> bool:
23 """
24 Detect if Python is running in 32-bit mode.
25 Returns True if running on 32-bit Python, False for 64-bit.
26 """
27 # Method 1: Check pointer size
28 bits = struct.calcsize("P") * 8
30 # Method 2: Check platform architecture string
31 try:
32 architecture = platform.architecture()[0]
33 except RuntimeError:
34 architecture = None
36 # Method 3: Check maxsize
37 is_small_maxsize = sys.maxsize <= 2**32
39 # Evaluate all available methods
40 is_32 = False
42 if bits == 32:
43 is_32 = True
44 elif architecture and "32" in architecture:
45 is_32 = True
46 elif is_small_maxsize:
47 is_32 = True
49 return is_32
52try:
53 # https://github.com/python/cpython/issues/101069 detection
54 if is_32bit():
55 datetime.datetime.fromtimestamp(3999999999)
56 OVERFLOW32B_MODE = False
57except OverflowError:
58 OVERFLOW32B_MODE = True
61UTC_DT = datetime.timezone.utc
62EPOCH = datetime.datetime.fromtimestamp(0, UTC_DT)
64M_ALPHAS: dict[str, Union[int, str]] = {
65 "jan": 1,
66 "feb": 2,
67 "mar": 3,
68 "apr": 4,
69 "may": 5,
70 "jun": 6,
71 "jul": 7,
72 "aug": 8,
73 "sep": 9,
74 "oct": 10,
75 "nov": 11,
76 "dec": 12,
77}
78DOW_ALPHAS: dict[str, Union[int, str]] = {
79 "sun": 0,
80 "mon": 1,
81 "tue": 2,
82 "wed": 3,
83 "thu": 4,
84 "fri": 5,
85 "sat": 6,
86}
88MINUTE_FIELD = 0
89HOUR_FIELD = 1
90DAY_FIELD = 2
91MONTH_FIELD = 3
92DOW_FIELD = 4
93SECOND_FIELD = 5
94YEAR_FIELD = 6
96UNIX_FIELDS = (MINUTE_FIELD, HOUR_FIELD, DAY_FIELD, MONTH_FIELD, DOW_FIELD)
97SECOND_FIELDS = (MINUTE_FIELD, HOUR_FIELD, DAY_FIELD, MONTH_FIELD, DOW_FIELD, SECOND_FIELD)
98YEAR_FIELDS = (
99 MINUTE_FIELD,
100 HOUR_FIELD,
101 DAY_FIELD,
102 MONTH_FIELD,
103 DOW_FIELD,
104 SECOND_FIELD,
105 YEAR_FIELD,
106)
108step_search_re = re.compile(r"^([^-]+)-([^-/]+)(/(\d+))?$")
109only_int_re = re.compile(r"^\d+$")
111DAYS = (31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31)
112WEEKDAYS = "|".join(DOW_ALPHAS.keys())
113MONTHS = "|".join(M_ALPHAS.keys())
114star_or_int_re = re.compile(r"^(\d+|\*)$")
115special_dow_re = re.compile(
116 rf"^(?P<pre>((?P<he>(({WEEKDAYS})(-({WEEKDAYS}))?)"
117 rf"|(({MONTHS})(-({MONTHS}))?)|\w+)#)|l)(?P<last>\d+)$"
118)
119nearest_weekday_re = re.compile(r"^(?:(\d+)w|w(\d+))$")
120re_star = re.compile("[*]")
121hash_expression_re = re.compile(
122 r"^(?P<hash_type>h|r)(\((?P<range_begin>\d+)-(?P<range_end>\d+)\))?(\/(?P<divisor>\d+))?$"
123)
125CRON_FIELDS = {
126 "unix": UNIX_FIELDS,
127 "second": SECOND_FIELDS,
128 "year": YEAR_FIELDS,
129 len(UNIX_FIELDS): UNIX_FIELDS,
130 len(SECOND_FIELDS): SECOND_FIELDS,
131 len(YEAR_FIELDS): YEAR_FIELDS,
132}
133UNIX_CRON_LEN = len(UNIX_FIELDS)
134SECOND_CRON_LEN = len(SECOND_FIELDS)
135YEAR_CRON_LEN = len(YEAR_FIELDS)
136# retrocompat
137VALID_LEN_EXPRESSION = {a for a in CRON_FIELDS if isinstance(a, int)}
139MARKER = object()
142def datetime_to_timestamp(d):
143 if d.tzinfo is not None:
144 d = d.replace(tzinfo=None) - d.utcoffset()
146 return (d - datetime.datetime(1970, 1, 1)).total_seconds()
149def _is_leap(year: int) -> bool:
150 return year % 400 == 0 or (year % 4 == 0 and year % 100 != 0)
153def _last_day_of_month(year: int, month: int) -> int:
154 """Calculate the last day of the given month (honor leap years)."""
155 last_day = DAYS[month - 1]
156 if month == 2 and _is_leap(year):
157 last_day += 1
158 return last_day
161def _is_successor(
162 date: datetime.datetime, previous_date: datetime.datetime, is_prev: bool
163) -> bool:
164 """Check if the given date is a successor (after/before) of the previous date."""
165 if is_prev:
166 return date.astimezone(UTC_DT) < previous_date.astimezone(UTC_DT)
167 return date.astimezone(UTC_DT) > previous_date.astimezone(UTC_DT)
170def _timezone_delta(date1: datetime.datetime, date2: datetime.datetime) -> datetime.timedelta:
171 """Calculate the timezone difference of the given dates."""
172 offset1 = date1.utcoffset()
173 offset2 = date2.utcoffset()
174 assert offset1 is not None
175 assert offset2 is not None
176 return offset2 - offset1
179def _add_tzinfo(
180 date: datetime.datetime, previous_date: datetime.datetime, is_prev: bool
181) -> tuple[datetime.datetime, bool]:
182 """Add the tzinfo from the previous date to the given date.
184 In case the new date is ambiguous, determine the correct date
185 based on it being closer to the previous date but still a successor
186 (after/before based on `is_prev`).
188 In case the date does not exist, jump forward to the next existing date.
189 """
190 localize = getattr(previous_date.tzinfo, "localize", None)
191 if localize is not None:
192 # pylint: disable-next=import-outside-toplevel
193 import pytz
195 try:
196 result = localize(date, is_dst=None)
197 except pytz.NonExistentTimeError:
198 while True:
199 date += datetime.timedelta(minutes=1)
200 try:
201 result = localize(date, is_dst=None)
202 except pytz.NonExistentTimeError:
203 continue
204 break
205 return result, False
206 except pytz.AmbiguousTimeError:
207 closer = localize(date, is_dst=not is_prev)
208 farther = localize(date, is_dst=is_prev)
209 # TODO: Check negative DST
210 assert (closer.astimezone(UTC_DT) > farther.astimezone(UTC_DT)) == is_prev
211 if _is_successor(closer, previous_date, is_prev):
212 result = closer
213 else:
214 assert _is_successor(farther, previous_date, is_prev)
215 result = farther
216 return result, True
218 result = date.replace(fold=1 if is_prev else 0, tzinfo=previous_date.tzinfo)
219 if not datetime_exists(result):
220 while not datetime_exists(result):
221 result += datetime.timedelta(minutes=1)
222 return result, False
224 # result is closer to the previous date
225 farther = date.replace(fold=0 if is_prev else 1, tzinfo=previous_date.tzinfo)
226 # Comparing the UTC offsets in the check for the date being ambiguous.
227 if result.utcoffset() != farther.utcoffset():
228 # TODO: Check negative DST
229 assert (result.astimezone(UTC_DT) > farther.astimezone(UTC_DT)) == is_prev
230 if not _is_successor(result, previous_date, is_prev):
231 assert _is_successor(farther, previous_date, is_prev)
232 result = farther
233 return result, True
236class CroniterError(ValueError):
237 """General top-level Croniter base exception"""
240class CroniterBadTypeRangeError(TypeError):
241 """."""
244class CroniterBadCronError(CroniterError):
245 """Syntax, unknown value, or range error within a cron expression"""
248class CroniterUnsupportedSyntaxError(CroniterBadCronError):
249 """Valid cron syntax, but likely to produce inaccurate results"""
251 # Extending CroniterBadCronError, which may be contridatory, but this allows
252 # catching both errors with a single exception. From a user perspective
253 # these will likely be handled the same way.
256class CroniterBadDateError(CroniterError):
257 """Unable to find next/prev timestamp match"""
260class CroniterNotAlphaError(CroniterBadCronError):
261 """Cron syntax contains an invalid day or month abbreviation"""
264class croniter:
265 MONTHS_IN_YEAR = 12
267 # This helps with expanding `*` fields into `lower-upper` ranges. Each item
268 # in this tuple maps to the corresponding field index
269 RANGES = ((0, 59), (0, 23), (1, 31), (1, 12), (0, 6), (0, 59), (1970, 2099))
271 ALPHACONV: tuple[dict[str, Union[int, str]], ...] = (
272 {}, # 0: min
273 {}, # 1: hour
274 {"l": "l"}, # 2: dom
275 # 3: mon
276 copy.deepcopy(M_ALPHAS),
277 # 4: dow
278 copy.deepcopy(DOW_ALPHAS),
279 # 5: second
280 {},
281 # 6: year
282 {},
283 )
285 LOWMAP: tuple[dict[int, int], ...] = ({}, {}, {0: 1}, {0: 1}, {7: 0}, {}, {})
287 LEN_MEANS_ALL = (60, 24, 31, 12, 7, 60, 130)
289 def __init__(
290 self,
291 expr_format: str,
292 start_time: Optional[Union[datetime.datetime, float]] = None,
293 ret_type: type = float,
294 day_or: bool = True,
295 max_years_between_matches: Optional[int] = None,
296 is_prev: bool = False,
297 hash_id: Optional[Union[bytes, str]] = None,
298 implement_cron_bug: bool = False,
299 second_at_beginning: bool = False,
300 expand_from_start_time: bool = False,
301 ) -> None:
302 self._ret_type = ret_type
303 self._day_or = day_or
304 self._implement_cron_bug = implement_cron_bug
305 self.second_at_beginning = bool(second_at_beginning)
306 self._expand_from_start_time = expand_from_start_time
308 if hash_id is not None:
309 if not isinstance(hash_id, (bytes, str)):
310 raise TypeError("hash_id must be bytes or UTF-8 string")
311 if not isinstance(hash_id, bytes):
312 hash_id = hash_id.encode("UTF-8")
314 self._max_years_btw_matches_explicitly_set = max_years_between_matches is not None
315 if max_years_between_matches is None:
316 max_years_between_matches = 50
317 self._max_years_between_matches = max(int(max_years_between_matches), 1)
319 if start_time is None:
320 start_time = time()
322 self.tzinfo: Optional[datetime.tzinfo] = None
324 self.start_time = 0.0
325 self.dst_start_time = 0.0
326 self.cur = 0.0
327 self.set_current(start_time, force=True)
329 self.expanded, self.nth_weekday_of_month, self.expressions, self.nearest_weekday = self._expand(
330 expr_format,
331 hash_id=hash_id,
332 from_timestamp=self.dst_start_time if self._expand_from_start_time else None,
333 from_timestamp_tz=self.tzinfo if self._expand_from_start_time else None,
334 second_at_beginning=second_at_beginning,
335 )
336 self.fields = CRON_FIELDS[len(self.expanded)]
337 self._is_prev = is_prev
339 @classmethod
340 def _alphaconv(cls, index, key, expressions):
341 try:
342 return cls.ALPHACONV[index][key]
343 except KeyError:
344 raise CroniterNotAlphaError(f"[{' '.join(expressions)}] is not acceptable")
346 def get_next(self, ret_type=None, start_time=None, update_current=True):
347 if start_time and self._expand_from_start_time:
348 raise ValueError(
349 "start_time is not supported when using expand_from_start_time = True."
350 )
351 return self._get_next(
352 ret_type=ret_type, start_time=start_time, is_prev=False, update_current=update_current
353 )
355 def get_prev(self, ret_type=None, start_time=None, update_current=True):
356 return self._get_next(
357 ret_type=ret_type, start_time=start_time, is_prev=True, update_current=update_current
358 )
360 def get_current(self, ret_type=None):
361 ret_type = ret_type or self._ret_type
362 if issubclass(ret_type, datetime.datetime):
363 return self.timestamp_to_datetime(self.cur)
364 return self.cur
366 def set_current(
367 self, start_time: Optional[Union[datetime.datetime, float]], force: bool = True
368 ) -> float:
369 if (force or (self.cur is None)) and start_time is not None:
370 if isinstance(start_time, datetime.datetime):
371 self.tzinfo = start_time.tzinfo
372 start_time = self.datetime_to_timestamp(start_time)
374 self.start_time = start_time
375 self.dst_start_time = start_time
376 self.cur = start_time
377 return self.cur
379 @staticmethod
380 def datetime_to_timestamp(d: datetime.datetime) -> float:
381 """
382 Converts a `datetime` object `d` into a UNIX timestamp.
383 """
384 return datetime_to_timestamp(d)
386 _datetime_to_timestamp = datetime_to_timestamp # retrocompat
388 def timestamp_to_datetime(self, timestamp: float, tzinfo: Any = MARKER) -> datetime.datetime:
389 """
390 Converts a UNIX `timestamp` into a `datetime` object.
391 """
392 if tzinfo is MARKER: # allow to give tzinfo=None even if self.tzinfo is set
393 tzinfo = self.tzinfo
394 if OVERFLOW32B_MODE:
395 # degraded mode to workaround Y2038
396 # see https://github.com/python/cpython/issues/101069
397 result = EPOCH.replace(tzinfo=None) + datetime.timedelta(seconds=timestamp)
398 else:
399 result = datetime.datetime.fromtimestamp(timestamp, tz=tzutc()).replace(tzinfo=None)
400 if tzinfo:
401 result = result.replace(tzinfo=UTC_DT).astimezone(tzinfo)
402 return result
404 _timestamp_to_datetime = timestamp_to_datetime # retrocompat
406 def _get_next(self, ret_type=None, start_time=None, is_prev=None, update_current=None):
407 if update_current is None:
408 update_current = True
409 self.set_current(start_time, force=True)
410 if is_prev is None:
411 is_prev = self._is_prev
412 self._is_prev = is_prev
414 ret_type = ret_type or self._ret_type
416 if not issubclass(ret_type, (float, datetime.datetime)):
417 raise TypeError("Invalid ret_type, only 'float' or 'datetime' is acceptable.")
419 result = self._calc_next(is_prev)
420 timestamp = self.datetime_to_timestamp(result)
421 if update_current:
422 self.cur = timestamp
423 if issubclass(ret_type, datetime.datetime):
424 return result
425 return timestamp
427 # iterator protocol, to enable direct use of croniter
428 # objects in a loop, like "for dt in croniter("5 0 * * *'): ..."
429 # or for combining multiple croniters into single
430 # dates feed using 'itertools' module
431 def all_next(self, ret_type=None, start_time=None, update_current=None):
432 """
433 Returns a generator yielding consecutive dates.
435 May be used instead of an implicit call to __iter__ whenever a
436 non-default `ret_type` needs to be specified.
437 """
438 # In a Python 3.7+ world: contextlib.suppress and contextlib.nullcontext could
439 # be used instead
440 try:
441 while True:
442 self._is_prev = False
443 yield self._get_next(
444 ret_type=ret_type, start_time=start_time, update_current=update_current
445 )
446 start_time = None
447 except CroniterBadDateError:
448 if self._max_years_btw_matches_explicitly_set:
449 return
450 raise
452 def all_prev(self, ret_type=None, start_time=None, update_current=None):
453 """
454 Returns a generator yielding previous dates.
455 """
456 try:
457 while True:
458 self._is_prev = True
459 yield self._get_next(
460 ret_type=ret_type, start_time=start_time, update_current=update_current
461 )
462 start_time = None
463 except CroniterBadDateError:
464 if self._max_years_btw_matches_explicitly_set:
465 return
466 raise
468 def iter(self, *args, **kwargs):
469 return self.all_prev if self._is_prev else self.all_next
471 def __iter__(self):
472 return self
474 __next__ = next = _get_next
476 def _calc_next(self, is_prev: bool) -> datetime.datetime:
477 current = self.timestamp_to_datetime(self.cur)
478 expanded = self.expanded[:]
479 nth_weekday_of_month = self.nth_weekday_of_month.copy()
481 # exception to support day of month and day of week as defined in cron
482 if (expanded[DAY_FIELD][0] != "*" and expanded[DOW_FIELD][0] != "*") and self._day_or:
483 # If requested, handle a bug in vixie cron/ISC cron where day_of_month and
484 # day_of_week form an intersection (AND) instead of a union (OR) if either
485 # field is an asterisk or starts with an asterisk (https://crontab.guru/cron-bug.html)
486 if self._implement_cron_bug and (
487 re_star.match(self.expressions[DAY_FIELD])
488 or re_star.match(self.expressions[DOW_FIELD])
489 ):
490 # To produce a schedule identical to the cron bug, we'll bypass the code
491 # that makes a union of DOM and DOW, and instead skip to the code that
492 # does an intersect instead
493 pass
494 else:
495 # Under OR semantics an unsatisfiable side -- the 31st in a month
496 # that has no 31st, say -- contributes no dates rather than ruling
497 # out the expression, so the other side must still be able to
498 # match. Only both sides failing means there is no such date.
499 #
500 # That only holds while each side is a clean operand. '#' and 'W'
501 # are not carried by the DAY/DOW fields but by nth_weekday_of_month
502 # and self.nearest_weekday, so blanking the fields above does not
503 # remove them and each side stays an intersection (the semantics
504 # test_issue_k33 pins down). Swallowing a failure there would drop
505 # the other field from the schedule instead, so let it propagate.
506 clean_split = not nth_weekday_of_month and not self.nearest_weekday
508 bak = expanded[DOW_FIELD]
509 expanded[DOW_FIELD] = ["*"]
510 try:
511 t1 = self._calc(current, expanded, nth_weekday_of_month, is_prev)
512 except CroniterBadDateError:
513 if not clean_split:
514 raise
515 t1 = None
516 expanded[DOW_FIELD] = bak
517 expanded[DAY_FIELD] = ["*"]
519 try:
520 t2 = self._calc(current, expanded, nth_weekday_of_month, is_prev)
521 except CroniterBadDateError:
522 if not clean_split:
523 raise
524 t2 = None
526 if t1 is None:
527 if t2 is None:
528 raise CroniterBadDateError(
529 "failed to find prev date" if is_prev else "failed to find next date"
530 )
531 return t2
532 if t2 is None:
533 return t1
534 if is_prev:
535 return t1 if t1 > t2 else t2
536 return t1 if t1 < t2 else t2
538 return self._calc(current, expanded, nth_weekday_of_month, is_prev)
540 def _calc(
541 self,
542 now: datetime.datetime,
543 expanded: list[ExpandedExpression],
544 nth_weekday_of_month: dict[int, set[int]],
545 is_prev: bool,
546 ) -> datetime.datetime:
547 if is_prev:
548 nearest_diff_method = self._get_prev_nearest_diff
549 offset = relativedelta(microseconds=-1)
550 else:
551 nearest_diff_method = self._get_next_nearest_diff
552 if len(expanded) > UNIX_CRON_LEN:
553 offset = relativedelta(seconds=1)
554 else:
555 offset = relativedelta(minutes=1)
556 # Calculate the next cron time in local time a.k.a. timezone unaware time.
557 unaware_time = now.replace(tzinfo=None) + offset
558 if len(expanded) > UNIX_CRON_LEN:
559 unaware_time = unaware_time.replace(microsecond=0)
560 else:
561 unaware_time = unaware_time.replace(second=0, microsecond=0)
563 month = unaware_time.month
564 year = current_year = unaware_time.year
566 def proc_year(d):
567 if len(expanded) == YEAR_CRON_LEN:
568 try:
569 expanded[YEAR_FIELD].index("*")
570 except ValueError:
571 # use None as range_val to indicate no loop
572 diff_year = nearest_diff_method(d.year, expanded[YEAR_FIELD], None)
573 if diff_year is None:
574 return None, d
575 if diff_year != 0:
576 if is_prev:
577 d += relativedelta(
578 years=diff_year, month=12, day=31, hour=23, minute=59, second=59
579 )
580 else:
581 d += relativedelta(
582 years=diff_year, month=1, day=1, hour=0, minute=0, second=0
583 )
584 return True, d
585 return False, d
587 def proc_month(d):
588 try:
589 expanded[MONTH_FIELD].index("*")
590 except ValueError:
591 diff_month = nearest_diff_method(
592 d.month, expanded[MONTH_FIELD], self.MONTHS_IN_YEAR
593 )
594 reset_day = 1
596 if diff_month is not None and diff_month != 0:
597 if is_prev:
598 d += relativedelta(months=diff_month)
599 reset_day = _last_day_of_month(d.year, d.month)
600 d += relativedelta(day=reset_day, hour=23, minute=59, second=59)
601 else:
602 d += relativedelta(
603 months=diff_month, day=reset_day, hour=0, minute=0, second=0
604 )
605 return True, d
606 return False, d
608 def proc_day_of_month(d):
609 try:
610 expanded[DAY_FIELD].index("*")
611 except ValueError:
612 days = _last_day_of_month(year, month)
613 if "l" in expanded[DAY_FIELD] and days == d.day:
614 return False, d
616 if is_prev:
617 prev_month = (month - 2) % self.MONTHS_IN_YEAR + 1
618 prev_year = year - 1 if month == 1 else year
619 days_in_prev_month = _last_day_of_month(prev_year, prev_month)
620 diff_day = nearest_diff_method(d.day, expanded[DAY_FIELD], days_in_prev_month)
621 else:
622 diff_day = nearest_diff_method(d.day, expanded[DAY_FIELD], days)
624 if diff_day is not None and diff_day != 0:
625 if is_prev:
626 d += relativedelta(days=diff_day, hour=23, minute=59, second=59)
627 else:
628 d += relativedelta(days=diff_day, hour=0, minute=0, second=0)
629 return True, d
630 return False, d
632 def proc_day_of_week(d):
633 try:
634 expanded[DOW_FIELD].index("*")
635 except ValueError:
636 diff_day_of_week = nearest_diff_method(d.isoweekday() % 7, expanded[DOW_FIELD], 7)
637 if diff_day_of_week is not None and diff_day_of_week != 0:
638 if is_prev:
639 d += relativedelta(days=diff_day_of_week, hour=23, minute=59, second=59)
640 else:
641 d += relativedelta(days=diff_day_of_week, hour=0, minute=0, second=0)
642 return True, d
643 return False, d
645 def proc_day_of_week_nth(d):
646 if "*" in nth_weekday_of_month:
647 s = nth_weekday_of_month["*"]
648 for i in range(0, 7):
649 if i in nth_weekday_of_month:
650 nth_weekday_of_month[i].update(s)
651 else:
652 nth_weekday_of_month[i] = s
653 del nth_weekday_of_month["*"]
655 candidates = []
656 for wday, nth in nth_weekday_of_month.items():
657 c = self._get_nth_weekday_of_month(d.year, d.month, wday)
658 for n in nth:
659 if n == "l":
660 candidate = c[-1]
661 elif len(c) < n:
662 continue
663 else:
664 candidate = c[n - 1]
665 if (is_prev and candidate <= d.day) or (not is_prev and d.day <= candidate):
666 candidates.append(candidate)
668 if not candidates:
669 if is_prev:
670 d += relativedelta(days=-d.day, hour=23, minute=59, second=59)
671 else:
672 days = _last_day_of_month(year, month)
673 d += relativedelta(days=(days - d.day + 1), hour=0, minute=0, second=0)
674 return True, d
676 candidates.sort()
677 diff_day = (candidates[-1] if is_prev else candidates[0]) - d.day
678 if diff_day != 0:
679 if is_prev:
680 d += relativedelta(days=diff_day, hour=23, minute=59, second=59)
681 else:
682 d += relativedelta(days=diff_day, hour=0, minute=0, second=0)
683 return True, d
684 return False, d
686 def proc_nearest_weekday(d):
687 """Process W (nearest weekday) day-of-month entries."""
688 candidates = []
689 for w_day in self.nearest_weekday:
690 candidate = self._get_nearest_weekday(d.year, d.month, w_day)
691 if (is_prev and candidate <= d.day) or (not is_prev and d.day <= candidate):
692 candidates.append(candidate)
694 if not candidates:
695 if is_prev:
696 d += relativedelta(days=-d.day, hour=23, minute=59, second=59)
697 else:
698 days = _last_day_of_month(year, month)
699 d += relativedelta(days=(days - d.day + 1), hour=0, minute=0, second=0)
700 return True, d
702 candidates.sort()
703 diff_day = (candidates[-1] if is_prev else candidates[0]) - d.day
704 if diff_day != 0:
705 if is_prev:
706 d += relativedelta(days=diff_day, hour=23, minute=59, second=59)
707 else:
708 d += relativedelta(days=diff_day, hour=0, minute=0, second=0)
709 return True, d
710 return False, d
712 def proc_hour(d):
713 try:
714 expanded[HOUR_FIELD].index("*")
715 except ValueError:
716 diff_hour = nearest_diff_method(d.hour, expanded[HOUR_FIELD], 24)
717 if diff_hour is not None and diff_hour != 0:
718 if is_prev:
719 d += relativedelta(hours=diff_hour, minute=59, second=59)
720 else:
721 d += relativedelta(hours=diff_hour, minute=0, second=0)
722 return True, d
723 return False, d
725 def proc_minute(d):
726 try:
727 expanded[MINUTE_FIELD].index("*")
728 except ValueError:
729 diff_min = nearest_diff_method(d.minute, expanded[MINUTE_FIELD], 60)
730 if diff_min is not None and diff_min != 0:
731 if is_prev:
732 d += relativedelta(minutes=diff_min, second=59)
733 else:
734 d += relativedelta(minutes=diff_min, second=0)
735 return True, d
736 return False, d
738 def proc_second(d):
739 if len(expanded) > UNIX_CRON_LEN:
740 try:
741 expanded[SECOND_FIELD].index("*")
742 except ValueError:
743 diff_sec = nearest_diff_method(d.second, expanded[SECOND_FIELD], 60)
744 if diff_sec is not None and diff_sec != 0:
745 d += relativedelta(seconds=diff_sec)
746 return True, d
747 else:
748 d += relativedelta(second=0)
749 return False, d
751 procs = [
752 proc_year,
753 proc_month,
754 (proc_nearest_weekday if self.nearest_weekday else proc_day_of_month),
755 (proc_day_of_week_nth if nth_weekday_of_month else proc_day_of_week),
756 proc_hour,
757 proc_minute,
758 proc_second,
759 ]
761 while abs(year - current_year) <= self._max_years_between_matches:
762 next = False
763 stop = False
764 for proc in procs:
765 (changed, unaware_time) = proc(unaware_time)
766 # `None` can be set mostly for year processing
767 # so please see proc_year / _get_prev_nearest_diff / _get_next_nearest_diff
768 if changed is None:
769 stop = True
770 break
771 if changed:
772 month, year = unaware_time.month, unaware_time.year
773 next = True
774 break
775 if stop:
776 break
777 if next:
778 continue
780 unaware_time = unaware_time.replace(microsecond=0)
781 if now.tzinfo is None:
782 return unaware_time
784 # Add timezone information back and handle DST changes
785 aware_time, exists = _add_tzinfo(unaware_time, now, is_prev)
787 if not exists and (
788 not _is_successor(aware_time, now, is_prev) or "*" in expanded[HOUR_FIELD]
789 ):
790 # The calculated local date does not exist and moving the time forward
791 # to the next valid time isn't the correct solution. Search for the
792 # next matching cron time that exists.
793 while not exists:
794 unaware_time = self._calc(
795 unaware_time, expanded, nth_weekday_of_month, is_prev
796 )
797 aware_time, exists = _add_tzinfo(unaware_time, now, is_prev)
799 offset_delta = _timezone_delta(now, aware_time)
800 if not offset_delta:
801 # There was no DST change.
802 return aware_time
804 # There was a DST change. So check if there is a alternative cron time
805 # for the other UTC offset.
806 alternative_unaware_time = now.replace(tzinfo=None) + offset_delta
807 alternative_unaware_time = self._calc(
808 alternative_unaware_time, expanded, nth_weekday_of_month, is_prev
809 )
810 alternative_aware_time, exists = _add_tzinfo(alternative_unaware_time, now, is_prev)
812 if not _is_successor(alternative_aware_time, now, is_prev):
813 # The alternative time is an ancestor of now. Thus it is not an alternative.
814 return aware_time
816 if _is_successor(aware_time, alternative_aware_time, is_prev):
817 return alternative_aware_time
819 return aware_time
821 if is_prev:
822 raise CroniterBadDateError("failed to find prev date")
823 raise CroniterBadDateError("failed to find next date")
825 @staticmethod
826 def _get_next_nearest_diff(x, to_check, range_val):
827 """
828 `range_val` is the range of a field.
829 If no available time, we can move to next loop(like next month).
830 `range_val` can also be set to `None` to indicate that there is no loop.
831 ( Currently, should only used for `year` field )
832 """
833 for i, d in enumerate(to_check):
834 if range_val is not None:
835 if d == "l":
836 # if 'l' then it is the last day of month
837 # => its value of range_val
838 d = range_val
839 elif d > range_val:
840 continue
841 if d >= x:
842 return d - x
843 # When range_val is None and x not exists in to_check,
844 # `None` will be returned to suggest no more available time
845 if range_val is None:
846 return None
847 return to_check[0] - x + range_val
849 @staticmethod
850 def _get_prev_nearest_diff(x, to_check, range_val):
851 """
852 `range_val` is the range of a field.
853 If no available time, we can move to previous loop(like previous month).
854 Range_val can also be set to `None` to indicate that there is no loop.
855 ( Currently should only used for `year` field )
856 """
857 candidates = to_check[:]
858 candidates.reverse()
859 for d in candidates:
860 if d != "l" and d <= x:
861 return d - x
862 if "l" in candidates:
863 return -x
864 # When range_val is None and x not exists in to_check,
865 # `None` will be returned to suggest no more available time
866 if range_val is None:
867 return None
868 candidate = candidates[0]
869 for c in candidates:
870 # fixed: c < range_val
871 # this code will reject all 31 day of month, 12 month, 59 second,
872 # 23 hour and so on.
873 # if candidates has just a element, this will not harmful.
874 # but candidates have multiple elements, then values equal to
875 # range_val will rejected.
876 if c <= range_val:
877 candidate = c
878 break
879 # fix crontab "0 6 30 3 *" condidates only a element, then get_prev error
880 # return 2021-03-02 06:00:00
881 if candidate > range_val:
882 return -range_val
883 return candidate - x - range_val
885 @staticmethod
886 def _get_nth_weekday_of_month(year: int, month: int, day_of_week: int) -> tuple[int, ...]:
887 """For a given year/month return a list of days in nth-day-of-month order.
888 The last weekday of the month is always [-1].
889 """
890 w = (day_of_week + 6) % 7
891 c = calendar.Calendar(w).monthdayscalendar(year, month)
892 if c[0][0] == 0:
893 c.pop(0)
894 return tuple(i[0] for i in c)
896 @staticmethod
897 def _get_nearest_weekday(year, month, day):
898 """Get the nearest weekday (Mon-Fri) to the given day in the given month.
900 Rules:
901 - If the day is a weekday, return it.
902 - If Saturday, return Friday (day-1), unless that crosses into previous month,
903 then return Monday (day+2).
904 - If Sunday, return Monday (day+1), unless that crosses into next month,
905 then return Friday (day-2).
906 """
907 last_day = _last_day_of_month(year, month)
908 day = min(day, last_day)
909 weekday = calendar.weekday(year, month, day) # 0=Mon, 6=Sun
910 if weekday < 5: # Mon-Fri
911 return day
912 if weekday == 5: # Saturday
913 if day > 1:
914 return day - 1 # Friday
915 else:
916 return day + 2 # Monday (1st is Sat, so 3rd is Mon)
917 # Sunday
918 if day < last_day:
919 return day + 1 # Monday
920 else:
921 return day - 2 # Friday (last day is Sun, go back to Fri)
923 @classmethod
924 def value_alias(cls, val, field_index, len_expressions=UNIX_CRON_LEN):
925 if isinstance(len_expressions, (list, dict, tuple, set)):
926 len_expressions = len(len_expressions)
927 if val in cls.LOWMAP[field_index] and not (
928 # do not support 0 as a month either for classical 5 fields cron,
929 # 6fields second repeat form or 7 fields year form
930 # but still let conversion happen if day field is shifted
931 (field_index in [DAY_FIELD, MONTH_FIELD] and len_expressions == UNIX_CRON_LEN)
932 or (field_index in [MONTH_FIELD, DOW_FIELD] and len_expressions == SECOND_CRON_LEN)
933 or (
934 field_index in [DAY_FIELD, MONTH_FIELD, DOW_FIELD]
935 and len_expressions == YEAR_CRON_LEN
936 )
937 ):
938 val = cls.LOWMAP[field_index][val]
939 return val
941 # Maximum days in each month (non-leap year for Feb)
942 DAYS_IN_MONTH = {1: 31, 2: 28, 3: 31, 4: 30, 5: 31, 6: 30, 7: 31, 8: 31, 9: 30, 10: 31, 11: 30, 12: 31}
944 @classmethod
945 def _expand(cls, expr_format, hash_id=None, second_at_beginning=False, from_timestamp=None, from_timestamp_tz=None, strict=False, strict_year=None):
946 # Split the expression in components, and normalize L -> l, MON -> mon,
947 # etc. Keep expr_format untouched so we can use it in the exception
948 # messages.
949 expr_aliases = {
950 "@midnight": ("0 0 * * *", "h h(0-2) * * * h"),
951 "@hourly": ("0 * * * *", "h * * * * h"),
952 "@daily": ("0 0 * * *", "h h * * * h"),
953 "@weekly": ("0 0 * * 0", "h h * * h h"),
954 "@monthly": ("0 0 1 * *", "h h h * * h"),
955 "@yearly": ("0 0 1 1 *", "h h h h * h"),
956 "@annually": ("0 0 1 1 *", "h h h h * h"),
957 }
959 efl = expr_format.lower()
960 hash_id_expr = 1 if hash_id is not None else 0
961 try:
962 efl = expr_aliases[efl][hash_id_expr]
963 except KeyError:
964 pass
966 expressions = efl.split()
968 if len(expressions) not in VALID_LEN_EXPRESSION:
969 raise CroniterBadCronError(
970 "Exactly 5, 6 or 7 columns has to be specified for iterator expression."
971 )
973 if len(expressions) > UNIX_CRON_LEN and second_at_beginning:
974 # move second to it's own(6th) field to process by same logical
975 expressions.insert(SECOND_FIELD, expressions.pop(0))
977 expanded = []
978 nth_weekday_of_month = {}
979 nearest_weekday = set()
981 for field_index, expr in enumerate(expressions):
982 for expanderid, expander in EXPANDERS.items():
983 expr = expander(cls).expand(
984 efl, field_index, expr, hash_id=hash_id, from_timestamp=from_timestamp
985 )
987 if "?" in expr:
988 if expr != "?":
989 raise CroniterBadCronError(
990 f"[{expr_format}] is not acceptable."
991 f" Question mark can not used with other characters"
992 )
993 if field_index not in [DAY_FIELD, DOW_FIELD]:
994 raise CroniterBadCronError(
995 f"[{expr_format}] is not acceptable. "
996 f"Question mark can only used in day_of_month or day_of_week"
997 )
998 # currently just trade `?` as `*`
999 expr = "*"
1001 e_list = expr.split(",")
1002 res = []
1003 seen = set()
1005 while len(e_list) > 0:
1006 e = e_list.pop()
1007 nth = None
1009 if field_index == DOW_FIELD:
1010 # Handle special case in the dow expression: 2#3, l3
1011 special_dow_rem = special_dow_re.match(str(e))
1012 if special_dow_rem:
1013 g = special_dow_rem.groupdict()
1014 he, last = g.get("he", ""), g.get("last", "")
1015 if he:
1016 e = he
1017 try:
1018 nth = int(last)
1019 assert 5 >= nth >= 1
1020 except (KeyError, ValueError, AssertionError):
1021 raise CroniterBadCronError(
1022 f"[{expr_format}] is not acceptable."
1023 f" Invalid day_of_week value: '{nth}'"
1024 )
1025 elif last:
1026 e = last
1027 nth = g["pre"] # 'l'
1029 if field_index == DAY_FIELD:
1030 # Handle W (nearest weekday) in day-of-month: 15w, w15
1031 w_match = nearest_weekday_re.match(str(e))
1032 if w_match:
1033 w_day = int(w_match.group(1) or w_match.group(2))
1034 if w_day < 1 or w_day > 31:
1035 raise CroniterBadCronError(
1036 f"[{expr_format}] is not acceptable,"
1037 f" nearest weekday day value '{w_day}' out of range"
1038 )
1039 if len(e_list) > 0 or len(res) > 0:
1040 raise CroniterBadCronError(
1041 f"[{expr_format}] is not acceptable."
1042 f" 'W' can only be used with a single day value,"
1043 f" not in a list or range"
1044 )
1045 nearest_weekday.add(w_day)
1046 res.append(w_day)
1047 continue
1049 # Before matching step_search_re, normalize "*" to "{min}-{max}".
1050 # Example: in the minute field, "*/5" normalizes to "0-59/5"
1051 t = re.sub(
1052 r"^\*(\/.+)$",
1053 r"%d-%d\1" % (cls.RANGES[field_index][0], cls.RANGES[field_index][1]),
1054 str(e),
1055 )
1056 m = step_search_re.search(t)
1058 # "{start}/{step}" is its own shape, not a range. It is normalized
1059 # below to "{start}-{max}/{step}" so that one regex parses both, but
1060 # the two must not be conflated afterwards: "Jan-Jan" is an explicit
1061 # equal range that croniter deliberately expands to the whole cycle,
1062 # whereas "DEC/3" is a start with a step and denotes just [DEC].
1063 start_with_step = False
1064 if not m:
1065 # Before matching step_search_re,
1066 # normalize "{start}/{step}" to "{start}-{max}/{step}".
1067 # Example: in the minute field, "10/5" normalizes to "10-59/5"
1068 t = re.sub(r"^(.+)\/(.+)$", r"\1-%d/\2" % (cls.RANGES[field_index][1]), str(e))
1069 m = step_search_re.search(t)
1070 start_with_step = bool(m)
1072 if m:
1073 # early abort if low/high are out of bounds
1074 (low, high, step) = m.group(1), m.group(2), m.group(4) or 1
1075 if field_index == DAY_FIELD and high == "l":
1076 high = "31"
1078 if not only_int_re.search(low):
1079 low = str(cls._alphaconv(field_index, low, expressions))
1081 if not only_int_re.search(high):
1082 high = str(cls._alphaconv(field_index, high, expressions))
1084 # normally, it's already guarded by the RE that should not accept
1085 # not-int values.
1086 if not only_int_re.search(str(step)):
1087 raise CroniterBadCronError(
1088 f"[{expr_format}] step '{step}'"
1089 f" in field {field_index} is not acceptable"
1090 )
1091 step = int(step)
1092 if step == 0:
1093 raise CroniterBadCronError(
1094 f"[{expr_format}] step '{step}'"
1095 f" in field {field_index} is not acceptable"
1096 )
1098 for band in low, high:
1099 if not only_int_re.search(str(band)):
1100 raise CroniterBadCronError(
1101 f"[{expr_format}] bands '{low}-{high}'"
1102 f" in field {field_index} are not acceptable"
1103 )
1105 low, high = (
1106 cls.value_alias(int(_val), field_index, expressions)
1107 for _val in (low, high)
1108 )
1110 if max(low, high) > max(
1111 cls.RANGES[field_index][0], cls.RANGES[field_index][1]
1112 ):
1113 raise CroniterBadCronError(f"{expr_format} is out of bands")
1115 # "{start}/{step}" normalizes to "{start}-{max}/{step}", so when
1116 # the start IS the field maximum the two bounds collide and the
1117 # token becomes indistinguishable from an explicitly written equal
1118 # range. That is the whole bug: "59/15" arrived at the ``low ==
1119 # high`` branch below -- which exists for "Jan-Jan" and expands to
1120 # the whole cycle -- and so fired at :00/:15/:30/:45 instead of
1121 # :59. Recognising the collision here is what keeps the two apart.
1122 #
1123 # Deliberately ``low == high`` and not the wider ``low + step >
1124 # high``. Both fix the reported bug, but the wider form also
1125 # suppresses the re-base below for starts that are not at the
1126 # maximum, silently changing ~365 additional
1127 # ``expand_from_start_time`` schedules that were never broken.
1128 # That is a separate question about what an explicit lower bound
1129 # should mean under that flag, and it is not this fix's to answer.
1130 start_at_field_max = start_with_step and low == high
1132 # ``from_timestamp`` re-bases the start of a *cycle* on the start
1133 # time. A single point has no cycle to re-base, and rewriting its
1134 # bound here would leave the bug reachable in
1135 # ``expand_from_start_time`` mode while looking fixed by default.
1136 if from_timestamp and not start_at_field_max:
1137 low = cls._get_low_from_current_date_number(
1138 field_index, int(step), int(from_timestamp), from_timestamp_tz
1139 )
1141 # Handle when the second bound of the range is in backtracking order:
1142 # eg: X-Sun or X-7 (Sat-Sun) in DOW, or X-Jan (Apr-Jan) in MONTH
1143 if start_at_field_max:
1144 rng = [low]
1145 elif low > high:
1146 whole_field_range = list(
1147 range(cls.RANGES[field_index][0], cls.RANGES[field_index][1] + 1, 1)
1148 )
1149 # Add FirstBound -> ENDRANGE, respecting step
1150 rng = list(range(low, cls.RANGES[field_index][1] + 1, step))
1151 # Then 0 -> SecondBound, but skipping n first occurences according to step
1152 # EG to respect such expressions : Apr-Jan/3
1153 to_skip = 0
1154 if rng:
1155 already_skipped = list(reversed(whole_field_range)).index(rng[-1])
1156 curpos = whole_field_range.index(rng[-1])
1157 if ((curpos + step) > len(whole_field_range)) and (
1158 already_skipped < step
1159 ):
1160 to_skip = step - already_skipped
1161 rng += list(range(cls.RANGES[field_index][0] + to_skip, high + 1, step))
1162 # if we include a range type: Jan-Jan, or Sun-Sun,
1163 # it means the whole cycle (all days of week, # all monthes of year, etc)
1164 elif low == high:
1165 rng = list(
1166 range(cls.RANGES[field_index][0], cls.RANGES[field_index][1] + 1, step)
1167 )
1168 else:
1169 try:
1170 rng = list(range(low, high + 1, step))
1171 except ValueError as exc:
1172 raise CroniterBadCronError(f"invalid range: {exc}")
1174 if field_index == DOW_FIELD and nth and nth != "l":
1175 rng = [f"{item}#{nth}" for item in rng]
1176 e_list += [a for a in rng if a not in seen]
1177 seen.update(rng)
1178 else:
1179 if t.startswith("-"):
1180 raise CroniterBadCronError(
1181 f"[{expr_format}] is not acceptable, negative numbers not allowed"
1182 )
1183 if not star_or_int_re.search(t):
1184 t = cls._alphaconv(field_index, t, expressions)
1186 try:
1187 t = int(t)
1188 except ValueError:
1189 pass
1191 t = cls.value_alias(t, field_index, expressions)
1193 if t not in ["*", "l"] and (
1194 int(t) < cls.RANGES[field_index][0] or int(t) > cls.RANGES[field_index][1]
1195 ):
1196 raise CroniterBadCronError(
1197 f"[{expr_format}] is not acceptable, out of range"
1198 )
1200 res.append(t)
1202 if field_index == DOW_FIELD and nth:
1203 if t not in nth_weekday_of_month:
1204 nth_weekday_of_month[t] = set()
1205 nth_weekday_of_month[t].add(nth)
1207 res = set(res)
1208 res = sorted(res, key=lambda i: f"{i:02}" if isinstance(i, int) else i)
1209 if len(res) == cls.LEN_MEANS_ALL[field_index]:
1210 # Make sure the wildcard is used in the correct way (avoid over-optimization)
1211 if (field_index == DAY_FIELD and "*" not in expressions[DOW_FIELD]) or (
1212 field_index == DOW_FIELD and "*" not in expressions[DAY_FIELD]
1213 ):
1214 pass
1215 else:
1216 res = ["*"]
1218 expanded.append(["*"] if (len(res) == 1 and res[0] == "*") else res)
1220 # Check to make sure the dow combo in use is supported
1221 if nth_weekday_of_month:
1222 dow_expanded_set = set(expanded[DOW_FIELD])
1223 dow_expanded_set = dow_expanded_set.difference(nth_weekday_of_month.keys())
1224 dow_expanded_set.discard("*")
1225 # Skip: if it's all weeks instead of wildcard
1226 if dow_expanded_set and len(set(expanded[DOW_FIELD])) != cls.LEN_MEANS_ALL[DOW_FIELD]:
1227 raise CroniterUnsupportedSyntaxError(
1228 f"day-of-week field does not support mixing literal values and nth"
1229 f" day of week syntax. Cron: '{expr_format}'"
1230 f" dow={dow_expanded_set} vs nth={nth_weekday_of_month}"
1231 )
1233 if strict:
1234 # Cross-validate day-of-month against month (and optionally year)
1235 # to reject impossible combinations like "0 0 31 2 *" (Feb 31st).
1236 days = expanded[DAY_FIELD]
1237 months = expanded[MONTH_FIELD]
1238 if days != ["*"] and days != ["l"] and months != ["*"]:
1239 int_days = [d for d in days if isinstance(d, int)]
1240 int_months = [m for m in months if isinstance(m, int)]
1241 if int_days and int_months:
1242 # Determine max days per month, accounting for leap years
1243 days_in_month = dict(cls.DAYS_IN_MONTH)
1244 if 2 in int_months:
1245 has_leap_year = True # assume possible by default
1246 if strict_year is not None:
1247 # Year explicitly provided as parameter
1248 if isinstance(strict_year, int):
1249 has_leap_year = calendar.isleap(strict_year)
1250 else:
1251 has_leap_year = any(calendar.isleap(y) for y in strict_year)
1252 elif len(expanded) > YEAR_FIELD:
1253 years = expanded[YEAR_FIELD]
1254 if years != ["*"]:
1255 int_years = [y for y in years if isinstance(y, int)]
1256 if int_years:
1257 has_leap_year = any(calendar.isleap(y) for y in int_years)
1258 if has_leap_year:
1259 days_in_month[2] = 29
1260 min_day = min(int_days)
1261 max_possible = max(days_in_month[m] for m in int_months)
1262 if min_day > max_possible:
1263 raise CroniterBadCronError(
1264 f"[{expr_format}] is not acceptable. Day(s) {int_days}"
1265 f" can never occur in month(s) {int_months}"
1266 )
1268 return expanded, nth_weekday_of_month, expressions, nearest_weekday
1270 @classmethod
1271 def expand(
1272 cls,
1273 expr_format: str,
1274 hash_id: Optional[Union[bytes, str]] = None,
1275 second_at_beginning: bool = False,
1276 from_timestamp: Optional[float] = None,
1277 from_timestamp_tz: Optional[datetime.tzinfo] = None,
1278 strict: bool = False,
1279 strict_year: Optional[Union[int, list[int]]] = None,
1280 ) -> tuple[list[ExpandedExpression], dict[int, set[int]]]:
1281 """
1282 Expand a cron expression format into a noramlized format of
1283 list[list[int | 'l' | '*']]. The first list representing each element
1284 of the epxression, and each sub-list representing the allowed values
1285 for that expression component.
1287 A tuple is returned, the first value being the expanded epxression
1288 list, and the second being a `nth_weekday_of_month` mapping.
1290 Examples:
1292 # Every minute
1293 >>> croniter.expand('* * * * *')
1294 ([['*'], ['*'], ['*'], ['*'], ['*']], {})
1296 # On the hour
1297 >>> croniter.expand('0 0 * * *')
1298 ([[0], [0], ['*'], ['*'], ['*']], {})
1300 # Hours 0-5 and 10 monday through friday
1301 >>> croniter.expand('0-5,10 * * * mon-fri')
1302 ([[0, 1, 2, 3, 4, 5, 10], ['*'], ['*'], ['*'], [1, 2, 3, 4, 5]], {})
1304 Note that some special values such as nth day of week are expanded to a
1305 special mapping format for later processing:
1307 # Every minute on the 3rd tuesday of the month
1308 >>> croniter.expand('* * * * 2#3')
1309 ([['*'], ['*'], ['*'], ['*'], [2]], {2: {3}})
1311 # Every hour on the last day of the month
1312 >>> croniter.expand('0 * l * *')
1313 ([[0], ['*'], ['l'], ['*'], ['*']], {})
1315 # On the hour every 15 seconds
1316 >>> croniter.expand('0 0 * * * */15')
1317 ([[0], [0], ['*'], ['*'], ['*'], [0, 15, 30, 45]], {})
1318 """
1319 try:
1320 expanded, nth_weekday_of_month, _expressions, _nearest_weekday = cls._expand(
1321 expr_format,
1322 hash_id=hash_id,
1323 second_at_beginning=second_at_beginning,
1324 from_timestamp=from_timestamp,
1325 from_timestamp_tz=from_timestamp_tz,
1326 strict=strict,
1327 strict_year=strict_year,
1328 )
1329 return expanded, nth_weekday_of_month
1330 except (ValueError,) as exc:
1331 if isinstance(exc, CroniterError):
1332 raise
1333 trace = _traceback.format_exc()
1334 raise CroniterBadCronError(trace)
1336 @classmethod
1337 def _get_low_from_current_date_number(cls, field_index, step, from_timestamp, tzinfo=None):
1338 # Read the start time back in its own timezone. A naive start_time was
1339 # converted to a timestamp as if it were UTC, so UTC round-trips it to the
1340 # same wall clock and nothing changes for that case.
1341 dt = datetime.datetime.fromtimestamp(from_timestamp, tz=tzinfo or UTC_DT)
1342 if field_index == MINUTE_FIELD:
1343 return dt.minute % step
1344 if field_index == HOUR_FIELD:
1345 return dt.hour % step
1346 if field_index == DAY_FIELD:
1347 return ((dt.day - 1) % step) + 1
1348 if field_index == MONTH_FIELD:
1349 return ((dt.month - 1) % step) + 1
1350 if field_index == DOW_FIELD:
1351 return (dt.isoweekday() % 7) % step
1352 if field_index == SECOND_FIELD:
1353 return dt.second % step
1354 if field_index == YEAR_FIELD:
1355 # Like day and month, the year field does not start at 0, so the phase is
1356 # taken from the field minimum rather than from the value itself.
1357 year_start = cls.RANGES[YEAR_FIELD][0]
1358 return ((dt.year - year_start) % step) + year_start
1360 raise ValueError(f"Can't get current date number for field index {field_index}")
1362 @classmethod
1363 def is_valid(cls, expression, hash_id=None, encoding="UTF-8", second_at_beginning=False, strict=False, strict_year=None):
1364 if hash_id:
1365 if not isinstance(hash_id, (bytes, str)):
1366 raise TypeError("hash_id must be bytes or UTF-8 string")
1367 if not isinstance(hash_id, bytes):
1368 hash_id = hash_id.encode(encoding)
1369 try:
1370 cls.expand(expression, hash_id=hash_id, second_at_beginning=second_at_beginning, strict=strict, strict_year=strict_year)
1371 except CroniterError:
1372 return False
1373 return True
1375 @classmethod
1376 def match(
1377 cls,
1378 cron_expression,
1379 testdate,
1380 day_or=True,
1381 second_at_beginning=False,
1382 precision_in_seconds=None,
1383 ):
1384 return cls.match_range(
1385 cron_expression, testdate, testdate, day_or, second_at_beginning, precision_in_seconds
1386 )
1388 @classmethod
1389 def match_range(
1390 cls,
1391 cron_expression,
1392 from_datetime,
1393 to_datetime,
1394 day_or=True,
1395 second_at_beginning=False,
1396 precision_in_seconds=None,
1397 ):
1398 cron = cls(
1399 cron_expression,
1400 to_datetime,
1401 ret_type=datetime.datetime,
1402 day_or=day_or,
1403 second_at_beginning=second_at_beginning,
1404 )
1405 tdp = cron.get_current(datetime.datetime)
1406 if not tdp.microsecond:
1407 tdp += relativedelta(microseconds=1)
1408 cron.set_current(tdp, force=True)
1409 try:
1410 tdt = cron.get_prev()
1411 except CroniterBadDateError:
1412 return False
1413 if precision_in_seconds is None:
1414 precision_in_seconds = 1 if len(cron.expanded) > UNIX_CRON_LEN else 60
1415 duration_in_second = (to_datetime - from_datetime).total_seconds() + precision_in_seconds
1416 return (max(tdp, tdt) - min(tdp, tdt)).total_seconds() < duration_in_second
1419def croniter_range(
1420 start,
1421 stop,
1422 expr_format,
1423 ret_type=None,
1424 day_or=True,
1425 exclude_ends=False,
1426 _croniter=None,
1427 second_at_beginning=False,
1428 expand_from_start_time=False,
1429):
1430 """
1431 Generator that provides all times from start to stop matching the given cron expression.
1432 If the cron expression matches either 'start' and/or 'stop', those times will be returned as
1433 well unless 'exclude_ends=True' is passed.
1435 You can think of this function as sibling to the builtin range function for datetime objects.
1436 Like range(start,stop,step), except that here 'step' is a cron expression.
1437 """
1438 _croniter = _croniter or croniter
1439 auto_rt = datetime.datetime
1440 # type is used in first if branch for perfs reasons
1441 if type(start) is not type(stop) and not (
1442 isinstance(start, type(stop)) or isinstance(stop, type(start))
1443 ):
1444 raise CroniterBadTypeRangeError(
1445 f"The start and stop must be same type. {type(start)} != {type(stop)}"
1446 )
1447 if isinstance(start, (float, int)):
1448 start, stop = (
1449 datetime.datetime.fromtimestamp(t, tzutc()).replace(tzinfo=None) for t in (start, stop)
1450 )
1451 auto_rt = float
1452 if ret_type is None:
1453 ret_type = auto_rt
1454 if not exclude_ends:
1455 ms1 = relativedelta(microseconds=1)
1456 if start < stop: # Forward (normal) time order
1457 start -= ms1
1458 stop += ms1
1459 else: # Reverse time order
1460 start += ms1
1461 stop -= ms1
1462 year_span = math.floor(abs(stop.year - start.year)) + 1
1463 ic = _croniter(
1464 expr_format,
1465 start,
1466 ret_type=datetime.datetime,
1467 day_or=day_or,
1468 max_years_between_matches=year_span,
1469 second_at_beginning=second_at_beginning,
1470 expand_from_start_time=expand_from_start_time,
1471 )
1472 # define a continue (cont) condition function and step function for the main while loop
1473 if start < stop: # Forward
1475 def cont(v):
1476 return v < stop
1478 step = ic.get_next
1479 else: # Reverse
1481 def cont(v):
1482 return v > stop
1484 step = ic.get_prev
1485 try:
1486 dt = step()
1487 while cont(dt):
1488 if ret_type is float:
1489 yield ic.get_current(float)
1490 else:
1491 yield dt
1492 dt = step()
1493 except CroniterBadDateError:
1494 # Stop iteration when this exception is raised; no match found within the given year range
1495 return
1498class HashExpander:
1499 def __init__(self, cronit):
1500 self.cron = cronit
1502 def do(self, idx, hash_type="h", hash_id=None, range_end=None, range_begin=None):
1503 """Return a hashed/random integer given range/hash information"""
1504 if range_end is None:
1505 range_end = self.cron.RANGES[idx][1]
1506 if range_begin is None:
1507 range_begin = self.cron.RANGES[idx][0]
1508 if hash_type == "r":
1509 crc = random.randint(0, 0xFFFFFFFF)
1510 else:
1511 crc = binascii.crc32(hash_id) & 0xFFFFFFFF
1512 return ((crc >> idx) % (range_end - range_begin + 1)) + range_begin
1514 def match(self, efl, idx, expr, hash_id=None, **kw):
1515 return hash_expression_re.match(expr)
1517 def _expand_divisor(self, idx, m, hash_id, range_begin, range_end):
1518 """Hash a start offset into the first period, then step to range_end.
1520 The offset is drawn from the first period, [range_begin, range_begin +
1521 divisor - 1], but never past range_end: a divisor wider than the range has
1522 only the range itself to draw from. And when the offset lands on range_end
1523 the step cannot reach a second value, so the result is that single value --
1524 emitting "{end}-{end}/{divisor}" instead would read as an explicit equal
1525 range, which croniter expands to the whole field.
1526 """
1527 divisor = int(m["divisor"])
1528 x = self.do(
1529 idx,
1530 hash_type=m["hash_type"],
1531 hash_id=hash_id,
1532 range_begin=range_begin,
1533 range_end=min(range_begin + divisor - 1, range_end),
1534 )
1535 if x == range_end:
1536 return str(x)
1537 return f"{x}-{range_end}/{divisor}"
1539 def expand(self, efl, idx, expr, hash_id=None, match="", **kw):
1540 """Expand a hashed/random expression to its normal representation"""
1541 if match == "":
1542 match = self.match(efl, idx, expr, hash_id, **kw)
1543 if not match:
1544 return expr
1545 m = match.groupdict()
1547 if m["hash_type"] == "h" and hash_id is None:
1548 raise CroniterBadCronError("Hashed definitions must include hash_id")
1550 if m["range_begin"] and m["range_end"]:
1551 if int(m["range_begin"]) >= int(m["range_end"]):
1552 raise CroniterBadCronError("Range end must be greater than range begin")
1554 if m["range_begin"] and m["range_end"] and m["divisor"]:
1555 # Example: H(30-59)/10 -> 34-59/10 (i.e. 34,44,54)
1556 if int(m["divisor"]) == 0:
1557 raise CroniterBadCronError(f"Bad expression: {expr}")
1559 return self._expand_divisor(
1560 idx, m, hash_id, int(m["range_begin"]), int(m["range_end"])
1561 )
1562 if m["range_begin"] and m["range_end"]:
1563 # Example: H(0-29) -> 12
1564 return str(
1565 self.do(
1566 idx,
1567 hash_type=m["hash_type"],
1568 hash_id=hash_id,
1569 range_end=int(m["range_end"]),
1570 range_begin=int(m["range_begin"]),
1571 )
1572 )
1573 if m["divisor"]:
1574 # Example: H/15 -> 7-59/15 (i.e. 7,22,37,52)
1575 if int(m["divisor"]) == 0:
1576 raise CroniterBadCronError(f"Bad expression: {expr}")
1578 return self._expand_divisor(
1579 idx, m, hash_id, self.cron.RANGES[idx][0], self.cron.RANGES[idx][1]
1580 )
1582 # Example: H -> 32
1583 return str(self.do(idx, hash_type=m["hash_type"], hash_id=hash_id))
1586EXPANDERS = {"hash": HashExpander}