From 7567f248ee1386134432ff6c0a1b782ca0fdd524 Mon Sep 17 00:00:00 2001
From: kaburagisec <135169292+kaburagisec@users.noreply.github.com>
Date: Sun, 4 Oct 2026 09:22:04 +0700
Subject: [PATCH] feat(operator): Add `legacy_call` method for compatibility
 with renamed ONVIF operations

---
 onvif/operator.py | 94 ++++++++++++++++++++++++++++++++++++++++++++++-
 1 file changed, 92 insertions(+), 2 deletions(-)

diff --git a/onvif/operator.py b/onvif/operator.py
index d96601c..1443c7f 100644
--- a/onvif/operator.py
+++ b/onvif/operator.py
@@ -5,6 +5,7 @@
 import logging
 import os
 import warnings
+from copy import copy
 from enum import Enum
 from typing import Any
 
@@ -274,8 +275,8 @@ def call(self, method: str, *args, **kwargs) -> Any:
 
         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
+            *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`
@@ -304,6 +305,95 @@ def call(self, method: str, *args, **kwargs) -> Any:
         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:
+        """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:
         """Create a type instance from WSDL schema for the given type name.