from datetime import UTC, datetime from typing import Optional import pytz from croniter import croniter def calculate_next_run_at( cron_expression: str, timezone: str, base_time: Optional[datetime] = None, ) -> datetime: """ Calculate the next run time for a cron expression in a specific timezone. Args: cron_expression: Cron expression string (supports croniter extensions like 'L') timezone: Timezone string (e.g., 'UTC', 'America/New_York') base_time: Base time to calculate from (defaults to current UTC time) Returns: Next run time in UTC Note: Supports croniter's extended syntax including: - 'L' for last day of month - Standard 5-field cron expressions """ tz = pytz.timezone(timezone) if base_time is None: base_time = datetime.now(UTC) base_time_tz = base_time.astimezone(tz) cron = croniter(cron_expression, base_time_tz) next_run_tz = cron.get_next(datetime) next_run_utc = next_run_tz.astimezone(UTC) return next_run_utc def convert_12h_to_24h(time_str: str) -> tuple[int, int]: """ Parse 12-hour time format to 24-hour format for cron compatibility. Args: time_str: Time string in format "HH:MM AM/PM" (e.g., "12:30 PM") Returns: Tuple of (hour, minute) in 24-hour format Raises: ValueError: If time string format is invalid or values are out of range Examples: - "12:00 AM" -> (0, 0) # Midnight - "12:00 PM" -> (12, 0) # Noon - "1:30 PM" -> (13, 30) - "11:59 PM" -> (23, 59) """ if not time_str or not time_str.strip(): raise ValueError("Time string cannot be empty") parts = time_str.strip().split() if len(parts) != 2: raise ValueError(f"Invalid time format: '{time_str}'. Expected 'HH:MM AM/PM'") time_part, period = parts period = period.upper() if period not in ["AM", "PM"]: raise ValueError(f"Invalid period: '{period}'. Must be 'AM' or 'PM'") time_parts = time_part.split(":") if len(time_parts) != 2: raise ValueError(f"Invalid time format: '{time_part}'. Expected 'HH:MM'") try: hour = int(time_parts[0]) minute = int(time_parts[1]) except ValueError as e: raise ValueError(f"Invalid time values: {e}") if hour < 1 or hour > 12: raise ValueError(f"Invalid hour: {hour}. Must be between 1 and 12") if minute < 0 or minute > 59: raise ValueError(f"Invalid minute: {minute}. Must be between 0 and 59") # Handle 12-hour to 24-hour edge cases if period == "PM" and hour != 12: hour += 12 elif period == "AM" and hour == 12: hour = 0 return hour, minute