crontool.py

#!/usr/bin/env python3
"""
crontool -- Cron expression Swiss knife. Explain, next runs, templates. Zero deps.

Decode cryptic cron expressions, preview execution times, find templates.

Usage:
    py crontool.py explain "*/5 * * * *"            # "Every 5 minutes"
    py crontool.py explain "0 9 * * 1-5"            # "At 09:00, Monday through Friday"
    py crontool.py next "0 9 * * 1-5"               # Next 5 execution times
    py crontool.py next "*/15 * * * *" 10            # Next 10 runs
    py crontool.py validate "0 25 * * *"             # Check syntax
    py crontool.py templates                         # Common cron patterns
    py crontool.py info "30 2 15 */3 *"              # Full breakdown
"""

import argparse
import os
import re
import sys
from datetime import datetime, timedelta

# Colors
RESET = "\033[0m"
BOLD = "\033[1m"
DIM = "\033[2m"
GREEN = "\033[32m"
YELLOW = "\033[33m"
CYAN = "\033[36m"
RED = "\033[31m"


def color_supported() -> bool:
    if os.environ.get("NO_COLOR"):
        return False
    if sys.platform == "win32":
        return bool(os.environ.get("TERM") or os.environ.get("WT_SESSION"))
    return hasattr(sys.stdout, "isatty") and sys.stdout.isatty()


USE_COLOR = color_supported()


def c(code: str, text: str) -> str:
    return f"{code}{text}{RESET}" if USE_COLOR else text


# --- Field parsing ---

FIELD_NAMES = ["minute", "hour", "day of month", "month", "day of week"]
FIELD_RANGES = [(0, 59), (0, 23), (1, 31), (1, 12), (0, 7)]  # 0 and 7 = Sunday

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,
}
DOW_LABELS = ["Sun", "Mon", "Tue", "Wed", "Thu", "Fri", "Sat"]
MONTH_LABELS = ["", "Jan", "Feb", "Mar", "Apr", "May", "Jun",
                "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"]


def parse_field(field: str, idx: int) -> list[int]:
    """Parse a single cron field into a list of valid values."""
    lo, hi = FIELD_RANGES[idx]
    values = set()

    # Replace names
    field_lower = field.lower()
    if idx == 3:  # month
        for name, num in MONTH_NAMES.items():
            field_lower = field_lower.replace(name, str(num))
    elif idx == 4:  # dow
        for name, num in DOW_NAMES.items():
            field_lower = field_lower.replace(name, str(num))
    field = field_lower

    for part in field.split(","):
        part = part.strip()
        # */n or n
        m = re.match(r"^\*/(\d+)$", part)
        if m:
            step = int(m.group(1))
            values.update(range(lo, hi + 1, step))
            continue

        # n-m/s
        m = re.match(r"^(\d+)-(\d+)/(\d+)$", part)
        if m:
            start, end, step = int(m.group(1)), int(m.group(2)), int(m.group(3))
            values.update(range(start, end + 1, step))
            continue

        # n-m
        m = re.match(r"^(\d+)-(\d+)$", part)
        if m:
            start, end = int(m.group(1)), int(m.group(2))
            values.update(range(start, end + 1))
            continue

        # *
        if part == "*":
            values.update(range(lo, hi + 1))
            continue

        # plain number
        if part.isdigit():
            val = int(part)
            # Normalize Sunday: 7 -> 0
            if idx == 4 and val == 7:
                val = 0
            values.add(val)
            continue

        raise ValueError(f"Invalid token '{part}' in field {FIELD_NAMES[idx]}")

    return sorted(values)


def parse_cron(expr: str) -> list[list[int]]:
    """Parse full 5-field cron expression."""
    parts = expr.strip().split()
    if len(parts) != 5:
        raise ValueError(f"Expected 5 fields, got {len(parts)}: {expr}")
    return [parse_field(parts[i], i) for i in range(5)]


def validate_cron(expr: str) -> tuple[bool, str]:
    """Validate cron expression. Returns (valid, message)."""
    try:
        fields = parse_cron(expr)
        for i, vals in enumerate(fields):
            lo, hi = FIELD_RANGES[i]
            for v in vals:
                if v < lo or v > hi:
                    return False, f"Value {v} out of range ({lo}-{hi}) for {FIELD_NAMES[i]}"
            if not vals:
                return False, f"Empty value set for {FIELD_NAMES[i]}"
        return True, "Valid"
    except ValueError as e:
        return False, str(e)


# --- Explanation ---

def explain_field(values: list[int], idx: int, raw: str) -> str:
    """Generate human-readable description for one field."""
    lo, hi = FIELD_RANGES[idx]
    all_vals = list(range(lo, hi + 1))

    if raw.strip() == "*":
        return "every"

    if values == all_vals:
        return "every"

    # Check for step pattern */n
    m = re.match(r"^\*/(\d+)$", raw.strip())
    if m:
        n = int(m.group(1))
        return f"every {n}"

    # Range
    if len(values) >= 2 and values == list(range(values[0], values[-1] + 1)):
        if idx == 4:
            return f"{DOW_LABELS[values[0]]} through {DOW_LABELS[values[-1]]}"
        if idx == 3:
            return f"{MONTH_LABELS[values[0]]} through {MONTH_LABELS[values[-1]]}"
        return f"{values[0]} through {values[-1]}"

    # List
    if idx == 4:
        return ", ".join(DOW_LABELS[v] for v in values)
    if idx == 3:
        return ", ".join(MONTH_LABELS[v] for v in values)
    return ", ".join(str(v) for v in values)


def explain_cron(expr: str) -> str:
    """Generate full human-readable explanation."""
    parts = expr.strip().split()
    if len(parts) != 5:
        return f"Invalid: expected 5 fields, got {len(parts)}"

    fields = parse_cron(expr)
    descs = [explain_field(fields[i], i, parts[i]) for i in range(5)]
    minutes, hours, dom, month, dow = descs
    raw_parts = parts

    segments = []

    # Time
    if raw_parts[0] == "*" and raw_parts[1] == "*":
        segments.append("Every minute")
    elif raw_parts[0].startswith("*/"):
        n = raw_parts[0].split("/")[1]
        segments.append(f"Every {n} minutes")
    elif raw_parts[1] == "*":
        segments.append(f"At minute {minutes} of every hour")
    elif raw_parts[0] == "0" and not raw_parts[1].startswith("*/"):
        if len(fields[1]) == 1:
            segments.append(f"At {fields[1][0]:02d}:00")
        else:
            segments.append(f"At minute 0 past hour {hours}")
    elif raw_parts[1].startswith("*/"):
        n = raw_parts[1].split("/")[1]
        segments.append(f"At minute {minutes} past every {n} hours")
    else:
        if len(fields[0]) == 1 and len(fields[1]) == 1:
            segments.append(f"At {fields[1][0]:02d}:{fields[0][0]:02d}")
        else:
            segments.append(f"At minute {minutes} past hour {hours}")

    # Day of month
    if raw_parts[2] != "*":
        if len(fields[2]) == 1:
            segments.append(f"on day {fields[2][0]} of the month")
        else:
            segments.append(f"on days {dom} of the month")

    # Month
    if raw_parts[3] != "*":
        if raw_parts[3].startswith("*/"):
            n = raw_parts[3].split("/")[1]
            segments.append(f"every {n} months")
        else:
            segments.append(f"in {month}")

    # Day of week
    if raw_parts[4] != "*":
        segments.append(f"on {dow}")

    return ", ".join(segments)


# --- Next execution ---

def next_runs(expr: str, count: int = 5, after: datetime = None) -> list[datetime]:
    """Calculate next N execution times."""
    fields = parse_cron(expr)
    minutes, hours, doms, months, dows = fields

    if after is None:
        after = datetime.now()

    # Start from next minute
    current = after.replace(second=0, microsecond=0) + timedelta(minutes=1)
    results = []
    max_iterations = 525600 * 2  # 2 years of minutes

    for _ in range(max_iterations):
        if (current.minute in minutes and
            current.hour in hours and
            current.day in doms and
            current.month in months and
            current.weekday() in _convert_dow(dows)):
            results.append(current)
            if len(results) >= count:
                break
        current += timedelta(minutes=1)

    return results


def _convert_dow(cron_dows: list[int]) -> set[int]:
    """Convert cron day-of-week (0=Sun) to Python weekday (0=Mon)."""
    mapping = {0: 6, 1: 0, 2: 1, 3: 2, 4: 3, 5: 4, 6: 5}
    return {mapping.get(d, d) for d in cron_dows}


# --- Templates ---

TEMPLATES = [
    ("* * * * *", "Every minute"),
    ("*/5 * * * *", "Every 5 minutes"),
    ("*/15 * * * *", "Every 15 minutes"),
    ("*/30 * * * *", "Every 30 minutes"),
    ("0 * * * *", "Every hour (at :00)"),
    ("0 */2 * * *", "Every 2 hours"),
    ("0 */6 * * *", "Every 6 hours"),
    ("0 0 * * *", "Daily at midnight"),
    ("0 6 * * *", "Daily at 6:00 AM"),
    ("0 9 * * *", "Daily at 9:00 AM"),
    ("0 12 * * *", "Daily at noon"),
    ("0 9 * * 1-5", "Weekdays at 9:00 AM"),
    ("0 9 * * 1,3,5", "Mon/Wed/Fri at 9:00 AM"),
    ("0 0 * * 0", "Weekly on Sunday at midnight"),
    ("0 0 * * 1", "Weekly on Monday at midnight"),
    ("0 0 1 * *", "Monthly on the 1st at midnight"),
    ("0 0 15 * *", "Monthly on the 15th at midnight"),
    ("0 0 1 1 *", "Yearly on January 1st"),
    ("0 0 1 */3 *", "Every 3 months (quarterly)"),
    ("30 2 * * *", "Daily at 2:30 AM (off-peak)"),
]


# --- Commands ---

def cmd_explain(expr: str):
    desc = explain_cron(expr)
    print(f"\n  {c(CYAN, expr)}")
    print(f"  {c(BOLD, desc)}\n")

    # Also show field breakdown
    parts = expr.strip().split()
    if len(parts) == 5:
        fields = parse_cron(expr)
        print(c(DIM, "  Field breakdown:"))
        for i, (name, vals, raw) in enumerate(zip(FIELD_NAMES, fields, parts)):
            desc = explain_field(vals, i, raw)
            print(f"    {name:<15} {c(YELLOW, raw):<12} -> {desc}")
        print()


def cmd_next(expr: str, count: int):
    valid, msg = validate_cron(expr)
    if not valid:
        print(c(RED, f"  Invalid: {msg}"))
        return

    desc = explain_cron(expr)
    print(f"\n  {c(CYAN, expr)}  ({desc})\n")

    runs = next_runs(expr, count)
    if not runs:
        print(c(RED, "  Could not calculate next runs (expression may never match)"))
        return

    now = datetime.now()
    for i, dt in enumerate(runs, 1):
        delta = dt - now
        days = delta.days
        hours = delta.seconds // 3600
        mins = (delta.seconds % 3600) // 60

        if days > 0:
            relative = f"in {days}d {hours}h"
        elif hours > 0:
            relative = f"in {hours}h {mins}m"
        else:
            relative = f"in {mins}m"

        dow = DOW_LABELS[dt.isoweekday() % 7]
        print(f"  {i:>3}.  {dt.strftime('%Y-%m-%d %H:%M')}  {c(DIM, dow):<5}  {c(DIM, relative)}")
    print()


def cmd_validate(expr: str):
    valid, msg = validate_cron(expr)
    if valid:
        desc = explain_cron(expr)
        print(f"\n  {c(GREEN, 'VALID')}  {expr}")
        print(f"  {desc}\n")
    else:
        print(f"\n  {c(RED, 'INVALID')}  {expr}")
        print(f"  {c(RED, msg)}\n")


def cmd_templates():
    print(f"\n  {c(BOLD, 'Common Cron Templates')}\n")
    for expr, desc in TEMPLATES:
        print(f"  {c(CYAN, expr):<22}  {desc}")
    print(f"\n  {c(DIM, 'Format: minute hour day-of-month month day-of-week')}")
    print(c(DIM, "  Ranges: min(0-59) hour(0-23) dom(1-31) month(1-12) dow(0-6, 0=Sun)"))
    print(c(DIM, "  Special: * (any)  */n (every n)  n-m (range)  n,m (list)"))
    print()


def cmd_info(expr: str):
    """Full info: explain + next runs + field breakdown."""
    cmd_explain(expr)
    valid, msg = validate_cron(expr)
    if valid:
        print(f"  {c(BOLD, 'Next 5 runs:')}")
        runs = next_runs(expr, 5)
        now = datetime.now()
        for i, dt in enumerate(runs, 1):
            delta = dt - now
            days = delta.days
            hours = delta.seconds // 3600
            mins = (delta.seconds % 3600) // 60
            if days > 0:
                relative = f"in {days}d {hours}h"
            elif hours > 0:
                relative = f"in {hours}h {mins}m"
            else:
                relative = f"in {mins}m"
            dow = DOW_LABELS[dt.isoweekday() % 7]
            print(f"    {i}. {dt.strftime('%Y-%m-%d %H:%M')}  {c(DIM, dow)}  {c(DIM, relative)}")
        print()


def main():
    parser = argparse.ArgumentParser(
        description="crontool -- cron expression Swiss knife",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    sub = parser.add_subparsers(dest="command")

    p = sub.add_parser("explain", help="Explain cron expression in English")
    p.add_argument("expr", help="Cron expression (5 fields)")

    p = sub.add_parser("next", help="Show next execution times")
    p.add_argument("expr", help="Cron expression")
    p.add_argument("count", nargs="?", type=int, default=5, help="Number of runs (default: 5)")

    p = sub.add_parser("validate", help="Validate cron expression")
    p.add_argument("expr", help="Cron expression")

    p = sub.add_parser("info", help="Full breakdown + next runs")
    p.add_argument("expr", help="Cron expression")

    sub.add_parser("templates", help="Show common cron templates")

    args = parser.parse_args()

    if args.command == "explain":
        cmd_explain(args.expr)
    elif args.command == "next":
        cmd_next(args.expr, args.count)
    elif args.command == "validate":
        cmd_validate(args.expr)
    elif args.command == "info":
        cmd_info(args.expr)
    elif args.command == "templates":
        cmd_templates()
    else:
        parser.print_help()


if __name__ == "__main__":
    main()