Add source code files
Some checks failed
CI / test (push) Failing after 15s

This commit is contained in:
2026-02-01 15:06:54 +00:00
parent 27f8a69fcf
commit c33bc1128f

91
src/cronparse/parser.py Normal file
View File

@@ -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