add ntfy

records/test_records.yaml

@@ -4,4 +4,4 @@ records:
   ttl_seconds: 60
 - record: dummy.reeselink.com
   provider: unifi
-  ttl_seconds: 14400
+  ttl_seconds: 60

update.py

@@ -34,6 +34,7 @@ import subprocess
 import sys
 from typing import Literal, TypedDict
 
+import requests
 import yaml
 from dotenv import load_dotenv
 
@@ -67,6 +68,10 @@ UNIFI_SITE_ID = os.getenv("UNIFI_SITE_ID")
 UNIFI_API_TOKEN = os.getenv("UNIFI_API_TOKEN")
 UNIFI_VERIFY_SSL = os.getenv("UNIFI_VERIFY_SSL", "false").lower() == "true"
 
+NTFY_URL = os.getenv("NTFY_URL", "")
+NTFY_TOPIC = os.getenv("NTFY_TOPIC", "")
+NTFY_API_KEY = os.getenv("NTFY_API_KEY", "")
+
 
 class Route53RecordType(TypedDict):
     record: str
@@ -86,6 +91,28 @@ class RecordYamlStruct(TypedDict):
     records: list[Route53RecordType | UnifiRecordType]
 
 
+def send_ntfy_notification(title: str, message: str, priority: int = 3) -> None:
+    if not NTFY_URL or not NTFY_TOPIC:
+        return
+    try:
+        headers = {
+            "Title": title,
+            "Priority": str(priority),
+        }
+        if NTFY_API_KEY:
+            headers["Authorization"] = f"Bearer {NTFY_API_KEY}"
+        response = requests.post(
+            f"{NTFY_URL}/{NTFY_TOPIC}",
+            data=message.encode(),
+            headers=headers,
+            timeout=10,
+        )
+        response.raise_for_status()
+        logger.debug("NTFY notification sent: %s", title)
+    except Exception as e:
+        logger.warning("Failed to send NTFY notification: %s", e)
+
+
 def get_ipv4() -> str:
     logger.debug("Executing: curl -4 ifconfig.me")
     result = subprocess.run(["curl", "-4", "ifconfig.me"], capture_output=True)
@@ -216,20 +243,29 @@ def main() -> None:
             exit(1)
         logger.info("Public IPv6 is %s", public_ipv6)
 
+    route53_success = True
+    unifi_success = True
+
     if route53_records:
         if not ROUTE53_HOSTED_ZONE_ID:
             logger.error("ROUTE53_HOSTED_ZONE_ID must be set for Route53 records!")
-            exit(1)
+            route53_success = False
+        else:
             logger.info("=== Starting Route53 updates (hosted_zone_id=%s) ===", ROUTE53_HOSTED_ZONE_ID)
+            try:
                 _update_route53_records(route53_records, public_ipv4, public_ipv6)
                 logger.info("=== Finished Route53 updates ===")
+            except Exception as e:
+                logger.error("Route53 updates failed: %s", e)
+                route53_success = False
 
     if unifi_records:
         logger.info("=== Starting UniFi updates ===")
         if not all([UNIFI_HOST, UNIFI_SITE_ID, UNIFI_API_TOKEN]):
             logger.error("UNIFI_HOST, UNIFI_SITE_ID, and UNIFI_API_TOKEN must be set for UniFi records!")
-            exit(1)
-
+            unifi_success = False
+        else:
+            try:
                 unifi_config: UnifiConfig = {
                     "host": UNIFI_HOST,  # type: ignore
                     "site_id": UNIFI_SITE_ID,  # type: ignore
@@ -240,6 +276,42 @@ def main() -> None:
 
                 _update_unifi_records(unifi_config, public_ipv4, public_ipv6)
                 logger.info("=== Finished UniFi updates ===")
+            except Exception as e:
+                logger.error("UniFi updates failed: %s", e)
+                unifi_success = False
+
+    route53_domains = [r["record"] for r in route53_records]
+    unifi_domains = [r["record"] for r in unifi_records]
+
+    route53_lines = []
+    if public_ipv4:
+        for domain in route53_domains:
+            route53_lines.append(f"{domain} (A): {public_ipv4}")
+    if public_ipv6:
+        for domain in route53_domains:
+            route53_lines.append(f"{domain} (AAAA): {public_ipv6}")
+    route53_text = "\n".join(route53_lines) if route53_lines else "No IPs available"
+
+    unifi_lines = []
+    if public_ipv4:
+        for domain in unifi_domains:
+            unifi_lines.append(f"{domain} (A): {public_ipv4}")
+    if public_ipv6:
+        for domain in unifi_domains:
+            unifi_lines.append(f"{domain} (AAAA): {public_ipv6}")
+    unifi_text = "\n".join(unifi_lines) if unifi_lines else "No IPs available"
+
+    if route53_records:
+        if route53_success:
+            send_ntfy_notification("Route53 Update Successful", f"Records updated:\n{route53_text}", priority=4)
+        else:
+            send_ntfy_notification("Route53 Update Failed", f"Records:\n{route53_text}", priority=2)
+
+    if unifi_records:
+        if unifi_success:
+            send_ntfy_notification("UniFi Update Successful", f"Records updated:\n{unifi_text}", priority=4)
+        else:
+            send_ntfy_notification("UniFi Update Failed", f"Records:\n{unifi_text}", priority=2)
 
     logger.info("=== DDNS Update Complete ===")