cron_util.py 4.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173
  1. import re
  2. from datetime import datetime
  3. class CronUtil:
  4. """
  5. Cron表达式工具类
  6. """
  7. @classmethod
  8. def __valid_range(cls, search_str: str, start_range: int, end_range: int):
  9. match = re.match(r'^(\d+)-(\d+)$', search_str)
  10. if match:
  11. start, end = int(match.group(1)), int(match.group(2))
  12. return start_range <= start < end <= end_range
  13. return False
  14. @classmethod
  15. def __valid_sum(
  16. cls, search_str: str, start_range_a: int, start_range_b: int, end_range_a: int, end_range_b: int, sum_range: int
  17. ):
  18. match = re.match(r'^(\d+)/(\d+)$', search_str)
  19. if match:
  20. start, end = int(match.group(1)), int(match.group(2))
  21. return (
  22. start_range_a <= start <= start_range_b
  23. and end_range_a <= end <= end_range_b
  24. and start + end <= sum_range
  25. )
  26. return False
  27. @classmethod
  28. def validate_second_or_minute(cls, second_or_minute: str):
  29. """
  30. 校验秒或分钟值是否正确
  31. :param second_or_minute: 秒或分钟值
  32. :return: 校验结果
  33. """
  34. if (
  35. second_or_minute == '*'
  36. or ('-' in second_or_minute and cls.__valid_range(second_or_minute, 0, 59))
  37. or ('/' in second_or_minute and cls.__valid_sum(second_or_minute, 0, 58, 1, 59, 59))
  38. or re.match(r'^(?:[0-5]?\d|59)(?:,[0-5]?\d|59)*$', second_or_minute)
  39. ):
  40. return True
  41. return False
  42. @classmethod
  43. def validate_hour(cls, hour: str):
  44. """
  45. 校验小时值是否正确
  46. :param hour: 小时值
  47. :return: 校验结果
  48. """
  49. if (
  50. hour == '*'
  51. or ('-' in hour and cls.__valid_range(hour, 0, 23))
  52. or ('/' in hour and cls.__valid_sum(hour, 0, 22, 1, 23, 23))
  53. or re.match(r'^(?:0|[1-9]|1\d|2[0-3])(?:,(?:0|[1-9]|1\d|2[0-3]))*$', hour)
  54. ):
  55. return True
  56. return False
  57. @classmethod
  58. def validate_day(cls, day: str):
  59. """
  60. 校验日值是否正确
  61. :param day: 日值
  62. :return: 校验结果
  63. """
  64. if (
  65. day in ['*', '?', 'L']
  66. or ('-' in day and cls.__valid_range(day, 1, 31))
  67. or ('/' in day and cls.__valid_sum(day, 1, 30, 1, 30, 31))
  68. or ('W' in day and re.match(r'^(?:[1-9]|1\d|2\d|3[01])W$', day))
  69. or re.match(r'^(?:0|[1-9]|1\d|2[0-9]|3[0-1])(?:,(?:0|[1-9]|1\d|2[0-9]|3[0-1]))*$', day)
  70. ):
  71. return True
  72. return False
  73. @classmethod
  74. def validate_month(cls, month: str):
  75. """
  76. 校验月值是否正确
  77. :param month: 月值
  78. :return: 校验结果
  79. """
  80. if (
  81. month == '*'
  82. or ('-' in month and cls.__valid_range(month, 1, 12))
  83. or ('/' in month and cls.__valid_sum(month, 1, 11, 1, 11, 12))
  84. or re.match(r'^(?:0|[1-9]|1[0-2])(?:,(?:0|[1-9]|1[0-2]))*$', month)
  85. ):
  86. return True
  87. return False
  88. @classmethod
  89. def validate_week(cls, week: str):
  90. """
  91. 校验周值是否正确
  92. :param week: 周值
  93. :return: 校验结果
  94. """
  95. if (
  96. week in ['*', '?']
  97. or ('-' in week and cls.__valid_range(week, 1, 7))
  98. or ('#' in week and re.match(r'^[1-7]#[1-4]$', week))
  99. or ('L' in week and re.match(r'^[1-7]L$', week))
  100. or re.match(r'^[1-7](,[1-7])*$', week)
  101. ):
  102. return True
  103. return False
  104. @classmethod
  105. def validate_year(cls, year: str):
  106. """
  107. 校验年值是否正确
  108. :param year: 年值
  109. :return: 校验结果
  110. """
  111. current_year = int(datetime.now().year)
  112. future_years = [current_year + i for i in range(9)]
  113. if (
  114. year == '*'
  115. or ('-' in year and cls.__valid_range(year, current_year, 2099))
  116. or ('/' in year and cls.__valid_sum(year, current_year, 2098, 1, 2099 - current_year, 2099))
  117. or ('#' in year and re.match(r'^[1-7]#[1-4]$', year))
  118. or ('L' in year and re.match(r'^[1-7]L$', year))
  119. or (
  120. (len(year) == 4 or ',' in year)
  121. and all(int(item) in future_years and current_year <= int(item) <= 2099 for item in year.split(','))
  122. )
  123. ):
  124. return True
  125. return False
  126. @classmethod
  127. def validate_cron_expression(cls, cron_expression: str):
  128. """
  129. 校验Cron表达式是否正确
  130. :param cron_expression: Cron表达式
  131. :return: 校验结果
  132. """
  133. values = cron_expression.split()
  134. if len(values) != 6 and len(values) != 7:
  135. return False
  136. second_validation = cls.validate_second_or_minute(values[0])
  137. minute_validation = cls.validate_second_or_minute(values[1])
  138. hour_validation = cls.validate_hour(values[2])
  139. day_validation = cls.validate_day(values[3])
  140. month_validation = cls.validate_month(values[4])
  141. week_validation = cls.validate_week(values[5])
  142. validation = (
  143. second_validation
  144. and minute_validation
  145. and hour_validation
  146. and day_validation
  147. and month_validation
  148. and week_validation
  149. )
  150. if len(values) == 6:
  151. return validation
  152. if len(values) == 7:
  153. year_validation = cls.validate_year(values[6])
  154. return validation and year_validation
  155. return None