beam: cron expression parser with next-run + tests

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))

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)