Python script to transform Quartz CRON expressions into APScheduler CronTrigger

quartz_apscheduler.py

from datetime import datetime, timedelta

from apscheduler.triggers.cron import CronTrigger

class QuartzExpressionParser:
    """Parser for converting Quartz CRON expressions to APScheduler kwargs.

    This class takes into account the special characters used in Quartz ('L', 'W', and '#')
    and translates them into the format that APScheduler understands.
    """

    def __init__(self, quartz_cron):
        """Initialize the parser with a Quartz CRON expression."""
        self.quartz_cron = quartz_cron.strip()
        self.parts = self.quartz_cron.split()

        if len(self.parts) < 6 or len(self.parts) > 7:
            raise ValueError("Invalid Quartz CRON expression")

    def to_apscheduler_kwargs(self):
        """Convert the Quartz CRON expression to APScheduler kwargs."""
        kwargs = {
            'second': self.parts[0],
            'minute': self.parts[1],
            'hour': self.parts[2],
            'month': self.parts[4],
            'day': self._parse_day(),
            'year': self._parse_year()
        }

        # Omit 'day_of_week' if 'day' already contains special expressions
        if not kwargs.get('day') or not self._is_special_day_expression(kwargs['day']):
            kwargs['day_of_week'] = self._parse_day_of_week()

        return kwargs

    def _parse_weekday_nearest(self, day):
        """Calculate the nearest weekday for a given day of the month or last weekday of the month."""
        current_year = datetime.now().year
        current_month = datetime.now().month

        # If year or month are not specified, use current year and month
        year = int(self.parts[6]) if len(self.parts) == 7 and self.parts[6].isdigit() else current_year
        month = int(self.parts[4]) if self.parts[4].isdigit() else current_month

        if 'LW' == day:
            # Calculate the last day of the month
            next_month = datetime(year, month % 12 + 1, 1)
            last_day_of_month = (next_month - timedelta(days=1)).day
            target_date = datetime(year, month, last_day_of_month)
        elif 'W' in day:
            day_number = int(day.replace('W', ''))
            target_date = datetime(year, month, day_number)
        else:
            return day

        # Adjust if the target date falls on a weekend
        weekday = target_date.weekday()
        if weekday == 5:
            # If it's Saturday, move to Friday
            adjusted_date = target_date - timedelta(days=1)
        elif weekday == 6:
            # If it's Sunday, move to Monday (but check for month boundary)
            adjusted_date = target_date + timedelta(days=1) if target_date.day == 1 else target_date - timedelta(days=2)
        else:
            # It's already a weekday
            adjusted_date = target_date

        return adjusted_date.day


    def _parse_day(self):
        """Parse the day field to handle special Quartz characters."""
        # Handle 'L' and '#' in the day-of-month field
        day_of_month = self.parts[3]
        day_of_week = self.parts[5]

        if 'L' in day_of_month and 'W' not in day_of_month:
            return 'last'
        elif 'W' in day_of_month:
            # Handle 'LW' and 'xW' in the day-of-month field
            return str(self._parse_weekday_nearest(day_of_month))
        elif day_of_week.endswith('L'):
            # The 'L' character is used to specify the last occurrence of a day in a month in Quartz.
            return 'last ' + self._day_of_week_from_quartz(day_of_week[0])
        elif '#' in day_of_week:
            # The '#' character is used to specify the "nth" occurrence of a particular weekday.
            weekday, nth = day_of_week.split('#')
            if int(nth) == 1:
                nth = '1st'
            elif int(nth) == 2:
                nth = '2nd'
            elif int(nth) == 3:
                nth = '3rd'
            else:
                nth = f'{nth}th'
            return nth + ' ' + self._day_of_week_from_quartz(weekday)
        elif day_of_month in ('?', '*'):
            return None
        return day_of_month

    def _parse_day_of_week(self):
        """Parse the day_of_week field and return as a comma-separated string."""
        day_of_week = self.parts[5]

        if day_of_week in ('?', '*'):
            return None
        if '-' in day_of_week:
            # Handle ranges
            start_day, end_day = day_of_week.split('-')
            return '-'.join(map(self._day_of_week_from_quartz, [start_day, end_day]))

        return self._day_of_week_from_quartz(day_of_week)

    def _parse_year(self):
        """Parse the year field from the Quartz CRON expression."""
        if len(self.parts) == 7 and self.parts[6] != '*':
            return self.parts[6]
        return None


    def _day_of_week_from_quartz(self, quartz_day):
        """Convert Quartz day of the week to APScheduler format."""
        weekdays = {'1': 'sun', '2': 'mon', '3': 'tue', '4': 'wed', '5': 'thu', '6': 'fri', '7': 'sat'}
        return weekdays.get(quartz_day, quartz_day.lower())

    def _is_special_day_expression(self, day_expression):
        """Check if the 'day' field contains a special expression like 'last' or 'nth'."""
        return 'last' in day_expression or any(char.isdigit() for char in day_expression)


# The following cron expressions are based on the screenshot provided
quartz_cron_expressions = [
    # <second> <minute> <hour> <day of the month> <month> <day of the week> <year>
    "0 20 20 ? * * *",        # Daily at 20:20 UTC
    "0 0 10 ? * MON-FRI *",   # Daily on weekdays at 10:00 UTC
    "0 0 10 ? * MON *",       # Weekly on Monday at 10:00 UTC
    "0 0 10 ? * MON,TUE *",   # Weekly on Mon and Tue at 10:00 UTC
    "0 0 10 ? * MON-WED *",   # Weekly on Mon, Tue and Wed at 10:00 UTC
    "0 0 10 ? * MON-FRI *",   # Weekly on weekdays at 10:00 UTC
    "0 0 10 ? * SAT,SUN *",   # Weekly on weekends at 10:00 UTC
    "0 0 10 ? * 2#2 *",       # Monthly, every second Monday at 10:00 UTC
    "0 0 10 ? * 3L *",        # Monthly, every last Tuesday at 10:00 UTC
    "0 0 10 ? * 3#3 *",       # Monthly, every third Wednesday at 10:00 UTC
    "0 0 10 ? */2 4#4 *",       # Every 2 months, on the fourth weekday, at 10:00 UTC,
    "0 20 20 */2 * ? *",      # Every 2 days, at 20:20 UTC
    "0 0 10 1 * ?",           # Monthly on the 1st at 10:00 UTC
    "0 0 10 1W * ?",         # Monthly on the nearest weekday to the 1st at 10:00 UTC
    "0 0 10 L * ?",           # Monthly on the last day of the month at 10:00 UTC
    "0 0 10 LW * ?",          # Monthly on the last weekday of the month at 10:00 UTC
]

if __name__ == "__main__":
    for quartz_cron in quartz_cron_expressions:
        try:
            parser = QuartzExpressionParser(quartz_cron)
            apscheduler_kwargs = parser.to_apscheduler_kwargs()
            trigger = CronTrigger(**apscheduler_kwargs)
            print(f"Validated Quartz CRON Expression: '{quartz_cron}' -> {trigger}")
        except ValueError as e:
            print(f"Error with Quartz CRON Expression '{quartz_cron}': {e}")

Output:

Validated Quartz CRON Expression: '0 20 20 ? * * *' -> cron[month='*', hour='20', minute='20', second='0']
Validated Quartz CRON Expression: '0 0 10 ? * MON-FRI *' -> cron[month='*', day_of_week='mon-fri', hour='10', minute='0', second='0']
Validated Quartz CRON Expression: '0 0 10 ? * MON *' -> cron[month='*', day_of_week='mon', hour='10', minute='0', second='0']
Validated Quartz CRON Expression: '0 0 10 ? * MON,TUE *' -> cron[month='*', day_of_week='mon,tue', hour='10', minute='0', second='0']
Validated Quartz CRON Expression: '0 0 10 ? * MON-WED *' -> cron[month='*', day_of_week='mon-wed', hour='10', minute='0', second='0']
Validated Quartz CRON Expression: '0 0 10 ? * MON-FRI *' -> cron[month='*', day_of_week='mon-fri', hour='10', minute='0', second='0']
Validated Quartz CRON Expression: '0 0 10 ? * SAT,SUN *' -> cron[month='*', day_of_week='sat,sun', hour='10', minute='0', second='0']
Validated Quartz CRON Expression: '0 0 10 ? * 2#2 *' -> cron[month='*', day='2nd mon', hour='10', minute='0', second='0']
Validated Quartz CRON Expression: '0 0 10 ? * 3L *' -> cron[month='*', day='last tue', hour='10', minute='0', second='0']
Validated Quartz CRON Expression: '0 0 10 ? * 3#3 *' -> cron[month='*', day='3rd tue', hour='10', minute='0', second='0']
Validated Quartz CRON Expression: '0 0 10 ? */2 4#4 *' -> cron[month='*/2', day='4th wed', hour='10', minute='0', second='0']
Validated Quartz CRON Expression: '0 20 20 */2 * ? *' -> cron[month='*', day='*/2', hour='20', minute='20', second='0']
Validated Quartz CRON Expression: '0 0 10 1 * ?' -> cron[month='*', day='1', hour='10', minute='0', second='0']
Validated Quartz CRON Expression: '0 0 10 1W * ?' -> cron[month='*', day='1', hour='10', minute='0', second='0']
Validated Quartz CRON Expression: '0 0 10 L * ?' -> cron[month='*', day='last', hour='10', minute='0', second='0']
Validated Quartz CRON Expression: '0 0 10 LW * ?' -> cron[month='*', day='29', hour='10', minute='0', second='0']

This doesn't handle the W quite accurately yet, but all other Quartz features work