feat: initial ZIPAIR Singapore sale monitor

Polls ZIPAIR sitemap every 10 min via Gitea Actions, detects new
Singapore/winter notifications, and fires an ntfy push alert.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

.gitea/workflows/zipair-monitor.yml

@@ -0,0 +1,41 @@
+name: ZIPAIR Singapore Sale Monitor
+
+on:
+  schedule:
+    # Runs every 10 minutes (cron is UTC)
+    - cron: '*/10 * * * *'
+  workflow_dispatch:   # allow manual trigger from Gitea UI
+
+jobs:
+  check:
+    runs-on: ubuntu-latest   # or your self-hosted runner label
+
+    steps:
+      - name: Checkout repo
+        uses: actions/checkout@v4
+        with:
+          # Need write access to commit state file back
+          token: ${{ secrets.GITEA_TOKEN }}
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.11'
+
+      # No pip deps needed — script uses stdlib only
+
+      - name: Run ZIPAIR monitor
+        env:
+          NTFY_URL:   ${{ secrets.NTFY_URL }}
+          NTFY_TOKEN: ${{ secrets.NTFY_TOKEN }}
+          STATE_FILE: last_seen.txt
+        run: python check_zipair.py
+
+      - name: Commit updated state
+        run: |
+          git config user.name  "zipair-bot"
+          git config user.email "zipair-bot@localhost"
+          git add last_seen.txt
+          # Only commit if the file actually changed
+          git diff --cached --quiet || git commit -m "chore: update last_seen to $(cat last_seen.txt)"
+          git push

check_zipair.py

@@ -0,0 +1,202 @@
+#!/usr/bin/env python3
+"""
+ZIPAIR Singapore Winter Sale Monitor
+Checks ZIPAIR's sitemap for new notifications about Singapore ticket sales.
+Sends a push notification via ntfy when detected.
+
+State is persisted in last_seen.txt (committed back to repo by the workflow).
+"""
+
+import os
+import re
+import sys
+import json
+import time
+import urllib.request
+import urllib.error
+from datetime import datetime
+
+# ── Config (set via environment variables / Gitea secrets) ──────────────────
+NTFY_URL    = os.environ.get("NTFY_URL", "https://ntfy.example.com/zipair-alert")
+NTFY_TOKEN  = os.environ.get("NTFY_TOKEN", "")   # optional, if your ntfy requires auth
+STATE_FILE  = os.environ.get("STATE_FILE", "last_seen.txt")
+
+ZIPAIR_SITEMAP  = "https://www.zipair.net/sitemap.xml"
+ZIPAIR_NOTIF    = "https://www.zipair.net/en/notification/{id}"
+
+# Keywords that must ALL appear (case-insensitive) in a notification page
+# to trigger an alert. Tune these as needed.
+TRIGGER_KEYWORDS = ["singapore", "winter"]
+
+# Browser-like headers to avoid 403
+HEADERS = {
+    "User-Agent": (
+        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+        "AppleWebKit/537.36 (KHTML, like Gecko) "
+        "Chrome/124.0.0.0 Safari/537.36"
+    ),
+    "Accept":          "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
+    "Accept-Language": "en-US,en;q=0.9",
+    "Accept-Encoding": "gzip, deflate, br",
+    "Connection":      "keep-alive",
+}
+
+# ── Helpers ─────────────────────────────────────────────────────────────────
+
+def fetch(url: str, timeout: int = 15) -> str:
+    """Fetch a URL and return the decoded body, or empty string on error."""
+    req = urllib.request.Request(url, headers=HEADERS)
+    try:
+        with urllib.request.urlopen(req, timeout=timeout) as resp:
+            raw = resp.read()
+            # handle gzip transparently (urlopen usually does, but just in case)
+            try:
+                import gzip
+                return gzip.decompress(raw).decode("utf-8", errors="replace")
+            except Exception:
+                return raw.decode("utf-8", errors="replace")
+    except urllib.error.HTTPError as e:
+        print(f"  HTTP {e.code} for {url}", file=sys.stderr)
+        return ""
+    except Exception as e:
+        print(f"  Error fetching {url}: {e}", file=sys.stderr)
+        return ""
+
+
+def get_notification_ids_from_sitemap() -> list[int]:
+    """Parse the ZIPAIR sitemap and return all notification IDs found."""
+    print("Fetching sitemap …")
+    xml = fetch(ZIPAIR_SITEMAP)
+    if not xml:
+        print("  Sitemap unavailable.", file=sys.stderr)
+        return []
+    ids = [int(m) for m in re.findall(r"/notification/(\d+)", xml)]
+    ids = sorted(set(ids))
+    print(f"  Found {len(ids)} notification IDs in sitemap (max={ids[-1] if ids else 'n/a'})")
+    return ids
+
+
+def read_last_seen() -> int:
+    """Read the last-seen notification ID from the state file."""
+    try:
+        with open(STATE_FILE) as f:
+            return int(f.read().strip())
+    except Exception:
+        return 0
+
+
+def write_last_seen(n: int):
+    """Persist the last-seen notification ID."""
+    with open(STATE_FILE, "w") as f:
+        f.write(str(n))
+    print(f"State updated: last_seen = {n}")
+
+
+def matches_keywords(text: str) -> bool:
+    """Return True if all TRIGGER_KEYWORDS appear in text."""
+    lower = text.lower()
+    return all(kw in lower for kw in TRIGGER_KEYWORDS)
+
+
+def send_ntfy(notif_id: int, snippet: str):
+    """Fire a push notification via ntfy."""
+    title   = "✈️ ZIPAIR SIN→TYO Tickets On Sale!"
+    message = (
+        f"A new ZIPAIR announcement about Singapore winter sales was detected "
+        f"(notification #{notif_id}). "
+        f"Check: https://www.zipair.net/en/notification/{notif_id}"
+    )
+    payload = json.dumps({
+        "topic":    NTFY_URL.rstrip("/").rsplit("/", 1)[-1],
+        "title":    title,
+        "message":  message,
+        "priority": 5,
+        "tags":     ["airplane", "moneybag"],
+        "click":    f"https://www.zipair.net/en/notification/{notif_id}",
+        "actions": [{
+            "action": "view",
+            "label":  "Open ZIPAIR",
+            "url":    f"https://www.zipair.net/en/notification/{notif_id}",
+        }],
+    }).encode()
+
+    # Build the POST request to the ntfy server base URL
+    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 response: {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.utcnow().isoformat()}Z] ZIPAIR monitor starting …")
+    print(f"  Keywords : {TRIGGER_KEYWORDS}")
+    print(f"  ntfy URL : {NTFY_URL}")
+
+    last_seen = read_last_seen()
+    print(f"  Last seen notification ID: {last_seen}")
+
+    ids = get_notification_ids_from_sitemap()
+    if not ids:
+        print("No notification IDs found; exiting.")
+        sys.exit(0)
+
+    new_ids = [i for i in ids if i > last_seen]
+    if not new_ids:
+        print("No new notifications since last check. All good.")
+        write_last_seen(max(ids))
+        sys.exit(0)
+
+    print(f"  {len(new_ids)} new notification(s) to check: {new_ids}")
+    found_match = None
+
+    for nid in new_ids:
+        url = ZIPAIR_NOTIF.format(id=nid)
+        print(f"  Fetching notification #{nid} …")
+        text = fetch(url)
+        time.sleep(1)  # be polite
+
+        if not text:
+            print(f"    Could not fetch #{nid}, skipping.")
+            continue
+
+        if matches_keywords(text):
+            print(f"  ✅ MATCH in notification #{nid}!")
+            # Grab a short snippet for context
+            lower = text.lower()
+            pos = lower.find("singapore")
+            snippet = text[max(0, pos - 50): pos + 200].strip()
+            found_match = (nid, snippet)
+            break
+        else:
+            print(f"    No match in #{nid}.")
+
+    # Always advance the state to the latest ID we've seen
+    write_last_seen(max(ids))
+
+    if found_match:
+        nid, snippet = found_match
+        print(f"\n🚨 Sending ntfy push for notification #{nid}")
+        send_ntfy(nid, snippet)
+        print("Done — notification sent!")
+    else:
+        print("\nNo Singapore winter sale announcement found yet.")
+
+    sys.exit(0)
+
+
+if __name__ == "__main__":
+    main()

last_seen.txt

@@ -0,0 +1 @@
+0