#!/usr/bin/env python3 """Copy your Healthchecks.io checks into CronPup (Python 3.8+, no packages). export HC_API_KEY=... # Healthchecks.io: Project Settings -> API Access (read-only key is fine) export CRONPUP_API_KEY=cp_... # CronPup: Settings -> Create API key python3 cronpup-import-healthchecks.py # dry run: shows what would be created python3 cronpup-import-healthchecks.py --apply # create monitors + write a URL mapping CSV Copies name, schedule (period or cron + timezone), grace and tags. Checks using OnCalendar schedules are listed and skipped (CronPup supports 5-field cron). Re-running is safe: monitors are matched by slug. Nothing in your Healthchecks.io account is changed. Afterwards, replace each old ping URL with the new one from the CSV. Options: --hc-base URL (self-hosted Healthchecks), --base URL (CronPup), --out FILE (default cronpup-import.csv) """ import argparse import csv import hashlib import json import os import re import sys import urllib.request CRON_RE = re.compile(r"^(@(yearly|annually|monthly|weekly|daily|midnight|hourly)|(\S+\s+){4}\S+)$") SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,62}$") def get_json(url, headers): req = urllib.request.Request(url, headers={**headers, "User-Agent": "cronpup-import/1.0"}) with urllib.request.urlopen(req, timeout=30) as r: return json.loads(r.read()) class ApiError(Exception): pass def tb(base, key, method, path, body=None): req = urllib.request.Request(base + path, method=method, data=json.dumps(body).encode() if body else None, headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json", "User-Agent": "cronpup-import/1.0"}) try: with urllib.request.urlopen(req, timeout=30) as r: return json.loads(r.read()) except urllib.error.HTTPError as e: try: msg = json.loads(e.read()).get("error", "") except Exception: # noqa: BLE001 msg = "" raise ApiError(f"CronPup API error {e.code}: {msg}") from None def to_minutes(seconds, lo=1): return max(lo, round(int(seconds or 0) / 60)) def slug_for(check): s = (check.get("slug") or "").strip().lower() if SLUG_RE.match(s) and s not in ("start", "fail"): return s base = re.sub(r"[^a-z0-9_-]+", "-", (check.get("name") or "check").lower()).strip("-")[:50] or "check" ident = check.get("uuid") or check.get("unique_key") or check.get("name") or "" return f"{base}-{hashlib.sha1(ident.encode()).hexdigest()[:6]}" def plan_for(check): """Return (body, problem) for one Healthchecks check.""" name = (check.get("name") or "").strip() or "unnamed check" body = {"name": name[:80], "grace": to_minutes(check.get("grace"), 1) * 60, "tags": [t for t in (check.get("tags") or "").split() if t], "slug": slug_for(check)} sched = (check.get("schedule") or "").strip() if sched: if not CRON_RE.match(sched): return None, f"OnCalendar schedule '{sched}' isn't supported; add this one by hand" body.update(cron=sched, tz=check.get("tz") or "UTC") else: minutes = min(max(to_minutes(check.get("timeout"), 1), 1), 90 * 1440) body["period"] = minutes * 60 return body, None def main(): ap = argparse.ArgumentParser() ap.add_argument("--apply", action="store_true") ap.add_argument("--hc-base", default=os.environ.get("HC_BASE", "https://healthchecks.io")) ap.add_argument("--base", default=os.environ.get("CRONPUP_URL", os.environ.get("TARDYBELL_URL", "https://cronpup.com"))) ap.add_argument("--out", default="cronpup-import.csv") a = ap.parse_args() hc_key = os.environ.get("HC_API_KEY") if not hc_key: raise SystemExit("set HC_API_KEY (Healthchecks.io: Project Settings -> API Access)") checks = get_json(a.hc_base.rstrip("/") + "/api/v3/checks/", {"X-Api-Key": hc_key}).get("checks", []) plans, used = [], set() for c in checks: body, problem = plan_for(c) if problem: print(f"skip {c.get('name')}: {problem}") continue slug, n = body["slug"], 2 while body["slug"] in used: # two checks sharing a slug get distinct ones body["slug"] = f"{slug[:58]}-{n}" n += 1 used.add(body["slug"]) plans.append((c, body)) sched = body.get("cron") and f"cron {body['cron']} ({body['tz']})" or f"every {body['period'] // 60} min" print(f"plan {body['name']:30} {sched:28} grace {body['grace'] // 60} min tags {','.join(body['tags'])}") if not a.apply: print(f"\nDry run: {len(plans)} monitor(s) would be created. Re-run with --apply.") return 0 key = os.environ.get("CRONPUP_API_KEY") or os.environ.get("TARDYBELL_API_KEY") if not key: raise SystemExit("set CRONPUP_API_KEY (CronPup: Settings -> Create API key)") base = a.base.rstrip("/") try: existing = {m.get("slug"): m for m in tb(base, key, "GET", "/api/v1/monitors")["monitors"] if m.get("slug")} except ApiError as ex: raise SystemExit(str(ex)) from None rows, failed = [], 0 for c, body in plans: prior = existing.get(body["slug"]) if prior and prior.get("name") == body["name"]: # a previous import of this same check m, verb = prior, "reused" else: if prior: # an unrelated monitor already uses this slug: pick another body["slug"] = f"{body['slug'][:54]}-hc{hashlib.sha1(body['name'].encode()).hexdigest()[:4]}" try: m, verb = tb(base, key, "POST", "/api/v1/monitors", body), "created" except ApiError as ex: print(f"error {body['name']}: {ex} (continuing)") failed += 1 continue rows.append({"name": body["name"], "old_ping_url": c.get("ping_url", "(read-only key: see Healthchecks)"), "new_ping_url": m["ping_url"]}) print(f"{verb} {body['name']}: {m['ping_url']}") if failed: print(f"\n{failed} check(s) could not be created (see errors above); fix them and re-run. Others are safe to re-run.") with open(a.out, "w", newline="") as f: w = csv.DictWriter(f, fieldnames=["name", "old_ping_url", "new_ping_url"]) w.writeheader() w.writerows(rows) print(f"\nWrote {a.out}. Swap each old ping URL for the new one; both services can run side by side meanwhile.") return 0 if __name__ == "__main__": sys.exit(main())