"""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\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\d+)(?:-(?P\d+))?(?:/(?P\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[a-z]+)(?:-(?P[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[a-z]+)(?:-(?P[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{}) +(?P(?:\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__}()"