# The MIT License (MIT) # # Copyright (c) 2016 Adam Schubert # # Permission is hereby granted, free of charge, to any person obtaining a copy # of this software and associated documentation files (the "Software"), to deal # in the Software without restriction, including without limitation the rights # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell # copies of the Software, and to permit persons to whom the Software is # furnished to do so, subject to the following conditions: # # The above copyright notice and this permission notice shall be included in all # copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE # SOFTWARE. import re from .Exception import MissingFieldException, FormatException class ExpressionParser(object): _expression = '' _options = None _cron_days = { 0: 'SUN', 1: 'MON', 2: 'TUE', 3: 'WED', 4: 'THU', 5: 'FRI', 6: 'SAT' } _cron_months = { 1: 'JAN', 2: 'FEB', 3: 'MAR', 4: 'APR', 5: 'MAY', 6: 'JUN', 7: 'JUL', 8: 'AUG', 9: 'SEP', 10: 'OCT', 11: 'NOV', 12: 'DEC' } def __init__(self, expression, options): """Initializes a new instance of the ExpressionParser class Args: expression: The cron expression string options: Parsing options """ self._expression = expression self._options = options def parse(self): """Parses the cron expression string Returns: A 7 part string array, one part for each component of the cron expression (seconds, minutes, etc.) Raises: MissingFieldException: if _expression is empty or None FormatException: if _expression has wrong format """ # Initialize all elements of parsed array to empty strings parsed = ['', '', '', '', '', '', ''] if self._expression is None or len(self._expression) == 0: raise MissingFieldException("ExpressionDescriptor.expression") else: expression_parts_temp = self._expression.split() expression_parts_temp_length = len(expression_parts_temp) if expression_parts_temp_length < 5: raise FormatException( "Error: Expression only has {0} parts. At least 5 part are required.".format( expression_parts_temp_length ) ) elif expression_parts_temp_length == 5: # 5 part cron so shift array past seconds element for i, expression_part_temp in enumerate(expression_parts_temp): parsed[i + 1] = expression_part_temp elif expression_parts_temp_length == 6: # We will detect if this 6 part expression has a year specified and if so we will shift the parts and treat the # first part as a minute part rather than a second part. # Ways we detect: # 1. Last part is a literal year (i.e. 2020) # 2. 3rd or 5th part is specified as "?" (DOM or DOW) year_regex = re.compile(r"\d{4}$") is_year_with_no_seconds_part = bool(year_regex.search(expression_parts_temp[5])) or "?" in [expression_parts_temp[4], expression_parts_temp[2]] for i, expression_part_temp in enumerate(expression_parts_temp): if is_year_with_no_seconds_part: # Shift parts over by one parsed[i + 1] = expression_part_temp else: parsed[i] = expression_part_temp elif expression_parts_temp_length == 7: parsed = expression_parts_temp else: raise FormatException( "Error: Expression has too many parts ({0}). Expression must not have more than 7 parts.".format( expression_parts_temp_length ) ) self.normalize_expression(parsed) return parsed def normalize_expression(self, expression_parts): """Converts cron expression components into consistent, predictable formats. Args: expression_parts: A 7 part string array, one part for each component of the cron expression Returns: None """ # convert ? to * only for DOM and DOW expression_parts[3] = expression_parts[3].replace("?", "*") expression_parts[5] = expression_parts[5].replace("?", "*") # convert 0/, 1/ to */ if expression_parts[0].startswith("0/"): expression_parts[0] = expression_parts[0].replace("0/", "*/") # seconds if expression_parts[1].startswith("0/"): expression_parts[1] = expression_parts[1].replace("0/", "*/") # minutes if expression_parts[2].startswith("0/"): expression_parts[2] = expression_parts[2].replace("0/", "*/") # hours if expression_parts[3].startswith("1/"): expression_parts[3] = expression_parts[3].replace("1/", "*/") # DOM if expression_parts[4].startswith("1/"): expression_parts[4] = expression_parts[4].replace("1/", "*/") # Month if expression_parts[5].startswith("1/"): expression_parts[5] = expression_parts[5].replace("1/", "*/") # DOW if expression_parts[6].startswith("1/"): expression_parts[6] = expression_parts[6].replace("1/", "*/") # Years # Adjust DOW based on dayOfWeekStartIndexZero option def digit_replace(match): match_value = match.group() dow_digits = re.sub(r'\D', "", match_value) dow_digits_adjusted = dow_digits if self._options.day_of_week_start_index_zero: if dow_digits == "7": dow_digits_adjusted = "0" else: dow_digits_adjusted = str(int(dow_digits) - 1) return match_value.replace(dow_digits, dow_digits_adjusted) expression_parts[5] = re.sub(r'(^\d)|([^#/\s]\d)', digit_replace, expression_parts[5]) # Convert DOM '?' to '*' if expression_parts[3] == "?": expression_parts[3] = "*" # convert SUN-SAT format to 0-6 format for day_number in self._cron_days: expression_parts[5] = expression_parts[5].upper().replace(self._cron_days[day_number], str(day_number)) # convert JAN-DEC format to 1-12 format for month_number in self._cron_months: expression_parts[4] = expression_parts[4].upper().replace( self._cron_months[month_number], str(month_number)) # convert 0 second to (empty) if expression_parts[0] == "0": expression_parts[0] = '' # If time interval is specified for seconds or minutes and next time part is single item, make it a "self-range" so # the expression can be interpreted as an interval 'between' range. # For example: # 0-20/3 9 * * * => 0-20/3 9-9 * * * (9 => 9-9) # */5 3 * * * => */5 3-3 * * * (3 => 3-3) star_and_slash = ['*', '/'] has_part_zero_star_and_slash = any(ext in expression_parts[0] for ext in star_and_slash) has_part_one_star_and_slash = any(ext in expression_parts[1] for ext in star_and_slash) has_part_two_special_chars = any(ext in expression_parts[2] for ext in ['*', '-', ',', '/']) if not has_part_two_special_chars and (has_part_zero_star_and_slash or has_part_one_star_and_slash): expression_parts[2] += '-{}'.format(expression_parts[2]) # Loop through all parts and apply global normalization length = len(expression_parts) for i in range(length): # convert all '*/1' to '*' if expression_parts[i] == "*/1": expression_parts[i] = "*" """ Convert Month,DOW,Year step values with a starting value (i.e. not '*') to between expressions. This allows us to reuse the between expression handling for step values. For Example: - month part '3/2' will be converted to '3-12/2' (every 2 months between March and December) - DOW part '3/2' will be converted to '3-6/2' (every 2 days between Tuesday and Saturday) """ if "/" in expression_parts[i] and not any(exp in expression_parts[i] for exp in ['*', '-', ',']): choices = { 4: "12", 5: "6", 6: "9999" } step_range_through = choices.get(i) if step_range_through is not None: parts = expression_parts[i].split('/') expression_parts[i] = "{0}-{1}/{2}".format(parts[0], step_range_through, parts[1])