Cron Expression Explainer — Python source

"""Pure 5-field cron parser, natural-language explainer, builder, and next-run
calculator — Python polyglot port.

Language: Python
CosmoDev polyglot showcase port of the ``cron-explainer`` tool.
Ported from src/lib/cron-explainer.ts — display source, part of CosmoDev's
polyglot tool pages.

Zero deps (stdlib only). Deterministic. Times are interpreted as UTC so
results are unambiguous and DST-independent (the caller controls the instant).

The public surface mirrors the TypeScript reference: ``explain_cron``,
``build_cron``, ``next_run``.
"""

from __future__ import annotations

import re
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from typing import List, Optional


# ─── Field model ────────────────────────────────────────────────────────────
#
# The five cron fields in positional order, each with its numeric range and
# whether it accepts named tokens (JAN..DEC / SUN..SAT). ``wrap_max`` is True
# only for day-of-week, where 7 is treated as an alias for 0 (Sunday).

@dataclass(frozen=True)
class FieldMeta:
    name: str            # 'minute' | 'hour' | 'day-of-month' | 'month' | 'day-of-week'
    label: str           # same as name; surfaced in error messages
    min: int
    max: int
    named: bool          # accepts JAN..DEC / SUN..SAT tokens
    wrap_max: bool       # max value wraps to min (dow: 7 → 0 / Sunday)


FIELDS: List[FieldMeta] = [
    FieldMeta('minute',       'minute',       0, 59, named=False, wrap_max=False),
    FieldMeta('hour',         'hour',         0, 23, named=False, wrap_max=False),
    FieldMeta('day-of-month', 'day-of-month', 1, 31, named=False, wrap_max=False),
    FieldMeta('month',        'month',        1, 12, named=True,  wrap_max=False),
    FieldMeta('day-of-week',  'day-of-week',  0, 7,  named=True,  wrap_max=True),
]

MONTH_NAMES = [
    'January', 'February', 'March', 'April', 'May', 'June',
    'July', 'August', 'September', 'October', 'November', 'December',
]
DOW_NAMES = ['Sunday', 'Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday']

# Insertion order is preserved (Python 3.7+); mirrors the TS object iteration
# order used during token substitution.
MONTH_TOKENS = {
    '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_TOKENS = {'SUN': 0, 'MON': 1, 'TUE': 2, 'WED': 3, 'THU': 4, 'FRI': 5, 'SAT': 6}

_DIGITS_RE = re.compile(r'\d+')


def _pad2(n: int) -> str:
    return f'{n:02d}'


def _month_name(m: int) -> str:
    return MONTH_NAMES[m - 1]


def _dow_name(d: int) -> str:
    return DOW_NAMES[d % 7]


def _range(lo: int, hi: int) -> List[int]:
    """Inclusive integer range, e.g. _range(1, 5) -> [1, 2, 3, 4, 5]."""
    return list(range(lo, hi + 1))


class CronError(ValueError):
    """Raised for a malformed single field. Carries a human-readable message."""


def _parse_int_strict(s: str, label: str) -> int:
    """Parse a strictly-numeric token (digits only). Rejects named tokens,
    signs, and surrounding garbage so malformed fields surface clearly."""
    t = s.strip()
    if not _DIGITS_RE.fullmatch(t):
        raise CronError(f'{label}: invalid number "{s}"')
    return int(t)


def _normalize(value: str, meta: FieldMeta) -> str:
    """Replace named tokens (JAN..DEC / SUN..SAT) with their numeric values.
    Uses global substring replacement so ranges like 'JUN-AUG' and lists like
    'MON,WED,FRI' normalize in a single pass."""
    v = value.strip().upper()
    if not meta.named:
        return v
    tokens = MONTH_TOKENS if meta.name == 'month' else DOW_TOKENS
    for tok, num in tokens.items():
        v = v.replace(tok, str(num))
    return v


@dataclass
class ParsedField:
    meta: FieldMeta
    raw: str
    values: List[int]
    wildcard: bool


def _expand_field(value: str, meta: FieldMeta) -> ParsedField:
    """Expand one field value into the explicit set of numbers it matches.

    Handles ``*``, ``*/N``, ``A-B``, ``A-B/N``, ``A`` (single), ``A/N`` (A to
    field max), and comma-separated lists of any of these. Returns the deduped,
    sorted values plus a ``wildcard`` flag distinguishing a bare ``*``.
    """
    norm = _normalize(value, meta)
    if norm == '':
        raise CronError(f'{meta.label}: empty field')
    if norm == '*':
        return ParsedField(meta, value, _range(meta.min, meta.max), True)

    out = set()
    for term in norm.split(','):
        if term == '':
            raise CronError(f'{meta.label}: empty list item')
        slash_idx = term.find('/')
        base = term
        step = 1
        if slash_idx != -1:
            base = term[:slash_idx]
            step = _parse_int_strict(term[slash_idx + 1:], meta.label)
            if step <= 0:
                raise CronError(f'{meta.label}: step must be a positive number')

        if base == '*':
            lo = meta.min
            hi = meta.max
        elif '-' in base:
            dash = base.find('-')
            lo = _parse_int_strict(base[:dash], meta.label)
            hi = _parse_int_strict(base[dash + 1:], meta.label)
        else:
            lo = _parse_int_strict(base, meta.label)
            # "A/step" runs from A to the field max; a bare "A" is a single value.
            hi = meta.max if slash_idx != -1 else lo

        if lo > hi:
            raise CronError(f'{meta.label}: range start {lo} is greater than end {hi}')
        if lo < meta.min:
            raise CronError(f'{meta.label}: value {lo} is below minimum {meta.min}')
        if hi > meta.max:
            raise CronError(f'{meta.label}: value {hi} is above maximum {meta.max}')

        v = lo
        while v <= hi:
            out.add(meta.min if (meta.wrap_max and v == meta.max) else v)
            v += step

    return ParsedField(meta, value, sorted(out), False)


def _parse_expr(expr: str):
    """Parse all five fields. Returns (parts, None) or (None, error_message)."""
    tokens = [t for t in expr.strip().split() if t]
    if len(tokens) != 5:
        return None, (
            f'Expected 5 fields (minute hour day-of-month month day-of-week), '
            f'got {len(tokens)}'
        )
    parts: List[ParsedField] = []
    for i in range(5):
        try:
            parts.append(_expand_field(tokens[i], FIELDS[i]))
        except CronError as e:
            return None, str(e)
        except Exception:
            return None, f'Invalid {FIELDS[i].name} field'
    return parts, None


def _is_contiguous(values: List[int]) -> bool:
    """True when a sorted value list is a contiguous run (e.g. [3, 4, 5, 6])."""
    return all(values[i] - values[i - 1] == 1 for i in range(1, len(values)))


def _single_value(n: int, meta: FieldMeta) -> str:
    """Describe a single value in the field's own vocabulary."""
    if meta.name == 'minute':
        return f'minute {n}'
    if meta.name == 'hour':
        return f'hour {n}'
    if meta.name == 'day-of-month':
        return f'day {n} of the month'
    if meta.name == 'month':
        return _month_name(n)
    return _dow_name(n)  # day-of-week


def _describe_field(p: ParsedField) -> str:
    """Describe a parsed field as a human phrase (no leading preposition).

    ``raw`` is consulted to distinguish step syntax (star/N or A-B/N) from
    plain lists, since two different raw forms can expand to the same set.
    """
    meta, raw, values = p.meta, p.raw, p.values
    if p.wildcard:
        return {
            'minute': 'every minute',
            'hour': 'every hour',
            'day-of-month': 'every day of the month',
            'month': 'every month',
            'day-of-week': 'every day of the week',
        }[meta.name]

    # Step syntax is reported as "every N <units>".
    if '/' in raw and len(values) >= 1:
        step = _parse_int_strict(raw[raw.index('/') + 1:], meta.label)
        start = values[0]
        if meta.name == 'day-of-month':
            unit_plural = 'days of the month'
        elif meta.name == 'day-of-week':
            unit_plural = 'days of the week'
        else:
            unit_plural = f'{meta.name}s'
        if start == meta.min:
            return f'every {step} {unit_plural}'
        return f'every {step} {unit_plural} starting at {_single_value(start, meta)}'

    if len(values) == 1:
        return _single_value(values[0], meta)

    if _is_contiguous(values):
        a, b = values[0], values[-1]
        if meta.name == 'month':
            return f'{_month_name(a)} through {_month_name(b)}'
        if meta.name == 'day-of-week':
            return f'{_dow_name(a)} through {_dow_name(b)}'
        unit_plural = 'days' if meta.name == 'day-of-month' else f'{meta.name}s'
        return f'{unit_plural} {a} through {b}'

    # Explicit list of discrete values.
    if meta.name == 'month':
        return ', '.join(_month_name(v) for v in values)
    if meta.name == 'day-of-week':
        return ', '.join(_dow_name(v) for v in values)
    if meta.name == 'minute':
        return f'minutes {", ".join(str(v) for v in values)}'
    if meta.name == 'hour':
        return f'hours {", ".join(str(v) for v in values)}'
    return f'days {", ".join(str(v) for v in values)} of the month'


def _prepend(prefix: str, phrase: str) -> str:
    """Prepend a preposition, but never before a phrase that already leads with
    'every' (e.g. 'every day of the week' reads wrong as 'on every …')."""
    return phrase if phrase.startswith('every') else f'{prefix} {phrase}'


def _time_clause(minute: ParsedField, hour: ParsedField) -> str:
    """Compose the opening time-of-day clause from minute and hour fields."""
    m_all = minute.wildcard
    h_all = hour.wildcard
    m_single = not m_all and len(minute.values) == 1
    h_single = not h_all and len(hour.values) == 1

    if m_all and h_all:
        return 'Every minute'
    if m_all and h_single:
        return f'Every minute of hour {hour.values[0]}'
    if m_single and h_all:
        return f'At minute {minute.values[0]} of every hour'
    if m_single and h_single:
        return f'At {_pad2(hour.values[0])}:{_pad2(minute.values[0])}'

    # Mixed: describe each non-wildcard field, hour first.
    clauses = []
    if not h_all:
        clauses.append(_describe_field(hour))
    if not m_all:
        clauses.append(_describe_field(minute))
    s = ', '.join(clauses)
    return s[0].upper() + s[1:]


def _compose_description(parts: List[ParsedField]) -> str:
    minute, hour, dom, month, dow = parts
    clauses = [_time_clause(minute, hour)]
    if not dom.wildcard:
        clauses.append(_prepend('on', _describe_field(dom)))
    if not month.wildcard:
        clauses.append(_prepend('in', _describe_field(month)))
    if not dow.wildcard:
        clauses.append(_prepend('on', _describe_field(dow)))
    return ', '.join(clauses)


# ─── Public API ─────────────────────────────────────────────────────────────


@dataclass
class CronFieldInfo:
    field: str         # one of the five positional field names
    value: str         # raw field value as written in the expression
    meaning: str       # human-readable description of what this field matches


@dataclass
class CronExplanation:
    valid: bool
    description: str                               # '' when invalid
    fields: List[CronFieldInfo] = field(default_factory=list)
    error: Optional[str] = None                    # present only when valid is False


def explain_cron(expr: str) -> CronExplanation:
    """Parse and explain a 5-field cron expression in plain English.

    >>> explain_cron('30 14 * * *').description
    'At 14:30'
    """
    parts, error = _parse_expr(expr)
    if error is not None:
        return CronExplanation(valid=False, description='', fields=[], error=error)
    fields = [
        CronFieldInfo(field=p.meta.name, value=p.raw, meaning=_describe_field(p))
        for p in parts
    ]
    return CronExplanation(
        valid=True,
        description=_compose_description(parts),
        fields=fields,
    )


@dataclass
class BuildCronOptions:
    minute: Optional[str] = None
    hour: Optional[str] = None
    dom: Optional[str] = None
    month: Optional[str] = None
    dow: Optional[str] = None


def build_cron(opts: Optional[BuildCronOptions] = None) -> str:
    """Assemble a 5-field cron expression from per-field specs.

    Each field defaults to ``*`` when empty/omitted; invalid fields raise
    ``CronError`` so callers cannot build a malformed expression.

    >>> build_cron(BuildCronOptions(minute='30', hour='14'))
    '30 14 * * *'
    """
    opts = opts or BuildCronOptions()
    specs = [
        (FIELDS[0], opts.minute),
        (FIELDS[1], opts.hour),
        (FIELDS[2], opts.dom),
        (FIELDS[3], opts.month),
        (FIELDS[4], opts.dow),
    ]
    out: List[str] = []
    for meta, value in specs:
        v = (value or '').strip()
        if v == '':
            out.append('*')
            continue
        _expand_field(v, meta)  # validates; raises on bad input
        out.append(v)
    return ' '.join(out)


def next_run(expr: str, after: datetime) -> Optional[datetime]:
    """Next time the expression fires, strictly after ``after``, in UTC.

    Implements standard Vixie-cron day matching: when BOTH day-of-month and
    day-of-week are restricted, a match on either suffices (OR); otherwise
    both must match (AND). Returns ``None`` if no firing occurs within ~3
    years. ``after`` is interpreted in UTC; a naive datetime is assumed UTC.
    """
    parts, error = _parse_expr(expr)
    if error is not None:
        return None

    minute, hour, dom, month, dow = parts
    m_set = set(minute.values)
    h_set = set(hour.values)
    dom_set = set(dom.values)
    mon_set = set(month.values)
    dow_set = set(dow.values)
    dom_wild = dom.wildcard
    dow_wild = dow.wildcard

    # Work in UTC; treat naive datetimes as UTC (caller controls the instant).
    if after.tzinfo is None:
        cur = after.replace(tzinfo=timezone.utc)
    else:
        cur = after.astimezone(timezone.utc)

    # Start at the top of the minute following `after`, seconds zeroed.
    cur = cur.replace(second=0, microsecond=0) + timedelta(minutes=1)
    limit = cur.year + 3  # hard stop ~3 years out

    while cur.year < limit:
        if cur.month not in mon_set:
            # setUTCMonth(getUTCMonth()+1, 1) + zero time → next month, day 1.
            if cur.month == 12:
                cur = cur.replace(year=cur.year + 1, month=1, day=1,
                                  hour=0, minute=0, second=0, microsecond=0)
            else:
                cur = cur.replace(month=cur.month + 1, day=1,
                                  hour=0, minute=0, second=0, microsecond=0)
            continue
        dom_ok = cur.day in dom_set
        # Python weekday() is 0=Mon..6=Sun; cron day-of-week is 0=Sun..6=Sat,
        # so rotate by one to align Sunday with 0.
        cron_dow = (cur.weekday() + 1) % 7
        dow_ok = cron_dow in dow_set
        day_ok = (dom_ok and dow_ok) if (dom_wild or dow_wild) else (dom_ok or dow_ok)
        if not day_ok:
            cur = cur.replace(hour=0, minute=0, second=0, microsecond=0) + timedelta(days=1)
            continue
        if cur.hour not in h_set:
            cur = cur.replace(minute=0, second=0, microsecond=0) + timedelta(hours=1)
            continue
        if cur.minute not in m_set:
            cur = cur.replace(second=0, microsecond=0) + timedelta(minutes=1)
            continue
        return cur
    return None