onvif.utils.plugins

def egress(
    self,
    envelope: etree._Element,
    http_headers: dict[str, str],
    operation: Any,
    binding_options: dict[str, object],
) -> tuple[Any, dict[str, str]]:
    """
    <!-- md:version 0.0.6 -->
    <!-- md:internal -->

    Capture a SOAP request before it is sent to the device.

    This Zeep ``egress`` hook serializes the outgoing SOAP envelope and stores
    it in `last_sent_xml`. The request is also appended to `history` together
    with the operation name and HTTP headers.

    The original envelope and HTTP headers are returned unchanged, so this
    hook only observes the outgoing request and does not modify it.

    Args:
        envelope: SOAP request envelope that is about to be sent
        http_headers: HTTP request headers that will be sent with the request
        operation: Zeep operation being invoked
        binding_options: Zeep binding options for the current request

    Returns:
        A tuple containing the unchanged ``envelope`` and ``http_headers``
    """
    logger.debug(
        "Capturing outgoing SOAP request for operation: %s", operation.name
    )

    # Serialize XML with proper pretty printing
    if self.pretty_print:
        self.last_sent_xml = self._format_xml(envelope)
    else:
        self.last_sent_xml = etree.tostring(
            envelope, pretty_print=False, encoding="unicode"
        )

    self.last_operation = operation.name

    # Store in history
    self.history.append(
        {
            "type": "request",
            "operation": operation.name,
            "xml": self.last_sent_xml,
            "http_headers": dict(http_headers) if http_headers else {},
        }
    )

    logger.debug(
        "Captured SOAP request for %s (%d chars)",
        operation.name,
        len(self.last_sent_xml),
    )
    return envelope, http_headers

def ingress(
    self, envelope: etree._Element, http_headers: dict[str, str], operation: Any
) -> tuple[Any, dict[str, str]]:
    """
    <!-- md:version 0.0.6 -->
    <!-- md:internal -->

    Capture a SOAP response after it is received from the device.

    This Zeep ``ingress`` hook serializes the incoming SOAP envelope and stores
    it in `last_received_xml`. The response is also appended to `history`
    together with the operation name and HTTP headers.

    The original envelope and HTTP headers are returned unchanged, so this
    hook only observes the incoming response and does not modify it.

    Args:
        envelope: Raw SOAP response envelope received from the device
        http_headers: HTTP response headers returned by the device
        operation: Zeep operation associated with the response

    Returns:
        A tuple containing the unchanged ``envelope`` and ``http_headers``
    """
    logger.debug(
        "Capturing incoming SOAP response for operation: %s", operation.name
    )

    # Serialize XML with proper pretty printing
    if self.pretty_print:
        self.last_received_xml = self._format_xml(envelope)
    else:
        self.last_received_xml = etree.tostring(
            envelope, pretty_print=False, encoding="unicode"
        )

    # Store in history
    self.history.append(
        {
            "type": "response",
            "operation": operation.name,
            "xml": self.last_received_xml,
            "http_headers": dict(http_headers) if http_headers else {},
        }
    )

    logger.debug(
        "Captured SOAP response for %s (%d chars)",
        operation.name,
        len(self.last_received_xml),
    )
    return envelope, http_headers

def get_last_request(self) -> str | None:
    """
    <!-- md:version 0.0.6 -->
    <!-- md:utility -->

    Get the last captured request XML.

    Returns:
        Last captured request XML
    """
    return self.last_sent_xml

def get_last_response(self) -> str | None:
    """
    <!-- md:version 0.0.6 -->
    <!-- md:utility -->

    Get the last captured response XML.

    Returns:
        Last captured response XML
    """
    return self.last_received_xml

def get_history(self) -> list:
    """
    <!-- md:version 0.0.6 -->
    <!-- md:utility -->

    Get all captured requests and responses.

    Returns:
        List of all captured requests and responses.
    """
    return self.history

def clear_history(self) -> None:
    """
    <!-- md:version 0.0.6 -->
    <!-- md:utility -->

    Clear the capture history.
    """
    history_count = len(self.history)
    self.history = []
    self.last_sent_xml = None
    self.last_received_xml = None
    self.last_operation = None
    logger.debug("Cleared XML capture history (%d items)", history_count)

def save_to_file(
    self, request_file: str | None = None, response_file: str | None = None
) -> None:
    """
    <!-- md:version 0.0.6 -->
    <!-- md:utility -->

    Save captured XML to files.

    Args:
        request_file (str | None): Path to save request XML
        response_file (str | None): Path to save response XML
    """
    if request_file and self.last_sent_xml:
        try:
            with open(request_file, "w", encoding="utf-8") as f:
                f.write(self.last_sent_xml)
            logger.info("Saved SOAP request XML to: %s", request_file)
        except OSError as e:
            logger.error("Failed to save request XML to %s: %s", request_file, e)

    if response_file and self.last_received_xml:
        try:
            with open(response_file, "w", encoding="utf-8") as f:
                f.write(self.last_received_xml)
            logger.info("Saved SOAP response XML to: %s", response_file)
        except OSError as e:
            logger.error("Failed to save response XML to %s: %s", response_file, e)

def ingress(
    self, envelope: etree._Element, http_headers: dict[str, str], operation: Any
) -> tuple[Any, dict[str, str]]:
    """
    <!-- md:version 0.2.2 -->
    <!-- md:internal -->

    Process a SOAP response before Zeep deserializes it.

    This Zeep ``ingress`` hook inspects the raw SOAP response envelope and
    extracts text from elements matching the configured XPath expressions.
    The extracted values are stored temporarily and can be retrieved with
    `get_extracted_texts`.

    The extraction cache is cleared before processing each response, so it
    always contains values from the most recently processed SOAP response.

    Args:
        envelope: Raw SOAP response envelope as an `lxml.etree.Element`
        http_headers: HTTP response headers returned by the device
        operation: Zeep operation associated with the SOAP response

    Returns:
        A tuple containing the unchanged ``envelope`` and ``http_headers``.
            Returning both values allows Zeep to continue processing the response
            and allows other plugins in the chain to process it as well.
    """
    # Auto-clear cache from previous response
    self._extracted_elements = {}

    try:
        # Extract elements using XPath from raw XML envelope
        for name, xpath in self.extract_xpaths.items():
            elements = envelope.findall(xpath)

            # Extract text content from found elements
            texts = [elem.text for elem in elements]

            if texts:
                self._extracted_elements[name] = texts
                logger.debug(
                    "ONVIFParser: Extracted %d '%s' elements", len(texts), name
                )

    except (ValueError, TypeError, AttributeError) as e:
        logger.warning("ONVIFParser: Failed to extract elements: %s", e)

    return envelope, http_headers

def get_extracted_texts(self, name: str, count: int) -> list[str | None]:
    """
    <!-- md:version 0.2.2 -->
    <!-- md:utility -->

    Get extracted element texts by name.

    Args:
        name (str): Name of the extracted elements (key from `extract_xpaths` dict)
        count (int): Number of elements to return

    Returns:
        List of element text values, padded with None if fewer elements were found.
            Example: If 3 elements found but `count=5`, returns
            [text1, text2, text3, None, None]
    """
    texts = self._extracted_elements.get(name, [])[:count]

    # Pad with None if not enough elements
    while len(texts) < count:
        texts.append(None)

    return texts

def egress(
    self,
    envelope: etree._Element,
    http_headers: dict[str, str],
    operation: Any,
    binding_options: dict[str, object],
) -> tuple[Any, dict[str, str]]:
    """
    <!-- md:version 0.3.1 -->
    <!-- md:internal -->

    Inject WS-Addressing reference parameters into a SOAP request.

    This Zeep ``egress`` hook adds the configured reference parameters to the
    SOAP ``Header`` before the request is sent to the device.

    If the SOAP envelope does not contain a ``Header`` element, one is created.
    Reference parameters are deep-copied before being inserted so the original
    XML elements remain unchanged and can safely be reused.

    Args:
        envelope: SOAP request envelope that is about to be sent
        http_headers: HTTP request headers that will be sent with the request
        operation: Zeep operation being invoked
        binding_options: Zeep binding options for the current request

    Returns:
        A tuple containing the modified ``envelope`` and the unchanged ``http_headers``

    Raises:
        RuntimeError: If the SOAP envelope uses an unsupported SOAP namespace
    """
    soap_namespace = etree.QName(envelope).namespace

    if soap_namespace not in self.SOAP_NAMESPACES:
        raise RuntimeError(f"Unsupported SOAP envelope namespace: {soap_namespace}")

    header = envelope.find(f"{{{soap_namespace}}}Header")

    if header is None:
        header = etree.Element(f"{{{soap_namespace}}}Header")
        envelope.insert(0, header)

    for parameter in self.reference_parameters:
        header.append(deepcopy(parameter))

    return envelope, http_headers