about summary refs log tree commit diff
path: root/.venv/lib/python3.12/site-packages/apscheduler/triggers/cron/expressions.py
diff options
context:
space:
mode:
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.py285
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__}()"