"""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)) # (low, high, names) per field; the day of the week also takes 7 for Sunday. _BOUNDS = [(0, 59, None), (0, 23, None), (1, 31, None), (1, 12, _MONTHS), (0, 6, _DAYS)] # Words people put around a pasted line. Anything else in the text leaves it to the model. _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]") # L, W, #, ? and H, once the names are taken out 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 # five plain numbers ("1 2 3 4 5") are more likely a list than a cron line if "-" in fields or any(f[:3] in _MONTHS or f[:3] in _DAYS for f in fields[:3]): return None # a spaced dash, or a name where cron takes a number ("9 - 5 mon - fri", "mon tue wed thu 9-5") 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, [] # a date that never comes ("0 0 30 2 *") return "schedule", [" ".join(fields)]