"""SOAP calls for the Task Designer's SOAP node.

Two modes: "wsdl" (zeep builds the request from the service description) and
"raw" (the author's own envelope is posted as written, for services whose WSDL
is broken or unavailable). Both run synchronously, on a worker thread.

Every URL — the WSDL, anything it imports, the address it posts to — must be
http(s) and not link-local. zeep will otherwise happily read a local file path
as a WSDL, and a crafted WSDL could aim the call at the cloud metadata address.
Private addresses stay allowed, as for the REST node: internal SOAP services
are the normal case. XML is parsed with entities, network access and huge trees
off.
"""

import base64
import datetime
import decimal
import ipaddress
import socket
import uuid
from dataclasses import dataclass, field
from typing import Any
from urllib.parse import urlsplit

import requests
import zeep
from lxml import etree
from requests.auth import HTTPBasicAuth, HTTPDigestAuth
from zeep.exceptions import Fault
from zeep.helpers import serialize_object
from zeep.settings import Settings
from zeep.transports import Transport
from zeep.wsse.username import UsernameToken

MAX_RESPONSE_BYTES = 20 * 1024 * 1024
MAX_XML_TEXT = 200_000
MIN_TIMEOUT, MAX_TIMEOUT = 1, 300


class SoapError(Exception):
    """A problem worth showing the task's author, already in plain words."""


@dataclass
class SoapSpec:
    wsdl_url: str = ""
    operation: str = ""
    endpoint: str = ""  # wsdl: overrides the WSDL's address; raw: where to post
    service: str = ""
    port: str = ""
    auth: str = "none"  # none | basic | digest | wsse_text | wsse_digest | bearer
    username: str = ""
    password: str = ""
    headers: dict[str, str] = field(default_factory=dict)
    header_xml: str = ""
    timeout: int = 30
    verify_tls: bool = True
    # raw
    envelope: str = ""
    soap_action: str = ""
    version: str = "1.1"


def check_url(url: str) -> None:
    parts = urlsplit(url or "")
    if parts.scheme not in ("http", "https") or not parts.hostname:
        raise SoapError(f"the address must start with http:// or https://: {url[:200]}")
    try:
        infos = socket.getaddrinfo(parts.hostname, parts.port or (443 if parts.scheme == "https" else 80))
    except socket.gaierror as exc:
        raise SoapError(f"couldn't find the host {parts.hostname}") from exc
    for info in infos:
        address = ipaddress.ip_address(info[4][0])
        if address.is_link_local or address.is_multicast or address.is_unspecified:
            raise SoapError(f"{parts.hostname} can't be called from a task")


class _SafeTransport(Transport):
    """zeep's transport, with every load and post checked."""

    def load(self, url):  # noqa: ANN001, ANN201 - zeep's signature
        check_url(str(url))
        return super().load(url)

    def post(self, address, message, headers):  # noqa: ANN001, ANN201 - zeep's signature
        check_url(str(address))
        return super().post(address, message, headers)

    def get(self, address, params, headers):  # noqa: ANN001, ANN201 - zeep's signature
        check_url(str(address))
        return super().get(address, params, headers)


def _session(spec: SoapSpec) -> requests.Session:
    session = requests.Session()
    session.verify = spec.verify_tls
    session.max_redirects = 3
    if spec.auth == "basic":
        session.auth = HTTPBasicAuth(spec.username, spec.password)
    elif spec.auth == "digest":
        session.auth = HTTPDigestAuth(spec.username, spec.password)
    elif spec.auth == "bearer":
        session.headers["Authorization"] = f"Bearer {spec.password}"
    session.headers.update({k: v for k, v in spec.headers.items() if k and v})
    return session


def _timeout(spec: SoapSpec) -> int:
    return max(MIN_TIMEOUT, min(MAX_TIMEOUT, int(spec.timeout or 30)))


def _client(spec: SoapSpec) -> zeep.Client:
    if not spec.wsdl_url.strip():
        raise SoapError("set the WSDL URL")
    check_url(spec.wsdl_url)
    wsse = None
    if spec.auth in ("wsse_text", "wsse_digest"):
        wsse = UsernameToken(spec.username, spec.password, use_digest=spec.auth == "wsse_digest")
    transport = _SafeTransport(session=_session(spec), timeout=_timeout(spec), operation_timeout=_timeout(spec))
    settings = Settings(strict=False, xml_huge_tree=False, forbid_entities=True, forbid_external=True)
    return zeep.Client(wsdl=spec.wsdl_url.strip(), transport=transport, settings=settings, wsse=wsse, service_name=spec.service or None, port_name=spec.port or None)


def jsonable(value: Any) -> Any:
    """zeep's return values — nested dicts, Decimals, dates, raw elements — as
    what a task's context can hold."""
    if value is None or isinstance(value, (str, int, float, bool)):
        return value
    if isinstance(value, decimal.Decimal):
        return int(value) if value == value.to_integral_value() else float(value)
    if isinstance(value, (datetime.datetime, datetime.date, datetime.time)):
        return value.isoformat()
    if isinstance(value, uuid.UUID):
        return str(value)
    if isinstance(value, (bytes, bytearray)):
        return base64.b64encode(bytes(value)).decode()
    if isinstance(value, dict):
        return {str(k): jsonable(v) for k, v in value.items()}
    if isinstance(value, (list, tuple, set)):
        return [jsonable(v) for v in value]
    if etree.iselement(value):
        return element_to_data(value)
    return str(value)


def _parser() -> etree.XMLParser:
    return etree.XMLParser(resolve_entities=False, no_network=True, huge_tree=False, load_dtd=False, dtd_validation=False, remove_comments=True, remove_pis=True)


def _parse(xml: bytes | str) -> etree._Element:
    data = xml.encode() if isinstance(xml, str) else xml
    return etree.fromstring(data, parser=_parser())


def _local(tag: Any) -> str:
    return etree.QName(tag).localname if isinstance(tag, str) else ""


def element_to_data(element: etree._Element) -> Any:
    """XML as plain data: namespaces dropped, a repeated name becomes a list,
    a leaf is its text; attributes appear under "@name"."""
    children = [c for c in element if isinstance(c.tag, str)]
    attributes = {f"@{etree.QName(k).localname}": v for k, v in element.attrib.items()}
    if not children:
        text = (element.text or "").strip()
        if not attributes:
            return text or None
        return {**attributes, "#text": text} if text else attributes
    data: dict[str, Any] = dict(attributes)
    for child in children:
        name = _local(child.tag)
        value = element_to_data(child)
        if name in data:
            if not isinstance(data[name], list):
                data[name] = [data[name]]
            data[name].append(value)
        else:
            data[name] = value
    return data


def _headers_from_xml(xml: str) -> list:
    text = (xml or "").strip()
    if not text:
        return []
    try:
        root = _parse(f'<wrapper>{text}</wrapper>')
    except etree.XMLSyntaxError as exc:
        raise SoapError(f"the SOAP header isn't valid XML: {exc}") from exc
    return list(root)


def _fault_data(code: Any, message: Any, detail: Any = None) -> dict:
    return {"code": str(code) if code is not None else "", "message": str(message) if message is not None else "", "detail": detail}


def call_wsdl(spec: SoapSpec, args: dict) -> dict:
    """{"ok": True, "value": …} or {"ok": False, "fault": {code, message, detail}}.
    A fault is the service answering "no"; anything else raises."""
    if not spec.operation.strip():
        raise SoapError("set the operation")
    client = _client(spec)
    service = client.service
    if spec.endpoint.strip():
        check_url(spec.endpoint)
        service._binding_options["address"] = spec.endpoint.strip()
    try:
        method = service[spec.operation.strip()]
    except AttributeError as exc:
        names = sorted(service._binding._operations.keys())
        raise SoapError(f"no operation \"{spec.operation}\" — this service has: {', '.join(names[:40])}") from exc
    kwargs = dict(args)
    headers = _headers_from_xml(spec.header_xml)
    if headers:
        kwargs["_soapheaders"] = headers
    try:
        value = method(**kwargs)
    except Fault as fault:
        detail = element_to_data(fault.detail) if fault.detail is not None and etree.iselement(fault.detail) else None
        return {"ok": False, "fault": _fault_data(fault.code, fault.message, detail)}
    except TypeError as exc:
        # zeep's way of saying the arguments don't fit the operation.
        raise SoapError(f"the inputs don't fit {spec.operation}: {exc}") from exc
    return {"ok": True, "value": jsonable(serialize_object(value, target_cls=dict))}


def _fault_in(body: etree._Element | None) -> dict | None:
    if body is None:
        return None
    fault = next((c for c in body if _local(c.tag) == "Fault"), None)
    if fault is None:
        return None
    parts = {_local(c.tag): c for c in fault if isinstance(c.tag, str)}
    if "faultcode" in parts or "faultstring" in parts:  # SOAP 1.1
        code = (parts["faultcode"].text if "faultcode" in parts else "") or ""
        message = (parts["faultstring"].text if "faultstring" in parts else "") or ""
        detail = element_to_data(parts["detail"]) if "detail" in parts else None
        return _fault_data(code.strip(), message.strip(), detail)
    code = element_to_data(parts["Code"]) if "Code" in parts else ""  # SOAP 1.2
    reason = element_to_data(parts["Reason"]) if "Reason" in parts else ""
    code_text = code.get("Value", "") if isinstance(code, dict) else code
    reason_text = reason.get("Text", "") if isinstance(reason, dict) else reason
    if isinstance(reason_text, dict):
        reason_text = reason_text.get("#text", "")
    detail = element_to_data(parts["Detail"]) if "Detail" in parts else None
    return _fault_data(code_text, reason_text, detail)


def call_raw(spec: SoapSpec) -> dict:
    url = spec.endpoint.strip()
    if not url:
        raise SoapError("set the service address")
    check_url(url)
    envelope = (spec.envelope or "").strip()
    if not envelope:
        raise SoapError("set the SOAP envelope")
    try:
        _parse(envelope)
    except etree.XMLSyntaxError as exc:
        raise SoapError(f"the SOAP envelope isn't valid XML once filled in: {exc}") from exc
    action = spec.soap_action.strip()
    if spec.version == "1.2":
        content_type = "application/soap+xml; charset=utf-8" + (f'; action="{action}"' if action else "")
        headers = {"Content-Type": content_type}
    else:
        headers = {"Content-Type": "text/xml; charset=utf-8", "SOAPAction": f'"{action}"'}
    with _session(spec) as session:
        with session.post(url, data=envelope.encode("utf-8"), headers=headers, timeout=_timeout(spec), allow_redirects=False, stream=True) as resp:
            content = b""
            for chunk in resp.iter_content(64 * 1024):
                content += chunk
                if len(content) > MAX_RESPONSE_BYTES:
                    raise SoapError("the response is over the 20 MB limit")
            status = resp.status_code
    try:
        root = _parse(content)
    except etree.XMLSyntaxError as exc:
        preview = content[:300].decode("utf-8", errors="replace")
        raise SoapError(f"HTTP {status}, and the answer wasn't XML: {preview}") from exc
    body = next((c for c in root if _local(c.tag) == "Body"), None)
    text = content.decode("utf-8", errors="replace")[:MAX_XML_TEXT]
    fault = _fault_in(body)
    if fault:
        return {"ok": False, "fault": fault, "status": status, "xml": text}
    if status >= 400:
        raise SoapError(f"HTTP {status} from the service")
    if body is None:
        raise SoapError("the answer has no SOAP Body")
    children = [c for c in body if isinstance(c.tag, str)]
    value: Any = element_to_data(children[0]) if len(children) == 1 else element_to_data(body)
    return {"ok": True, "value": value, "status": status, "xml": text}


def _element_inputs(operation: Any) -> list[dict]:
    """The operation's top-level arguments. Best effort: zeep has no public
    API for this, and it doesn't fit every WSDL style."""
    try:
        elements = operation.input.body.type.elements
    except AttributeError:
        return []
    inputs = []
    for name, element in elements:
        try:
            inputs.append({"name": name, "type": str(getattr(element.type, "name", "") or "complex"), "required": not getattr(element, "is_optional", False)})
        except AttributeError:
            continue
    return inputs


def inspect(spec: SoapSpec) -> dict:
    """What the Task Designer's picker lists: every service, port and
    operation in a WSDL."""
    client = _client(spec)
    operations = []
    for service in client.wsdl.services.values():
        for port in service.ports.values():
            for name, operation in sorted(port.binding._operations.items()):
                try:
                    signature = operation.input.signature(as_output=False)
                except Exception:  # noqa: BLE001 - only a display hint, and zeep raises many types
                    signature = ""
                operations.append({"name": name, "service": service.name, "port": port.name, "address": port.binding_options.get("address") or "", "signature": signature[:500], "inputs": _element_inputs(operation)})
    return {"operations": operations}