| """A request that is already a cron line, read exactly before the model; standard library only. |
| |
| "*/5 * * * *", "run 0 9 * * mon-fri" and "@daily" are parsed, not guessed. Seconds fields, |
| L, W, #, ?, H, @reboot, @every and 6 or 7 fields are UNSUPPORTED; a line with a value out |
| of range, or a date that never comes, is INVALID. Any other text goes to the model. |
| |
| A port may skip this file: the model still answers pasted cron, less often right. |
| """ |
|
|
| import re |
|
|
| UNSUPPORTED, INVALID = "UNSUPPORTED", "INVALID" |
|
|
| _MACROS = {"@yearly": "0 0 1 1 *", "@annually": "0 0 1 1 *", "@monthly": "0 0 1 * *", |
| "@weekly": "0 0 * * 0", "@daily": "0 0 * * *", "@midnight": "0 0 * * *", "@hourly": "0 * * * *"} |
| _MONTHS = {n: i for i, n in enumerate("jan feb mar apr may jun jul aug sep oct nov dec".split(), start=1)} |
| _DAYS = {n: i for i, n in enumerate("sun mon tue wed thu fri sat".split())} |
| _LONGEST = dict(enumerate([31, 29, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31], start=1)) |
| _NAMES = "|".join(list(_MONTHS) + list(_DAYS)) |
| |
| _BOUNDS = [(0, 59, None), (0, 23, None), (1, 31, None), (1, 12, _MONTHS), (0, 6, _DAYS)] |
|
|
| |
| _WRAPPER = re.compile(r"^(?:run(?: it)?(?: at)?|cron(?:tab)?(?: line| entry| job)?:?|schedule:?|expression:?)\s+", re.I) |
| _TOKEN = re.compile(rf"^(?:[\d*/,\-?#]+[lw]?|[lw]|h(?:\([\d\-]+\))?(?:/\d+)?|(?:{_NAMES})(?:[\-,/](?:{_NAMES}|\d+))*)$", re.A) |
| _OUTSIDE = re.compile(r"[lw#?h]") |
|
|
|
|
| def _values(text, low, high, names): |
| """The values one field fires on, or None when the field is not valid cron.""" |
| top = 7 if names is _DAYS else high |
| values = set() |
| for item in text.split(","): |
| rng, slash, step = item.partition("/") |
| ends = [low, high] if rng == "*" else rng.split("-", 1) |
| try: |
| ends = [e if isinstance(e, int) else names[e] if names and e in names else int(e) for e in ends] |
| step = int(step) if slash else 1 |
| except ValueError: |
| return None |
| a, b = ends[0], ends[-1] if len(ends) == 2 else (top if slash else ends[0]) |
| if step < 1 or not (low <= a <= b <= top): |
| return None |
| values.update(range(a, b + 1, step)) |
| if names is _DAYS and 7 in values: |
| values.discard(7) |
| values.add(0) |
| return values |
|
|
|
|
| def read_cron(text): |
| """For text that is only a cron line or macro: (kind, lines) with kind "schedule" and |
| the line, or (UNSUPPORTED or INVALID, []). None for any other text.""" |
| t = text.strip().strip("`'\"").strip() |
| t = _WRAPPER.sub("", t).strip("`'\" .") |
| low = t.lower() |
| if re.match(r"@(?:reboot|every)\b", low): |
| return UNSUPPORTED, [] |
| if low in _MACROS: |
| return "schedule", [_MACROS[low]] |
| fields = low.split() |
| if not 5 <= len(fields) <= 7 or not all(_TOKEN.match(f) for f in fields): |
| return None |
| if not any(c in "*/,-" for f in fields for c in f): |
| return None |
| if "-" in fields or any(f[:3] in _MONTHS or f[:3] in _DAYS for f in fields[:3]): |
| return None |
| if len(fields) > 5 or _OUTSIDE.search(re.sub(_NAMES, "", low)): |
| return UNSUPPORTED, [] |
| values = [_values(f, *b) for f, b in zip(fields, _BOUNDS)] |
| if None in values: |
| return INVALID, [] |
| doms, months = values[2], values[3] |
| if fields[4].startswith("*") and not any(d <= _LONGEST[m] for d in doms for m in months): |
| return INVALID, [] |
| return "schedule", [" ".join(fields)] |
|
|
