Compute Upcoming Cron Run Times

from __future__ import annotations

from datetime import datetime, timedelta


def parse_field(
    text: str,
    low: int,
    high: int,
    *,
    sunday_alias: bool = False,
) -> set[int]:
    values: set[int] = set()
    for part in text.split(","):
        step = 1
        if "/" in part:
            part, raw_step = part.split("/", 1)
            step = int(raw_step)
        if part == "*":
            start, end = low, high
        elif "-" in part:
            raw_start, raw_end = part.split("-", 1)
            start, end = int(raw_start), int(raw_end)
        else:
            start = end = int(part)
        if not (low <= start <= end <= high) or step <= 0:
            raise ValueError(f"invalid cron field: {text}")
        values.update(range(start, end + 1, step))
    if sunday_alias:
        values = {0 if value == 7 else value for value in values}
    return values


def next_runs(expression: str, start: datetime, count: int = 3) -> list[datetime]:
    fields = expression.split()
    if len(fields) != 5:
        raise ValueError("expected five cron fields")
    minutes = parse_field(fields[0], 0, 59)
    hours = parse_field(fields[1], 0, 23)
    days = parse_field(fields[2], 1, 31)
    months = parse_field(fields[3], 1, 12)
    weekdays = parse_field(fields[4], 0, 7, sunday_alias=True)
    day_restricted = fields[2] != "*"
    weekday_restricted = fields[4] != "*"

    current = start.replace(second=0, microsecond=0) + timedelta(minutes=1)
    deadline = current + timedelta(days=366 * 2)
    results: list[datetime] = []
    while len(results) < count and current <= deadline:
        cron_weekday = (current.weekday() + 1) % 7
        if day_restricted and weekday_restricted:
            day_matches = current.day in days or cron_weekday in weekdays
        else:
            day_matches = current.day in days and cron_weekday in weekdays
        if (
            current.minute in minutes
            and current.hour in hours
            and current.month in months
            and day_matches
        ):
            results.append(current)
        current += timedelta(minutes=1)
    if len(results) < count:
        raise ValueError("no matching run time within two years")
    return results


def main() -> None:
    start = datetime(2026, 10, 21, 8, 50)
    for run in next_runs("*/15 9-17 * * 1-5", start):
        print(run.isoformat(timespec="minutes"))
    first_of_month = next_runs("0 0 1 * *", start, 1)[0]
    print(first_of_month.isoformat(timespec="minutes"))
    assert first_of_month == datetime(2026, 11, 1, 0, 0)


if __name__ == "__main__":
    main()