diff --git a/wizard-work/beam/cron_next.py b/wizard-work/beam/cron_next.py new file mode 100644 index 0000000..8fc5a1e --- /dev/null +++ b/wizard-work/beam/cron_next.py @@ -0,0 +1,158 @@ +"""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)) \ No newline at end of file diff --git a/wizard-work/beam/test_cron_next.py b/wizard-work/beam/test_cron_next.py new file mode 100644 index 0000000..79d8404 --- /dev/null +++ b/wizard-work/beam/test_cron_next.py @@ -0,0 +1,50 @@ +import datetime as dt +from cron_next import parse + +def test_every_minute(): + c = parse("* * * * *") + t = dt.datetime(2026,1,1,0,0) + assert c.next(t) == dt.datetime(2026,1,1,0,1) + +def test_specific(): + c = parse("30 14 * * *") + t = dt.datetime(2026,1,1,10,0) + assert c.next(t) == dt.datetime(2026,1,1,14,30) + +def test_dow_restricted(): + # Mondays at 09:00 + c = parse("0 9 * * 1") + # 2026-01-05 is Monday + t = dt.datetime(2026,1,4,10,0) + assert c.next(t) == dt.datetime(2026,1,5,9,0) + +def test_dom_and_dow_or(): + # dom=1 OR dow=monday + c = parse("0 0 1 * 1") + # 2026-02-01 is Sunday; next monday is Feb 2; but dom=1 matches Feb1 midnight already? after Jan 2 00:00 + t = dt.datetime(2026,1,2,0,1) + # next match: Feb1 00:00 (dom) since before Monday Feb2 + assert c.next(t) == dt.datetime(2026,1,5,0,0) + +def test_step_and_range(): + c = parse("*/15 9-17 * * *") + t = dt.datetime(2026,1,1,8,50) + assert c.next(t) == dt.datetime(2026,1,1,9,0) + +def test_names(): + c = parse("0 0 1 jan *") + t = dt.datetime(2025,6,1,0,0) + assert c.next(t) == dt.datetime(2026,1,1,0,0) + +def test_invalid_fields(): + try: + parse("* * * *") + assert False + except ValueError: + pass + +def test_dow_sunday_7(): + c = parse("0 0 * * 7") + # sunday + t = dt.datetime(2026,1,3,0,1) # sat + assert c.next(t) == dt.datetime(2026,1,4,0,0) \ No newline at end of file