expressions.py 6.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198
  1. """This module contains the expressions applicable for CronTrigger's fields."""
  2. from calendar import monthrange
  3. import re
  4. from apscheduler.util import asint
  5. __all__ = ('AllExpression', 'RangeExpression', 'WeekdayRangeExpression',
  6. 'WeekdayPositionExpression', 'LastDayOfMonthExpression')
  7. WEEKDAYS = ['mon', 'tue', 'wed', 'thu', 'fri', 'sat', 'sun']
  8. class AllExpression(object):
  9. value_re = re.compile(r'\*(?:/(?P<step>\d+))?$')
  10. def __init__(self, step=None):
  11. self.step = asint(step)
  12. if self.step == 0:
  13. raise ValueError('Increment must be higher than 0')
  14. def get_next_value(self, date, field):
  15. start = field.get_value(date)
  16. minval = field.get_min(date)
  17. maxval = field.get_max(date)
  18. start = max(start, minval)
  19. if not self.step:
  20. next = start
  21. else:
  22. distance_to_next = (self.step - (start - minval)) % self.step
  23. next = start + distance_to_next
  24. if next <= maxval:
  25. return next
  26. def __eq__(self, other):
  27. return isinstance(other, self.__class__) and self.step == other.step
  28. def __str__(self):
  29. if self.step:
  30. return '*/%d' % self.step
  31. return '*'
  32. def __repr__(self):
  33. return "%s(%s)" % (self.__class__.__name__, self.step)
  34. class RangeExpression(AllExpression):
  35. value_re = re.compile(
  36. r'(?P<first>\d+)(?:-(?P<last>\d+))?(?:/(?P<step>\d+))?$')
  37. def __init__(self, first, last=None, step=None):
  38. AllExpression.__init__(self, step)
  39. first = asint(first)
  40. last = asint(last)
  41. if last is None and step is None:
  42. last = first
  43. if last is not None and first > last:
  44. raise ValueError('The minimum value in a range must not be higher than the maximum')
  45. self.first = first
  46. self.last = last
  47. def get_next_value(self, date, field):
  48. start = field.get_value(date)
  49. minval = field.get_min(date)
  50. maxval = field.get_max(date)
  51. # Apply range limits
  52. minval = min(maxval, max(minval, self.first))
  53. if self.last is not None:
  54. maxval = min(maxval, self.last)
  55. start = max(start, minval)
  56. if not self.step:
  57. next = start
  58. else:
  59. distance_to_next = (self.step - (start - minval)) % self.step
  60. next = start + distance_to_next
  61. if next <= maxval:
  62. return next
  63. def __eq__(self, other):
  64. return (isinstance(other, self.__class__) and self.first == other.first and
  65. self.last == other.last)
  66. def __str__(self):
  67. if self.last != self.first and self.last is not None:
  68. range = '%d-%d' % (self.first, self.last)
  69. else:
  70. range = str(self.first)
  71. if self.step:
  72. return '%s/%d' % (range, self.step)
  73. return range
  74. def __repr__(self):
  75. args = [str(self.first)]
  76. if self.last != self.first and self.last is not None or self.step:
  77. args.append(str(self.last))
  78. if self.step:
  79. args.append(str(self.step))
  80. return "%s(%s)" % (self.__class__.__name__, ', '.join(args))
  81. class WeekdayRangeExpression(RangeExpression):
  82. value_re = re.compile(r'(?P<first>[a-z]+)(?:-(?P<last>[a-z]+))?', re.IGNORECASE)
  83. def __init__(self, first, last=None):
  84. try:
  85. first_num = WEEKDAYS.index(first.lower())
  86. except ValueError:
  87. raise ValueError('Invalid weekday name "%s"' % first)
  88. if last:
  89. try:
  90. last_num = WEEKDAYS.index(last.lower())
  91. except ValueError:
  92. raise ValueError('Invalid weekday name "%s"' % last)
  93. else:
  94. last_num = None
  95. RangeExpression.__init__(self, first_num, last_num)
  96. def __str__(self):
  97. if self.last != self.first and self.last is not None:
  98. return '%s-%s' % (WEEKDAYS[self.first], WEEKDAYS[self.last])
  99. return WEEKDAYS[self.first]
  100. def __repr__(self):
  101. args = ["'%s'" % WEEKDAYS[self.first]]
  102. if self.last != self.first and self.last is not None:
  103. args.append("'%s'" % WEEKDAYS[self.last])
  104. return "%s(%s)" % (self.__class__.__name__, ', '.join(args))
  105. class WeekdayPositionExpression(AllExpression):
  106. options = ['1st', '2nd', '3rd', '4th', '5th', 'last']
  107. value_re = re.compile(r'(?P<option_name>%s) +(?P<weekday_name>(?:\d+|\w+))' %
  108. '|'.join(options), re.IGNORECASE)
  109. def __init__(self, option_name, weekday_name):
  110. try:
  111. self.option_num = self.options.index(option_name.lower())
  112. except ValueError:
  113. raise ValueError('Invalid weekday position "%s"' % option_name)
  114. try:
  115. self.weekday = WEEKDAYS.index(weekday_name.lower())
  116. except ValueError:
  117. raise ValueError('Invalid weekday name "%s"' % weekday_name)
  118. def get_next_value(self, date, field):
  119. # Figure out the weekday of the month's first day and the number of days in that month
  120. first_day_wday, last_day = monthrange(date.year, date.month)
  121. # Calculate which day of the month is the first of the target weekdays
  122. first_hit_day = self.weekday - first_day_wday + 1
  123. if first_hit_day <= 0:
  124. first_hit_day += 7
  125. # Calculate what day of the month the target weekday would be
  126. if self.option_num < 5:
  127. target_day = first_hit_day + self.option_num * 7
  128. else:
  129. target_day = first_hit_day + ((last_day - first_hit_day) / 7) * 7
  130. if target_day <= last_day and target_day >= date.day:
  131. return target_day
  132. def __eq__(self, other):
  133. return (super(WeekdayPositionExpression, self).__eq__(other) and
  134. self.option_num == other.option_num and self.weekday == other.weekday)
  135. def __str__(self):
  136. return '%s %s' % (self.options[self.option_num], WEEKDAYS[self.weekday])
  137. def __repr__(self):
  138. return "%s('%s', '%s')" % (self.__class__.__name__, self.options[self.option_num],
  139. WEEKDAYS[self.weekday])
  140. class LastDayOfMonthExpression(AllExpression):
  141. value_re = re.compile(r'last', re.IGNORECASE)
  142. def __init__(self):
  143. pass
  144. def get_next_value(self, date, field):
  145. return monthrange(date.year, date.month)[1]
  146. def __str__(self):
  147. return 'last'
  148. def __repr__(self):
  149. return "%s()" % self.__class__.__name__