1#@!#!123s

: / opt / imh-python / lib / python3.9 / site-packages / apscheduler / triggers / cron /
Filename : fields.py
back
"""Fields represent CronTrigger options which map to :class:`~datetime.datetime` fields."""

__all__ = (
    "MIN_VALUES",
    "MAX_VALUES",
    "DEFAULT_VALUES",
    "BaseField",
    "WeekField",
    "DayOfMonthField",
    "DayOfWeekField",
)

import re
from calendar import monthrange

from apscheduler.triggers.cron.expressions import (
    AllExpression,
    LastDayOfMonthExpression,
    MonthRangeExpression,
    RangeExpression,
    WeekdayPositionExpression,
    WeekdayRangeExpression,
)

MIN_VALUES = {
    "year": 1970,
    "month": 1,
    "day": 1,
    "week": 1,
    "day_of_week": 0,
    "hour": 0,
    "minute": 0,
    "second": 0,
}
MAX_VALUES = {
    "year": 9999,
    "month": 12,
    "day": 31,
    "week": 53,
    "day_of_week": 6,
    "hour": 23,
    "minute": 59,
    "second": 59,
}
DEFAULT_VALUES = {
    "year": "*",
    "month": 1,
    "day": 1,
    "week": "*",
    "day_of_week": "*",
    "hour": 0,
    "minute": 0,
    "second": 0,
}
SEPARATOR = re.compile(" *, *")


class BaseField:
    REAL = True
    COMPILERS = [AllExpression, RangeExpression]

    def __init__(self, name, exprs, is_default=False):
        self.name = name
        self.is_default = is_default
        self.compile_expressions(exprs)

    def get_min(self, dateval):
        return MIN_VALUES[self.name]

    def get_max(self, dateval):
        return MAX_VALUES[self.name]

    def get_value(self, dateval):
        return getattr(dateval, self.name)

    def get_next_value(self, dateval):
        smallest = None
        for expr in self.expressions:
            value = expr.get_next_value(dateval, self)
            if smallest is None or (value is not None and value < smallest):
                smallest = value

        return smallest

    def compile_expressions(self, exprs):
        self.expressions = []

        # Split a comma-separated expression list, if any
        for expr in SEPARATOR.split(str(exprs).strip()):
            self.compile_expression(expr)

    def compile_expression(self, expr):
        for compiler in self.COMPILERS:
            match = compiler.value_re.match(expr)
            if match:
                compiled_expr = compiler(**match.groupdict())

                try:
                    compiled_expr.validate_range(self.name)
                except ValueError as e:
                    raise ValueError(
                        f"Error validating expression {expr!r}: {e}"
                    ) from None

                self.expressions.append(compiled_expr)
                return

        raise ValueError(f'Unrecognized expression "{expr}" for field "{self.name}"')

    def __eq__(self, other):
        return (
            isinstance(self, self.__class__) and self.expressions == other.expressions
        )

    def __str__(self):
        expr_strings = (str(e) for e in self.expressions)
        return ",".join(expr_strings)

    def __repr__(self):
        return f"{self.__class__.__name__}('{self.name}', '{self}')"


class WeekField(BaseField):
    REAL = False

    def get_value(self, dateval):
        return dateval.isocalendar()[1]


class DayOfMonthField(BaseField):
    COMPILERS = BaseField.COMPILERS + [
        WeekdayPositionExpression,
        LastDayOfMonthExpression,
    ]

    def get_max(self, dateval):
        return monthrange(dateval.year, dateval.month)[1]


class DayOfWeekField(BaseField):
    REAL = False
    COMPILERS = BaseField.COMPILERS + [WeekdayRangeExpression]

    def get_value(self, dateval):
        return dateval.weekday()


class MonthField(BaseField):
    COMPILERS = BaseField.COMPILERS + [MonthRangeExpression]