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