diff --git a/src/cronparse/parser.py b/src/cronparse/parser.py new file mode 100644 index 0000000..4712897 --- /dev/null +++ b/src/cronparse/parser.py @@ -0,0 +1,91 @@ +"""Cron expression parsing and validation module.""" + +from typing import Dict, Any, Tuple, Optional +from croniter import croniter + + +def validate_cron(expression: str) -> Tuple[bool, Optional[str]]: + """Validate a cron expression. + + Args: + expression: The cron expression to validate. + + Returns: + Tuple of (is_valid, error_message). + """ + try: + is_valid = croniter.is_valid(expression) + if is_valid: + return True, None + else: + return False, "Invalid cron expression format" + except Exception as e: + return False, str(e) + + +def parse_cron(expression: str) -> Tuple[bool, Dict[str, Any]]: + """Parse a cron expression and return detailed breakdown. + + Args: + expression: The cron expression to parse. + + Returns: + Tuple of (is_valid, result_dict). + result_dict contains: minute, hour, day, month, day_of_week, command. + """ + try: + parts = expression.split() + + if len(parts) < 5: + raise ValueError("Invalid cron expression: must have at least 5 fields") + + result = { + "minute": parts[0], + "hour": parts[1], + "day": parts[2], + "month": parts[3], + "day_of_week": parts[4], + } + + if len(parts) > 5: + command = " ".join(parts[5:]) + if command: + result["command"] = command + + return True, result + except Exception as e: + return False, {"error": str(e)} + + +def parse_field_value(field: str, value: str) -> list: + """Parse a single cron field value into a list of values. + + Args: + field: The field name (minute, hour, day, month, dow). + value: The field value string. + + Returns: + List of integer values or special strings. + """ + values = [] + parts = value.split(",") + + for part in parts: + part = part.strip() + if "/" in part: + base, step = part.split("/") + base = base.strip() if base.strip() else "*" + step = int(step) + values.append(f"{base}/step:{step}") + elif "-" in part: + start, end = part.split("-") + values.extend(range(int(start), int(end) + 1)) + elif part == "*": + values.append("*") + else: + try: + values.append(int(part)) + except ValueError: + values.append(part) + + return values