"""Minimal cron expression parser + next-run calculator. Supports standard 5-field cron: minute hour day-of-month month day-of-week. Fields accept: *, */n, a-b, a-b/n, comma lists, and names for month (jan..dec) and day-of-week (sun..sat, 0/7 = sunday). Day-of-month and day-of-week interaction follows Vixie cron: if BOTH are restricted (not '*'), a match occurs when EITHER matches; if only one is restricted, only that one applies. Public API: parse(expr) -> Cron Cron.next(after) -> datetime """ from __future__ import annotations import datetime as dt MONTH_NAMES = { "jan": 1, "feb": 2, "mar": 3, "apr": 4, "may": 5, "jun": 6, "jul": 7, "aug": 8, "sep": 9, "oct": 10, "nov": 11, "dec": 12, } DOW_NAMES = { "sun": 0, "mon": 1, "tue": 2, "wed": 3, "thu": 4, "fri": 5, "sat": 6, } RANGES = { "minute": (0, 59), "hour": (0, 23), "dom": (1, 31), "month": (1, 12), "dow": (0, 6), } def _name_to_int(tok, names): t = tok.lower()[:3] if t in names: return names[t] return None def _expand_field(field, lo, hi, names=None): """Return set of ints in [lo,hi] that the field matches.""" out = set() for part in field.split(","): part = part.strip() if not part: raise ValueError("empty field part") # step step = 1 rng = part if "/" in part: rng, step_s = part.split("/", 1) step = int(step_s) if step < 1: raise ValueError("step must be >=1") # wildcard or range if rng == "*": start, end = lo, hi elif "-" in rng: a, b = rng.split("-", 1) # names va = _name_to_int(a, names) if names else None if va is None: va = int(a) vb = _name_to_int(b, names) if names else None if vb is None: vb = int(b) start, end = va, vb else: v = _name_to_int(rng, names) if names else None if v is None: v = int(rng) start = end = v if not (lo <= start <= hi and lo <= end <= hi): # allow dow 7 as Sunday if not (names is DOW_NAMES and start in (0,7) and end in (0,7)): raise ValueError(f"value out of range: {part}") # dow 7 -> 0 if names is DOW_NAMES: if start == 7: start = 0 if end == 7: end = 0 # handle wrap (e.g. 5-1 dow? not standard; skip wrap except dow 7->0 already) if start <= end: r = range(start, end + 1, step) else: # wrap-around range (only meaningful for dow); build both halves r = list(range(start, hi + 1, step)) + list(range(lo, end + 1, step)) for v in r: out.add(v) # normalize dow 7 if names is DOW_NAMES and 7 in out: out.discard(7) out.add(0) return out class Cron: def __init__(self, expr): self.expr = expr parts = expr.split() if len(parts) != 5: raise ValueError(f"expected 5 fields, got {len(parts)}") mn, hr, dom, mo, dow = parts self.minute = _expand_field(mn, *RANGES["minute"], None) self.hour = _expand_field(hr, *RANGES["hour"], None) self.dom = _expand_field(dom, *RANGES["dom"], None) self.month = _expand_field(mo, *RANGES["month"], MONTH_NAMES) self.dow = _expand_field(dow, *RANGES["dow"], DOW_NAMES) # track restricted flags self.dom_restricted = (dom.strip() != "*") self.dow_restricted = (dow.strip() != "*") def matches(self, t): if t.minute not in self.minute: return False if t.hour not in self.hour: return False if t.month not in self.month: return False dom_ok = t.day in self.dom # python weekday: monday=0; convert to cron dow (sunday=0) cron_dow = (t.weekday() + 1) % 7 dow_ok = cron_dow in self.dow if self.dom_restricted and self.dow_restricted: if not (dom_ok or dow_ok): return False elif self.dom_restricted: if not dom_ok: return False elif self.dow_restricted: if not dow_ok: return False # dom validity: if day > month's days, cron can't match anyway; month days handled by candidate iteration return True def next(self, after): # after: datetime; return next run strictly after 'after' cur = after.replace(second=0, microsecond=0) + dt.timedelta(minutes=1) cap = after + dt.timedelta(days=365 * 5) while cur <= cap: if self.matches(cur): return cur cur += dt.timedelta(minutes=1) raise RuntimeError("no matching time within 5 years") def parse(expr): return Cron(expr) if __name__ == "__main__": import sys c = parse(" ".join(sys.argv[1:6]) if len(sys.argv) > 6 else " ".join(sys.argv[1:])) now = dt.datetime.now() print(c.next(now))