From 209d260b189afd2a8ac7927670427a92f345bdd8 Mon Sep 17 00:00:00 2001 From: ducoterra Date: Wed, 3 Jun 2026 21:10:41 -0400 Subject: [PATCH] add ntfy --- records/test_records.yaml | 2 +- update.py | 101 ++++++++++++++++++++++++++++++++------ 2 files changed, 88 insertions(+), 15 deletions(-) diff --git a/records/test_records.yaml b/records/test_records.yaml index a45e5b8..140a1b7 100644 --- a/records/test_records.yaml +++ b/records/test_records.yaml @@ -4,4 +4,4 @@ records: ttl_seconds: 60 - record: dummy.reeselink.com provider: unifi - ttl_seconds: 14400 + ttl_seconds: 60 diff --git a/update.py b/update.py index f2f5e99..e7464ff 100644 --- a/update.py +++ b/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,29 @@ 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}" + logger.info("Sending NTFY notification: %s", title) + response = requests.post( + f"{NTFY_URL}/{NTFY_TOPIC}", + data=message.encode(), + headers=headers, + timeout=10, + ) + response.raise_for_status() + logger.info("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,30 +244,75 @@ 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) - logger.info("=== Starting Route53 updates (hosted_zone_id=%s) ===", ROUTE53_HOSTED_ZONE_ID) - _update_route53_records(route53_records, public_ipv4, public_ipv6) - logger.info("=== Finished Route53 updates ===") + 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 + "api_token": UNIFI_API_TOKEN, # type: ignore + "verify_ssl": UNIFI_VERIFY_SSL, + "records": unifi_records, # type: ignore[arg-type] + } - unifi_config: UnifiConfig = { - "host": UNIFI_HOST, # type: ignore - "site_id": UNIFI_SITE_ID, # type: ignore - "api_token": UNIFI_API_TOKEN, # type: ignore - "verify_ssl": UNIFI_VERIFY_SSL, - "records": unifi_records, # type: ignore[arg-type] - } + _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 - _update_unifi_records(unifi_config, public_ipv4, public_ipv6) - logger.info("=== Finished UniFi updates ===") + 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 ===")