diff options
Diffstat (limited to '.venv/lib/python3.12/site-packages/apscheduler/triggers/cron/expressions.py')
-rw-r--r-- | .venv/lib/python3.12/site-packages/apscheduler/triggers/cron/expressions.py | 285 |
1 files changed, 285 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/apscheduler/triggers/cron/expressions.py b/.venv/lib/python3.12/site-packages/apscheduler/triggers/cron/expressions.py new file mode 100644 index 00000000..0d84ec23 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/apscheduler/triggers/cron/expressions.py @@ -0,0 +1,285 @@ +"""This module contains the expressions applicable for CronTrigger's fields.""" + +__all__ = ( + "AllExpression", + "RangeExpression", + "WeekdayRangeExpression", + "WeekdayPositionExpression", + "LastDayOfMonthExpression", +) + +import re +from calendar import monthrange + +from apscheduler.util import asint + +WEEKDAYS = ["mon", "tue", "wed", "thu", "fri", "sat", "sun"] +MONTHS = [ + "jan", + "feb", + "mar", + "apr", + "may", + "jun", + "jul", + "aug", + "sep", + "oct", + "nov", + "dec", +] + + +class AllExpression: + value_re = re.compile(r"\*(?:/(?P<step>\d+))?$") + + def __init__(self, step=None): + self.step = asint(step) + if self.step == 0: + raise ValueError("Increment must be higher than 0") + + def validate_range(self, field_name): + from apscheduler.triggers.cron.fields import MAX_VALUES, MIN_VALUES + + value_range = MAX_VALUES[field_name] - MIN_VALUES[field_name] + if self.step and self.step > value_range: + raise ValueError( + f"the step value ({self.step}) is higher than the total range of the " + f"expression ({value_range})" + ) + + def get_next_value(self, date, field): + start = field.get_value(date) + minval = field.get_min(date) + maxval = field.get_max(date) + start = max(start, minval) + + if not self.step: + next = start + else: + distance_to_next = (self.step - (start - minval)) % self.step + next = start + distance_to_next + + if next <= maxval: + return next + + def __eq__(self, other): + return isinstance(other, self.__class__) and self.step == other.step + + def __str__(self): + if self.step: + return "*/%d" % self.step + return "*" + + def __repr__(self): + return f"{self.__class__.__name__}({self.step})" + + +class RangeExpression(AllExpression): + value_re = re.compile(r"(?P<first>\d+)(?:-(?P<last>\d+))?(?:/(?P<step>\d+))?$") + + def __init__(self, first, last=None, step=None): + super().__init__(step) + first = asint(first) + last = asint(last) + if last is None and step is None: + last = first + if last is not None and first > last: + raise ValueError( + "The minimum value in a range must not be higher than the maximum" + ) + self.first = first + self.last = last + + def validate_range(self, field_name): + from apscheduler.triggers.cron.fields import MAX_VALUES, MIN_VALUES + + super().validate_range(field_name) + if self.first < MIN_VALUES[field_name]: + raise ValueError( + f"the first value ({self.first}) is lower than the minimum value ({MIN_VALUES[field_name]})" + ) + if self.last is not None and self.last > MAX_VALUES[field_name]: + raise ValueError( + f"the last value ({self.last}) is higher than the maximum value ({MAX_VALUES[field_name]})" + ) + value_range = (self.last or MAX_VALUES[field_name]) - self.first + if self.step and self.step > value_range: + raise ValueError( + f"the step value ({self.step}) is higher than the total range of the " + f"expression ({value_range})" + ) + + def get_next_value(self, date, field): + startval = field.get_value(date) + minval = field.get_min(date) + maxval = field.get_max(date) + + # Apply range limits + minval = max(minval, self.first) + maxval = min(maxval, self.last) if self.last is not None else maxval + nextval = max(minval, startval) + + # Apply the step if defined + if self.step: + distance_to_next = (self.step - (nextval - minval)) % self.step + nextval += distance_to_next + + return nextval if nextval <= maxval else None + + def __eq__(self, other): + return ( + isinstance(other, self.__class__) + and self.first == other.first + and self.last == other.last + ) + + def __str__(self): + if self.last != self.first and self.last is not None: + range = "%d-%d" % (self.first, self.last) + else: + range = str(self.first) + + if self.step: + return "%s/%d" % (range, self.step) + + return range + + def __repr__(self): + args = [str(self.first)] + if self.last != self.first and self.last is not None or self.step: + args.append(str(self.last)) + + if self.step: + args.append(str(self.step)) + + return "{}({})".format(self.__class__.__name__, ", ".join(args)) + + +class MonthRangeExpression(RangeExpression): + value_re = re.compile(r"(?P<first>[a-z]+)(?:-(?P<last>[a-z]+))?", re.IGNORECASE) + + def __init__(self, first, last=None): + try: + first_num = MONTHS.index(first.lower()) + 1 + except ValueError: + raise ValueError(f'Invalid month name "{first}"') + + if last: + try: + last_num = MONTHS.index(last.lower()) + 1 + except ValueError: + raise ValueError(f'Invalid month name "{last}"') + else: + last_num = None + + super().__init__(first_num, last_num) + + def __str__(self): + if self.last != self.first and self.last is not None: + return f"{MONTHS[self.first - 1]}-{MONTHS[self.last - 1]}" + return MONTHS[self.first - 1] + + def __repr__(self): + args = [f"'{MONTHS[self.first]}'"] + if self.last != self.first and self.last is not None: + args.append(f"'{MONTHS[self.last - 1]}'") + return "{}({})".format(self.__class__.__name__, ", ".join(args)) + + +class WeekdayRangeExpression(RangeExpression): + value_re = re.compile(r"(?P<first>[a-z]+)(?:-(?P<last>[a-z]+))?", re.IGNORECASE) + + def __init__(self, first, last=None): + try: + first_num = WEEKDAYS.index(first.lower()) + except ValueError: + raise ValueError(f'Invalid weekday name "{first}"') + + if last: + try: + last_num = WEEKDAYS.index(last.lower()) + except ValueError: + raise ValueError(f'Invalid weekday name "{last}"') + else: + last_num = None + + super().__init__(first_num, last_num) + + def __str__(self): + if self.last != self.first and self.last is not None: + return f"{WEEKDAYS[self.first]}-{WEEKDAYS[self.last]}" + return WEEKDAYS[self.first] + + def __repr__(self): + args = [f"'{WEEKDAYS[self.first]}'"] + if self.last != self.first and self.last is not None: + args.append(f"'{WEEKDAYS[self.last]}'") + return "{}({})".format(self.__class__.__name__, ", ".join(args)) + + +class WeekdayPositionExpression(AllExpression): + options = ["1st", "2nd", "3rd", "4th", "5th", "last"] + value_re = re.compile( + r"(?P<option_name>{}) +(?P<weekday_name>(?:\d+|\w+))".format("|".join(options)), + re.IGNORECASE, + ) + + def __init__(self, option_name, weekday_name): + super().__init__(None) + try: + self.option_num = self.options.index(option_name.lower()) + except ValueError: + raise ValueError(f'Invalid weekday position "{option_name}"') + + try: + self.weekday = WEEKDAYS.index(weekday_name.lower()) + except ValueError: + raise ValueError(f'Invalid weekday name "{weekday_name}"') + + def get_next_value(self, date, field): + # Figure out the weekday of the month's first day and the number of days in that month + first_day_wday, last_day = monthrange(date.year, date.month) + + # Calculate which day of the month is the first of the target weekdays + first_hit_day = self.weekday - first_day_wday + 1 + if first_hit_day <= 0: + first_hit_day += 7 + + # Calculate what day of the month the target weekday would be + if self.option_num < 5: + target_day = first_hit_day + self.option_num * 7 + else: + target_day = first_hit_day + ((last_day - first_hit_day) // 7) * 7 + + if target_day <= last_day and target_day >= date.day: + return target_day + + def __eq__(self, other): + return ( + super().__eq__(other) + and self.option_num == other.option_num + and self.weekday == other.weekday + ) + + def __str__(self): + return f"{self.options[self.option_num]} {WEEKDAYS[self.weekday]}" + + def __repr__(self): + return f"{self.__class__.__name__}('{self.options[self.option_num]}', '{WEEKDAYS[self.weekday]}')" + + +class LastDayOfMonthExpression(AllExpression): + value_re = re.compile(r"last", re.IGNORECASE) + + def __init__(self): + super().__init__(None) + + def get_next_value(self, date, field): + return monthrange(date.year, date.month)[1] + + def __str__(self): + return "last" + + def __repr__(self): + return f"{self.__class__.__name__}()" |