aboutsummaryrefslogtreecommitdiff
path: root/.venv/lib/python3.12/site-packages/apscheduler/triggers/cron/expressions.py
diff options
context:
space:
mode:
authorS. Solomon Darnell2025-03-28 21:52:21 -0500
committerS. Solomon Darnell2025-03-28 21:52:21 -0500
commit4a52a71956a8d46fcb7294ac71734504bb09bcc2 (patch)
treeee3dc5af3b6313e921cd920906356f5d4febc4ed /.venv/lib/python3.12/site-packages/apscheduler/triggers/cron/expressions.py
parentcc961e04ba734dd72309fb548a2f97d67d578813 (diff)
downloadgn-ai-master.tar.gz
two version of R2R are hereHEADmaster
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__}()"