"""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.I) _OUTSIDE = re.compile(r"[lw#?h]", re.I) # 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("/") if slash and not (step.isdigit() and int(step) >= 1): return None 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] except ValueError: return None a, b = ends[0], ends[-1] if len(ends) == 2 else (top if slash else ends[0]) if not (low <= a <= b <= top): return None values.update(range(a, b + 1, int(step) if slash else 1)) 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 low.startswith(("@reboot", "@every")): 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 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)] # Hours the model left out of a run that the text never excludes. Each hour is its own # output bit, so a few bits can drop below the threshold: "every 5 minutes" comes out as # "*/5 0-16,19-21,23 * * *". A hole of at most _MAX_GAP hours in a field of _MIN_HOURS or # more is filled, unless the text has a word that leaves hours out; a field of 22 or more # hours becomes "*". On the release's design sets (2026-10-25) this fixed 13 rows and # changed no right answer. A port may skip it. _EXCLUDE = re.compile(r"\b(?:except|exclud\w*|not|skip\w*|but|other than|apart from|avoid\w*|besides|outside|" r"without|n't|no)\b", re.I) _MAX_GAP, _MIN_HOURS = 2, 6 def _fill(hours): """The hour field with its small holes filled, or the field as it was.""" if hours == "*" or "/" in hours: return hours on = sorted(_values(hours, 0, 23, None) or ()) if len(on) < _MIN_HOURS: return hours steps = [b - a for a, b in zip(on, on[1:])] if max(steps) == 1 or max(steps) - 1 > _MAX_GAP or len(set(steps)) == 1: return hours # no hole, a hole too wide, or an even progression cuts = [0] + [i + 1 for i, s in enumerate(steps) if s > 1] + [len(on)] if len(cuts) == 3 and min(b - a for a, b in zip(cuts, cuts[1:])) >= 3 and on[-1] - on[0] < 20: return hours # two windows of 3 hours or more ("8 to 12 and 2 to 6") if on[-1] - on[0] >= 21: return "*" return f"{on[0]}-{on[-1]}" def fill_hours(text, lines): """The cron line with unasked-for hour holes filled (see _fill). Several lines are left as they are: their hours often fit together on purpose ("hourly on weekdays, every 3 hours on weekends").""" if len(lines) != 1 or _EXCLUDE.search(text): return lines f = lines[0].split() f[1] = _fill(f[1]) return [" ".join(f)]