Handle cameras that generate broken subscription urls (#36)

Author: bdraco

onvif/client.py

@@ -29,7 +29,7 @@
 from onvif.definition import SERVICES
 from onvif.exceptions import ONVIFAuthError, ONVIFError, ONVIFTimeoutError
 
-from .util import extract_subcodes_as_strings, is_auth_error, stringify_onvif_error
+from .util import normalize_url, stringify_onvif_error
 
 logger = logging.getLogger("onvif")
 logging.basicConfig(level=logging.INFO)
@@ -577,7 +577,7 @@ async def _start(self) -> float:
         # pylint: disable=protected-access
         device.xaddrs[
             "http://www.onvif.org/ver10/events/wsdl/NotificationConsumer"
-        ] = result.SubscriptionReference.Address._value_1
+        ] = normalize_url(result.SubscriptionReference.Address._value_1)
         # Create subscription manager
         # 5.2.3 BASIC NOTIFICATION INTERFACE - NOTIFY
         # Call SetSynchronizationPoint to generate a notification message
@@ -657,7 +657,7 @@ async def _start(self) -> float:
         # pylint: disable=protected-access
         device.xaddrs[
             "http://www.onvif.org/ver10/events/wsdl/PullPointSubscription"
-        ] = result.SubscriptionReference.Address._value_1
+        ] = normalize_url(result.SubscriptionReference.Address._value_1)
         # Create subscription manager
         self._subscription = await device.create_subscription_service(
             "PullPointSubscription"

onvif/util.py

@@ -2,10 +2,25 @@
 from __future__ import annotations
 
 from typing import Any
+from urllib.parse import urlparse, urlunparse
 
 from zeep.exceptions import Fault
 
 
+def normalize_url(url: str) -> str:
+    """Normalize URL.
+
+    Some cameras respond with <wsa5:Address>http://192.168.1.106:8106:8106/onvif/Subscription?Idx=43</wsa5:Address>
+    https://github.com/home-assistant/core/issues/92603#issuecomment-1537213126
+    """
+    parsed = urlparse(url)
+    if "[" not in parsed.netloc and parsed.netloc.count(":") > 1:
+        net_location = parsed.netloc.split(":", 3)
+        net_location.pop()
+        return urlunparse(parsed._replace(netloc=":".join(net_location)))
+    return url
+
+
 def extract_subcodes_as_strings(subcodes: Any) -> list[str]:
     """Stringify ONVIF subcodes."""
     if isinstance(subcodes, list):