Something went wrong. Try again.
forked niri
Something went wrong. Try again.
Python
1234567891011121314151617181920212223242526272829303132333435363738394041424344#!/usr/bin/env python3import timeimport reimport osimport urllib.requestimport urllib.error
CONF = os.environ.get("NIRI_HEARTBEAT_CONF", "/home/regent/Developer/niri/home/heartbeat.conf")SERVER = os.environ.get("NIRI_SERVER_URL", "http://localhost:3000").rstrip("/")AGENT = os.environ.get("NIRI_AGENT_ID", "niri")TRIGGER = f"{SERVER}/agents/{AGENT}/trigger/cron"DEFAULT_INTERVAL = 100
def read_interval() -> int: try: with open(CONF) as f: content = f.read() match = re.search(r"^\d+$", content, re.MULTILINE) if match: return int(match.group()) except OSError: pass return DEFAULT_INTERVAL
def trigger() -> bool: try: req = urllib.request.Request(TRIGGER, method="POST") with urllib.request.urlopen(req) as resp: return resp.status < 400 except urllib.error.URLError: return False
if __name__ == "__main__": print("Heartbeat started. Press Ctrl+C to stop.") while True: interval = read_interval() success = trigger() status = "OK" if success else "FAILED" print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Trigger {status} — next in {interval}m") time.sleep(interval * 60)