only update records and notify if IP has changed

records/test_records.yaml

@@ -5,4 +5,3 @@ records:
 - record: dummy.reeselink.com
   provider: unifi
   ttl_seconds: 60
-  skip_ipv4: true

route53_update.py

@@ -36,7 +36,26 @@ def _get_route53_client() -> Route53Client:
     return boto3.client("route53")
 
 
-def update_ipv4(hosted_zone_id: str, record: str, public_ipv4: str) -> None:
+def _get_current_record(hosted_zone_id: str, record: str, record_type: str) -> str | None:
+    client = _get_route53_client()
+    try:
+        response = client.list_resource_record_sets(HostedZoneId=hosted_zone_id)
+        record_normalized = record.rstrip(".")
+        for rrset in response["ResourceRecordSets"]:
+            if rrset["Name"].rstrip(".") == record_normalized and rrset["Type"] == record_type:
+                records = rrset.get("ResourceRecords", [])
+                if records:
+                    return records[0]["Value"].rstrip(".")
+    except Exception as e:
+        logger.warning("Failed to get current %s record for %s: %s", record_type, record, e)
+    return None
+
+
+def update_ipv4(hosted_zone_id: str, record: str, public_ipv4: str) -> bool:
+    current = _get_current_record(hosted_zone_id, record, "A")
+    if current == public_ipv4:
+        logger.info("IPv4 for %s is already %s, no update needed", record, public_ipv4)
+        return False
     logger.debug("Creating Route53 client for hosted zone %s", hosted_zone_id)
     client: Route53Client = _get_route53_client()
     logger.debug("Building ChangeBatch for IPv4: record=%s, ip=%s, ttl=300", record, public_ipv4)
@@ -60,12 +79,17 @@ def update_ipv4(hosted_zone_id: str, record: str, public_ipv4: str) -> None:
             },
         )
         logger.info("Successfully updated IPv4 for %s -> %s", record, public_ipv4)
+        return True
     except Exception as e:
         logger.error("Error updating IPv4 for %s: %s", record, e)
         raise e
 
 
-def update_ipv6(hosted_zone_id: str, record: str, public_ipv6: str) -> None:
+def update_ipv6(hosted_zone_id: str, record: str, public_ipv6: str) -> bool:
+    current = _get_current_record(hosted_zone_id, record, "AAAA")
+    if current == public_ipv6:
+        logger.info("IPv6 for %s is already %s, no update needed", record, public_ipv6)
+        return False
     logger.debug("Creating Route53 client for hosted zone %s", hosted_zone_id)
     client = _get_route53_client()
     logger.debug("Building ChangeBatch for IPv6: record=%s, ip=%s, ttl=300", record, public_ipv6)
@@ -89,6 +113,7 @@ def update_ipv6(hosted_zone_id: str, record: str, public_ipv6: str) -> None:
             },
         )
         logger.info("Successfully updated IPv6 for %s -> %s", record, public_ipv6)
+        return True
     except Exception as e:
         logger.error("Error updating IPv6 for %s: %s", record, e)
         raise e

test_unifi_update.py

@@ -85,7 +85,7 @@ class TestListDnsPolicies:
         assert policies[0]["ipv4Address"] == "1.2.3.4"
 
         mock_session.get.assert_called_once_with(
-            "https://unifi.example.com/api/s/site123/dns/policies",
+            "https://unifi.example.com/api/sites/site123/dns/policies",
             verify=False,
         )
 
@@ -119,7 +119,7 @@ class TestGetSession:
 
             mock_session_cls.assert_called_once()
             mock_session.headers.update.assert_called_once_with(
-                {"X-CSRF-Token": "my-api-token"}
+                {"X-API-Key": "my-api-token"}
             )
             assert session is mock_session
 
@@ -311,7 +311,7 @@ class TestUpdateRecords:
 
         mock_session.get.assert_called_once()
         call_url = mock_session.get.call_args[0][0]
-        assert "https://unifi.example.com/api/s/site123/dns/policies" == call_url
+        assert "https://unifi.example.com/proxy/network/integration/v1/sites/site123/dns/policies" == call_url
 
     def test_update_records_raises_on_post_failure(self) -> None:
         mock_session = MagicMock()

unifi_update.py

@@ -93,7 +93,7 @@ def _create_or_update_policy(
     ip_address: str,
     ttl_seconds: int,
     existing_policy: dict | None,
-) -> None:
+) -> bool:
     payload: dict = {
         "type": record_type,
         "enabled": True,
@@ -109,10 +109,13 @@ def _create_or_update_policy(
     logger.debug("Payload for %s on %s: %s", record_type, domain, payload)
 
     if existing_policy and existing_policy.get("id"):
+        current_ip = existing_policy.get("ipv4Address") or existing_policy.get("ipv6Address")
+        if current_ip == ip_address:
+            logger.info("%s policy for %s is already %s, no update needed", record_type, domain, ip_address)
+            return False
         policy_id = existing_policy["id"]
         logger.info("Updating existing %s policy for %s (id=%s, current_ip=%s)",
-                     record_type, domain, policy_id,
-                     existing_policy.get("ipv4Address") or existing_policy.get("ipv6Address"))
+                     record_type, domain, policy_id, current_ip)
         url = f"{api_base}/sites/{site_id}/dns/policies/{policy_id}"
         logger.debug("Sending PUT to %s", url)
         response = session.put(url, json=payload, verify=session.verify)
@@ -124,13 +127,14 @@ def _create_or_update_policy(
 
     response.raise_for_status()
     logger.info("Successfully updated %s policy for %s -> %s", record_type, domain, ip_address)
+    return True
 
 
 def update_records(
     unifi_config: UnifiConfig,
     ipv4: str | None = None,
     ipv6: str | None = None,
-) -> None:
+) -> tuple[set[str], set[str]]:
     base_url = unifi_config["host"]
     site_id = unifi_config["site_id"]
     api_token = unifi_config["api_token"]
@@ -144,6 +148,9 @@ def update_records(
     policies = list_dns_policies(session, api_base, site_id)
     policy_map = _get_policy_map(policies)
 
+    updated_ipv4: set[str] = set()
+    updated_ipv6: set[str] = set()
+
     for record in records:
         domain = record["record"]
         ttl = record.get("ttl_seconds", 14400)
@@ -156,11 +163,13 @@ def update_records(
                              domain, existing["id"], existing.get("ipv4Address"))
             else:
                 logger.debug("No existing A_RECORD policy for %s, will create new", domain)
-            _create_or_update_policy(
+            changed = _create_or_update_policy(
                 session, api_base, site_id,
                 "A_RECORD", domain, ipv4, ttl,
                 existing,
             )
+            if changed:
+                updated_ipv4.add(domain)
         elif ipv4 and record.get("skip_ipv4"):
             logger.info("Skipping IPv4 for %s (skip_ipv4=true)", domain)
 
@@ -171,12 +180,16 @@ def update_records(
                              domain, existing["id"], existing.get("ipv6Address"))
             else:
                 logger.debug("No existing AAAA_RECORD policy for %s, will create new", domain)
-            _create_or_update_policy(
+            changed = _create_or_update_policy(
                 session, api_base, site_id,
                 "AAAA_RECORD", domain, ipv6, ttl,
                 existing,
             )
+            if changed:
+                updated_ipv6.add(domain)
         elif ipv6 and record.get("skip_ipv6"):
             logger.info("Skipping IPv6 for %s (skip_ipv6=true)", domain)
 
         logger.info("=== Done processing UniFi record: %s ===", domain)
+
+    return updated_ipv4, updated_ipv6

update.py

@@ -154,12 +154,13 @@ def _update_route53_records(
             logger.info("Skipping IPv4 for %s (global skip or no IPv4 available)", record["record"])
         else:
             logger.info("Updating IPv4 for %s -> %s", record["record"], public_ipv4)
-            route53_update_ipv4(
+            changed = route53_update_ipv4(
                 hosted_zone_id=ROUTE53_HOSTED_ZONE_ID,  # type: ignore[arg-type]
                 record=record["record"],
                 public_ipv4=public_ipv4,
             )
-            updated_ipv4.add(record["record"])
+            if changed:
+                updated_ipv4.add(record["record"])
 
         if record.get("skip_ipv6"):
             logger.info("Skipping IPv6 for %s (skip_ipv6=true)", record["record"])
@@ -167,12 +168,13 @@ def _update_route53_records(
             logger.info("Skipping IPv6 for %s (global skip or no IPv6 available)", record["record"])
         else:
             logger.info("Updating IPv6 for %s -> %s", record["record"], public_ipv6)
-            route53_update_ipv6(
+            changed = route53_update_ipv6(
                 hosted_zone_id=ROUTE53_HOSTED_ZONE_ID,  # type: ignore[arg-type]
                 record=record["record"],
                 public_ipv6=public_ipv6,
             )
-            updated_ipv6.add(record["record"])
+            if changed:
+                updated_ipv6.add(record["record"])
 
         logger.info("=== Done processing Route53 record: %s ===", record["record"])
 
@@ -185,15 +187,7 @@ def _update_unifi_records(
     public_ipv6: str | None,
 ) -> tuple[set[str], set[str]]:
     logger.info("Processing %d UniFi record(s)", len(unifi_config["records"]))
-    updated_ipv4: set[str] = set()
-    updated_ipv6: set[str] = set()
-    for record in unifi_config["records"]:
-        domain = record["record"]
-        if public_ipv4 and not record.get("skip_ipv4"):
-            updated_ipv4.add(domain)
-        if public_ipv6 and not record.get("skip_ipv6"):
-            updated_ipv6.add(domain)
-    unifi_update_records(
+    updated_ipv4, updated_ipv6 = unifi_update_records(
         unifi_config=unifi_config,
         ipv4=public_ipv4,
         ipv6=public_ipv6,
@@ -329,13 +323,19 @@ def main() -> None:
 
     if route53_records:
         if route53_success:
-            send_ntfy_notification("Route53 Update Successful", f"Records updated:\n{route53_text}", priority=4)
+            has_changes = bool(route53_updated_ipv4 or route53_updated_ipv6)
+            ntfy_priority = 4 if has_changes else 2
+            ntfy_title = "Route53 IP Changed" if has_changes else "Route53 IP Unchanged"
+            send_ntfy_notification(ntfy_title, f"Records updated:\n{route53_text}", priority=ntfy_priority)
         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)
+            has_changes = bool(unifi_updated_ipv4 or unifi_updated_ipv6)
+            ntfy_priority = 4 if has_changes else 2
+            ntfy_title = "UniFi IP Changed" if has_changes else "UniFi IP Unchanged"
+            send_ntfy_notification(ntfy_title, f"Records updated:\n{unifi_text}", priority=ntfy_priority)
         else:
             send_ntfy_notification("UniFi Update Failed", f"Records:\n{unifi_text}", priority=2)