#!/usr/bin/env python3 """ ZIPAIR Notification Monitor Calls the ZIPAIR BFF API via FlareSolverr, compares notification IDs against last_seen state, and fires an ntfy push for each new entry. State is persisted in last_seen.txt (committed back to repo by the workflow). """ import os import re import sys import json import urllib.request import urllib.error from datetime import datetime, timezone # ── Config ─────────────────────────────────────────────────────────────────── NTFY_URL = os.environ.get("NTFY_URL") or "https://ntfy.isky-homelab.com/zipair" NTFY_TOKEN = os.environ.get("NTFY_TOKEN", "") STATE_FILE = os.environ.get("STATE_FILE", "last_seen.txt") FLARESOLVERR_URL = os.environ.get("FLARESOLVERR_URL", "http://192.168.10.76:8191") BFF_API = "https://bff.zipair.net/v1/information" NOTIF_BASE = "https://www.zipair.net/en/notification" # ── FlareSolverr fetch ─────────────────────────────────────────────────────── def fs_fetch(url: str, timeout_ms: int = 60000) -> str: """Fetch a URL via FlareSolverr and return the response body, or '' on error.""" payload = json.dumps({ "cmd": "request.get", "url": url, "maxTimeout": timeout_ms, }).encode() req = urllib.request.Request( f"{FLARESOLVERR_URL.rstrip('/')}/v1", data=payload, headers={"Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(req, timeout=timeout_ms // 1000 + 10) as resp: data = json.loads(resp.read()) http_status = data.get("solution", {}).get("status", 0) body = data.get("solution", {}).get("response", "") print(f" FlareSolverr: {data.get('status')} | HTTP {http_status} | {len(body)} bytes") return body except Exception as e: print(f" FlareSolverr error for {url}: {e}", file=sys.stderr) return "" def fs_fetch_json(url: str) -> dict | None: """Fetch a JSON API via FlareSolverr. Chromium wraps JSON in 
, so strip that."""
    body = fs_fetch(url)
    if not body:
        return None
    # Chromium renders raw JSON as 
{...}

    m = re.search(r"]*>(.*)
", body, re.DOTALL) raw = m.group(1) if m else body try: return json.loads(raw) except Exception as e: print(f" JSON parse error: {e}", file=sys.stderr) return None # ── State ──────────────────────────────────────────────────────────────────── def read_seen_ids() -> set: try: with open(STATE_FILE) as f: data = json.loads(f.read().strip()) if isinstance(data, list): return set(data) except Exception: pass return set() def write_seen_ids(ids: set): with open(STATE_FILE, "w") as f: json.dump(sorted(ids), f) print(f"State updated: {len(ids)} notification ID(s) tracked.") # ── ntfy ───────────────────────────────────────────────────────────────────── def send_ntfy(notif: dict): notif_id = notif["id"] title = notif.get("title", f"Notification #{notif_id}") url = f"{NOTIF_BASE}/{notif_id}" payload = json.dumps({ "topic": NTFY_URL.rstrip("/").rsplit("/", 1)[-1], "title": f"✈️ New ZIPAIR Notification", "message": title, "priority": 5, "tags": ["airplane", "moneybag"], "click": url, "actions": [{"action": "view", "label": "Open ZIPAIR", "url": url}], }).encode() base_url = NTFY_URL.rstrip("/").rsplit("/", 1)[0] req = urllib.request.Request( f"{base_url}/", data=payload, headers={"Content-Type": "application/json"}, method="POST", ) if NTFY_TOKEN: req.add_header("Authorization", f"Bearer {NTFY_TOKEN}") try: with urllib.request.urlopen(req, timeout=10) as resp: print(f" ntfy: {resp.status} {resp.reason}") except Exception as e: print(f" Failed to send ntfy: {e}", file=sys.stderr) sys.exit(1) # ── Main ───────────────────────────────────────────────────────────────────── def main(): print(f"\n[{datetime.now(timezone.utc).isoformat()}] ZIPAIR monitor starting ...") print(f" ntfy URL : {NTFY_URL}") print(f" FlareSolverr : {FLARESOLVERR_URL}") seen_ids = read_seen_ids() print(f" Known IDs : {len(seen_ids)}") api_url = f"{BFF_API}?language=en&page=1" print(f"Fetching {api_url} via FlareSolverr ...") data = fs_fetch_json(api_url) if not data or "information" not in data: print("Could not retrieve notification list; exiting.") sys.exit(0) notifications = data["information"] total = data.get("informationTotal", len(notifications)) all_ids = {n["id"] for n in notifications} new_notifications = [n for n in notifications if n["id"] not in seen_ids] print(f" API total: {total} | Page 1: {len(notifications)} | New: {len(new_notifications)}") if not new_notifications: print("No new notifications. All good.") write_seen_ids(all_ids | seen_ids) sys.exit(0) # First run: no prior state — seed without alerting to avoid a flood if not seen_ids: print(f"First run: seeding state with {len(all_ids)} notification ID(s), no alerts sent.") write_seen_ids(all_ids) sys.exit(0) for notif in new_notifications: print(f" NEW: [{notif['id']}] {notif['title']} -- sending ntfy ...") send_ntfy(notif) write_seen_ids(all_ids | seen_ids) print("Done.") sys.exit(0) if __name__ == "__main__": main()