ai_v/venv/Lib/site-packages/apscheduler/triggers/cron/expressions.py

"""This module contains the expressions applicable for CronTrigger's fields."""

__all__ = (
    "AllExpression",
    "LastDayOfMonthExpression",
    "RangeExpression",
    "WeekdayPositionExpression",
    "WeekdayRangeExpression",
)

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 f"*/{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 = f"{self.first}-{self.last}"
        else:
            range = str(self.first)

        if self.step:
            return f"{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__}()"