Book a Demo!
CoCalc Logo Icon
StoreFeaturesDocsShareSupportNewsAboutPoliciesSign UpSign In
Der-Henning
GitHub Repository: Der-Henning/tgtg
Path: blob/main/tgtg_scanner/models/cron.py
725 views
1
import logging
2
from typing import Union
3
4
import pycron
5
from cron_descriptor import Options, get_description
6
7
log = logging.getLogger("tgtg")
8
9
10
class Cron:
11
def __init__(self, cron_str: Union[str, None] = None) -> None:
12
self.crons = list(dict.fromkeys([cron.strip() for cron in cron_str.split(";")])) if cron_str else ["* * * * *"]
13
self.options = Options()
14
self.options.use_24hour_time_format = True
15
self.options.day_of_week_start_index_zero = True
16
try:
17
self.is_now
18
except ValueError as err:
19
raise ValueError(f"Cron expression parsing error - {err}") from err
20
for cron in self.crons:
21
_, _, _, _, dow = cron.split()
22
if any(int(day) > 6 for day in dow.split("-") if day.isdigit()):
23
raise ValueError("Cron expression parsing error - day of week must be between 0 and 6 (Sunday=0)")
24
25
@property
26
def is_now(self) -> bool:
27
"""Returns True if the cron expression matches the current time"""
28
return any(pycron.is_now(cron) for cron in self.crons)
29
30
def get_description(self, locale: str = "en") -> str:
31
"""Returns a human-readable description of the cron expression"""
32
self.options.locale_code = locale
33
return "; ".join(get_description(cron, options=self.options) for cron in self.crons)
34
35
def __eq__(self, __o: object) -> bool:
36
return getattr(__o, "crons") == self.crons
37
38
def __repr__(self) -> str:
39
return f"Cron({self.crons})"
40
41