ONVIFOperator

class ONVIFOperator:
    """
    <!-- md:version 0.0.1 -->
    <!-- md:internal -->

    Low-level ONVIF service operator using Zeep SOAP client.

    This class handles the actual SOAP communication with ONVIF devices. It manages
    WSDL loading, service binding, authentication, caching, and error handling.

    !!! danger
        ``ONVIFOperator`` is typically **used internally** by service classes
        such as Device, Media, and PTZ, and is not intended to be
        instantiated directly by end users.

        Use [`ONVIFClient`](onvif_client.md) instead.

    Attributes:
        wsdl_path (str): Path to the WSDL file
        host (str): Device hostname or IP address
        port (int): Device port number
        username (str | None): ONVIF username
        password (str | None): ONVIF password
        http_digest (bool): Whether to use **HTTP Digest** or **WS-UsernameToken** for auth
        timeout (int): Request timeout in seconds
        apply_patch (bool): Whether to apply ``xsd:any`` flattening patch
        address (str): Service endpoint URL (XAddr)
        client (Client): Zeep SOAP client instance
        service (ServiceProxy): Zeep service proxy for making SOAP calls
        service_name (str): Name of the ONVIF service (e.g., "Device", "Media")

    !!! tip "Version History"
        - No version tracking for attribute changes because this class is used internally
    """

    def __init__(
        self,
        wsdl_path: str,
        host: str,
        port: int,
        username: str | None = None,
        password: str | None = None,
        http_digest: bool = False,  # True = use HTTP Digest / False = use WS-UsernameToken
        timeout: int = 10,
        binding: str | None = None,
        service_path: str | None = None,
        xaddr: str | None = None,
        cache: CacheMode = CacheMode.DB,  #  db | mem | none
        cache_path: str | None = None,
        use_https: bool = False,
        verify_ssl: bool = False,
        apply_patch: bool = True,
        plugins: list | None = None,
    ):
        logger.debug(
            "Creating ONVIFOperator for %s:%d with WSDL: %s", host, port, wsdl_path
        )

        self.wsdl_path: str = wsdl_path
        self.host: str = host
        self.port: int = port
        self.username: str | None = username
        self.password: str | None = password
        self.http_digest: bool = http_digest
        self.timeout: int = timeout
        self.apply_patch: bool = apply_patch

        if xaddr:
            self.address: str = xaddr
        else:
            protocol = "https" if use_https else "http"
            path = service_path or "device_service"  # default fallback
            self.address = f"{protocol}://{self.host}:{self.port}/onvif/{path}"

        logger.debug("Service endpoint: %s", self.address)

        # Session reuse with retry strategy
        session: Session = self._create_session(verify_ssl=verify_ssl)

        transport_kwargs = {"session": session, "operation_timeout": self.timeout}

        if cache == CacheMode.MEM:
            logger.debug("Using in-memory WSDL cache")
            transport_kwargs["cache"] = InMemoryCache()
        elif cache == CacheMode.DB:
            if cache_path is None:
                user_cache_dir = os.path.expanduser("~/.onvif-python")
                os.makedirs(user_cache_dir, exist_ok=True)
                cache_path = os.path.join(user_cache_dir, "onvif_zeep_cache.sqlite")

            logger.debug("Using SQLite cache: %s", cache_path)
            transport_kwargs["cache"] = SqliteCache(path=cache_path)
        elif cache != CacheMode.NONE:
            raise ValueError(f"Unknown cache option: {cache}")

        transport = Transport(**transport_kwargs)

        # zeep settings
        settings = Settings(strict=False, xml_huge_tree=True)
        wsse: UsernameToken | None = self._create_wsse()

        logger.debug("Using cache mode: %s", cache.value)

        self.client: Client = Client(
            wsdl=self.wsdl_path,
            transport=transport,
            settings=settings,
            wsse=wsse,
            plugins=plugins,
        )

        if not binding:
            raise ValueError("Bindings must be set according to the WSDL service")

        self.service: ServiceProxy = self.client.create_service(
            binding_name=binding, address=self.address
        )
        self.service_name: str = binding.split("}")[-1].replace(
            "Binding", ""
        )  # Store cleaned service name for logging context
        logger.info("ONVIFOperator initialized %s at %s", binding, self.address)

    def _create_session(
        self,
        verify_ssl: bool,
    ) -> Session:
        """Create and configure the HTTP session.

        Args:
            verify_ssl: Whether SSL certificates should be verified.

        Returns:
            Configured requests session.
        """
        session = requests.Session()
        session.verify = verify_ssl

        if not verify_ssl:
            # Format SSL warnings to be more concise when verify_ssl is False
            logger.debug("SSL verification disabled")
            warnings.simplefilter(
                "once",
                urllib3.exceptions.InsecureRequestWarning,
            )

        if self.http_digest and self.username and self.password:
            logger.debug("Configuring HTTP Digest authentication")

            session.auth = HTTPDigestAuth(
                username=self.username,
                password=self.password,
            )

        return session

    def _create_wsse(
        self,
    ) -> UsernameToken | None:
        """Create WS-Security authentication configuration.

        Returns:
            Zeep WS-Security UsernameToken or None.
        """
        if self.http_digest:
            return None

        if not self.username or not self.password:
            return None

        logger.debug("Configuring WS-Security UsernameToken authentication")

        return UsernameToken(
            username=self.username,
            password=self.password,
            use_digest=True,
        )

    def call(self, method: str, *args, **kwargs) -> Any:
        """
        <!-- md:version 0.0.1 -->
        <!-- md:internal -->

        Call an ONVIF service operation.

        This method invokes a SOAP operation on the ONVIF device service and handles
        errors gracefully. It automatically flattens `xsd:any` fields in the response
        when `apply_patch` is enabled.

        Args:
            method (str): Name of the ONVIF operation to call (e.g., "GetDeviceInformation")
            *args (Any): Positional arguments to pass to the operation
            **kwargs (Any): Keyword arguments to pass to the operation

        Returns:
            The operation result with `xsd:any` fields flattened if `apply_patch=True`

        Raises:
            ONVIFOperationException: If the operation fails (wraps original exception)
        """
        logger.debug("Calling ONVIF method: %s.%s", self.service_name, method)

        try:
            func = getattr(self.service, method)
        except AttributeError as e:
            raise ONVIFOperationException(operation=method, original_exception=e) from e

        try:
            result = func(*args, **kwargs)
            logger.debug("ONVIF call %s.%s succeeded", self.service_name, method)

            # Post-process to flatten xsd:any fields if enabled (> v0.0.4 patch)
            if self.apply_patch:
                result = ZeepPatcher.flatten_xsd_any_fields(result)
            return result

        except Fault as e:
            raise ONVIFOperationException(operation=method, original_exception=e) from e
        except Exception as e:
            raise ONVIFOperationException(operation=method, original_exception=e) from e

    def legacy_call(
        self,
        legacy_method: str,
        current_method: str,
        *args,
        **kwargs,
    ) -> Any:
        """
        <!-- md:version 0.4.3 -->
        <!-- md:internal -->

        Call a legacy ONVIF operation using the current WSDL definition.

        This provides compatibility with devices which implemented an older
        operation name that was subsequently renamed in the ONVIF WSDL.

        The legacy operation is created as a temporary alias of the current
        Zeep SOAP operation. The alias reuses the current operation's request
        and response definitions while overriding the operation name and SOAP
        action.

        Args:
            legacy_method: Name of the legacy ONVIF operation
            current_method: Name of the operation defined by the current WSDL
            *args (Any): Positional arguments to pass to the operation
            **kwargs (Any): Keyword arguments to pass to the operation

        Returns:
           The operation result with `xsd:any` fields flattened if `apply_patch=True`

        Raises:
            ONVIFOperationException: If either operation cannot be resolved or
                the SOAP operation fails
        """
        logger.debug(
            "Calling legacy ONVIF method: %s.%s -> %s",
            self.service_name,
            legacy_method,
            current_method,
        )

        try:
            binding = self.service._binding  # pylint: disable=protected-access

            current_operation = binding.get(current_method)
            if current_operation is None:
                raise AttributeError(
                    f"Current WSDL operation '{current_method}' was not found"
                )

            legacy_operation = copy(current_operation)
            legacy_operation.name = legacy_method

            if legacy_operation.soapaction:
                legacy_operation.soapaction = legacy_operation.soapaction.replace(
                    current_method,
                    legacy_method,
                )

            binding._operations[legacy_method] = (  # pylint: disable=protected-access
                legacy_operation
            )

            try:
                func = getattr(self.service, legacy_method)
                result = func(*args, **kwargs)
            finally:
                binding._operations.pop(  # pylint: disable=protected-access
                    legacy_method, None
                )

            logger.debug(
                "Legacy ONVIF call %s.%s succeeded",
                self.service_name,
                legacy_method,
            )

            if self.apply_patch:
                result = ZeepPatcher.flatten_xsd_any_fields(result)

            return result

        except Fault as e:
            raise ONVIFOperationException(
                operation=legacy_method, original_exception=e
            ) from e
        except ONVIFOperationException:
            raise
        except Exception as e:
            raise ONVIFOperationException(
                operation=legacy_method, original_exception=e
            ) from e

    def create_type(self, type_name: str) -> Any:
        """
        <!-- md:version 0.1.9 -->
        <!-- md:internal -->

        Create a type instance from WSDL schema for the given type name.

        Recursively initializes nested complex types so that fields like TimeZone, DateTime,
        Date, and Time are properly instantiated as objects rather than None.

        Args:
            type_name (str): Name of the type to create (e.g., 'SetHostname', 'SetIPAddressFilter')

        Returns:
            Type instance that can be populated with data

        Raises:
            AttributeError: If type not found in WSDL schema
        """
        logger.debug("Creating type instance for: %s", type_name)

        # Method 1: Try to get element from WSDL (works for operation parameters)
        # Common namespace prefixes for ONVIF services
        namespaces_to_try = [
            "ns0",  # Default namespace
            "tt",  # Common types (User, NetworkInterface, etc.)
        ]

        # Try to get element with namespace prefix
        for ns in namespaces_to_try:
            try:
                element = self.client.get_element(f"{ns}:{type_name}")
                instance = element()
                logger.debug(
                    "Successfully created type %s using namespace %s", type_name, ns
                )
                return self._initialize_nested_types(instance)
            except (AttributeError, TypeError, ValueError) as e:
                logger.debug(
                    "Failed to create type %s with namespace %s: %s", type_name, ns, e
                )
                continue

        # Method 2: Try without namespace prefix
        try:
            element = self.client.get_element(type_name)
            instance = element()
            logger.debug("Successfully created type %s without namespace", type_name)
            return self._initialize_nested_types(instance)
        except (AttributeError, TypeError, ValueError) as e:
            logger.debug(
                "Failed to create element %s without namespace: %s", type_name, e
            )

        # Method 3: Try to get type from schema (for complex types)
        try:
            for ns in namespaces_to_try:
                try:
                    type_obj = self.client.get_type(f"{ns}:{type_name}")
                    instance = type_obj()
                    logger.debug(
                        "Successfully created complex type %s using namespace %s",
                        type_name,
                        ns,
                    )
                    return self._initialize_nested_types(instance)
                except (AttributeError, TypeError, ValueError) as e:
                    logger.debug(
                        "Failed to create complex type %s with namespace %s: %s",
                        type_name,
                        ns,
                        e,
                    )
                    continue

            # Try without namespace
            type_obj = self.client.get_type(type_name)
            instance = type_obj()
            logger.debug(
                "Successfully created complex type %s without namespace", type_name
            )
            return self._initialize_nested_types(instance)
        except (AttributeError, TypeError, ValueError) as e:
            logger.debug(
                "Failed to create complex type %s without namespace: %s", type_name, e
            )

        # If all methods fail, log the error and raise
        logger.error("Type '%s' not found in WSDL schema using any method", type_name)
        raise AttributeError(f"Type '{type_name}' not found in WSDL schema.")

    def _initialize_nested_types(self, instance):
        """Recursively initialize nested complex types in a Zeep object.

        This ensures that fields like TimeZone, DateTime, Date, and Time are
        properly instantiated as objects rather than None values.

        Args:
            instance: A Zeep object instance to initialize

        Returns:
            The instance with all nested complex types initialized
        """
        # pylint: disable=too-many-nested-blocks
        try:
            # Get the XSD type from the instance's class
            if hasattr(instance.__class__, "_xsd_type"):
                xsd_type = (
                    instance.__class__._xsd_type  # pylint: disable=protected-access
                )

                # Iterate through elements defined in the XSD type
                if hasattr(xsd_type, "elements"):
                    for element_name, element_obj in xsd_type.elements:
                        current_value = getattr(instance, element_name, None)

                        # Only initialize if the value is None and the element has a type
                        if current_value is None and hasattr(element_obj, "type"):
                            element_type = element_obj.type

                            # Check if this is a complex type (has elements)
                            if hasattr(element_type, "elements"):
                                try:
                                    # Complex type - instantiate it and recursively initialize
                                    nested_instance = element_type()
                                    nested_instance = self._initialize_nested_types(
                                        nested_instance
                                    )
                                    setattr(instance, element_name, nested_instance)
                                    logger.debug(
                                        "Initialized nested type for element: %s",
                                        element_name,
                                    )
                                except (AttributeError, TypeError, ValueError) as e:
                                    # Log specific nested type initialization failures
                                    logger.debug(
                                        "Failed to initialize nested type for %s: %s",
                                        element_name,
                                        e,
                                    )
                                    # Continue with other elements instead of failing completely
                                    continue
        except (AttributeError, TypeError, ValueError) as e:
            # Log the error but don't fail - the top-level object is still usable
            logger.debug("Error during nested type initialization: %s", e)
            # The important thing is that the top-level object is created

        return instance