Add devterm tools and TUI menu
Some checks failed
CI / test (3.11) (push) Has been cancelled
CI / test (3.12) (push) Has been cancelled
CI / test (3.8) (push) Has been cancelled
CI / test (3.9) (push) Has been cancelled
CI / lint (push) Has been cancelled
CI / typecheck (push) Has been cancelled
CI / build-package (push) Has been cancelled
CI / test (3.10) (push) Has been cancelled
Some checks failed
CI / test (3.11) (push) Has been cancelled
CI / test (3.12) (push) Has been cancelled
CI / test (3.8) (push) Has been cancelled
CI / test (3.9) (push) Has been cancelled
CI / lint (push) Has been cancelled
CI / typecheck (push) Has been cancelled
CI / build-package (push) Has been cancelled
CI / test (3.10) (push) Has been cancelled
This commit is contained in:
98
devterm/tools/cron_tool.py
Normal file
98
devterm/tools/cron_tool.py
Normal file
@@ -0,0 +1,98 @@
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
|
||||
def validate_cron(cron_expr: str) -> bool:
|
||||
parts = cron_expr.strip().split()
|
||||
if len(parts) not in [5, 6]:
|
||||
return False
|
||||
|
||||
try:
|
||||
minute, hour, day, month, dow = parts[:5]
|
||||
_validate_cron_field(minute, 0, 59)
|
||||
_validate_cron_field(hour, 0, 23)
|
||||
_validate_cron_field(day, 1, 31)
|
||||
_validate_cron_field(month, 1, 12)
|
||||
_validate_cron_field(dow, 0, 7)
|
||||
return True
|
||||
except (ValueError, IndexError):
|
||||
return False
|
||||
|
||||
|
||||
def _validate_cron_field(value: str, min_val: int, max_val: int) -> bool:
|
||||
if value == "*":
|
||||
return True
|
||||
if "/" in value:
|
||||
value, step = value.split("/", 1)
|
||||
step = int(step)
|
||||
if "-" in value:
|
||||
start, end = value.split("-", 1)
|
||||
if not (min_val <= int(start) <= max_val and min_val <= int(end) <= max_val):
|
||||
raise ValueError
|
||||
return True
|
||||
if "," in value:
|
||||
for v in value.split(","):
|
||||
if not _validate_cron_field(v, min_val, max_val):
|
||||
raise ValueError
|
||||
return True
|
||||
int(value)
|
||||
return True
|
||||
|
||||
|
||||
def get_next_run(cron_expr: str, current_time: datetime = None) -> str:
|
||||
if current_time is None:
|
||||
current_time = datetime.now()
|
||||
|
||||
parts = cron_expr.strip().split()
|
||||
minute, hour, day, month, dow = parts[:5]
|
||||
second = parts[5] if len(parts) == 6 else "0"
|
||||
|
||||
next_time = current_time.replace(second=int(second), microsecond=0)
|
||||
next_time += timedelta(minutes=1)
|
||||
|
||||
max_iterations = 366 * 24 * 60
|
||||
for _ in range(max_iterations):
|
||||
if _matches_cron(next_time, minute, hour, day, month, dow):
|
||||
return next_time.strftime("%Y-%m-%d %H:%M:%S")
|
||||
next_time += timedelta(minutes=1)
|
||||
|
||||
return "Could not calculate next run"
|
||||
|
||||
|
||||
def _matches_cron(dt: datetime, minute: str, hour: str, day: str, month: str, dow: str) -> bool:
|
||||
return (
|
||||
_matches_field(dt.minute, minute) and
|
||||
_matches_field(dt.hour, hour) and
|
||||
_matches_field(dt.day, day) and
|
||||
_matches_field(dt.month, month) and
|
||||
_matches_dow(dt.weekday(), dow)
|
||||
)
|
||||
|
||||
|
||||
def _matches_field(value: int, pattern: str) -> bool:
|
||||
if pattern == "*":
|
||||
return True
|
||||
if "/" in pattern:
|
||||
pattern, step = pattern.split("/", 1)
|
||||
if pattern == "*":
|
||||
return value % int(step) == 0
|
||||
if "-" in pattern:
|
||||
start, end = pattern.split("-", 1)
|
||||
return int(start) <= value <= int(end)
|
||||
if "," in pattern:
|
||||
return str(value) in pattern.split(",")
|
||||
return value == int(pattern)
|
||||
|
||||
|
||||
def _matches_dow(weekday: int, pattern: str) -> bool:
|
||||
if pattern == "*":
|
||||
return True
|
||||
if "/" in pattern:
|
||||
pattern, step = pattern.split("/", 1)
|
||||
if pattern == "*":
|
||||
return weekday % int(step) == 0
|
||||
if "-" in pattern:
|
||||
start, end = pattern.split("-", 1)
|
||||
return int(start) <= weekday <= int(end)
|
||||
if "," in pattern:
|
||||
return str(weekday) in pattern.split(",")
|
||||
return weekday == int(pattern)
|
||||
Reference in New Issue
Block a user