dify/api/libs/schedule_utils.py

from datetime import UTC, datetime

import pytz
from croniter import croniter


def calculate_next_run_at(
    cron_expression: str,
    timezone: str,
    base_time: datetime | None = None,
) -> datetime:
    """
    Calculate the next run time for a cron expression in a specific timezone.

    Args:
        cron_expression: Standard 5-field cron expression or predefined expression
        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 enhanced cron syntax including:
        - Month abbreviations: JAN, FEB, MAR-JUN, JAN,JUN,DEC
        - Day abbreviations: MON, TUE, MON-FRI, SUN,WED,FRI
        - Predefined expressions: @daily, @weekly, @monthly, @yearly, @hourly
        - Special characters: ? wildcard, L (last day), Sunday as 7
        - Standard 5-field format only (minute hour day month dayOfWeek)
    """
    # Validate cron expression format to match frontend behavior
    parts = cron_expression.strip().split()

    # Support both 5-field format and predefined expressions (matching frontend)
    if len(parts) != 5 and not cron_expression.startswith("@"):
        raise ValueError(
            f"Cron expression must have exactly 5 fields or be a predefined expression "
            f"(@daily, @weekly, etc.). Got {len(parts)} fields: '{cron_expression}'"
        )

    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