import importlib
import json
import os
import sys
import urllib.parse
import urllib.request
from typing import Any, Optional

from common.customer_ownership import (
    MISSING_IDENTITY_MESSAGE,
    PARTNER_RECHECK_MESSAGE as NON_PARTNER_MESSAGE,
    SERVER_ERROR_MESSAGE,
    WELCOME_MESSAGE,
    classify_tls_certificate_exception,
    remote_customer_verify as shared_remote_customer_verify,
)


BASE_DIR = os.path.dirname(os.path.abspath(__file__))
WWW_ROOT = os.path.dirname(BASE_DIR)
HUAWEI_SERVICE_DIR = os.path.join(WWW_ROOT, "WaaCloudEcs", "huaweicloud")
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)


class PartnerCustomerLookupError(RuntimeError):
    pass


def _compact(value: Any) -> str:
    return str(value or "").strip().lower()


def _field_token(value: Any) -> str:
    return str(value or "").replace("_", "").replace("-", "").strip().lower()


def classify_customer_guard_error(exc: BaseException) -> str:
    text = str(exc or "").lower()
    if MISSING_IDENTITY_MESSAGE.lower() in text:
        return MISSING_IDENTITY_MESSAGE
    tls_message = classify_tls_certificate_exception(exc)
    if tls_message:
        return tls_message
    return SERVER_ERROR_MESSAGE


IDENTITY_KEY_FIELDS = {
    _field_token(field)
    for field in (
        "account_id",
        "accountId",
        "identity_id",
        "identityId",
        "trn",
        "user_id",
        "userId",
        "customer_id",
        "customerId",
        "customer_name",
        "customerName",
        "id",
        "name",
    )
}
IDENTITY_KEY_PARTS = ("accountid", "identityid", "userid", "customerid", "customername")
CUSTOMER_MATCH_FIELDS = IDENTITY_KEY_FIELDS | {
    _field_token(field)
    for field in (
        "account",
        "account_name",
        "accountName",
        "phone",
        "mobile",
        "mobile_phone",
        "phone_number",
        "email",
        "login_name",
        "user_name",
        "username",
        "display_name",
        "real_name",
    )
}
CUSTOMER_MATCH_PARTS = IDENTITY_KEY_PARTS + ("accountname", "phonenumber", "mobilephone", "loginname", "username")


def _append_unique(values: list[str], value: Any) -> None:
    item = _compact(value)
    if item and item not in values:
        values.append(item)


def _matching_scalar_values(payload: Any, field_names: set[str], field_parts: tuple[str, ...] = ()) -> list[str]:
    values: list[str] = []

    def visit(value: Any, key: str = "") -> None:
        if isinstance(value, dict):
            for child_key, child_value in value.items():
                visit(child_value, str(child_key))
            return
        if isinstance(value, (list, tuple, set)):
            for child_value in value:
                visit(child_value, key)
            return
        token = _field_token(key)
        if not token:
            return
        if token in field_names or any(part in token for part in field_parts):
            _append_unique(values, value)

    visit(payload)
    return values


def _search_keywords(keys: list[str]) -> list[str]:
    keywords: list[str] = []
    for key in keys:
        _append_unique(keywords, key)
        if len(key) >= 16:
            _append_unique(keywords, key[-8:])
    return keywords


def _customer_items(result: Any) -> list[dict[str, Any]]:
    items: list[dict[str, Any]] = []

    def visit(value: Any, depth: int = 0) -> None:
        if depth > 4:
            return
        if isinstance(value, list):
            for item in value:
                if isinstance(item, dict):
                    items.append(item)
            return
        if not isinstance(value, dict):
            return
        for field in ("items", "customers", "list", "records", "rows", "data", "result"):
            child = value.get(field)
            if isinstance(child, list):
                visit(child, depth + 1)
            elif isinstance(child, dict):
                visit(child, depth + 1)

    visit(result)
    return items


def _ensure_partner_backend_ready() -> None:
    old_path = list(sys.path)
    try:
        if HUAWEI_SERVICE_DIR in sys.path:
            sys.path.remove(HUAWEI_SERVICE_DIR)
        sys.path.insert(0, HUAWEI_SERVICE_DIR)
        volc_agent_service = importlib.import_module("volc_agent_service")
        sync_metrics = getattr(volc_agent_service, "sync_metrics", None)
        if callable(sync_metrics):
            result = sync_metrics()
            if isinstance(result, dict) and result.get("ok") is False:
                raise RuntimeError(result.get("message") or "数据接口返回失败")
    except Exception as exc:
        raise PartnerCustomerLookupError(f"火山引擎代理后台登录态或数据接口不可用: {exc}") from exc
    finally:
        sys.path[:] = old_path


def _to_dict(value: Any) -> dict[str, Any]:
    if value is None:
        return {}
    if isinstance(value, dict):
        return value
    to_dict = getattr(value, "to_dict", None)
    if callable(to_dict):
        data = to_dict()
        if isinstance(data, dict):
            return data
    return {}


def _remote_customer_verify(provider: str, **fields: Any) -> dict[str, Any]:
    try:
        return shared_remote_customer_verify(provider, **fields)
    except Exception as exc:
        raise PartnerCustomerLookupError(str(exc)) from exc


def _remote_customer_verify_for_identity(identity: dict[str, Any]) -> dict[str, Any]:
    account_id = str(identity.get("account_id") or identity.get("identity_id") or "").strip()
    return _remote_customer_verify("volc", account_id=account_id)


def _search_partner_customers(keyword: str, limit: int = 20) -> dict[str, Any]:
    old_path = list(sys.path)
    try:
        if HUAWEI_SERVICE_DIR in sys.path:
            sys.path.remove(HUAWEI_SERVICE_DIR)
        sys.path.insert(0, HUAWEI_SERVICE_DIR)
        volc_agent_service = importlib.import_module("volc_agent_service")
        return volc_agent_service.search_customers(keyword=keyword, limit=limit)
    finally:
        sys.path[:] = old_path


def _get_caller_identity(ak: str, sk: str) -> dict[str, Any]:
    import volcenginesdkcore
    import volcenginesdksts
    from volcenginesdksts.models.get_caller_identity_request import GetCallerIdentityRequest

    configuration = volcenginesdkcore.Configuration()
    configuration.ak = ak
    configuration.sk = sk
    configuration.region = "cn-beijing"
    api_client = volcenginesdkcore.ApiClient(configuration)
    api = volcenginesdksts.STSApi(api_client)
    response = api.get_caller_identity(GetCallerIdentityRequest())
    data = _to_dict(response)
    account_id = str(data.get("account_id") or data.get("AccountId") or getattr(response, "account_id", "") or "").strip()
    identity_id = str(data.get("identity_id") or data.get("IdentityId") or getattr(response, "identity_id", "") or "").strip()
    identity_type = str(data.get("identity_type") or data.get("IdentityType") or getattr(response, "identity_type", "") or "").strip()
    trn = str(data.get("trn") or data.get("Trn") or getattr(response, "trn", "") or "").strip()
    return {
        "provider": "volc",
        "account_id": account_id,
        "identity_id": identity_id,
        "identity_type": identity_type,
        "trn": trn,
        "raw": data,
    }


def _identity_keys(identity: dict[str, Any]) -> list[str]:
    keys: list[str] = []
    for value in _matching_scalar_values(identity, IDENTITY_KEY_FIELDS, IDENTITY_KEY_PARTS):
        _append_unique(keys, value)
    return keys


def _customer_matches(customer: dict[str, Any], keys: list[str]) -> bool:
    direct_values = set(_matching_scalar_values(customer, CUSTOMER_MATCH_FIELDS, CUSTOMER_MATCH_PARTS))
    for key in keys:
        if key in direct_values:
            return True
    haystack = json.dumps(customer, ensure_ascii=False).lower()
    return any(len(key) >= 4 and key in haystack for key in keys)


def _first_customer_value(customer: dict[str, Any], *keys: str) -> str:
    for key in keys:
        value = str(customer.get(key) or "").strip()
        if value:
            return value
    return ""


def customer_mark_from_customer(customer: dict[str, Any]) -> str:
    """把代理后台客户资料转换为界面 Mark，优先保留手机号尾号、客户名和后台账号。"""
    if not isinstance(customer, dict):
        return ""
    phone = _first_customer_value(customer, "phone", "mobile", "mobile_phone", "phone_number")
    phone_digits = "".join(ch for ch in phone if ch.isdigit())
    phone_tail = phone_digits[-4:] if len(phone_digits) >= 4 else phone_digits
    customer_name = _first_customer_value(
        customer,
        "customer_name",
        "customerName",
        "nick_name",
        "nickname",
        "display_name",
        "real_name",
        "name",
    )
    account = _first_customer_value(customer, "account", "account_name", "login_name", "loginName", "user_name", "username")
    if customer_name:
        prefix = "" if (phone_tail and customer_name.startswith(phone_tail)) else phone_tail
        mark = f"{prefix}{customer_name}"
        if account and account not in mark:
            mark = f"{mark}#{account}"
        return mark.strip()[:80]
    if account:
        return account.strip()[:80]
    return phone_tail[:80]


MARK_VALUE_FIELDS = {
    _field_token(field)
    for field in (
        "mark",
        "account",
        "account_name",
        "accountName",
        "login_name",
        "loginName",
        "user_name",
        "userName",
        "username",
        "customer_name",
        "customerName",
        "nick_name",
        "nickname",
        "display_name",
        "displayName",
        "real_name",
        "realName",
        "name",
    )
}
MARK_VALUE_PARTS = ("accountname", "loginname", "username", "customername", "nickname", "displayname", "realname")


def _safe_mark_value(value: Any) -> str:
    text = str(value or "").strip()
    if not text:
        return ""
    if text.lower() in {"true", "false", "none", "null", "unknown", "account", "root"}:
        return ""
    if text.isdigit() and len(text) >= 6:
        return ""
    return text[:80]


def _mark_from_lookup_payload(payload: Any) -> str:
    if isinstance(payload, dict):
        customer_mark = customer_mark_from_customer(payload)
        if customer_mark:
            return customer_mark
        for key, value in payload.items():
            token = _field_token(key)
            if (token in MARK_VALUE_FIELDS or any(part in token for part in MARK_VALUE_PARTS)) and not isinstance(value, (dict, list, tuple, set)):
                mark = _safe_mark_value(value)
                if mark:
                    return mark
        for value in payload.values():
            mark = _mark_from_lookup_payload(value)
            if mark:
                return mark
    elif isinstance(payload, (list, tuple, set)):
        for value in payload:
            mark = _mark_from_lookup_payload(value)
            if mark:
                return mark
    return ""


def _find_customer(keys: list[str]) -> Optional[dict[str, Any]]:
    last_error: Optional[Exception] = None
    _ensure_partner_backend_ready()
    for key in _search_keywords(keys):
        try:
            result = _search_partner_customers(key, 50)
        except Exception as exc:
            last_error = exc
            continue
        if not result.get("ok"):
            continue
        for item in _customer_items(result):
            if isinstance(item, dict) and _customer_matches(item, keys):
                return item
    try:
        result = _search_partner_customers("", 500)
    except Exception as exc:
        last_error = exc
    else:
        if result.get("ok"):
            for item in _customer_items(result):
                if isinstance(item, dict) and _customer_matches(item, keys):
                    return item
    if last_error is not None:
        raise PartnerCustomerLookupError(f"火山引擎代理后台客户查询失败: {last_error}")
    return None


def get_account_identity_for_credentials(ak: str, sk: str) -> dict[str, Any]:
    return _get_caller_identity(ak, sk)


def resolve_customer_mark_for_identity(identity: dict[str, Any]) -> tuple[str, dict[str, Any]]:
    detail: dict[str, Any] = {"identity": identity}
    result: dict[str, Any] = {}
    try:
        result = _remote_customer_verify_for_identity(identity)
    except PartnerCustomerLookupError as exc:
        detail["verify_error_type"] = type(exc).__name__
    else:
        detail["verify"] = result
        mark = _mark_from_lookup_payload(result)
        if mark:
            detail["mark_source"] = "remote_verify"
            return mark, detail

    keys = _identity_keys(identity)
    if not keys:
        return "", detail
    try:
        customer = _find_customer(keys)
    except PartnerCustomerLookupError as exc:
        detail["search_error_type"] = type(exc).__name__
        return "", detail
    if not customer:
        return "", detail
    mark = customer_mark_from_customer(customer).strip()
    if not mark:
        return "", detail
    detail["mark_source"] = "partner_customer_search"
    return mark, detail


def resolve_customer_mark_for_credentials(ak: str, sk: str) -> tuple[str, dict[str, Any]]:
    identity = get_account_identity_for_credentials(ak, sk)
    return resolve_customer_mark_for_identity(identity)


def validate_customer_ownership(ak: str, sk: str) -> tuple[bool, str, dict[str, Any]]:
    identity = get_account_identity_for_credentials(ak, sk)
    try:
        result = _remote_customer_verify_for_identity(identity)
    except PartnerCustomerLookupError as exc:
        message = classify_customer_guard_error(exc)
        return False, message, {"identity": identity, "error_type": type(exc).__name__}
    if not result.get("is_customer"):
        return False, NON_PARTNER_MESSAGE, {"identity": identity, "verify": result}
    return True, WELCOME_MESSAGE, {"identity": identity, "verify": result}
