from __future__ import annotations

import json
import os
import ssl
import urllib.parse
import urllib.request
from typing import Any


CLIENT_CUSTOMER_VERIFY_URL = os.getenv(
    "WAWA_CUSTOMER_VERIFY_URL",
    "https://www.wawaapi.com/api/client/customer-ownership/verify",
)
CUSTOMER_VERIFY_TIMEOUT_SECONDS = float(os.getenv("WAWA_CUSTOMER_VERIFY_TIMEOUT_SECONDS", "25") or 25)
HUAWEI_CUSTOMER_VERIFY_TIMEOUT_SECONDS = max(
    CUSTOMER_VERIFY_TIMEOUT_SECONDS,
    float(os.getenv("WAWA_HUAWEI_CUSTOMER_VERIFY_TIMEOUT_SECONDS", "60") or 60),
)

CHECKING_MESSAGE = "正在校验AK/SK合作伙伴归属，请稍等。"
WELCOME_MESSAGE = "您好合作伙伴，欢迎使用！查询账号资源中。"
READY_MESSAGE = "您好合作伙伴，欢迎使用！账号资源读取完毕。"
NON_PARTNER_MESSAGE = "校验AK/SK非合作伙伴，请在软件右上方注册链接内注册/绑定为合作伙伴后继续使用，或在软件左侧“联系客户经理”。"
PARTNER_RECHECK_MESSAGE = "校验AK/SK非合作伙伴，请在软件右上方注册链接内注册/绑定为合作伙伴后继续使用或联系客户经理（软件左侧相同）。"
AK_SK_ERROR_MESSAGE = "AK/SK错误，请检查后重试。"
SERVER_ERROR_MESSAGE = "服务器客户归属校验失败，请稍后重试。"
TLS_CERTIFICATE_ERROR_MESSAGE = "当前电脑无法验证软件服务器的HTTPS安全证书，请更新系统根证书或更换网络后重试。"
TLS_CERT_CHAIN_ERROR_MESSAGE = (
    "当前电脑无法验证软件服务器的HTTPS安全证书链。\n"
    "具体原因：电脑缺少或未信任证书签发机构的根证书。\n"
    "请修复或更新Windows根证书后重试；仍失败请关闭代理、VPN或杀毒软件的HTTPS扫描，或更换网络。"
)
TLS_CERT_TIME_ERROR_MESSAGE = (
    "当前电脑无法验证软件服务器的HTTPS安全证书。\n"
    "具体原因：电脑日期、时间或时区不正确，或者证书已过期、尚未生效。\n"
    "请校准Windows日期、时间和时区并更新根证书后重试；仍失败请更换网络。"
)
TLS_CERT_HOSTNAME_ERROR_MESSAGE = (
    "当前网络返回的HTTPS安全证书与软件服务器域名不一致。\n"
    "具体原因：代理、VPN、杀毒软件HTTPS扫描或当前网络可能拦截了连接。\n"
    "请关闭相关拦截后重试，或更换网络。为保护AK/SK，软件已停止本次校验。"
)
TLS_CERT_UNTRUSTED_ERROR_MESSAGE = (
    "当前网络返回了自签名或不受信任的HTTPS安全证书。\n"
    "具体原因：代理、VPN、杀毒软件HTTPS扫描或当前网络可能替换了服务器证书。\n"
    "请关闭相关拦截后重试，或更换网络。为保护AK/SK，软件不会忽略证书错误。"
)
TLS_CERT_GENERAL_ERROR_MESSAGE = (
    "当前电脑无法验证软件服务器的HTTPS安全证书。\n"
    "请修复或更新Windows根证书后重试；仍失败请关闭代理、VPN或杀毒软件的HTTPS扫描，或更换网络。"
)
MISSING_IDENTITY_MESSAGE = "AK/SK已通过，但未能读取可核对的账号归属信息，请联系客户经理确认绑定状态。"


class PartnerCustomerLookupError(RuntimeError):
    pass


def _trusted_ssl_contexts() -> list[ssl.SSLContext]:
    contexts: list[ssl.SSLContext] = []
    try:
        import certifi

        contexts.append(ssl.create_default_context(cafile=certifi.where()))
    except Exception:
        pass
    contexts.append(ssl.create_default_context())
    return contexts


def _clone_request(request: urllib.request.Request) -> urllib.request.Request:
    return urllib.request.Request(
        request.full_url,
        data=request.data,
        headers=dict(request.header_items()),
        origin_req_host=request.origin_req_host,
        unverifiable=request.unverifiable,
        method=request.get_method(),
    )


def _exception_chain_text(exc: BaseException) -> str:
    pending: list[BaseException] = [exc]
    seen: set[int] = set()
    parts: list[str] = []
    while pending:
        current = pending.pop(0)
        marker = id(current)
        if marker in seen:
            continue
        seen.add(marker)
        parts.append(str(current or ""))
        verify_message = getattr(current, "verify_message", None)
        if verify_message:
            parts.append(str(verify_message))
        reason = getattr(current, "reason", None)
        if isinstance(reason, BaseException):
            pending.append(reason)
        elif reason:
            parts.append(str(reason))
        for linked in (getattr(current, "__cause__", None), getattr(current, "__context__", None)):
            if isinstance(linked, BaseException):
                pending.append(linked)
    return "\n".join(parts).lower()


def classify_tls_certificate_exception(exc: BaseException) -> str:
    text = _exception_chain_text(exc)
    has_certificate_signal = TLS_CERTIFICATE_ERROR_MESSAGE.lower() in text or any(
        token in text
        for token in (
            "certificate",
            "sslcertverificationerror",
            "hostname mismatch",
            "ip address mismatch",
            "unknown ca",
        )
    )
    if not has_certificate_signal:
        return ""
    if any(
        token in text
        for token in (
            "hostname mismatch",
            "ip address mismatch",
            "certificate is not valid for",
        )
    ):
        return TLS_CERT_HOSTNAME_ERROR_MESSAGE
    if any(
        token in text
        for token in (
            "certificate has expired",
            "certificate is not yet valid",
            "expired certificate",
            "not yet valid",
        )
    ):
        return TLS_CERT_TIME_ERROR_MESSAGE
    if any(
        token in text
        for token in (
            "self signed certificate",
            "self-signed certificate",
            "unknown ca",
        )
    ):
        return TLS_CERT_UNTRUSTED_ERROR_MESSAGE
    if any(
        token in text
        for token in (
            "unable to get local issuer certificate",
            "unable to get issuer certificate",
            "unable to verify the first certificate",
            "unable to verify first certificate",
        )
    ):
        return TLS_CERT_CHAIN_ERROR_MESSAGE
    if TLS_CERTIFICATE_ERROR_MESSAGE.lower() in text or any(
        token in text
        for token in (
            "certificate_verify_failed",
            "certificate verify failed",
            "sslcertverificationerror",
        )
    ):
        return TLS_CERT_GENERAL_ERROR_MESSAGE
    return ""


def _urlopen_with_verified_fallback(request: urllib.request.Request, *, timeout: float):
    """Try the bundled public CA store and Windows trust store, with and without a proxy."""
    last_error: Exception | None = None
    for context in _trusted_ssl_contexts():
        try:
            return urllib.request.urlopen(_clone_request(request), timeout=timeout, context=context)
        except Exception as exc:
            last_error = exc
        direct_opener = urllib.request.build_opener(
            urllib.request.ProxyHandler({}),
            urllib.request.HTTPSHandler(context=context),
        )
        try:
            return direct_opener.open(_clone_request(request), timeout=timeout)
        except Exception as exc:
            last_error = exc
    assert last_error is not None
    raise last_error


def _is_tls_certificate_error(exc: BaseException) -> bool:
    return bool(classify_tls_certificate_exception(exc))


def remote_customer_verify(provider: str, **fields: Any) -> dict[str, Any]:
    params = {
        key: str(value).strip()
        for key, value in fields.items()
        if str(value or "").strip()
    }
    params["provider"] = provider
    if len(params) <= 1:
        raise PartnerCustomerLookupError(MISSING_IDENTITY_MESSAGE)
    query = urllib.parse.urlencode(params)
    request = urllib.request.Request(
        f"{CLIENT_CUSTOMER_VERIFY_URL}?{query}",
        headers={
            "Accept": "application/json",
            "User-Agent": "YunTools-Client/6.18",
        },
    )
    timeout = HUAWEI_CUSTOMER_VERIFY_TIMEOUT_SECONDS if provider == "huawei" else CUSTOMER_VERIFY_TIMEOUT_SECONDS
    try:
        with _urlopen_with_verified_fallback(request, timeout=timeout) as response:
            raw = response.read().decode("utf-8", errors="replace")
    except Exception as exc:
        tls_message = classify_tls_certificate_exception(exc)
        if tls_message:
            raise PartnerCustomerLookupError(tls_message) from exc
        raise PartnerCustomerLookupError(f"{SERVER_ERROR_MESSAGE} {exc}") from exc
    try:
        data = json.loads(raw)
    except Exception as exc:
        raise PartnerCustomerLookupError(SERVER_ERROR_MESSAGE) from exc
    if not isinstance(data, dict):
        raise PartnerCustomerLookupError(SERVER_ERROR_MESSAGE)
    if not data.get("ok"):
        message = str(data.get("message") or "").strip()
        raise PartnerCustomerLookupError(message or SERVER_ERROR_MESSAGE)
    return data


def ownership_message_from_result(result: dict[str, Any]) -> str:
    if result.get("is_customer"):
        return WELCOME_MESSAGE
    return NON_PARTNER_MESSAGE


def classify_auth_exception(exc: BaseException) -> str:
    tls_message = classify_tls_certificate_exception(exc)
    if tls_message:
        return tls_message
    text = _exception_chain_text(exc)
    if any(
        token in text
        for token in (
            "401",
            "403",
            "forbidden",
            "unauthorized",
            "signature",
            "invalid access",
            "invalidaccesskey",
            "incorrect iam authentication",
            "verify ak sk signature failed",
            "apigw.0301",
        )
    ):
        return AK_SK_ERROR_MESSAGE
    return SERVER_ERROR_MESSAGE
