""" cron_schedule.py ================ Minimal cron-expression parser — no external dependency. Supports the classic 5-field syntax:: ┌──────── minute (0-59) │ ┌────── hour (0-23) │ │ ┌──── day of month (1-31) │ │ │ ┌── month (1-12 or jan-dec) │ │ │ │ ┌ day of week (0-7 or sun-sat; 0 and 7 = Sunday) │ │ │ │ │ 0 19 * * 1,4 -> 19:00 every Monday and Thursday Field syntax: ``*``, single values, ranges (``a-b``), steps (``*/n``, ``a-b/n``) and comma lists. Month / weekday names (``jan``, ``mon``, …) are accepted case-insensitively. As in Vixie cron, when *both* day-of-month and day-of-week are restricted the job runs when **either** matches. Times are evaluated against the local system clock (``datetime.now()``) — in Docker set the ``TZ`` environment variable so "19:00" means local time. """ from __future__ import annotations from datetime import datetime, timedelta _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} _DAY_NAMES = {"sun": 0, "mon": 1, "tue": 2, "wed": 3, "thu": 4, "fri": 5, "sat": 6} def _parse_value(token: str, lo: int, hi: int, names: "dict[str, int] | None") -> int: token = token.strip().lower() if names and token in names: return names[token] try: value = int(token) except ValueError: raise ValueError(f"invalid cron value {token!r}") from None if not (lo <= value <= hi): raise ValueError(f"cron value {value} out of range {lo}-{hi}") return value def _parse_field(field: str, lo: int, hi: int, names: "dict[str, int] | None" = None) -> "set[int]": """Parses one cron field into the set of matching integer values.""" result: set[int] = set() for part in field.split(","): part = part.strip() if not part: raise ValueError(f"empty element in cron field {field!r}") step = 1 if "/" in part: part, step_text = part.split("/", 1) try: step = int(step_text) except ValueError: raise ValueError(f"invalid cron step {step_text!r}") from None if step < 1: raise ValueError(f"cron step must be >= 1, got {step}") if part == "*": start, end = lo, hi elif "-" in part: a, b = part.split("-", 1) start = _parse_value(a, lo, hi, names) end = _parse_value(b, lo, hi, names) if end < start: raise ValueError(f"inverted cron range {part!r}") else: start = end = _parse_value(part, lo, hi, names) result.update(range(start, end + 1, step)) return result class CronSchedule: """ Parsed 5-field cron expression with ``next_after()`` evaluation. Usage:: cron = CronSchedule("0 19 * * mon,thu") run_at = cron.next_after(datetime.now()) """ def __init__(self, expression: str): self.expression = expression.strip() fields = self.expression.split() if len(fields) != 5: raise ValueError( f"cron expression needs 5 fields " f"(minute hour dom month dow), got {len(fields)}: " f"{expression!r}") minute, hour, dom, month, dow = fields self._minutes = _parse_field(minute, 0, 59) self._hours = _parse_field(hour, 0, 23) self._dom = _parse_field(dom, 1, 31) self._months = _parse_field(month, 1, 12, _MONTH_NAMES) dow_values = _parse_field(dow, 0, 7, _DAY_NAMES) # 7 is an alias for Sunday (= 0) self._dow = {0 if v == 7 else v for v in dow_values} # Vixie-cron rule: dom/dow are OR-combined when both are restricted. self._dom_restricted = dom != "*" self._dow_restricted = dow != "*" def __repr__(self) -> str: return f"CronSchedule({self.expression!r})" # ------------------------------------------------------------------ def _day_matches(self, day: "datetime.date") -> bool: if day.month not in self._months: return False dom_ok = day.day in self._dom # Python: Monday=0 … Sunday=6 -> cron: Sunday=0 … Saturday=6 dow_ok = ((day.weekday() + 1) % 7) in self._dow if self._dom_restricted and self._dow_restricted: return dom_ok or dow_ok if self._dom_restricted: return dom_ok if self._dow_restricted: return dow_ok return True def next_after(self, dt: datetime) -> datetime: """ Returns the first matching time strictly after ``dt`` (second/microsecond precision is dropped). """ cand = (dt + timedelta(minutes=1)).replace(second=0, microsecond=0) hours = sorted(self._hours) minutes = sorted(self._minutes) # Walk day by day (covers rare dom/month combos like Feb 29). for _ in range(366 * 5): if self._day_matches(cand.date()): for h in hours: if h < cand.hour: continue for m in minutes: if h == cand.hour and m < cand.minute: continue return cand.replace(hour=h, minute=m) cand = (cand + timedelta(days=1)).replace(hour=0, minute=0) raise ValueError( f"cron {self.expression!r}: no occurrence within 5 years")