~yerinalexey/pcrond

ref: 1f29ae5fafdbbc2a45f0fabe4307dcfe297f8a10 pcrond/pcrond/job.py -rw-r--r-- 6.0 KiB
1f29ae5f — Alexey Yerin refactor: completely remove logging 7 months ago
                                                                                
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
from datetime import datetime

reboot_time = datetime.now()

ALIASES = {'@yearly':    '0 0 1 1 *',
           '@annually':  '0 0 1 1 *',
           '@monthly':   '0 0 1 * *',
           '@weekly':    '0 0 * * 0',
           '@daily':     '0 0 * * *',
           '@midnight':  '0 0 * * *',
           '@hourly':    '0 * * * *',
           '@reboot':    '%d %d %d %d * %d' % (reboot_time.minute,
                                               reboot_time.hour,
                                               reboot_time.day,
                                               reboot_time.month,
                                               reboot_time.year)
           }


class Job(object):
    """
    A periodic job as used by :class:`Scheduler`.
    """
    def __init__(self, crontab=None, job_func=None, scheduler=None):
        """
        Constructor
        :param crontab:
            string containing crontab pattern
            Its tokens may be either: 1 (if alias), 5 (without year token),
            6 (with year token)
            if None, you should set it later
        :param job_func:
            the job 0-ary function to run
            if None, you should set it later
        :param scheduler:
            scheduler to register with
            if None, you should set it later
        """
        self.job_func = job_func
        self.scheduler = scheduler
        self.running = False
        if crontab is not None:
            self.set_crontab(crontab)

    def set_crontab(self, crontab):
        if crontab is None:
            raise ValueError("given None crontab")

        crontab = crontab.lower().strip()

        if crontab in ALIASES.keys():
            crontab = ALIASES[crontab]

        crontab_lst = crontab.split()

        if len(crontab_lst) == 5:
            crontab_lst.append("*")
        if len(crontab_lst) != 6:
            raise ValueError(
                "Each crontab pattern *must* contain either 1, 5 or 6 items")

        from .cronparser import Parser
        parser = Parser()

        # Easy ones:
        [self.allowed_every_min, self.allowed_min] = parser.parse_minute(crontab_lst[0])
        [self.allowed_every_hour, self.allowed_hours] = parser.parse_hour(crontab_lst[1])
        [self.allowed_every_month, self.allowed_months] = parser.parse_month(crontab_lst[3])
        [self.allowed_every_year, self.allowed_years] = parser.parse_year(crontab_lst[5])

        # Day of month.
        # L = last day
        # 15W= nearest working day around the 15th, in the same month
        [self.allowed_every_dom, self.allowed_dom, self.allowed_last_dom, self.allowed_wdom] = \
            parser.parse_day_in_month(crontab_lst[2])

        # Day of week.
        # 5L = last friday of the month
        # 5#2 = second friday of the month
        [self.allowed_every_dow, self.allowed_dow, self.allowed_dowl, self.allowed_dow_sharp] = \
            parser.parse_day_in_week(crontab_lst[4])

        self.crontab_pattern = crontab_lst

    def get_last_dom(self, now):
        """ get last day in month determined by given datetime """
        import calendar
        last_day_of_month = calendar.monthrange(now.year, now.month)[1]
        return last_day_of_month

    def get_num_wom(self, now):
        """
        for a given date, return the number of the week in the month (1..5)
        intended for #
        """
        return ((now.day - 1) // 7) + 1

    def is_last_wom(self, now):
        """ true if given date is in the last week of the month """
        return now.day > self.get_last_dom(now) - 7

    def _check_w(self, now):
        """ used for checking 15w """
        w = now.weekday()
        if w >= 5:
            return False
        d = now.day
        if d in self.allowed_wdom:
            return True
        if w == 0:
            if (d - 1) in self.allowed_wdom:
                return True
            # 1w matches monday, 3rd
            if d == 3 and 1 in self.allowed_wdom:
                return True
        elif w == 4:
            if (d + 1) in self.allowed_wdom:
                return True
            # 31w matches friday, 29th
            lday = self.get_last_dom(now)
            if d == lday - 2 and lday in self.allowed_wdom:
                return True
        return False

    def should_run(self):
        """
        :return: ``True`` if the job should be run now.
        """
        now = datetime.now()
        return self._should_run_at(now)

    def _should_run_at(self, now):
        """
        :return: ``True`` if the job should be run at given datetime.
        """
        # warning: in Python, Monday is 0 and Sunday is 6
        #          in cron, Sunday=0
        w = now.weekday()
        num_wom = self.get_num_wom(now)
        return (not self.running
                and (self.allowed_every_year or now.year in self.allowed_years)
                and (self.allowed_every_month or now.month in self.allowed_months)
                and (self.allowed_every_hour or now.hour in self.allowed_hours)
                and (self.allowed_every_min or now.minute in self.allowed_min)
                and (self.allowed_every_dow
                     or (w in self.allowed_dow)
                     or (self.allowed_dowl
                         and w in self.allowed_dowl
                         and self.is_last_wom(now))
                     or (self.allowed_dow_sharp[num_wom]
                         and w in self.allowed_dow_sharp[num_wom]))
                and (self.allowed_every_dom
                     or now.day in self.allowed_dom
                     or (self.allowed_last_dom and now.day == self.get_last_dom(now))
                     or (self.allowed_wdom and self._check_w(now)))
                )

    def run(self):
        """
        Run the job.
        :return: The return value returned by the `job_func`
        """
        self.running = True
        ret = self.job_func()
        self.running = False
        return ret

    def run_if_should(self):
        """
        Run the job if needed.
        :return: The return value returned by the `job_func`
        """
        if self.should_run():
            return self.run()