Cron Expression Parser (python, written by Codex)

Codex wrote this python project from the task below. It installed and ran on a clean Ubuntu 22.04 machine as written.

Task given to the agent:

TASK: Cron Expression Parser

Write a program that parses, validates, and explains cron expressions, computing the next N execution times and generating human-readable descriptions of the schedule.

FUNCTIONAL REQUIREMENTS:
- Accept a cron expression as a command-line argument (standard 5-field format: minute, hour, day-of-month, month, day-of-week)
- Support extended 6-field format with seconds as the first field via --with-seconds flag
- Support special characters: * (all), , (list), - (range), / (step), L (last), W (nearest weekday), # (nth day of week), and ? (any)
- Validate the expression and report clear errors for invalid fields with the specific position and reason (e.g., "Field 2 (hour): value 25 exceeds maximum 23")
- Compute and display the next N execution times via --next flag (default: 10), formatted in ISO 8601 and local human-readable format, relative to current time or a specified start time via --from flag
- Generate a human-readable English description of the schedule (e.g., "At 03:30 AM, every Monday and Wednesday, in January through March")
- Support time zone handling via --timezone flag (default: UTC)
- Compute the interval statistics: average gap between executions, minimum gap, maximum gap, total executions per day/week/month/year
- Support checking if a specific datetime matches the cron expression via --check flag
- Support converting between cron expression and a verbal description via --describe flag (cron to English) and --parse flag (English-like input to cron expression for common patterns)
- Print results to console in a formatted display
- Save results as JSON with --output flag (default: cron_analysis.json)
- Support batch validation of multiple expressions from a file via --batch flag
- If no expression is given, demonstrate with several example expressions: "*/5 * * * *" (every 5 minutes), "0 9 * * 1-5" (weekdays at 9am), "0 0 1,15 * *" (1st and 15th at midnight), "30 4 * * SUN" (Sundays at 4:30am), showing next 5 runs and descriptions for each
- Handle errors: invalid syntax, out-of-range values, impossible combinations (e.g., Feb 31), and malformed special characters

Create a complete Python project for a clean Ubuntu 22.04 machine with only Python 3.10+ installed. Include:
- Source code
- requirements.txt with all dependencies (direct and transitive) pinned to exact versions
- README.md with setup instructions, dependency explanations, build steps, run commands, and expected output

[]

# Cron Expression Parser (Python)

## Requirements
- Python 3.10+

## Install
```bash
python -m pip install -r requirements.txt
```

## Run
```bash
python src/main.py "*/5 * * * *" --count 10
```

Demo mode:
```bash
python src/main.py --demo
```

## Dependencies
- `croniter==2.0.1`
- `python-dateutil==2.8.2`

croniter==2.0.1
python-dateutil==2.8.2

#!/usr/bin/env python3
"""Cron Expression Parser - Trial 1 (croniter + python-dateutil)

Parses and validates cron expressions, computes next execution times,
and generates human-readable descriptions.
"""

import argparse
import json
import sys
from datetime import datetime, timedelta

from croniter import croniter
from dateutil import parser as dateutil_parser
from dateutil.relativedelta import relativedelta
from dateutil.tz import tzlocal, tzutc


# ---------------------------------------------------------------------------
# Cron field descriptors for human-readable output
# ---------------------------------------------------------------------------
MONTH_NAMES = [
    "", "January", "February", "March", "April", "May", "June",
    "July", "August", "September", "October", "November", "December"
]

DAY_NAMES = [
    "Sunday", "Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday"
]

FIELD_NAMES = ["minute", "hour", "day of month", "month", "day of week"]


def validate_cron(expression: str) -> dict:
    """Validate a cron expression and return details about its validity."""
    expression = expression.strip()
    result = {
        "expression": expression,
        "valid": False,
        "fields": {},
        "error": None,
    }

    parts = expression.split()
    if len(parts) not in (5, 6, 7):
        result["error"] = (
            f"Expected 5 fields (standard cron), got {len(parts)}. "
            "Format: minute hour day-of-month month day-of-week"
        )
        return result

    # Use only first 5 fields for standard cron
    cron_expr = " ".join(parts[:5])

    try:
        croniter(cron_expr)
        result["valid"] = True
    except (ValueError, KeyError) as exc:
        result["error"] = str(exc)
        return result

    for i, name in enumerate(FIELD_NAMES):
        result["fields"][name] = parts[i] if i < len(parts) else "*"

    return result


def describe_field(value: str, field_name: str) -> str:
    """Generate a human-readable description for a single cron field."""
    if value == "*":
        return f"every {field_name}"

    if "/" in value:
        base, step = value.split("/", 1)
        if base == "*":
            return f"every {step} {field_name}(s)"
        return f"every {step} {field_name}(s) starting at {base}"

    if "-" in value and "," not in value:
        start, end = value.split("-", 1)
        if field_name == "month":
            try:
                start = MONTH_NAMES[int(start)]
                end = MONTH_NAMES[int(end)]
            except (ValueError, IndexError):
                pass
        elif field_name == "day of week":
            try:
                start = DAY_NAMES[int(start)]
                end = DAY_NAMES[int(end)]
            except (ValueError, IndexError):
                pass
        return f"{field_name} {start} through {end}"

    if "," in value:
        items = value.split(",")
        if field_name == "month":
            items = [MONTH_NAMES[int(i)] if i.isdigit() and 1 <= int(i) <= 12 else i for i in items]
        elif field_name == "day of week":
            items = [DAY_NAMES[int(i)] if i.isdigit() and 0 <= int(i) <= 6 else i for i in items]
        return f"{field_name} {', '.join(items)}"

    # Single value
    display = value
    if field_name == "month" and value.isdigit():
        idx = int(value)
        if 1 <= idx <= 12:
            display = MONTH_NAMES[idx]
    elif field_name == "day of week" and value.isdigit():
        idx = int(value)
        if 0 <= idx <= 6:
            display = DAY_NAMES[idx]

    return f"at {field_name} {display}"


def describe_cron(expression: str) -> str:
    """Generate a full human-readable description of a cron expression."""
    parts = expression.strip().split()
    if len(parts) < 5:
        return "Invalid cron expression"

    descriptions = []
    for i, name in enumerate(FIELD_NAMES):
        value = parts[i]
        if value != "*":
            descriptions.append(describe_field(value, name))

    if not descriptions:
        return "Every minute"

    # Build natural sentence
    minute_val = parts[0]
    hour_val = parts[1]

    if minute_val != "*" and hour_val != "*" and "/" not in minute_val and "/" not in hour_val:
        try:
            m = int(minute_val) if minute_val.isdigit() else None
            h = int(hour_val) if hour_val.isdigit() else None
            if m is not None and h is not None:
                time_str = f"At {h:02d}:{m:02d}"
                remaining = []
                for i in range(2, 5):
                    if parts[i] != "*":
                        remaining.append(describe_field(parts[i], FIELD_NAMES[i]))
                if remaining:
                    return f"{time_str}, {', '.join(remaining)}"
                return time_str
        except ValueError:
            pass

    return ", ".join(descriptions)


def next_executions(expression: str, count: int = 5, base_time: datetime = None) -> list:
    """Compute the next N execution times for a cron expression."""
    if base_time is None:
        base_time = datetime.now()

    parts = expression.strip().split()
    cron_expr = " ".join(parts[:5])

    try:
        cron = croniter(cron_expr, base_time)
    except (ValueError, KeyError) as exc:
        return [{"error": str(exc)}]

    results = []
    for _ in range(count):
        next_dt = cron.get_next(datetime)
        delta = relativedelta(next_dt, base_time)
        results.append({
            "datetime": next_dt.isoformat(),
            "relative": _format_relative(next_dt, base_time),
        })

    return results


def _format_relative(target: datetime, base: datetime) -> str:
    """Format the time difference between target and base as human-readable."""
    diff = target - base
    total_seconds = int(diff.total_seconds())

    if total_seconds < 60:
        return f"in {total_seconds} second(s)"
    elif total_seconds < 3600:
        minutes = total_seconds // 60
        return f"in {minutes} minute(s)"
    elif total_seconds < 86400:
        hours = total_seconds // 3600
        minutes = (total_seconds % 3600) // 60
        if minutes:
            return f"in {hours} hour(s) and {minutes} minute(s)"
        return f"in {hours} hour(s)"
    else:
        days = total_seconds // 86400
        hours = (total_seconds % 86400) // 3600
        if hours:
            return f"in {days} day(s) and {hours} hour(s)"
        return f"in {days} day(s)"


def time_window_executions(expression: str, start: str, end: str) -> list:
    """Find all execution times within a given time window."""
    start_dt = dateutil_parser.parse(start)
    end_dt = dateutil_parser.parse(end)

    parts = expression.strip().split()
    cron_expr = " ".join(parts[:5])

    try:
        cron = croniter(cron_expr, start_dt - timedelta(seconds=1))
    except (ValueError, KeyError) as exc:
        return [{"error": str(exc)}]

    results = []
    while True:
        next_dt = cron.get_next(datetime)
        if next_dt > end_dt:
            break
        results.append(next_dt.isoformat())
        if len(results) > 10000:
            break

    return results


def analyze_cron(expression: str, count: int = 5, base_time: str = None,
                 window_start: str = None, window_end: str = None) -> dict:
    """Full analysis of a cron expression."""
    base_dt = None
    if base_time:
        base_dt = dateutil_parser.parse(base_time)

    validation = validate_cron(expression)

    result = {
        "expression": expression,
        "validation": validation,
    }

    if not validation["valid"]:
        return result

    result["description"] = describe_cron(expression)
    result["next_executions"] = next_executions(expression, count, base_dt)

    if window_start and window_end:
        window_times = time_window_executions(expression, window_start, window_end)
        result["window"] = {
            "start": window_start,
            "end": window_end,
            "count": len(window_times),
            "executions": window_times[:100],
        }

    return result


# ---------------------------------------------------------------------------
# Console output
# ---------------------------------------------------------------------------
def print_analysis(analysis: dict) -> None:
    """Pretty-print analysis results."""
    sep = "=" * 70
    print(f"\n{sep}")
    print("  CRON EXPRESSION PARSER - ANALYSIS REPORT")
    print(sep)
    print(f"  Expression: {analysis['expression']}")

    v = analysis["validation"]
    print(f"  Valid:      {v['valid']}")
    if v.get("error"):
        print(f"  Error:      {v['error']}")
        print(sep)
        return

    if v.get("fields"):
        print("\n  -- Fields --")
        for name, value in v["fields"].items():
            print(f"    {name:<15} {value}")

    if "description" in analysis:
        print(f"\n  Description: {analysis['description']}")

    if "next_executions" in analysis:
        print("\n  -- Next Executions --")
        for i, ex in enumerate(analysis["next_executions"], 1):
            if "error" in ex:
                print(f"    Error: {ex['error']}")
                break
            print(f"    [{i}] {ex['datetime']}  ({ex['relative']})")

    if "window" in analysis:
        w = analysis["window"]
        print(f"\n  -- Time Window --")
        print(f"    From:  {w['start']}")
        print(f"    To:    {w['end']}")
        print(f"    Count: {w['count']} executions")
        for ts in w["executions"][:10]:
            print(f"      {ts}")
        if w["count"] > 10:
            print(f"      ... and {w['count'] - 10} more")

    print(sep)


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def main() -> None:
    parser = argparse.ArgumentParser(
        description="Cron Expression Parser (croniter + python-dateutil)"
    )
    parser.add_argument("expression", nargs="?", default=None,
                        help="Cron expression to parse (e.g., '*/5 * * * *')")
    parser.add_argument("-n", "--count", type=int, default=5,
                        help="Number of next executions to show (default: 5)")
    parser.add_argument("-b", "--base-time", default=None,
                        help="Base time for calculations (ISO format)")
    parser.add_argument("--window-start", default=None,
                        help="Window start time (ISO format)")
    parser.add_argument("--window-end", default=None,
                        help="Window end time (ISO format)")
    parser.add_argument("-o", "--output", default=None,
                        help="Output JSON file path")
    parser.add_argument("--demo", action="store_true",
                        help="Run demo with sample expressions")
    args = parser.parse_args()

    if args.demo or args.expression is None:
        demo_expressions = [
            "*/5 * * * *",
            "0 9 * * 1-5",
            "0 0 1 * *",
            "30 */2 * * *",
            "0 0 * * 0",
            "15 14 1 * *",
            "0 22 * * 1-5",
            "*/15 9-17 * * 1-5",
            "0 0 1,15 * *",
            "invalid cron",
        ]
        print("Running demo with sample cron expressions...\n")
        all_results = []
        for expr in demo_expressions:
            result = analyze_cron(expr, count=3)
            print_analysis(result)
            all_results.append(result)

        if args.output:
            with open(args.output, "w", encoding="utf-8") as fh:
                json.dump(all_results, fh, indent=2, default=str)
            print(f"\nJSON report written to {args.output}")
        return

    result = analyze_cron(
        args.expression,
        count=args.count,
        base_time=args.base_time,
        window_start=args.window_start,
        window_end=args.window_end,
    )
    print_analysis(result)

    if args.output:
        with open(args.output, "w", encoding="utf-8") as fh:
            json.dump(result, fh, indent=2, default=str)
        print(f"\nJSON report written to {args.output}")


if __name__ == "__main__":
    main()