120 lines
3.1 KiB
Python
120 lines
3.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Call a URL on a cron schedule."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
import time
|
|
from datetime import datetime
|
|
from typing import Any
|
|
from urllib.error import URLError
|
|
from urllib.request import Request, urlopen
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from croniter import croniter
|
|
|
|
CRON_MACROS = {
|
|
"@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 * * * *",
|
|
}
|
|
|
|
|
|
def log(msg: str) -> None:
|
|
print(f"[{datetime.now().astimezone().isoformat(timespec='seconds')}] {msg}", file=sys.stderr)
|
|
|
|
|
|
def zone() -> ZoneInfo:
|
|
name = os.environ.get("TZ") or "UTC"
|
|
try:
|
|
return ZoneInfo(name)
|
|
except Exception as exc:
|
|
raise SystemExit(f"Invalid TZ '{name}': {exc}") from exc
|
|
|
|
|
|
def resolve_cron_expr() -> str:
|
|
cron = os.environ.get("CRON", "").strip()
|
|
if not cron:
|
|
raise SystemExit("CRON must be set (e.g. '0 3 * * *')")
|
|
|
|
expr = CRON_MACROS.get(cron.lower(), cron)
|
|
if not croniter.is_valid(expr):
|
|
raise SystemExit(f"Invalid CRON '{expr}' (expected a 5-field cron expression)")
|
|
return expr
|
|
|
|
|
|
def resolve_url() -> str:
|
|
url = os.environ.get("URL", "").strip()
|
|
if not url:
|
|
raise SystemExit("URL must be set")
|
|
return url
|
|
|
|
|
|
def timeout_seconds() -> int:
|
|
raw = os.environ.get("TIMEOUT", "30").strip()
|
|
try:
|
|
value = int(raw)
|
|
except ValueError as exc:
|
|
raise SystemExit(f"Invalid TIMEOUT '{raw}' (expected integer)") from exc
|
|
if value <= 0:
|
|
raise SystemExit("Invalid TIMEOUT (expected a positive integer)")
|
|
return value
|
|
|
|
|
|
def run_on_startup() -> bool:
|
|
return os.environ.get("ON_STARTUP", "true").lower() in ("1", "true", "yes", "on")
|
|
|
|
|
|
def next_run(expr: str, after: datetime) -> datetime:
|
|
return croniter(expr, after).get_next(datetime)
|
|
|
|
|
|
def do_request(url: str, timeout: int) -> None:
|
|
req = Request(url, method="GET")
|
|
try:
|
|
with urlopen(req, timeout=timeout) as response:
|
|
log(f"GET {url} -> HTTP {response.status}")
|
|
except URLError as exc:
|
|
reason = exc.reason if hasattr(exc, "reason") else str(exc)
|
|
log(f"GET {url} failed: {reason}")
|
|
|
|
|
|
def sleep_until(target: datetime, tz: ZoneInfo) -> None:
|
|
while True:
|
|
remaining = (target - datetime.now(tz)).total_seconds()
|
|
if remaining <= 0:
|
|
return
|
|
time.sleep(min(remaining, 60.0))
|
|
|
|
|
|
def main() -> None:
|
|
url = resolve_url()
|
|
expr = resolve_cron_expr()
|
|
tz = zone()
|
|
timeout = timeout_seconds()
|
|
on_startup = run_on_startup()
|
|
|
|
log(f"Starting cron-url-caller (TZ={tz.key}, CRON='{expr}', URL='{url}', TIMEOUT={timeout}s)")
|
|
|
|
if on_startup:
|
|
do_request(url, timeout)
|
|
|
|
now = datetime.now(tz)
|
|
nxt = next_run(expr, now)
|
|
log(f"Next scheduled call at {nxt.isoformat(timespec='seconds')}")
|
|
|
|
while True:
|
|
sleep_until(nxt, tz)
|
|
do_request(url, timeout)
|
|
nxt = next_run(expr, datetime.now(tz))
|
|
log(f"Next scheduled call at {nxt.isoformat(timespec='seconds')}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|