Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/croniter/croniter.py: 81%
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+))?$", re.ASCII)
109only_int_re = re.compile(r"^\d+$", re.ASCII)
110# ``str.split()`` also splits on Unicode whitespace (no-break space, ideographic
111# space, ...), which cron does not treat as a field separator.
112field_re = re.compile(r"\S+", re.ASCII)
114DAYS = (31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31)
115WEEKDAYS = "|".join(DOW_ALPHAS.keys())
116MONTHS = "|".join(M_ALPHAS.keys())
117star_or_int_re = re.compile(r"^(\d+|\*)$", re.ASCII)
118special_dow_re = re.compile(
119 rf"^(?P<pre>((?P<he>(({WEEKDAYS})(-({WEEKDAYS}))?)"
120 rf"|(({MONTHS})(-({MONTHS}))?)|\w+)#)|l)(?P<last>\d+)$",
121 re.ASCII,
122)
123nearest_weekday_re = re.compile(r"^(?:(\d+)w|w(\d+))$", re.ASCII)
124re_star = re.compile("[*]")
125hash_expression_re = re.compile(
126 r"^(?P<hash_type>h|r)(\((?P<range_begin>\d+)-(?P<range_end>\d+)\))?(\/(?P<divisor>\d+))?$",
127 re.ASCII,
128)
130CRON_FIELDS = {
131 "unix": UNIX_FIELDS,
132 "second": SECOND_FIELDS,
133 "year": YEAR_FIELDS,
134 len(UNIX_FIELDS): UNIX_FIELDS,
135 len(SECOND_FIELDS): SECOND_FIELDS,
136 len(YEAR_FIELDS): YEAR_FIELDS,
137}
138UNIX_CRON_LEN = len(UNIX_FIELDS)
139SECOND_CRON_LEN = len(SECOND_FIELDS)
140YEAR_CRON_LEN = len(YEAR_FIELDS)
141# retrocompat
142VALID_LEN_EXPRESSION = {a for a in CRON_FIELDS if isinstance(a, int)}
144MARKER = object()
147def datetime_to_timestamp(d):
148 if d.tzinfo is not None:
149 d = d.replace(tzinfo=None) - d.utcoffset()
151 return (d - datetime.datetime(1970, 1, 1)).total_seconds()
154def _is_leap(year: int) -> bool:
155 return year % 400 == 0 or (year % 4 == 0 and year % 100 != 0)
158def _last_day_of_month(year: int, month: int) -> int:
159 """Calculate the last day of the given month (honor leap years)."""
160 last_day = DAYS[month - 1]
161 if month == 2 and _is_leap(year):
162 last_day += 1
163 return last_day
166def _is_successor(
167 date: datetime.datetime, previous_date: datetime.datetime, is_prev: bool
168) -> bool:
169 """Check if the given date is a successor (after/before) of the previous date."""
170 if is_prev:
171 return date.astimezone(UTC_DT) < previous_date.astimezone(UTC_DT)
172 return date.astimezone(UTC_DT) > previous_date.astimezone(UTC_DT)
175def _timezone_delta(date1: datetime.datetime, date2: datetime.datetime) -> datetime.timedelta:
176 """Calculate the timezone difference of the given dates."""
177 offset1 = date1.utcoffset()
178 offset2 = date2.utcoffset()
179 assert offset1 is not None
180 assert offset2 is not None
181 return offset2 - offset1
184def _add_tzinfo(
185 date: datetime.datetime, previous_date: datetime.datetime, is_prev: bool
186) -> tuple[datetime.datetime, bool]:
187 """Add the tzinfo from the previous date to the given date.
189 In case the new date is ambiguous, determine the correct date
190 based on it being closer to the previous date but still a successor
191 (after/before based on `is_prev`).
193 In case the date does not exist, jump forward to the next existing date.
194 """
195 localize = getattr(previous_date.tzinfo, "localize", None)
196 if localize is not None:
197 # pylint: disable-next=import-outside-toplevel
198 import pytz
200 try:
201 result = localize(date, is_dst=None)
202 except pytz.NonExistentTimeError:
203 while True:
204 date += datetime.timedelta(minutes=1)
205 try:
206 result = localize(date, is_dst=None)
207 except pytz.NonExistentTimeError:
208 continue
209 break
210 return result, False
211 except pytz.AmbiguousTimeError:
212 candidates = [
213 localize(date, is_dst=True),
214 localize(date, is_dst=False),
215 ]
216 # Depending on the sign of the zone's DST offset, either
217 # interpretation can be the earlier instant, so order by the
218 # actual absolute times instead of assuming is_dst ordering.
219 candidates.sort(key=lambda value: value.astimezone(UTC_DT))
220 if is_prev:
221 candidates.reverse()
222 for candidate in candidates:
223 if _is_successor(candidate, previous_date, is_prev):
224 return candidate, True
225 return candidates[-1], True
226 return result, True
228 result = date.replace(fold=1 if is_prev else 0, tzinfo=previous_date.tzinfo)
229 if not datetime_exists(result):
230 while not datetime_exists(result):
231 result += datetime.timedelta(minutes=1)
232 return result, False
234 # result is closer to the previous date
235 farther = date.replace(fold=0 if is_prev else 1, tzinfo=previous_date.tzinfo)
236 # Comparing the UTC offsets in the check for the date being ambiguous.
237 if result.utcoffset() != farther.utcoffset():
238 # TODO: Check negative DST
239 assert (result.astimezone(UTC_DT) > farther.astimezone(UTC_DT)) == is_prev
240 if not _is_successor(result, previous_date, is_prev):
241 assert _is_successor(farther, previous_date, is_prev)
242 result = farther
243 return result, True
246class CroniterError(ValueError):
247 """General top-level Croniter base exception"""
250class CroniterBadTypeRangeError(TypeError):
251 """."""
254class CroniterBadCronError(CroniterError):
255 """Syntax, unknown value, or range error within a cron expression"""
258class CroniterUnsupportedSyntaxError(CroniterBadCronError):
259 """Valid cron syntax, but likely to produce inaccurate results"""
261 # Extending CroniterBadCronError, which may be contridatory, but this allows
262 # catching both errors with a single exception. From a user perspective
263 # these will likely be handled the same way.
266class CroniterBadDateError(CroniterError):
267 """Unable to find next/prev timestamp match"""
270class CroniterNotAlphaError(CroniterBadCronError):
271 """Cron syntax contains an invalid day or month abbreviation"""
274class croniter:
275 MONTHS_IN_YEAR = 12
277 # This helps with expanding `*` fields into `lower-upper` ranges. Each item
278 # in this tuple maps to the corresponding field index
279 RANGES = ((0, 59), (0, 23), (1, 31), (1, 12), (0, 6), (0, 59), (1970, 2099))
281 ALPHACONV: tuple[dict[str, Union[int, str]], ...] = (
282 {}, # 0: min
283 {}, # 1: hour
284 {"l": "l"}, # 2: dom
285 # 3: mon
286 copy.deepcopy(M_ALPHAS),
287 # 4: dow
288 copy.deepcopy(DOW_ALPHAS),
289 # 5: second
290 {},
291 # 6: year
292 {},
293 )
295 LOWMAP: tuple[dict[int, int], ...] = ({}, {}, {0: 1}, {0: 1}, {7: 0}, {}, {})
297 LEN_MEANS_ALL = (60, 24, 31, 12, 7, 60, 130)
299 def __init__(
300 self,
301 expr_format: str,
302 start_time: Optional[Union[datetime.datetime, float]] = None,
303 ret_type: type = float,
304 day_or: bool = True,
305 max_years_between_matches: Optional[int] = None,
306 is_prev: bool = False,
307 hash_id: Optional[Union[bytes, str]] = None,
308 implement_cron_bug: bool = False,
309 second_at_beginning: bool = False,
310 expand_from_start_time: bool = False,
311 ) -> None:
312 self._ret_type = ret_type
313 self._day_or = day_or
314 self._implement_cron_bug = implement_cron_bug
315 self.second_at_beginning = bool(second_at_beginning)
316 self._expand_from_start_time = expand_from_start_time
318 if hash_id is not None:
319 if not isinstance(hash_id, (bytes, str)):
320 raise TypeError("hash_id must be bytes or UTF-8 string")
321 if not isinstance(hash_id, bytes):
322 hash_id = hash_id.encode("UTF-8")
324 self._max_years_btw_matches_explicitly_set = max_years_between_matches is not None
325 if max_years_between_matches is None:
326 max_years_between_matches = 50
327 elif isinstance(max_years_between_matches, bool) or not isinstance(
328 max_years_between_matches, int
329 ):
330 # bool subclasses int. int(True) used to become 1, floats were
331 # truncated, and numeric strings were parsed by int().
332 raise TypeError(
333 "max_years_between_matches must be an int >= 1 or None, "
334 f"not {type(max_years_between_matches).__name__} "
335 f"(got {max_years_between_matches!r})"
336 )
337 elif max_years_between_matches < 1:
338 raise ValueError(
339 "max_years_between_matches must be an int >= 1 or None "
340 f"(got {max_years_between_matches!r})"
341 )
342 self._max_years_between_matches = max_years_between_matches
344 if start_time is None:
345 start_time = time()
347 self.tzinfo: Optional[datetime.tzinfo] = None
349 self.start_time = 0.0
350 self.dst_start_time = 0.0
351 self.cur = 0.0
352 self.set_current(start_time, force=True)
354 self.expanded, self.nth_weekday_of_month, self.expressions, self.nearest_weekday = self._expand(
355 expr_format,
356 hash_id=hash_id,
357 from_timestamp=self.dst_start_time if self._expand_from_start_time else None,
358 from_timestamp_tz=self.tzinfo if self._expand_from_start_time else None,
359 second_at_beginning=second_at_beginning,
360 )
361 self.fields = CRON_FIELDS[len(self.expanded)]
362 self._is_prev = is_prev
364 @classmethod
365 def _alphaconv(cls, index, key, expressions):
366 try:
367 return cls.ALPHACONV[index][key]
368 except KeyError:
369 raise CroniterNotAlphaError(f"[{' '.join(expressions)}] is not acceptable")
371 def get_next(self, ret_type=None, start_time=None, update_current=True):
372 if start_time and self._expand_from_start_time:
373 raise ValueError(
374 "start_time is not supported when using expand_from_start_time = True."
375 )
376 return self._get_next(
377 ret_type=ret_type, start_time=start_time, is_prev=False, update_current=update_current
378 )
380 def get_prev(self, ret_type=None, start_time=None, update_current=True):
381 return self._get_next(
382 ret_type=ret_type, start_time=start_time, is_prev=True, update_current=update_current
383 )
385 def get_current(self, ret_type=None):
386 ret_type = ret_type or self._ret_type
387 if issubclass(ret_type, datetime.datetime):
388 return self.timestamp_to_datetime(self.cur)
389 return self.cur
391 def set_current(
392 self, start_time: Optional[Union[datetime.datetime, float]], force: bool = True
393 ) -> float:
394 if (force or (self.cur is None)) and start_time is not None:
395 if isinstance(start_time, datetime.datetime):
396 self.tzinfo = start_time.tzinfo
397 start_time = self.datetime_to_timestamp(start_time)
399 self.start_time = start_time
400 self.dst_start_time = start_time
401 self.cur = start_time
402 return self.cur
404 @staticmethod
405 def datetime_to_timestamp(d: datetime.datetime) -> float:
406 """
407 Converts a `datetime` object `d` into a UNIX timestamp.
408 """
409 return datetime_to_timestamp(d)
411 _datetime_to_timestamp = datetime_to_timestamp # retrocompat
413 def timestamp_to_datetime(self, timestamp: float, tzinfo: Any = MARKER) -> datetime.datetime:
414 """
415 Converts a UNIX `timestamp` into a `datetime` object.
416 """
417 if tzinfo is MARKER: # allow to give tzinfo=None even if self.tzinfo is set
418 tzinfo = self.tzinfo
419 if OVERFLOW32B_MODE:
420 # degraded mode to workaround Y2038
421 # see https://github.com/python/cpython/issues/101069
422 result = EPOCH.replace(tzinfo=None) + datetime.timedelta(seconds=timestamp)
423 else:
424 result = datetime.datetime.fromtimestamp(timestamp, tz=tzutc()).replace(tzinfo=None)
425 if tzinfo:
426 result = result.replace(tzinfo=UTC_DT).astimezone(tzinfo)
427 return result
429 _timestamp_to_datetime = timestamp_to_datetime # retrocompat
431 def _get_next(self, ret_type=None, start_time=None, is_prev=None, update_current=None):
432 if update_current is None:
433 update_current = True
434 self.set_current(start_time, force=True)
435 if is_prev is None:
436 is_prev = self._is_prev
437 self._is_prev = is_prev
439 ret_type = ret_type or self._ret_type
441 if not issubclass(ret_type, (float, datetime.datetime)):
442 raise TypeError("Invalid ret_type, only 'float' or 'datetime' is acceptable.")
444 result = self._calc_next(is_prev)
445 timestamp = self.datetime_to_timestamp(result)
446 if update_current:
447 self.cur = timestamp
448 if issubclass(ret_type, datetime.datetime):
449 return result
450 return timestamp
452 # iterator protocol, to enable direct use of croniter
453 # objects in a loop, like "for dt in croniter("5 0 * * *'): ..."
454 # or for combining multiple croniters into single
455 # dates feed using 'itertools' module
456 def all_next(self, ret_type=None, start_time=None, update_current=None):
457 """
458 Returns a generator yielding consecutive dates.
460 May be used instead of an implicit call to __iter__ whenever a
461 non-default `ret_type` needs to be specified.
462 """
463 # In a Python 3.7+ world: contextlib.suppress and contextlib.nullcontext could
464 # be used instead
465 try:
466 while True:
467 self._is_prev = False
468 yield self._get_next(
469 ret_type=ret_type, start_time=start_time, update_current=update_current
470 )
471 start_time = None
472 except CroniterBadDateError:
473 if self._max_years_btw_matches_explicitly_set:
474 return
475 raise
477 def all_prev(self, ret_type=None, start_time=None, update_current=None):
478 """
479 Returns a generator yielding previous dates.
480 """
481 try:
482 while True:
483 self._is_prev = True
484 yield self._get_next(
485 ret_type=ret_type, start_time=start_time, update_current=update_current
486 )
487 start_time = None
488 except CroniterBadDateError:
489 if self._max_years_btw_matches_explicitly_set:
490 return
491 raise
493 def iter(self, *args, **kwargs):
494 return self.all_prev if self._is_prev else self.all_next
496 def __iter__(self):
497 return self
499 __next__ = next = _get_next
501 def _calc_next(self, is_prev: bool) -> datetime.datetime:
502 current = self.timestamp_to_datetime(self.cur)
503 expanded = self.expanded[:]
504 nth_weekday_of_month = self.nth_weekday_of_month.copy()
506 # exception to support day of month and day of week as defined in cron
507 if (expanded[DAY_FIELD][0] != "*" and expanded[DOW_FIELD][0] != "*") and self._day_or:
508 # If requested, handle a bug in vixie cron/ISC cron where day_of_month and
509 # day_of_week form an intersection (AND) instead of a union (OR) if either
510 # field is an asterisk or starts with an asterisk (https://crontab.guru/cron-bug.html)
511 if self._implement_cron_bug and (
512 re_star.match(self.expressions[DAY_FIELD])
513 or re_star.match(self.expressions[DOW_FIELD])
514 ):
515 # To produce a schedule identical to the cron bug, we'll bypass the code
516 # that makes a union of DOM and DOW, and instead skip to the code that
517 # does an intersect instead
518 pass
519 else:
520 # Under OR semantics an unsatisfiable side -- the 31st in a month
521 # that has no 31st, say -- contributes no dates rather than ruling
522 # out the expression, so the other side must still be able to
523 # match. Only both sides failing means there is no such date.
524 #
525 # That only holds while each side is a clean operand. '#' and 'W'
526 # are not carried by the DAY/DOW fields but by nth_weekday_of_month
527 # and self.nearest_weekday, so blanking the fields above does not
528 # remove them and each side stays an intersection (the semantics
529 # test_issue_k33 pins down). Swallowing a failure there would drop
530 # the other field from the schedule instead, so let it propagate.
531 clean_split = not nth_weekday_of_month and not self.nearest_weekday
533 bak = expanded[DOW_FIELD]
534 expanded[DOW_FIELD] = ["*"]
535 try:
536 t1 = self._calc(current, expanded, nth_weekday_of_month, is_prev)
537 except CroniterBadDateError:
538 if not clean_split:
539 raise
540 t1 = None
541 expanded[DOW_FIELD] = bak
542 expanded[DAY_FIELD] = ["*"]
544 try:
545 t2 = self._calc(current, expanded, nth_weekday_of_month, is_prev)
546 except CroniterBadDateError:
547 if not clean_split:
548 raise
549 t2 = None
551 if t1 is None:
552 if t2 is None:
553 raise CroniterBadDateError(
554 "failed to find prev date" if is_prev else "failed to find next date"
555 )
556 return t2
557 if t2 is None:
558 return t1
559 if is_prev:
560 return t1 if t1 > t2 else t2
561 return t1 if t1 < t2 else t2
563 return self._calc(current, expanded, nth_weekday_of_month, is_prev)
565 def _calc(
566 self,
567 now: datetime.datetime,
568 expanded: list[ExpandedExpression],
569 nth_weekday_of_month: dict[int, set[int]],
570 is_prev: bool,
571 ) -> datetime.datetime:
572 if is_prev:
573 nearest_diff_method = self._get_prev_nearest_diff
574 offset = relativedelta(microseconds=-1)
575 else:
576 nearest_diff_method = self._get_next_nearest_diff
577 if len(expanded) > UNIX_CRON_LEN:
578 offset = relativedelta(seconds=1)
579 else:
580 offset = relativedelta(minutes=1)
581 # Calculate the next cron time in local time a.k.a. timezone unaware time.
582 unaware_time = now.replace(tzinfo=None) + offset
583 if len(expanded) > UNIX_CRON_LEN:
584 unaware_time = unaware_time.replace(microsecond=0)
585 else:
586 unaware_time = unaware_time.replace(second=0, microsecond=0)
588 month = unaware_time.month
589 year = current_year = unaware_time.year
591 def proc_year(d):
592 if len(expanded) == YEAR_CRON_LEN:
593 try:
594 expanded[YEAR_FIELD].index("*")
595 except ValueError:
596 # use None as range_val to indicate no loop
597 diff_year = nearest_diff_method(d.year, expanded[YEAR_FIELD], None)
598 if diff_year is None:
599 return None, d
600 if diff_year != 0:
601 if is_prev:
602 d += relativedelta(
603 years=diff_year, month=12, day=31, hour=23, minute=59, second=59
604 )
605 else:
606 d += relativedelta(
607 years=diff_year, month=1, day=1, hour=0, minute=0, second=0
608 )
609 return True, d
610 return False, d
612 def proc_month(d):
613 try:
614 expanded[MONTH_FIELD].index("*")
615 except ValueError:
616 diff_month = nearest_diff_method(
617 d.month, expanded[MONTH_FIELD], self.MONTHS_IN_YEAR
618 )
619 reset_day = 1
621 if diff_month is not None and diff_month != 0:
622 if is_prev:
623 d += relativedelta(months=diff_month)
624 reset_day = _last_day_of_month(d.year, d.month)
625 d += relativedelta(day=reset_day, hour=23, minute=59, second=59)
626 else:
627 d += relativedelta(
628 months=diff_month, day=reset_day, hour=0, minute=0, second=0
629 )
630 return True, d
631 return False, d
633 def proc_day_of_month(d):
634 try:
635 expanded[DAY_FIELD].index("*")
636 except ValueError:
637 days = _last_day_of_month(year, month)
638 if "l" in expanded[DAY_FIELD] and days == d.day:
639 return False, d
641 if is_prev:
642 prev_month = (month - 2) % self.MONTHS_IN_YEAR + 1
643 prev_year = year - 1 if month == 1 else year
644 days_in_prev_month = _last_day_of_month(prev_year, prev_month)
645 diff_day = nearest_diff_method(d.day, expanded[DAY_FIELD], days_in_prev_month)
646 else:
647 diff_day = nearest_diff_method(d.day, expanded[DAY_FIELD], days)
649 if diff_day is not None and diff_day != 0:
650 if is_prev:
651 d += relativedelta(days=diff_day, hour=23, minute=59, second=59)
652 else:
653 d += relativedelta(days=diff_day, hour=0, minute=0, second=0)
654 return True, d
655 return False, d
657 def proc_day_of_week(d):
658 try:
659 expanded[DOW_FIELD].index("*")
660 except ValueError:
661 diff_day_of_week = nearest_diff_method(d.isoweekday() % 7, expanded[DOW_FIELD], 7)
662 if diff_day_of_week is not None and diff_day_of_week != 0:
663 if is_prev:
664 d += relativedelta(days=diff_day_of_week, hour=23, minute=59, second=59)
665 else:
666 d += relativedelta(days=diff_day_of_week, hour=0, minute=0, second=0)
667 return True, d
668 return False, d
670 def proc_day_of_week_nth(d):
671 if "*" in nth_weekday_of_month:
672 s = nth_weekday_of_month["*"]
673 for i in range(0, 7):
674 if i in nth_weekday_of_month:
675 nth_weekday_of_month[i].update(s)
676 else:
677 nth_weekday_of_month[i] = s
678 del nth_weekday_of_month["*"]
680 candidates = []
681 for wday, nth in nth_weekday_of_month.items():
682 c = self._get_nth_weekday_of_month(d.year, d.month, wday)
683 for n in nth:
684 if n == "l":
685 candidate = c[-1]
686 elif len(c) < n:
687 continue
688 else:
689 candidate = c[n - 1]
690 if (is_prev and candidate <= d.day) or (not is_prev and d.day <= candidate):
691 candidates.append(candidate)
693 if not candidates:
694 if is_prev:
695 d += relativedelta(days=-d.day, hour=23, minute=59, second=59)
696 else:
697 days = _last_day_of_month(year, month)
698 d += relativedelta(days=(days - d.day + 1), hour=0, minute=0, second=0)
699 return True, d
701 candidates.sort()
702 diff_day = (candidates[-1] if is_prev else candidates[0]) - d.day
703 if diff_day != 0:
704 if is_prev:
705 d += relativedelta(days=diff_day, hour=23, minute=59, second=59)
706 else:
707 d += relativedelta(days=diff_day, hour=0, minute=0, second=0)
708 return True, d
709 return False, d
711 def proc_nearest_weekday(d):
712 """Process W (nearest weekday) day-of-month entries."""
713 candidates = []
714 for w_day in self.nearest_weekday:
715 candidate = self._get_nearest_weekday(d.year, d.month, w_day)
716 if (is_prev and candidate <= d.day) or (not is_prev and d.day <= candidate):
717 candidates.append(candidate)
719 if not candidates:
720 if is_prev:
721 d += relativedelta(days=-d.day, hour=23, minute=59, second=59)
722 else:
723 days = _last_day_of_month(year, month)
724 d += relativedelta(days=(days - d.day + 1), hour=0, minute=0, second=0)
725 return True, d
727 candidates.sort()
728 diff_day = (candidates[-1] if is_prev else candidates[0]) - d.day
729 if diff_day != 0:
730 if is_prev:
731 d += relativedelta(days=diff_day, hour=23, minute=59, second=59)
732 else:
733 d += relativedelta(days=diff_day, hour=0, minute=0, second=0)
734 return True, d
735 return False, d
737 def proc_hour(d):
738 try:
739 expanded[HOUR_FIELD].index("*")
740 except ValueError:
741 diff_hour = nearest_diff_method(d.hour, expanded[HOUR_FIELD], 24)
742 if diff_hour is not None and diff_hour != 0:
743 if is_prev:
744 d += relativedelta(hours=diff_hour, minute=59, second=59)
745 else:
746 d += relativedelta(hours=diff_hour, minute=0, second=0)
747 return True, d
748 return False, d
750 def proc_minute(d):
751 try:
752 expanded[MINUTE_FIELD].index("*")
753 except ValueError:
754 diff_min = nearest_diff_method(d.minute, expanded[MINUTE_FIELD], 60)
755 if diff_min is not None and diff_min != 0:
756 if is_prev:
757 d += relativedelta(minutes=diff_min, second=59)
758 else:
759 d += relativedelta(minutes=diff_min, second=0)
760 return True, d
761 return False, d
763 def proc_second(d):
764 if len(expanded) > UNIX_CRON_LEN:
765 try:
766 expanded[SECOND_FIELD].index("*")
767 except ValueError:
768 diff_sec = nearest_diff_method(d.second, expanded[SECOND_FIELD], 60)
769 if diff_sec is not None and diff_sec != 0:
770 d += relativedelta(seconds=diff_sec)
771 return True, d
772 else:
773 d += relativedelta(second=0)
774 return False, d
776 procs = [
777 proc_year,
778 proc_month,
779 (proc_nearest_weekday if self.nearest_weekday else proc_day_of_month),
780 (proc_day_of_week_nth if nth_weekday_of_month else proc_day_of_week),
781 proc_hour,
782 proc_minute,
783 proc_second,
784 ]
786 while abs(year - current_year) <= self._max_years_between_matches:
787 next = False
788 stop = False
789 for proc in procs:
790 (changed, unaware_time) = proc(unaware_time)
791 # `None` can be set mostly for year processing
792 # so please see proc_year / _get_prev_nearest_diff / _get_next_nearest_diff
793 if changed is None:
794 stop = True
795 break
796 if changed:
797 month, year = unaware_time.month, unaware_time.year
798 next = True
799 break
800 if stop:
801 break
802 if next:
803 continue
805 unaware_time = unaware_time.replace(microsecond=0)
806 if now.tzinfo is None:
807 return unaware_time
809 # Add timezone information back and handle DST changes
810 aware_time, exists = _add_tzinfo(unaware_time, now, is_prev)
812 if not exists and (
813 not _is_successor(aware_time, now, is_prev) or "*" in expanded[HOUR_FIELD]
814 ):
815 # The calculated local date does not exist and moving the time forward
816 # to the next valid time isn't the correct solution. Search for the
817 # next matching cron time that exists.
818 while not exists:
819 unaware_time = self._calc(
820 unaware_time, expanded, nth_weekday_of_month, is_prev
821 )
822 aware_time, exists = _add_tzinfo(unaware_time, now, is_prev)
824 offset_delta = _timezone_delta(now, aware_time)
825 if not offset_delta:
826 # There was no DST change.
827 return aware_time
829 # There was a DST change. So check if there is a alternative cron time
830 # for the other UTC offset.
831 alternative_unaware_time = now.replace(tzinfo=None) + offset_delta
832 alternative_unaware_time = self._calc(
833 alternative_unaware_time, expanded, nth_weekday_of_month, is_prev
834 )
835 alternative_aware_time, exists = _add_tzinfo(alternative_unaware_time, now, is_prev)
837 if not exists:
838 # The alternative crossed a DST gap and was shifted forward to
839 # the next existing wall clock. Such a manufactured time is not
840 # a fire time the expression names, so keep the regular
841 # candidate instead.
842 return aware_time
844 if not _is_successor(alternative_aware_time, now, is_prev):
845 # The alternative time is an ancestor of now. Thus it is not an alternative.
846 return aware_time
848 if _is_successor(aware_time, alternative_aware_time, is_prev):
849 return alternative_aware_time
851 return aware_time
853 if is_prev:
854 raise CroniterBadDateError("failed to find prev date")
855 raise CroniterBadDateError("failed to find next date")
857 @staticmethod
858 def _get_next_nearest_diff(x, to_check, range_val):
859 """
860 `range_val` is the range of a field.
861 If no available time, we can move to next loop(like next month).
862 `range_val` can also be set to `None` to indicate that there is no loop.
863 ( Currently, should only used for `year` field )
864 """
865 for i, d in enumerate(to_check):
866 if range_val is not None:
867 if d == "l":
868 # if 'l' then it is the last day of month
869 # => its value of range_val
870 d = range_val
871 elif d > range_val:
872 continue
873 if d >= x:
874 return d - x
875 # When range_val is None and x not exists in to_check,
876 # `None` will be returned to suggest no more available time
877 if range_val is None:
878 return None
879 return to_check[0] - x + range_val
881 @staticmethod
882 def _get_prev_nearest_diff(x, to_check, range_val):
883 """
884 `range_val` is the range of a field.
885 If no available time, we can move to previous loop(like previous month).
886 Range_val can also be set to `None` to indicate that there is no loop.
887 ( Currently should only used for `year` field )
888 """
889 candidates = to_check[:]
890 candidates.reverse()
891 for d in candidates:
892 if d != "l" and d <= x:
893 return d - x
894 if "l" in candidates:
895 return -x
896 # When range_val is None and x not exists in to_check,
897 # `None` will be returned to suggest no more available time
898 if range_val is None:
899 return None
900 candidate = candidates[0]
901 for c in candidates:
902 # fixed: c < range_val
903 # this code will reject all 31 day of month, 12 month, 59 second,
904 # 23 hour and so on.
905 # if candidates has just a element, this will not harmful.
906 # but candidates have multiple elements, then values equal to
907 # range_val will rejected.
908 if c <= range_val:
909 candidate = c
910 break
911 # fix crontab "0 6 30 3 *" condidates only a element, then get_prev error
912 # return 2021-03-02 06:00:00
913 if candidate > range_val:
914 return -range_val
915 return candidate - x - range_val
917 @staticmethod
918 def _get_nth_weekday_of_month(year: int, month: int, day_of_week: int) -> tuple[int, ...]:
919 """For a given year/month return a list of days in nth-day-of-month order.
920 The last weekday of the month is always [-1].
921 """
922 w = (day_of_week + 6) % 7
923 c = calendar.Calendar(w).monthdayscalendar(year, month)
924 if c[0][0] == 0:
925 c.pop(0)
926 return tuple(i[0] for i in c)
928 @staticmethod
929 def _get_nearest_weekday(year, month, day):
930 """Get the nearest weekday (Mon-Fri) to the given day in the given month.
932 Rules:
933 - If the day is a weekday, return it.
934 - If Saturday, return Friday (day-1), unless that crosses into previous month,
935 then return Monday (day+2).
936 - If Sunday, return Monday (day+1), unless that crosses into next month,
937 then return Friday (day-2).
938 """
939 last_day = _last_day_of_month(year, month)
940 day = min(day, last_day)
941 weekday = calendar.weekday(year, month, day) # 0=Mon, 6=Sun
942 if weekday < 5: # Mon-Fri
943 return day
944 if weekday == 5: # Saturday
945 if day > 1:
946 return day - 1 # Friday
947 else:
948 return day + 2 # Monday (1st is Sat, so 3rd is Mon)
949 # Sunday
950 if day < last_day:
951 return day + 1 # Monday
952 else:
953 return day - 2 # Friday (last day is Sun, go back to Fri)
955 @classmethod
956 def value_alias(cls, val, field_index, len_expressions=UNIX_CRON_LEN):
957 if isinstance(len_expressions, (list, dict, tuple, set)):
958 len_expressions = len(len_expressions)
959 if val in cls.LOWMAP[field_index] and not (
960 # do not support 0 as a month either for classical 5 fields cron,
961 # 6fields second repeat form or 7 fields year form
962 # but still let conversion happen if day field is shifted
963 (field_index in [DAY_FIELD, MONTH_FIELD] and len_expressions == UNIX_CRON_LEN)
964 or (
965 field_index in [DAY_FIELD, MONTH_FIELD]
966 and len_expressions == SECOND_CRON_LEN
967 )
968 or (
969 field_index in [DAY_FIELD, MONTH_FIELD]
970 and len_expressions == YEAR_CRON_LEN
971 )
972 ):
973 val = cls.LOWMAP[field_index][val]
974 return val
976 # Maximum days in each month (non-leap year for Feb)
977 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}
979 @classmethod
980 def _expand(cls, expr_format, hash_id=None, second_at_beginning=False, from_timestamp=None, from_timestamp_tz=None, strict=False, strict_year=None):
981 # Split the expression in components, and normalize L -> l, MON -> mon,
982 # etc. Keep expr_format untouched so we can use it in the exception
983 # messages.
984 expr_aliases = {
985 "@midnight": ("0 0 * * *", "h h(0-2) * * * h"),
986 "@hourly": ("0 * * * *", "h * * * * h"),
987 "@daily": ("0 0 * * *", "h h * * * h"),
988 "@weekly": ("0 0 * * 0", "h h * * h h"),
989 "@monthly": ("0 0 1 * *", "h h h * * h"),
990 "@yearly": ("0 0 1 1 *", "h h h h * h"),
991 "@annually": ("0 0 1 1 *", "h h h h * h"),
992 }
994 efl = expr_format.lower()
995 hash_id_expr = 1 if hash_id is not None else 0
996 try:
997 efl = expr_aliases[efl][hash_id_expr]
998 except KeyError:
999 pass
1001 expressions = field_re.findall(efl)
1003 if len(expressions) not in VALID_LEN_EXPRESSION:
1004 raise CroniterBadCronError(
1005 "Exactly 5, 6 or 7 columns has to be specified for iterator expression."
1006 )
1008 if len(expressions) > UNIX_CRON_LEN and second_at_beginning:
1009 # move second to it's own(6th) field to process by same logical
1010 expressions.insert(SECOND_FIELD, expressions.pop(0))
1012 expanded = []
1013 nth_weekday_of_month = {}
1014 nearest_weekday = set()
1016 for field_index, expr in enumerate(expressions):
1017 for expanderid, expander in EXPANDERS.items():
1018 expr = expander(cls).expand(
1019 efl, field_index, expr, hash_id=hash_id, from_timestamp=from_timestamp
1020 )
1022 if "?" in expr:
1023 if expr != "?":
1024 raise CroniterBadCronError(
1025 f"[{expr_format}] is not acceptable."
1026 f" Question mark can not used with other characters"
1027 )
1028 if field_index not in [DAY_FIELD, DOW_FIELD]:
1029 raise CroniterBadCronError(
1030 f"[{expr_format}] is not acceptable. "
1031 f"Question mark can only used in day_of_month or day_of_week"
1032 )
1033 # currently just trade `?` as `*`
1034 expr = "*"
1036 e_list = expr.split(",")
1037 res = []
1038 seen = set()
1040 while len(e_list) > 0:
1041 e = e_list.pop()
1042 nth = None
1044 if field_index == DOW_FIELD:
1045 # Handle special case in the dow expression: 2#3, l3
1046 special_dow_rem = special_dow_re.match(str(e))
1047 if special_dow_rem:
1048 g = special_dow_rem.groupdict()
1049 he, last = g.get("he", ""), g.get("last", "")
1050 if he:
1051 e = he
1052 try:
1053 nth = int(last)
1054 except ValueError:
1055 nth = None
1056 # A plain check, not an assert: ``python -O`` strips
1057 # asserts, which let ``1#0`` and ``1#7`` through.
1058 if nth is None or not 1 <= nth <= 5:
1059 raise CroniterBadCronError(
1060 f"[{expr_format}] is not acceptable."
1061 f" Invalid day_of_week value: '{last}'"
1062 )
1063 elif last:
1064 e = last
1065 nth = g["pre"] # 'l'
1067 if field_index == DAY_FIELD:
1068 # Handle W (nearest weekday) in day-of-month: 15w, w15
1069 w_match = nearest_weekday_re.match(str(e))
1070 if w_match:
1071 w_day = int(w_match.group(1) or w_match.group(2))
1072 if w_day < 1 or w_day > 31:
1073 raise CroniterBadCronError(
1074 f"[{expr_format}] is not acceptable,"
1075 f" nearest weekday day value '{w_day}' out of range"
1076 )
1077 if len(e_list) > 0 or len(res) > 0:
1078 raise CroniterBadCronError(
1079 f"[{expr_format}] is not acceptable."
1080 f" 'W' can only be used with a single day value,"
1081 f" not in a list or range"
1082 )
1083 nearest_weekday.add(w_day)
1084 res.append(w_day)
1085 continue
1087 # Before matching step_search_re, normalize "*" to "{min}-{max}".
1088 # Example: in the minute field, "*/5" normalizes to "0-59/5"
1089 t = re.sub(
1090 r"^\*(\/.+)$",
1091 r"%d-%d\1" % (cls.RANGES[field_index][0], cls.RANGES[field_index][1]),
1092 str(e),
1093 )
1094 m = step_search_re.search(t)
1096 # "{start}/{step}" is its own shape, not a range. It is normalized
1097 # below to "{start}-{max}/{step}" so that one regex parses both, but
1098 # the two must not be conflated afterwards: "Jan-Jan" is an explicit
1099 # equal range that croniter deliberately expands to the whole cycle,
1100 # whereas "DEC/3" is a start with a step and denotes just [DEC].
1101 start_with_step = False
1102 if not m:
1103 # Before matching step_search_re,
1104 # normalize "{start}/{step}" to "{start}-{max}/{step}".
1105 # Example: in the minute field, "10/5" normalizes to "10-59/5"
1106 t = re.sub(r"^(.+)\/(.+)$", r"\1-%d/\2" % (cls.RANGES[field_index][1]), str(e))
1107 m = step_search_re.search(t)
1108 start_with_step = bool(m)
1110 if m:
1111 # early abort if low/high are out of bounds
1112 (low, high, step) = m.group(1), m.group(2), m.group(4) or 1
1113 if field_index == DAY_FIELD and high == "l":
1114 high = "31"
1116 if not only_int_re.search(low):
1117 low = str(cls._alphaconv(field_index, low, expressions))
1119 if not only_int_re.search(high):
1120 high = str(cls._alphaconv(field_index, high, expressions))
1122 # normally, it's already guarded by the RE that should not accept
1123 # not-int values.
1124 if not only_int_re.search(str(step)):
1125 raise CroniterBadCronError(
1126 f"[{expr_format}] step '{step}'"
1127 f" in field {field_index} is not acceptable"
1128 )
1129 step = int(step)
1130 if step == 0:
1131 raise CroniterBadCronError(
1132 f"[{expr_format}] step '{step}'"
1133 f" in field {field_index} is not acceptable"
1134 )
1136 for band in low, high:
1137 if not only_int_re.search(str(band)):
1138 raise CroniterBadCronError(
1139 f"[{expr_format}] bands '{low}-{high}'"
1140 f" in field {field_index} are not acceptable"
1141 )
1143 low, high = (
1144 cls.value_alias(int(_val), field_index, expressions)
1145 for _val in (low, high)
1146 )
1148 if max(low, high) > max(
1149 cls.RANGES[field_index][0], cls.RANGES[field_index][1]
1150 ):
1151 raise CroniterBadCronError(f"{expr_format} is out of bands")
1153 # "{start}/{step}" normalizes to "{start}-{max}/{step}", so when
1154 # the start IS the field maximum the two bounds collide and the
1155 # token becomes indistinguishable from an explicitly written equal
1156 # range. That is the whole bug: "59/15" was read the way "Jan-Jan"
1157 # is -- the whole cycle -- and so fired at :00/:15/:30/:45 instead
1158 # of :59. Recognising the collision here is what keeps the two
1159 # apart, both in the widening just below and in the range built
1160 # further down.
1161 #
1162 # Deliberately ``low == high`` and not the wider ``low + step >
1163 # high``. Only the collision is a false equal range: a token whose
1164 # step merely overshoots its range -- "50-51/15" -- still describes
1165 # a cycle, and re-basing must stay free to supply its phase.
1166 start_at_field_max = start_with_step and low == high
1168 # An equal range that was *written* as one -- "Jan-Jan", "Sun-Sun"
1169 # -- means the whole cycle, so widen it to the field's own bounds
1170 # here rather than in a branch further down. Widening it before
1171 # re-basing is what makes the two spellings of one schedule agree:
1172 # "1-1/3" and "*/3" are the same expression, and both now take the
1173 # phase of their cycle from ``start_time``.
1174 #
1175 # ``start_with_step`` is load-bearing: "{start}/{step}" normalizes
1176 # to "{start}-{max}/{step}", so at the field maximum its bounds
1177 # collide without the user having written an equal range at all.
1178 # That is #246's bug, and widening "59/15" back to the whole field
1179 # here would reintroduce it.
1180 if low == high and not start_with_step:
1181 low, high = cls.RANGES[field_index]
1183 # ``from_timestamp`` re-bases the start of a *cycle* on the start
1184 # time. A single point has no cycle to re-base: "59/15" is minute
1185 # 59, not a cycle starting there. (The bounds test below rejects it
1186 # anyway, since the only phase inside ``low == high`` is ``low``
1187 # itself; skipping the work here says so outright.)
1188 if from_timestamp is not None and not start_at_field_max:
1189 # Re-basing supplies the *phase* of the cycle, not its bounds:
1190 # take the first value of that phase lying inside the range the
1191 # expression declares. From a start at minute 7, "10-50/15" is
1192 # 22,37 -- not 7,22,37, which fires below its own lower bound.
1193 # If no value of the phase fits, the expression is honoured as
1194 # written, so an expansion can never come out empty. Note that
1195 # a re-phased range can hold fewer periods than the one it
1196 # replaces -- "10-20/7" is 10,17 by default but 14 alone from
1197 # a start at minute 7 -- which is inherent to re-phasing inside
1198 # a bounded window, not a value being dropped.
1199 phase = cls._get_low_from_current_date_number(
1200 field_index, int(step), int(from_timestamp), from_timestamp_tz
1201 )
1202 if phase < low:
1203 phase += -(-(low - phase) // step) * step
1204 # This one test is what keeps re-basing honest, and it covers
1205 # the wrapping ranges (Apr-Jan, Sat-Sun) too: there ``high``
1206 # lies below ``low``, no phase can satisfy it, and the range is
1207 # left as written rather than silently unwrapped. Its lower
1208 # half restates what the advance above already guarantees; it
1209 # is written out so the test reads as the range check it is.
1210 if low <= phase <= high:
1211 low = phase
1213 # Handle when the second bound of the range is in backtracking order:
1214 # eg: X-Sun or X-7 (Sat-Sun) in DOW, or X-Jan (Apr-Jan) in MONTH
1215 if start_at_field_max:
1216 rng = [low]
1217 elif low > high:
1218 whole_field_range = list(
1219 range(cls.RANGES[field_index][0], cls.RANGES[field_index][1] + 1, 1)
1220 )
1221 # Add FirstBound -> ENDRANGE, respecting step
1222 rng = list(range(low, cls.RANGES[field_index][1] + 1, step))
1223 # Then 0 -> SecondBound, but skipping n first occurences according to step
1224 # EG to respect such expressions : Apr-Jan/3
1225 to_skip = 0
1226 if rng:
1227 already_skipped = list(reversed(whole_field_range)).index(rng[-1])
1228 curpos = whole_field_range.index(rng[-1])
1229 if ((curpos + step) > len(whole_field_range)) and (
1230 already_skipped < step
1231 ):
1232 to_skip = step - already_skipped
1233 rng += list(range(cls.RANGES[field_index][0] + to_skip, high + 1, step))
1234 else:
1235 try:
1236 rng = list(range(low, high + 1, step))
1237 except ValueError as exc:
1238 raise CroniterBadCronError(f"invalid range: {exc}")
1240 if field_index == DOW_FIELD and nth and nth != "l":
1241 rng = [f"{item}#{nth}" for item in rng]
1242 e_list += [a for a in rng if a not in seen]
1243 seen.update(rng)
1244 else:
1245 if t.startswith("-"):
1246 raise CroniterBadCronError(
1247 f"[{expr_format}] is not acceptable, negative numbers not allowed"
1248 )
1249 if not star_or_int_re.search(t):
1250 t = cls._alphaconv(field_index, t, expressions)
1252 try:
1253 t = int(t)
1254 except ValueError:
1255 pass
1257 t = cls.value_alias(t, field_index, expressions)
1259 if t not in ["*", "l"] and (
1260 int(t) < cls.RANGES[field_index][0] or int(t) > cls.RANGES[field_index][1]
1261 ):
1262 raise CroniterBadCronError(
1263 f"[{expr_format}] is not acceptable, out of range"
1264 )
1266 res.append(t)
1268 if field_index == DOW_FIELD and nth:
1269 if t not in nth_weekday_of_month:
1270 nth_weekday_of_month[t] = set()
1271 nth_weekday_of_month[t].add(nth)
1273 res = set(res)
1274 res = sorted(res, key=lambda i: f"{i:02}" if isinstance(i, int) else i)
1275 if len(res) == cls.LEN_MEANS_ALL[field_index]:
1276 # Make sure the wildcard is used in the correct way (avoid over-optimization)
1277 if (field_index == DAY_FIELD and "*" not in expressions[DOW_FIELD]) or (
1278 field_index == DOW_FIELD and "*" not in expressions[DAY_FIELD]
1279 ):
1280 pass
1281 else:
1282 res = ["*"]
1284 expanded.append(["*"] if (len(res) == 1 and res[0] == "*") else res)
1286 # Check to make sure the dow combo in use is supported
1287 if nth_weekday_of_month:
1288 dow_expanded_set = set(expanded[DOW_FIELD])
1289 dow_expanded_set = dow_expanded_set.difference(nth_weekday_of_month.keys())
1290 dow_expanded_set.discard("*")
1291 # Skip: if it's all weeks instead of wildcard
1292 if dow_expanded_set and len(set(expanded[DOW_FIELD])) != cls.LEN_MEANS_ALL[DOW_FIELD]:
1293 raise CroniterUnsupportedSyntaxError(
1294 f"day-of-week field does not support mixing literal values and nth"
1295 f" day of week syntax. Cron: '{expr_format}'"
1296 f" dow={dow_expanded_set} vs nth={nth_weekday_of_month}"
1297 )
1299 if strict:
1300 # Cross-validate day-of-month against month (and optionally year)
1301 # to reject impossible combinations like "0 0 31 2 *" (Feb 31st).
1302 days = expanded[DAY_FIELD]
1303 months = expanded[MONTH_FIELD]
1304 if days != ["*"] and days != ["l"] and months != ["*"]:
1305 int_days = [d for d in days if isinstance(d, int)]
1306 int_months = [m for m in months if isinstance(m, int)]
1307 if int_days and int_months:
1308 # Determine max days per month, accounting for leap years
1309 days_in_month = dict(cls.DAYS_IN_MONTH)
1310 if 2 in int_months:
1311 has_leap_year = True # assume possible by default
1312 if strict_year is not None:
1313 # Year explicitly provided as parameter
1314 if isinstance(strict_year, int):
1315 has_leap_year = calendar.isleap(strict_year)
1316 else:
1317 has_leap_year = any(calendar.isleap(y) for y in strict_year)
1318 elif len(expanded) > YEAR_FIELD:
1319 years = expanded[YEAR_FIELD]
1320 if years != ["*"]:
1321 int_years = [y for y in years if isinstance(y, int)]
1322 if int_years:
1323 has_leap_year = any(calendar.isleap(y) for y in int_years)
1324 if has_leap_year:
1325 days_in_month[2] = 29
1326 min_day = min(int_days)
1327 max_possible = max(days_in_month[m] for m in int_months)
1328 if min_day > max_possible:
1329 raise CroniterBadCronError(
1330 f"[{expr_format}] is not acceptable. Day(s) {int_days}"
1331 f" can never occur in month(s) {int_months}"
1332 )
1334 return expanded, nth_weekday_of_month, expressions, nearest_weekday
1336 @classmethod
1337 def expand(
1338 cls,
1339 expr_format: str,
1340 hash_id: Optional[Union[bytes, str]] = None,
1341 second_at_beginning: bool = False,
1342 from_timestamp: Optional[float] = None,
1343 from_timestamp_tz: Optional[datetime.tzinfo] = None,
1344 strict: bool = False,
1345 strict_year: Optional[Union[int, list[int]]] = None,
1346 ) -> tuple[list[ExpandedExpression], dict[int, set[int]]]:
1347 """
1348 Expand a cron expression format into a noramlized format of
1349 list[list[int | 'l' | '*']]. The first list representing each element
1350 of the epxression, and each sub-list representing the allowed values
1351 for that expression component.
1353 A tuple is returned, the first value being the expanded epxression
1354 list, and the second being a `nth_weekday_of_month` mapping.
1356 Examples:
1358 # Every minute
1359 >>> croniter.expand('* * * * *')
1360 ([['*'], ['*'], ['*'], ['*'], ['*']], {})
1362 # On the hour
1363 >>> croniter.expand('0 0 * * *')
1364 ([[0], [0], ['*'], ['*'], ['*']], {})
1366 # Hours 0-5 and 10 monday through friday
1367 >>> croniter.expand('0-5,10 * * * mon-fri')
1368 ([[0, 1, 2, 3, 4, 5, 10], ['*'], ['*'], ['*'], [1, 2, 3, 4, 5]], {})
1370 Note that some special values such as nth day of week are expanded to a
1371 special mapping format for later processing:
1373 # Every minute on the 3rd tuesday of the month
1374 >>> croniter.expand('* * * * 2#3')
1375 ([['*'], ['*'], ['*'], ['*'], [2]], {2: {3}})
1377 # Every hour on the last day of the month
1378 >>> croniter.expand('0 * l * *')
1379 ([[0], ['*'], ['l'], ['*'], ['*']], {})
1381 # On the hour every 15 seconds
1382 >>> croniter.expand('0 0 * * * */15')
1383 ([[0], [0], ['*'], ['*'], ['*'], [0, 15, 30, 45]], {})
1384 """
1385 try:
1386 expanded, nth_weekday_of_month, _expressions, _nearest_weekday = cls._expand(
1387 expr_format,
1388 hash_id=hash_id,
1389 second_at_beginning=second_at_beginning,
1390 from_timestamp=from_timestamp,
1391 from_timestamp_tz=from_timestamp_tz,
1392 strict=strict,
1393 strict_year=strict_year,
1394 )
1395 return expanded, nth_weekday_of_month
1396 except (ValueError,) as exc:
1397 if isinstance(exc, CroniterError):
1398 raise
1399 trace = _traceback.format_exc()
1400 raise CroniterBadCronError(trace)
1402 @classmethod
1403 def _get_low_from_current_date_number(cls, field_index, step, from_timestamp, tzinfo=None):
1404 # Read the start time back in its own timezone. A naive start_time was
1405 # converted to a timestamp as if it were UTC, so UTC round-trips it to the
1406 # same wall clock and nothing changes for that case.
1407 dt = datetime.datetime.fromtimestamp(from_timestamp, tz=tzinfo or UTC_DT)
1408 if field_index == MINUTE_FIELD:
1409 return dt.minute % step
1410 if field_index == HOUR_FIELD:
1411 return dt.hour % step
1412 if field_index == DAY_FIELD:
1413 return ((dt.day - 1) % step) + 1
1414 if field_index == MONTH_FIELD:
1415 return ((dt.month - 1) % step) + 1
1416 if field_index == DOW_FIELD:
1417 return (dt.isoweekday() % 7) % step
1418 if field_index == SECOND_FIELD:
1419 return dt.second % step
1420 if field_index == YEAR_FIELD:
1421 # Like day and month, the year field does not start at 0, so the phase is
1422 # taken from the field minimum rather than from the value itself.
1423 year_start = cls.RANGES[YEAR_FIELD][0]
1424 return ((dt.year - year_start) % step) + year_start
1426 raise ValueError(f"Can't get current date number for field index {field_index}")
1428 @classmethod
1429 def is_valid(cls, expression, hash_id=None, encoding="UTF-8", second_at_beginning=False, strict=False, strict_year=None):
1430 if hash_id:
1431 if not isinstance(hash_id, (bytes, str)):
1432 raise TypeError("hash_id must be bytes or UTF-8 string")
1433 if not isinstance(hash_id, bytes):
1434 hash_id = hash_id.encode(encoding)
1435 try:
1436 cls.expand(expression, hash_id=hash_id, second_at_beginning=second_at_beginning, strict=strict, strict_year=strict_year)
1437 except CroniterError:
1438 return False
1439 return True
1441 @classmethod
1442 def match(
1443 cls,
1444 cron_expression,
1445 testdate,
1446 day_or=True,
1447 second_at_beginning=False,
1448 precision_in_seconds=None,
1449 ):
1450 return cls.match_range(
1451 cron_expression, testdate, testdate, day_or, second_at_beginning, precision_in_seconds
1452 )
1454 @classmethod
1455 def match_range(
1456 cls,
1457 cron_expression,
1458 from_datetime,
1459 to_datetime,
1460 day_or=True,
1461 second_at_beginning=False,
1462 precision_in_seconds=None,
1463 ):
1464 cron = cls(
1465 cron_expression,
1466 to_datetime,
1467 ret_type=datetime.datetime,
1468 day_or=day_or,
1469 second_at_beginning=second_at_beginning,
1470 )
1471 tdp = cron.get_current(datetime.datetime)
1472 if not tdp.microsecond:
1473 tdp += relativedelta(microseconds=1)
1474 cron.set_current(tdp, force=True)
1475 try:
1476 tdt = cron.get_prev()
1477 except CroniterBadDateError:
1478 return False
1479 if precision_in_seconds is None:
1480 precision_in_seconds = 1 if len(cron.expanded) > UNIX_CRON_LEN else 60
1481 duration_in_second = (to_datetime - from_datetime).total_seconds() + precision_in_seconds
1482 return (max(tdp, tdt) - min(tdp, tdt)).total_seconds() < duration_in_second
1485def croniter_range(
1486 start,
1487 stop,
1488 expr_format,
1489 ret_type=None,
1490 day_or=True,
1491 exclude_ends=False,
1492 _croniter=None,
1493 second_at_beginning=False,
1494 expand_from_start_time=False,
1495):
1496 """
1497 Generator that provides all times from start to stop matching the given cron expression.
1498 If the cron expression matches either 'start' and/or 'stop', those times will be returned as
1499 well unless 'exclude_ends=True' is passed.
1501 You can think of this function as sibling to the builtin range function for datetime objects.
1502 Like range(start,stop,step), except that here 'step' is a cron expression.
1503 """
1504 _croniter = _croniter or croniter
1505 auto_rt = datetime.datetime
1506 # type is used in first if branch for perfs reasons
1507 if type(start) is not type(stop) and not (
1508 isinstance(start, type(stop)) or isinstance(stop, type(start))
1509 ):
1510 raise CroniterBadTypeRangeError(
1511 f"The start and stop must be same type. {type(start)} != {type(stop)}"
1512 )
1513 if isinstance(start, (float, int)):
1514 start, stop = (
1515 datetime.datetime.fromtimestamp(t, tzutc()).replace(tzinfo=None) for t in (start, stop)
1516 )
1517 auto_rt = float
1518 if ret_type is None:
1519 ret_type = auto_rt
1520 if not exclude_ends:
1521 ms1 = relativedelta(microseconds=1)
1522 if start < stop: # Forward (normal) time order
1523 start -= ms1
1524 stop += ms1
1525 else: # Reverse time order
1526 start += ms1
1527 stop -= ms1
1528 year_span = math.floor(abs(stop.year - start.year)) + 1
1529 ic = _croniter(
1530 expr_format,
1531 start,
1532 ret_type=datetime.datetime,
1533 day_or=day_or,
1534 max_years_between_matches=year_span,
1535 second_at_beginning=second_at_beginning,
1536 expand_from_start_time=expand_from_start_time,
1537 )
1538 # define a continue (cont) condition function and step function for the main while loop
1539 if start < stop: # Forward
1541 def cont(v):
1542 return v < stop
1544 step = ic.get_next
1545 else: # Reverse
1547 def cont(v):
1548 return v > stop
1550 step = ic.get_prev
1551 try:
1552 dt = step()
1553 while cont(dt):
1554 if ret_type is float:
1555 yield ic.get_current(float)
1556 else:
1557 yield dt
1558 dt = step()
1559 except CroniterBadDateError:
1560 # Stop iteration when this exception is raised; no match found within the given year range
1561 return
1564class HashExpander:
1565 def __init__(self, cronit):
1566 self.cron = cronit
1568 def do(self, idx, hash_type="h", hash_id=None, range_end=None, range_begin=None):
1569 """Return a hashed/random integer given range/hash information"""
1570 if range_end is None:
1571 range_end = self.cron.RANGES[idx][1]
1572 if range_begin is None:
1573 range_begin = self.cron.RANGES[idx][0]
1574 if hash_type == "r":
1575 crc = random.randint(0, 0xFFFFFFFF)
1576 else:
1577 crc = binascii.crc32(hash_id) & 0xFFFFFFFF
1578 return ((crc >> idx) % (range_end - range_begin + 1)) + range_begin
1580 def match(self, efl, idx, expr, hash_id=None, **kw):
1581 return hash_expression_re.match(expr)
1583 def _expand_divisor(self, idx, m, hash_id, range_begin, range_end, from_timestamp=None):
1584 """Hash a start offset into the first period, then step to range_end.
1586 The offset is drawn from the first period, [range_begin, range_begin +
1587 divisor - 1], but never past range_end: a divisor wider than the range has
1588 only the range itself to draw from. And when the offset lands on range_end
1589 the step cannot reach a second value, so the result is that single value --
1590 emitting "{end}-{end}/{divisor}" instead would read as an explicit equal
1591 range, which croniter expands to the whole field.
1593 Under ``expand_from_start_time`` the phase of the cycle comes from the start
1594 time, not from the hash, so the declared range is emitted as written and the
1595 re-basing in ``_expand`` supplies the phase inside it. Emitting the hashed
1596 offset there instead would make re-basing read that offset as a declared
1597 lower bound: ``H/15`` with an offset of 6 from a start at minute 2 would fire
1598 at 17,32,47 rather than 2,17,32,47, losing one period an hour.
1599 """
1600 divisor = int(m["divisor"])
1601 if from_timestamp is not None:
1602 return f"{range_begin}-{range_end}/{divisor}"
1603 x = self.do(
1604 idx,
1605 hash_type=m["hash_type"],
1606 hash_id=hash_id,
1607 range_begin=range_begin,
1608 range_end=min(range_begin + divisor - 1, range_end),
1609 )
1610 if x == range_end:
1611 return str(x)
1612 return f"{x}-{range_end}/{divisor}"
1614 def expand(self, efl, idx, expr, hash_id=None, match="", **kw):
1615 """Expand a hashed/random expression to its normal representation"""
1616 if match == "":
1617 match = self.match(efl, idx, expr, hash_id, **kw)
1618 if not match:
1619 return expr
1620 m = match.groupdict()
1622 if m["hash_type"] == "h" and hash_id is None:
1623 raise CroniterBadCronError("Hashed definitions must include hash_id")
1625 if m["range_begin"] and m["range_end"]:
1626 if int(m["range_begin"]) >= int(m["range_end"]):
1627 raise CroniterBadCronError("Range end must be greater than range begin")
1628 # Check the bounds themselves, not just the value drawn from them: otherwise
1629 # "H(0-99)" in the minute field is valid or not depending on the hash id, and
1630 # "R(0-99)" depending on the draw.
1631 field_min, field_max = self.cron.RANGES[idx]
1632 for bound in (int(m["range_begin"]), int(m["range_end"])):
1633 if not field_min <= self.cron.value_alias(bound, idx) <= field_max:
1634 raise CroniterBadCronError(f"[{expr}] is not acceptable, out of range")
1636 if m["range_begin"] and m["range_end"] and m["divisor"]:
1637 # Example: H(30-59)/10 -> 34-59/10 (i.e. 34,44,54)
1638 if int(m["divisor"]) == 0:
1639 raise CroniterBadCronError(f"Bad expression: {expr}")
1641 return self._expand_divisor(
1642 idx,
1643 m,
1644 hash_id,
1645 int(m["range_begin"]),
1646 int(m["range_end"]),
1647 from_timestamp=kw.get("from_timestamp"),
1648 )
1649 if m["range_begin"] and m["range_end"]:
1650 # Example: H(0-29) -> 12
1651 return str(
1652 self.do(
1653 idx,
1654 hash_type=m["hash_type"],
1655 hash_id=hash_id,
1656 range_end=int(m["range_end"]),
1657 range_begin=int(m["range_begin"]),
1658 )
1659 )
1660 if m["divisor"]:
1661 # Example: H/15 -> 7-59/15 (i.e. 7,22,37,52)
1662 if int(m["divisor"]) == 0:
1663 raise CroniterBadCronError(f"Bad expression: {expr}")
1665 return self._expand_divisor(
1666 idx,
1667 m,
1668 hash_id,
1669 self.cron.RANGES[idx][0],
1670 self.cron.RANGES[idx][1],
1671 from_timestamp=kw.get("from_timestamp"),
1672 )
1674 # Example: H -> 32
1675 return str(self.do(idx, hash_type=m["hash_type"], hash_id=hash_id))
1678EXPANDERS = {"hash": HashExpander}