isky
2ce66aba10
All checks were successful
ZIPAIR Singapore Sale Monitor / check (push) Successful in 22s
Replace HTML scraping with a direct call to bff.zipair.net/v1/information via FlareSolverr. The BFF returns clean JSON (id, title, category, publishedAt) — no regex parsing of rendered pages. Also adds first-run seeding to avoid alerting on all historical notifications. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
169 lines
6.2 KiB
Python
169 lines
6.2 KiB
Python
#!/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 <pre>, so strip that."""
|
|
body = fs_fetch(url)
|
|
if not body:
|
|
return None
|
|
# Chromium renders raw JSON as <html><body><pre>{...}</pre></body></html>
|
|
m = re.search(r"<pre[^>]*>(.*)</pre>", 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()
|