Trixy3/trixy_core/scheduler/cron_parser.py

"""
Cron Expression Parser for Trixy Scheduler System

This module provides comprehensive parsing and validation of standard cron expressions
with support for all standard cron syntax features including ranges, lists, steps,
and special characters.

Standard Cron Format:
    minute hour day_of_month month day_of_week
    
Field Ranges:
    - minute: 0-59
    - hour: 0-23
    - day_of_month: 1-31
    - month: 1-12 (or JAN-DEC)
    - day_of_week: 0-7 (0 and 7 = Sunday, or SUN-SAT)

Special Characters:
    - * : Any value
    - , : List separator (e.g., 1,3,5)
    - - : Range (e.g., 1-5)
    - / : Step values (e.g., */5 or 1-10/2)
    - ? : No specific value (day_of_month or day_of_week only)

Examples:
    - "0 9 * * 1-5" : 9 AM on weekdays
    - "*/15 * * * *" : Every 15 minutes
    - "0 0 1 * *" : First day of every month at midnight
    - "30 14 * * 0" : 2:30 PM every Sunday
    - "0 */2 * * *" : Every 2 hours
    - "0 9-17 * * 1-5" : Every hour from 9 AM to 5 PM on weekdays

Usage:
    from trixy_core.scheduler.cron_parser import CronParser
    
    # Parse cron expression
    parser = CronParser()
    cron_expr = parser.parse("0 9 * * 1-5")
    
    # Check if current time matches
    import time
    if cron_expr.matches(time.time()):
        print("Cron expression matches current time")
    
    # Get next execution time
    next_time = cron_expr.get_next_time()
    print(f"Next execution: {next_time}")
"""

import re
import time
import calendar
from datetime import datetime, timedelta
from typing import List, Set, Optional, Union, Tuple, Dict, Any
from dataclasses import dataclass
from enum import Enum


def pprint(message: str) -> None:
    """
    Cron parser logging function that adapts based on mode.
    Uses the same pattern as specified in CLAUDE.md.
    """
    print(f"[SCHEDULER.CRON] {message}")


class CronParseError(Exception):
    """Raised when cron expression cannot be parsed."""
    pass


class CronValidationError(Exception):
    """Raised when cron expression is invalid."""
    pass


class CronFieldType(Enum):
    """Types of cron fields."""
    MINUTE = "minute"
    HOUR = "hour"
    DAY_OF_MONTH = "day_of_month"
    MONTH = "month"
    DAY_OF_WEEK = "day_of_week"


@dataclass
class CronFieldSpec:
    """Specification for a cron field."""
    field_type: CronFieldType
    min_value: int
    max_value: int
    aliases: Dict[str, int]
    allow_question_mark: bool = False


# Cron field specifications
CRON_FIELD_SPECS = {
    CronFieldType.MINUTE: CronFieldSpec(
        field_type=CronFieldType.MINUTE,
        min_value=0,
        max_value=59,
        aliases={}
    ),
    CronFieldType.HOUR: CronFieldSpec(
        field_type=CronFieldType.HOUR,
        min_value=0,
        max_value=23,
        aliases={}
    ),
    CronFieldType.DAY_OF_MONTH: CronFieldSpec(
        field_type=CronFieldType.DAY_OF_MONTH,
        min_value=1,
        max_value=31,
        aliases={},
        allow_question_mark=True
    ),
    CronFieldType.MONTH: CronFieldSpec(
        field_type=CronFieldType.MONTH,
        min_value=1,
        max_value=12,
        aliases={
            'JAN': 1, 'FEB': 2, 'MAR': 3, 'APR': 4, 'MAY': 5, 'JUN': 6,
            'JUL': 7, 'AUG': 8, 'SEP': 9, 'OCT': 10, 'NOV': 11, 'DEC': 12
        }
    ),
    CronFieldType.DAY_OF_WEEK: CronFieldSpec(
        field_type=CronFieldType.DAY_OF_WEEK,
        min_value=0,
        max_value=7,  # 0 and 7 both represent Sunday
        aliases={
            'SUN': 0, 'MON': 1, 'TUE': 2, 'WED': 3, 'THU': 4, 'FRI': 5, 'SAT': 6
        },
        allow_question_mark=True
    )
}


class CronField:
    """
    Represents a single field in a cron expression.
    
    Handles parsing and validation of individual cron fields with support
    for wildcards, ranges, lists, steps, and aliases.
    """
    
    def __init__(self, field_type: CronFieldType, expression: str):
        """
        Initialize a cron field.
        
        Args:
            field_type: Type of the cron field
            expression: String expression for this field
            
        Raises:
            CronParseError: If expression cannot be parsed
            CronValidationError: If expression is invalid
        """
        self.field_type = field_type
        self.expression = expression.strip().upper()
        self.spec = CRON_FIELD_SPECS[field_type]
        
        # Parse the expression into a set of valid values
        self.values: Set[int] = set()
        self.is_wildcard = False
        self.is_question_mark = False
        
        self._parse_expression()
        self._validate_values()
    
    def _parse_expression(self) -> None:
        """Parse the field expression into valid values."""
        expr = self.expression
        
        # Handle wildcard
        if expr == '*':
            self.is_wildcard = True
            self.values = set(range(self.spec.min_value, self.spec.max_value + 1))
            return
        
        # Handle question mark
        if expr == '?':
            if not self.spec.allow_question_mark:
                raise CronParseError(f"Question mark not allowed in {self.field_type.value} field")
            self.is_question_mark = True
            return
        
        # Split by commas for list values
        parts = expr.split(',')
        
        for part in parts:
            part = part.strip()
            if not part:
                continue
            
            # Check for step values (/)
            if '/' in part:
                range_part, step_part = part.split('/', 1)
                try:
                    step = int(step_part)
                    if step <= 0:
                        raise CronParseError(f"Step value must be positive: {step}")
                except ValueError:
                    raise CronParseError(f"Invalid step value: {step_part}")
                
                # Parse the range part
                if range_part == '*':
                    start, end = self.spec.min_value, self.spec.max_value
                elif '-' in range_part:
                    start_str, end_str = range_part.split('-', 1)
                    start = self._parse_value(start_str.strip())
                    end = self._parse_value(end_str.strip())
                else:
                    start = self._parse_value(range_part)
                    end = self.spec.max_value
                
                # Add stepped values
                current = start
                while current <= end:
                    if self.spec.min_value <= current <= self.spec.max_value:
                        self.values.add(current)
                    current += step
            
            # Check for range values (-)
            elif '-' in part:
                start_str, end_str = part.split('-', 1)
                start = self._parse_value(start_str.strip())
                end = self._parse_value(end_str.strip())
                
                if start > end:
                    raise CronParseError(f"Invalid range: {start}-{end} (start > end)")
                
                for value in range(start, end + 1):
                    self.values.add(value)
            
            # Single value
            else:
                value = self._parse_value(part)
                self.values.add(value)
    
    def _parse_value(self, value_str: str) -> int:
        """
        Parse a single value, handling aliases.
        
        Args:
            value_str: String representation of the value
            
        Returns:
            int: Parsed integer value
            
        Raises:
            CronParseError: If value cannot be parsed
        """
        value_str = value_str.strip().upper()
        
        # Check aliases first
        if value_str in self.spec.aliases:
            return self.spec.aliases[value_str]
        
        # Try to parse as integer
        try:
            value = int(value_str)
        except ValueError:
            raise CronParseError(f"Invalid value: {value_str}")
        
        return value
    
    def _validate_values(self) -> None:
        """Validate that all parsed values are within valid range."""
        if self.is_question_mark or self.is_wildcard:
            return
        
        for value in self.values:
            if not (self.spec.min_value <= value <= self.spec.max_value):
                # Special case for day_of_week: 7 is also Sunday (same as 0)
                if self.field_type == CronFieldType.DAY_OF_WEEK and value == 7:
                    self.values.remove(7)
                    self.values.add(0)
                else:
                    raise CronValidationError(
                        f"Value {value} out of range for {self.field_type.value} "
                        f"(valid range: {self.spec.min_value}-{self.spec.max_value})"
                    )
    
    def matches(self, value: int) -> bool:
        """
        Check if a value matches this cron field.
        
        Args:
            value: Value to check
            
        Returns:
            bool: True if value matches this field
        """
        if self.is_question_mark:
            return True
        
        if self.is_wildcard:
            return self.spec.min_value <= value <= self.spec.max_value
        
        # Special handling for day_of_week where 7 = Sunday = 0
        if self.field_type == CronFieldType.DAY_OF_WEEK and value == 7:
            value = 0
        
        return value in self.values
    
    def get_next_value(self, current_value: int) -> Optional[int]:
        """
        Get the next valid value after the current value.
        
        Args:
            current_value: Current value
            
        Returns:
            Optional[int]: Next valid value or None if no more values
        """
        if self.is_question_mark:
            return current_value
        
        valid_values = sorted(self.values)
        
        for value in valid_values:
            if value > current_value:
                return value
        
        # Wrap around to the first value
        return valid_values[0] if valid_values else None
    
    def to_dict(self) -> Dict[str, Any]:
        """Convert field to dictionary representation."""
        return {
            'field_type': self.field_type.value,
            'expression': self.expression,
            'values': sorted(list(self.values)) if not (self.is_wildcard or self.is_question_mark) else None,
            'is_wildcard': self.is_wildcard,
            'is_question_mark': self.is_question_mark
        }
    
    def __str__(self) -> str:
        """String representation of the cron field."""
        return f"CronField({self.field_type.value}='{self.expression}')"
    
    def __repr__(self) -> str:
        """Detailed representation of the cron field."""
        return (
            f"CronField(type={self.field_type.value}, "
            f"expr='{self.expression}', "
            f"values={sorted(list(self.values)) if self.values else None})"
        )


class CronExpression:
    """
    Represents a complete cron expression.
    
    Parses and validates a 5-field cron expression and provides methods
    for checking matches and calculating next execution times.
    """
    
    def __init__(self, expression: str):
        """
        Initialize a cron expression.
        
        Args:
            expression: Complete cron expression string
            
        Raises:
            CronParseError: If expression cannot be parsed
            CronValidationError: If expression is invalid
        """
        self.original_expression = expression.strip()
        self.fields: List[CronField] = []
        
        # Parse the expression
        self._parse_expression()
        
        # Validate the combination
        self._validate_expression()
    
    def _parse_expression(self) -> None:
        """Parse the complete cron expression."""
        parts = self.original_expression.split()
        
        if len(parts) != 5:
            raise CronParseError(
                f"Cron expression must have exactly 5 fields, got {len(parts)}: {self.original_expression}"
            )
        
        field_types = [
            CronFieldType.MINUTE,
            CronFieldType.HOUR,
            CronFieldType.DAY_OF_MONTH,
            CronFieldType.MONTH,
            CronFieldType.DAY_OF_WEEK
        ]
        
        for i, (field_type, field_expr) in enumerate(zip(field_types, parts)):
            try:
                field = CronField(field_type, field_expr)
                self.fields.append(field)
            except (CronParseError, CronValidationError) as e:
                raise CronParseError(f"Error in field {i+1} ({field_type.value}): {e}")
    
    def _validate_expression(self) -> None:
        """Validate the complete cron expression."""
        # Check for conflicting day specifications
        day_of_month_field = self.fields[2]  # day_of_month
        day_of_week_field = self.fields[4]   # day_of_week
        
        # If both day_of_month and day_of_week are specified (not * or ?),
        # this creates an OR condition which is valid but worth noting
        if (not day_of_month_field.is_wildcard and not day_of_month_field.is_question_mark and
            not day_of_week_field.is_wildcard and not day_of_week_field.is_question_mark):
            pprint(f"Warning: Both day_of_month and day_of_week specified - schedule will run when EITHER matches")
    
    def matches(self, timestamp: Optional[float] = None) -> bool:
        """
        Check if the cron expression matches a given timestamp.
        
        Args:
            timestamp: Unix timestamp to check (defaults to current time)
            
        Returns:
            bool: True if expression matches the timestamp
        """
        if timestamp is None:
            timestamp = time.time()
        
        dt = datetime.fromtimestamp(timestamp)
        
        # Check each field
        values = [
            dt.minute,           # minute
            dt.hour,             # hour
            dt.day,              # day_of_month
            dt.month,            # month
            dt.weekday() + 1     # day_of_week (convert from Monday=0 to Sunday=0)
        ]
        
        # Convert Sunday from 1 to 0 for day_of_week
        if values[4] == 7:
            values[4] = 0
        
        # Special handling for day_of_month and day_of_week
        # If both are specified, it's an OR condition
        day_of_month_matches = self.fields[2].matches(values[2])
        day_of_week_matches = self.fields[4].matches(values[4])
        
        # Check other fields (minute, hour, month)
        other_fields_match = all(
            field.matches(value) 
            for i, (field, value) in enumerate(zip(self.fields, values))
            if i not in [2, 4]  # Skip day_of_month and day_of_week
        )
        
        if not other_fields_match:
            return False
        
        # Handle day_of_month and day_of_week logic
        day_of_month_specified = not (self.fields[2].is_wildcard or self.fields[2].is_question_mark)
        day_of_week_specified = not (self.fields[4].is_wildcard or self.fields[4].is_question_mark)
        
        if day_of_month_specified and day_of_week_specified:
            # Both specified: OR condition
            return day_of_month_matches or day_of_week_matches
        elif day_of_month_specified:
            # Only day_of_month specified
            return day_of_month_matches
        elif day_of_week_specified:
            # Only day_of_week specified
            return day_of_week_matches
        else:
            # Neither specified (both * or ?)
            return True
    
    def get_next_time(self, after_timestamp: Optional[float] = None) -> float:
        """
        Get the next time this cron expression will match.
        
        Args:
            after_timestamp: Find next time after this timestamp (defaults to current time)
            
        Returns:
            float: Unix timestamp of next execution
            
        Raises:
            CronValidationError: If no valid next time can be found
        """
        if after_timestamp is None:
            after_timestamp = time.time()
        
        # Start from the next minute
        dt = datetime.fromtimestamp(after_timestamp)
        dt = dt.replace(second=0, microsecond=0) + timedelta(minutes=1)
        
        # Limit search to avoid infinite loops
        max_iterations = 366 * 24 * 60  # One year worth of minutes
        iterations = 0
        
        while iterations < max_iterations:
            if self.matches(dt.timestamp()):
                return dt.timestamp()
            
            # Advance to next minute
            dt += timedelta(minutes=1)
            iterations += 1
        
        raise CronValidationError("Could not find next execution time within reasonable timeframe")
    
    def get_previous_time(self, before_timestamp: Optional[float] = None) -> Optional[float]:
        """
        Get the previous time this cron expression matched.
        
        Args:
            before_timestamp: Find previous time before this timestamp (defaults to current time)
            
        Returns:
            Optional[float]: Unix timestamp of previous execution or None if not found
        """
        if before_timestamp is None:
            before_timestamp = time.time()
        
        # Start from the current minute
        dt = datetime.fromtimestamp(before_timestamp)
        dt = dt.replace(second=0, microsecond=0)
        
        # Limit search to avoid infinite loops
        max_iterations = 366 * 24 * 60  # One year worth of minutes
        iterations = 0
        
        while iterations < max_iterations:
            if self.matches(dt.timestamp()):
                return dt.timestamp()
            
            # Go back one minute
            dt -= timedelta(minutes=1)
            iterations += 1
        
        return None
    
    def get_description(self) -> str:
        """
        Get a human-readable description of the cron expression.
        
        Returns:
            str: Human-readable description
        """
        descriptions = []
        
        # Minute
        minute_field = self.fields[0]
        if minute_field.is_wildcard:
            descriptions.append("every minute")
        elif len(minute_field.values) == 1:
            minute = list(minute_field.values)[0]
            descriptions.append(f"at minute {minute}")
        else:
            minutes = sorted(minute_field.values)
            descriptions.append(f"at minutes {', '.join(map(str, minutes))}")
        
        # Hour
        hour_field = self.fields[1]
        if not hour_field.is_wildcard:
            if len(hour_field.values) == 1:
                hour = list(hour_field.values)[0]
                descriptions.append(f"at {hour:02d}:00")
            else:
                hours = sorted(hour_field.values)
                descriptions.append(f"during hours {', '.join(map(str, hours))}")
        
        # Day descriptions
        day_descriptions = []
        
        # Day of month
        dom_field = self.fields[2]
        if not (dom_field.is_wildcard or dom_field.is_question_mark):
            if len(dom_field.values) == 1:
                day = list(dom_field.values)[0]
                day_descriptions.append(f"on day {day} of the month")
            else:
                days = sorted(dom_field.values)
                day_descriptions.append(f"on days {', '.join(map(str, days))} of the month")
        
        # Day of week
        dow_field = self.fields[4]
        if not (dow_field.is_wildcard or dow_field.is_question_mark):
            day_names = ['Sunday', 'Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday']
            if len(dow_field.values) == 1:
                day = list(dow_field.values)[0]
                day_descriptions.append(f"on {day_names[day]}")
            else:
                days = sorted(dow_field.values)
                day_descriptions.append(f"on {', '.join(day_names[d] for d in days)}")
        
        if day_descriptions:
            if len(day_descriptions) == 2:
                descriptions.append(f"({' OR '.join(day_descriptions)})")
            else:
                descriptions.extend(day_descriptions)
        
        # Month
        month_field = self.fields[3]
        if not month_field.is_wildcard:
            month_names = ['', 'January', 'February', 'March', 'April', 'May', 'June',
                          'July', 'August', 'September', 'October', 'November', 'December']
            if len(month_field.values) == 1:
                month = list(month_field.values)[0]
                descriptions.append(f"in {month_names[month]}")
            else:
                months = sorted(month_field.values)
                descriptions.append(f"in {', '.join(month_names[m] for m in months)}")
        
        return ' '.join(descriptions)
    
    def to_dict(self) -> Dict[str, Any]:
        """Convert cron expression to dictionary representation."""
        return {
            'expression': self.original_expression,
            'description': self.get_description(),
            'fields': [field.to_dict() for field in self.fields]
        }
    
    def __str__(self) -> str:
        """String representation of the cron expression."""
        return self.original_expression
    
    def __repr__(self) -> str:
        """Detailed representation of the cron expression."""
        return f"CronExpression('{self.original_expression}')"


class CronParser:
    """
    Main cron parser class for parsing and validating cron expressions.
    
    Provides static methods for common cron parsing operations and maintains
    a cache of parsed expressions for performance.
    """
    
    def __init__(self, cache_size: int = 100):
        """
        Initialize the cron parser.
        
        Args:
            cache_size: Maximum number of parsed expressions to cache
        """
        self._cache: Dict[str, CronExpression] = {}
        self._cache_size = cache_size
    
    def parse(self, expression: str) -> CronExpression:
        """
        Parse a cron expression.
        
        Args:
            expression: Cron expression string to parse
            
        Returns:
            CronExpression: Parsed cron expression
            
        Raises:
            CronParseError: If expression cannot be parsed
            CronValidationError: If expression is invalid
        """
        expression = expression.strip()
        
        # Check cache first
        if expression in self._cache:
            return self._cache[expression]
        
        # Parse new expression
        cron_expr = CronExpression(expression)
        
        # Add to cache
        if len(self._cache) >= self._cache_size:
            # Remove oldest entry (simple FIFO)
            oldest_key = next(iter(self._cache))
            del self._cache[oldest_key]
        
        self._cache[expression] = cron_expr
        return cron_expr
    
    def validate(self, expression: str) -> List[str]:
        """
        Validate a cron expression without parsing.
        
        Args:
            expression: Cron expression string to validate
            
        Returns:
            List[str]: List of validation errors (empty if valid)
        """
        try:
            self.parse(expression)
            return []
        except (CronParseError, CronValidationError) as e:
            return [str(e)]
    
    def clear_cache(self) -> None:
        """Clear the parser cache."""
        self._cache.clear()


# Convenience functions for direct use

def parse_cron_expression(expression: str) -> CronExpression:
    """
    Parse a cron expression directly.
    
    Args:
        expression: Cron expression string
        
    Returns:
        CronExpression: Parsed cron expression
    """
    return CronExpression(expression)


def validate_cron_expression(expression: str) -> bool:
    """
    Validate a cron expression.
    
    Args:
        expression: Cron expression string
        
    Returns:
        bool: True if valid, False otherwise
    """
    try:
        parse_cron_expression(expression)
        return True
    except (CronParseError, CronValidationError):
        return False


def get_next_cron_time(expression: str, after_timestamp: Optional[float] = None) -> float:
    """
    Get the next execution time for a cron expression.
    
    Args:
        expression: Cron expression string
        after_timestamp: Find next time after this timestamp (defaults to current time)
        
    Returns:
        float: Unix timestamp of next execution
    """
    cron_expr = parse_cron_expression(expression)
    return cron_expr.get_next_time(after_timestamp)


def cron_matches_time(expression: str, timestamp: Optional[float] = None) -> bool:
    """
    Check if a cron expression matches a given time.
    
    Args:
        expression: Cron expression string
        timestamp: Unix timestamp to check (defaults to current time)
        
    Returns:
        bool: True if expression matches the timestamp
    """
    cron_expr = parse_cron_expression(expression)
    return cron_expr.matches(timestamp)


# Module exports
__all__ = [
    'CronParser',
    'CronExpression', 
    'CronField',
    'CronParseError',
    'CronValidationError',
    'CronFieldType',
    'parse_cron_expression',
    'validate_cron_expression',
    'get_next_cron_time',
    'cron_matches_time',
    'pprint'
]