"""Simple cron scheduler for recurring probe execution.
Uses a minimal cron parser — no APScheduler dependency needed.
Runs as a blocking loop, suitable for Docker or systemd.
Usage:
metapod schedule --cron "0 6 * * *" "oracle.metadata.*"
"""
from __future__ import annotations
import logging
import time
from datetime import datetime
logger = logging.getLogger("metapod.scheduler")
[docs]
def matches_cron(cron_expr: str, dt: datetime) -> bool:
"""Check if a datetime matches a cron expression (minute hour day month weekday)."""
parts = cron_expr.strip().split()
if len(parts) != 5:
return False
fields = [dt.minute, dt.hour, dt.day, dt.month, dt.weekday()]
ranges = [
(0, 59), # minute
(0, 23), # hour
(1, 31), # day
(1, 12), # month
(0, 6), # weekday (0=Monday in Python, adjust for cron 0=Sunday)
]
for i, (part, value, (lo, hi)) in enumerate(zip(parts, fields, ranges)):
if part == "*":
continue
if "/" in part:
base, step = part.split("/")
base_val = lo if base == "*" else int(base)
if (value - base_val) % int(step) != 0:
return False
continue
if "," in part:
if value not in [int(v) for v in part.split(",")]:
return False
continue
if "-" in part:
a, b = part.split("-")
if value < int(a) or value > int(b):
return False
continue
if value != int(part):
return False
return True
[docs]
def run_schedule(cron_expr: str, callback, check_interval: int = 30):
"""Run callback when cron expression matches. Blocks forever.
Args:
cron_expr: Cron expression (minute hour day month weekday)
callback: Function to call (no args)
check_interval: Seconds between checks (default 30)
"""
last_run_minute = -1
while True:
now = datetime.now()
current_minute = now.hour * 60 + now.minute
if current_minute != last_run_minute and matches_cron(cron_expr, now):
last_run_minute = current_minute
try:
callback()
except Exception as e:
logger.error("Schedule error: %s", e)
time.sleep(check_interval)