from __future__ import annotations

import hashlib
import hmac
import importlib
import json
import os
import re
import sys
import time
import urllib.parse
import urllib.request
import xml.etree.ElementTree as ET
from typing import Any

from common.customer_ownership import (
    AK_SK_ERROR_MESSAGE,
    MISSING_IDENTITY_MESSAGE,
    NON_PARTNER_MESSAGE,
    SERVER_ERROR_MESSAGE,
    WELCOME_MESSAGE,
    classify_auth_exception,
    remote_customer_verify,
)
from CTYUN.ctyun_openapi import CtyunClient, ctyun_urlopen
from CTYUN.workflow_config import ENDPOINT_BILLING, ENDPOINT_ECS


BASE_DIR = os.path.dirname(os.path.abspath(__file__))
HUAWEI_API_DIR = os.path.abspath(os.path.join(BASE_DIR, "..", "HUAWEI", "api"))


class PartnerCustomerLookupError(RuntimeError):
    pass


CTYUN_IDENTITY_KEYS = {
    "loginname": "login_name",
    "login_name": "login_name",
    "username": "login_name",
    "user_name": "login_name",
    "useraccount": "login_name",
    "user_account": "login_name",
    "account": "account",
    "accountname": "account_name",
    "account_name": "account_name",
    "tenantname": "account_name",
    "tenant_name": "account_name",
    "membername": "account_name",
    "member_name": "account_name",
    "email": "email",
    "emailaddress": "email",
    "email_address": "email",
    "mail": "email",
    "userid": "user_id",
    "user_id": "user_id",
    "accountid": "account_id",
    "account_id": "account_id",
    "tenantid": "account_id",
    "tenant_id": "account_id",
    "customerid": "customer_id",
    "customer_id": "customer_id",
    "customername": "customer_name",
    "customer_name": "customer_name",
    "custname": "customer_name",
    "cust_name": "customer_name",
    "realname": "customer_name",
    "real_name": "customer_name",
    "displayname": "customer_name",
    "display_name": "customer_name",
    "contactname": "customer_name",
    "contact_name": "customer_name",
    "contactperson": "customer_name",
    "contact_person": "customer_name",
    "linkman": "customer_name",
    "link_man": "customer_name",
    "phone": "phone",
    "phonenumber": "phone",
    "tel": "phone",
    "telephone": "phone",
    "mobile": "phone",
    "mobilenumber": "phone",
    "mobilephone": "phone",
    "mobile_phone": "phone",
    "mobile_phone_number": "phone",
    "phone_number": "phone",
    "cellphone": "phone",
    "cell_phone": "phone",
    "cellphonenumber": "phone",
    "contactphone": "phone",
    "contact_phone": "phone",
    "contactphonenumber": "phone",
    "contact_phone_number": "phone",
    "contactmobile": "phone",
    "contact_mobile": "phone",
    "contactmobilenumber": "phone",
    "contact_mobile_number": "phone",
    "customerphone": "phone",
    "customer_phone": "phone",
    "customerphonenumber": "phone",
    "customer_phone_number": "phone",
    "customermobile": "phone",
    "customer_mobile": "phone",
    "customermobilenumber": "phone",
    "customer_mobile_number": "phone",
    "custphone": "phone",
    "cust_phone": "phone",
    "custmobile": "phone",
    "cust_mobile": "phone",
    "accesskey": "access_key",
    "access_key": "access_key",
    "accesskeyid": "access_key",
    "access_key_id": "access_key",
    "accesskeysha256": "access_key_sha256",
    "access_key_sha256": "access_key_sha256",
    "aksha256": "access_key_sha256",
    "ak_sha256": "access_key_sha256",
    "accesskeytail": "access_key_tail",
    "access_key_tail": "access_key_tail",
    "aktail": "access_key_tail",
    "ak_tail": "access_key_tail",
    "accesskeymasked": "access_key_masked",
    "access_key_masked": "access_key_masked",
    "akmasked": "access_key_masked",
    "ak_masked": "access_key_masked",
}
CTYUN_DISPLAY_FIELD_ORDER = (
    "login_name",
    "account",
    "account_name",
    "email",
    "customer_name",
    "account_id",
    "user_id",
    "customer_id",
)
CTYUN_MASK_CHARS = "*＊xX"
CTYUN_IDENTITY_SUMMARY_ORDER = (
    "login_name",
    "account",
    "account_name",
    "email",
    "customer_name",
    "phone",
    "phone_tail",
    "mobile_tail",
    "access_key_sha256",
    "account_id",
    "user_id",
    "customer_id",
)


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


CTYUN_CUSTOMER_MATCH_FIELDS = {
    _field_token(field)
    for field in (
        "login_name",
        "loginName",
        "user_name",
        "username",
        "account",
        "account_name",
        "accountName",
        "email",
        "customer_name",
        "customerName",
        "customer_id",
        "customerId",
        "phone",
        "mobile",
        "mobile_phone",
        "phone_number",
        "real_name",
        "display_name",
        "contact_name",
        "access_key",
        "accessKey",
        "access_key_id",
        "accessKeyId",
        "access_key_sha256",
        "accessKeySha256",
        "ak_sha256",
        "akSha256",
    )
}
CTYUN_CUSTOMER_MATCH_PARTS = (
    "loginname",
    "username",
    "accountname",
    "customername",
    "customerid",
    "phonenumber",
    "mobilephone",
    "accesskey",
    "aksha256",
)


def _is_ctyun_success(data: dict[str, Any]) -> bool:
    status = str(data.get("statusCode") or data.get("code") or data.get("status") or "").strip().lower()
    if status in {"800", "success", "0", "200"}:
        return True
    if data.get("success") is True or data.get("ok") is True:
        return True
    message = str(data.get("message") or data.get("description") or "").lower()
    return "success" in message or "成功" in message


def _validate_ctyun_credentials(ak: str, sk: str) -> dict[str, Any]:
    client = CtyunClient(ak, sk, ENDPOINT_ECS, dry_run=False, timeout=12)
    try:
        data = client.get("/v4/common/get-ecs-flavors", {"regionID": "200000001852"})
    except Exception as exc:
        message = classify_auth_exception(exc)
        raise PartnerCustomerLookupError(message) from exc
    if not isinstance(data, dict):
        raise PartnerCustomerLookupError(AK_SK_ERROR_MESSAGE)
    if not _is_ctyun_success(data):
        reason = json.dumps(data, ensure_ascii=False).lower()
        if any(token in reason for token in ("signature", "unauthorized", "forbidden", "ak", "sk", "401", "403")):
            raise PartnerCustomerLookupError(AK_SK_ERROR_MESSAGE)
        raise PartnerCustomerLookupError(SERVER_ERROR_MESSAGE)
    return {"provider": "ctyun", "credential_checked": True}


def _clean_account_mark(value: Any) -> str:
    text = "".join(str(value or "").split()).strip(" ,;:/\\|\"'<>[](){}")
    if not text or text.lower() in {"success", "ok", "true", "false", "none", "null"}:
        return ""
    if len(text) > 12 and all(ch in "0123456789abcdefABCDEF-" for ch in text):
        return ""
    return text[:80]


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


def _clean_identity_value(value: Any) -> str:
    text = "".join(str(value or "").split()).strip(" ,;:/\\|\"'<>[](){}")
    if not text or text.lower() in {"success", "ok", "true", "false", "none", "null"}:
        return ""
    return text[:160]


def _is_masked_value(value: Any) -> bool:
    return any(ch in str(value or "") for ch in CTYUN_MASK_CHARS)


def _phone_tail_from_value(value: Any) -> str:
    digits = re.sub(r"\D", "", str(value or ""))
    if len(digits) < 4:
        return ""
    return digits[-4:]


def _access_key_fingerprint_fields(ak: str) -> dict[str, str]:
    value = str(ak or "").strip()
    if not value:
        return {}
    fields = {
        "access_key_sha256": hashlib.sha256(value.encode("utf-8")).hexdigest(),
    }
    if len(value) >= 6:
        fields["access_key_tail"] = value[-6:]
    if len(value) >= 10:
        fields["access_key_masked"] = f"{value[:4]}...{value[-6:]}"
    return fields


def _add_identity_value(fields: dict[str, str], field: str, value: Any) -> None:
    cleaned = _clean_identity_value(value)
    if not cleaned:
        return
    if field == "phone":
        tail = _phone_tail_from_value(cleaned)
        if tail and not fields.get("phone_tail"):
            fields["phone_tail"] = tail
        if tail and not fields.get("mobile_tail"):
            fields["mobile_tail"] = tail
        if not _is_masked_value(cleaned) and not fields.get("phone"):
            fields["phone"] = cleaned
        return
    if not fields.get(field):
        fields[field] = cleaned


def _identity_display_value(fields: dict[str, str]) -> str:
    for field in CTYUN_DISPLAY_FIELD_ORDER:
        value = _clean_account_mark(fields.get(field))
        if value:
            return value
    return ""


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


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 token and (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, (dict, list)):
                visit(child, depth + 1)

    visit(result)
    return items


def _call_ctyun_agent_service(method_name: str, *args: Any, **kwargs: Any) -> Any:
    old_path = list(sys.path)
    try:
        if HUAWEI_API_DIR in sys.path:
            sys.path.remove(HUAWEI_API_DIR)
        sys.path.insert(0, HUAWEI_API_DIR)
        service = importlib.import_module("ctyun_agent_service")
        method = getattr(service, method_name)
        return method(*args, **kwargs)
    finally:
        sys.path[:] = old_path


def _ensure_partner_backend_ready() -> None:
    try:
        result = _call_ctyun_agent_service("sync_metrics")
    except AttributeError:
        return
    except Exception as exc:
        raise PartnerCustomerLookupError(f"天翼云代理后台登录态或数据接口不可用: {exc}") from exc
    if isinstance(result, dict) and result.get("ok") is False:
        raise PartnerCustomerLookupError(str(result.get("message") or "天翼云代理后台数据接口返回失败"))


def _search_partner_customers(keyword: str, limit: int = 50) -> dict[str, Any]:
    try:
        return _call_ctyun_agent_service("search_customers", keyword=keyword, limit=limit)
    except TypeError:
        return _call_ctyun_agent_service("search_customers", keyword=keyword)


def _identity_keys(fields: dict[str, str]) -> list[str]:
    keys: list[str] = []
    for field in CTYUN_IDENTITY_SUMMARY_ORDER:
        _append_unique(keys, fields.get(field))
    return keys


def _customer_matches(customer: dict[str, Any], keys: list[str]) -> bool:
    direct_values = set(_matching_scalar_values(customer, CTYUN_CUSTOMER_MATCH_FIELDS, CTYUN_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 _find_partner_customer(fields: dict[str, str]) -> dict[str, Any] | None:
    keys = _identity_keys(fields)
    if not keys:
        return None
    last_error: Exception | None = None
    _ensure_partner_backend_ready()
    for keyword in _search_keywords(keys):
        try:
            result = _search_partner_customers(keyword, 50)
        except Exception as exc:
            last_error = exc
            continue
        if not isinstance(result, dict) or 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 isinstance(result, dict) and 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 _identity_fields_from_payload(data: Any) -> dict[str, str]:
    fields: dict[str, str] = {}

    def walk(value: Any, key: str = "") -> None:
        if isinstance(value, dict):
            for child_key, child_value in value.items():
                walk(child_value, str(child_key))
            return
        if isinstance(value, list):
            for item in value:
                walk(item, key)
            return
        normalized = _normalize_identity_key(key)
        field = CTYUN_IDENTITY_KEYS.get(normalized)
        if field:
            _add_identity_value(fields, field, value)

    walk(data)
    return fields


def _first_account_mark_from_payload(data: Any) -> str:
    mark = _identity_display_value(_identity_fields_from_payload(data))
    if mark:
        return mark
    preferred_keys = {
        "loginname",
        "login_name",
        "username",
        "user_name",
        "useraccount",
        "user_account",
        "accountname",
        "account_name",
        "customername",
        "customer_name",
        "custname",
        "cust_name",
        "tenantname",
        "tenant_name",
        "membername",
        "member_name",
    }
    fallback_keys = {"name", "displayname", "display_name", "realname", "real_name"}

    def walk(value: Any, *, allow_fallback: bool) -> str:
        if isinstance(value, dict):
            for key, item in value.items():
                normalized = str(key).replace("-", "_").replace(" ", "_").lower()
                if normalized in preferred_keys and not isinstance(item, (dict, list)):
                    mark = _clean_account_mark(item)
                    if mark:
                        return mark
            if allow_fallback:
                for key, item in value.items():
                    normalized = str(key).replace("-", "_").replace(" ", "_").lower()
                    if normalized in fallback_keys and not isinstance(item, (dict, list)):
                        mark = _clean_account_mark(item)
                        if mark:
                            return mark
            for item in value.values():
                mark = walk(item, allow_fallback=allow_fallback)
                if mark:
                    return mark
        elif isinstance(value, list):
            for item in value:
                mark = walk(item, allow_fallback=allow_fallback)
                if mark:
                    return mark
        return ""

    return walk(data, allow_fallback=False) or walk(data, allow_fallback=True)


def _ctyun_request_json(
    client: CtyunClient,
    endpoint: str,
    method: str,
    path: str,
    *,
    body: dict[str, Any] | None = None,
    query: dict[str, Any] | None = None,
    extra_headers: dict[str, str] | None = None,
) -> dict[str, Any]:
    body = body or {}
    query = query or {}
    method_upper = method.upper()
    url = endpoint.rstrip("/") + path
    if query:
        query_text = "&".join(f"{key}={urllib.parse.quote(str(value), safe='')}" for key, value in sorted(query.items()))
        url = f"{url}?{query_text}"
    payload = None
    sign_body = {}
    sign_query = query if method_upper == "GET" else {}
    if method_upper != "GET":
        sign_body = body
        payload = json.dumps(body, ensure_ascii=False).encode("utf-8")
    headers = client.signed_headers(query=sign_query, body=sign_body)
    headers.update(extra_headers or {})
    req = urllib.request.Request(url, data=payload, headers=headers, method=method_upper)
    with ctyun_urlopen(req, timeout=8) as resp:
        text = resp.read().decode("utf-8", errors="replace")
    try:
        data = json.loads(text)
    except json.JSONDecodeError:
        data = {"raw": text}
    return data if isinstance(data, dict) else {"returnObj": data}


def _ctyun_v4_signing_key(sk: str, date: str, region: str, service: str) -> bytes:
    key = ("AWS4" + sk).encode("utf-8")
    for value in (date, region, service, "aws4_request"):
        key = hmac.new(key, value.encode("utf-8"), hashlib.sha256).digest()
    return key


def _ctyun_oos_iam_request(ak: str, sk: str, action: str, *, service: str = "sts", **params: Any) -> str:
    host = "oos-cn-iam.ctyunapi.cn"
    region = "cn"
    content_type = "application/octet-stream"
    payload_hash = "UNSIGNED-PAYLOAD"
    request_params: dict[str, Any] = {"Action": action, "Version": "2010-05-08"}
    request_params.update({key: value for key, value in params.items() if value not in (None, "")})
    body_text = urllib.parse.urlencode(sorted(request_params.items()))
    payload = body_text.encode("utf-8")

    amz_date = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime())
    date_stamp = amz_date[:8]
    signed_headers = "content-type;host;x-amz-content-sha256;x-amz-date"
    canonical_headers = (
        f"content-type:{content_type}\n"
        f"host:{host}\n"
        f"x-amz-content-sha256:{payload_hash}\n"
        f"x-amz-date:{amz_date}\n"
    )
    canonical_request = "\n".join(
        [
            "POST",
            "/",
            "",
            canonical_headers,
            signed_headers,
            payload_hash,
        ]
    )
    credential_scope = f"{date_stamp}/{region}/{service}/aws4_request"
    string_to_sign = "\n".join(
        [
            "AWS4-HMAC-SHA256",
            amz_date,
            credential_scope,
            hashlib.sha256(canonical_request.encode("utf-8")).hexdigest(),
        ]
    )
    signature = hmac.new(
        _ctyun_v4_signing_key(sk, date_stamp, region, service),
        string_to_sign.encode("utf-8"),
        hashlib.sha256,
    ).hexdigest()
    headers = {
        "Content-Type": content_type,
        "Host": host,
        "X-Amz-Content-Sha256": payload_hash,
        "X-Amz-Date": amz_date,
        "Authorization": (
            "AWS4-HMAC-SHA256 "
            f"Credential={ak}/{credential_scope}, "
            f"SignedHeaders={signed_headers}, "
            f"Signature={signature}"
        ),
        "Connection": "close",
    }
    req = urllib.request.Request(f"https://{host}/", data=payload, headers=headers, method="POST")
    with ctyun_urlopen(req, timeout=8) as resp:
        return resp.read().decode("utf-8", errors="replace")


def _xml_local_name(tag: str) -> str:
    return str(tag or "").rsplit("}", 1)[-1]


def _ctyun_xml_text(node: ET.Element, tag_name: str) -> str:
    wanted = tag_name.lower()
    for child in node.iter():
        if _xml_local_name(child.tag).lower() == wanted and child.text:
            return child.text.strip()
    return ""


def _ctyun_mark_from_oos_xml(xml_text: str, ak: str = "") -> str:
    try:
        root = ET.fromstring(xml_text)
    except ET.ParseError:
        return ""
    ak_value = (ak or "").strip()
    for node in root.iter():
        if _xml_local_name(node.tag).lower() != "member":
            continue
        member_ak = _ctyun_xml_text(node, "AccessKeyId")
        if ak_value and member_ak and member_ak != ak_value:
            continue
        mark = _clean_account_mark(_ctyun_xml_text(node, "UserName"))
        if mark:
            return mark
    return _clean_account_mark(_ctyun_xml_text(root, "UserName"))


def _ctyun_identity_from_oos_xml(xml_text: str, ak: str = "") -> dict[str, str]:
    try:
        root = ET.fromstring(xml_text)
    except ET.ParseError:
        return {}

    def fields_from_node(node: ET.Element) -> dict[str, str]:
        fields: dict[str, str] = {}
        for child in node.iter():
            key = _normalize_identity_key(_xml_local_name(child.tag))
            field = CTYUN_IDENTITY_KEYS.get(key)
            if field and child.text:
                _add_identity_value(fields, field, child.text)
        return fields

    ak_value = (ak or "").strip()
    for node in root.iter():
        if _xml_local_name(node.tag).lower() != "member":
            continue
        member_ak = _ctyun_xml_text(node, "AccessKeyId")
        if ak_value and member_ak and member_ak != ak_value:
            continue
        fields = fields_from_node(node)
        if fields:
            return fields
    return fields_from_node(root)


def resolve_login_name_for_credentials(ak: str, sk: str) -> tuple[str, dict[str, Any]]:
    ak_value = (ak or "").strip()
    sk_value = (sk or "").strip()
    detail: dict[str, Any] = {"resolved": False, "source": "", "attempts": []}
    if not ak_value or not sk_value:
        return "", detail

    oos_attempts = (
        ("GetAccessKeyLastUsed", {"AccessKeyId": ak_value}),
        ("ListAccessKeys", {"AccessKeyId": ak_value}),
        ("GetUser", {}),
    )
    for action, params in oos_attempts:
        for service in ("iam", "sts", "s3"):
            source = f"oos_iam:{service}:{action}"
            try:
                xml_text = _ctyun_oos_iam_request(ak_value, sk_value, action, service=service, **params)
            except Exception as exc:
                detail["attempts"].append({"source": source, "error_type": type(exc).__name__})
                continue
            identity_fields = _ctyun_identity_from_oos_xml(xml_text, ak_value)
            mark = _identity_display_value(identity_fields) or _ctyun_mark_from_oos_xml(xml_text, ak_value)
            detail["attempts"].append({"source": source, "matched": bool(mark or identity_fields)})
            if mark or identity_fields:
                detail.update({"resolved": True, "source": source, "identity_fields": identity_fields})
                return mark, detail

    billing_headers = {"regionid": "200000001852", "urlType": "CTAPI"}
    candidates = [
        (ENDPOINT_BILLING, "POST", "/v1/bill/queryAccountBalance", {}, {}, billing_headers),
        (ENDPOINT_BILLING, "POST", "/v1/account/queryAccountInfo", {}, {}, billing_headers),
        (ENDPOINT_BILLING, "POST", "/v1/account/getAccountInfo", {}, {}, billing_headers),
        (ENDPOINT_BILLING, "POST", "/v1/user/queryUserInfo", {}, {}, billing_headers),
        ("https://ctiam-global.ctapi.ctyun.cn", "GET", "/v4/iam/user", {}, {}, {}),
        ("https://ctiam-global.ctapi.ctyun.cn", "GET", "/v4/iam/user/get", {}, {}, {}),
        ("https://iam-global.ctapi.ctyun.cn", "GET", "/v4/iam/user", {}, {}, {}),
    ]
    for endpoint, method, path, body, query, headers in candidates:
        try:
            client = CtyunClient(ak_value, sk_value, endpoint, dry_run=False, timeout=8)
            data = _ctyun_request_json(client, endpoint, method, path, body=body, query=query, extra_headers=headers)
        except Exception as exc:
            detail["attempts"].append({"source": path, "error_type": type(exc).__name__})
            continue
        identity_fields = _identity_fields_from_payload(data)
        mark = _identity_display_value(identity_fields) or _first_account_mark_from_payload(data)
        detail["attempts"].append({"source": path, "matched": bool(mark or identity_fields)})
        if mark or identity_fields:
            detail.update({"resolved": True, "source": path, "identity_fields": identity_fields})
            return mark, detail
    return "", detail


def _remote_verify(fields: dict[str, str]) -> dict[str, Any]:
    try:
        return remote_customer_verify("ctyun", **fields)
    except Exception as exc:
        raise PartnerCustomerLookupError(str(exc)) from exc


def _remote_verification_fields(fields: dict[str, str]) -> dict[str, str]:
    unsupported = {"access_key", "access_key_id", "access_key_tail"}
    return {key: value for key, value in fields.items() if key not in unsupported}


def _remote_fingerprint_fields_unsupported(exc: Exception, fields: dict[str, str]) -> bool:
    if not fields:
        return False
    if set(fields) - {"access_key_sha256", "access_key_masked"}:
        return False
    return "HTTP Error 400" in str(exc or "")


def _verification_fields_from_identity(identity_fields: dict[str, str], ak: str = "") -> dict[str, str]:
    fields = {
        str(key): str(value).strip()
        for key, value in (identity_fields or {}).items()
        if str(value or "").strip()
    }
    ak_value = str(ak or "").strip()
    if ak_value:
        fields.update({key: value for key, value in _access_key_fingerprint_fields(ak_value).items() if value})
        fields.setdefault("access_key", ak_value)
        fields.setdefault("access_key_id", ak_value)
    if not fields.get("login_name"):
        for alias in ("account", "account_name", "email"):
            value = fields.get(alias)
            if value:
                fields["login_name"] = value
                break
    return fields


def _local_partner_verify(fields: dict[str, str]) -> dict[str, Any]:
    customer = _find_partner_customer(fields)
    return {"ok": True, "is_customer": bool(customer), "source": "ctyun_agent_service"}


def _identity_summary(resolve_detail: dict[str, Any], verify_fields: dict[str, str]) -> dict[str, Any]:
    attempts = resolve_detail.get("attempts") if isinstance(resolve_detail, dict) else []
    return {
        "source": str(resolve_detail.get("source") or "") if isinstance(resolve_detail, dict) else "",
        "fields": [field for field in CTYUN_IDENTITY_SUMMARY_ORDER if verify_fields.get(field)],
        "attempts": len(attempts) if isinstance(attempts, list) else 0,
        "resolved": bool(resolve_detail.get("resolved")) if isinstance(resolve_detail, dict) else False,
    }


def validate_customer_ownership(ak: str, sk: str, mark: str = "") -> tuple[bool, str, dict[str, Any]]:
    try:
        identity = _validate_ctyun_credentials(ak, sk)
    except PartnerCustomerLookupError as exc:
        text = str(exc or "")
        if AK_SK_ERROR_MESSAGE in text:
            return False, AK_SK_ERROR_MESSAGE, {"error_type": type(exc).__name__}
        return False, classify_auth_exception(exc), {"error_type": type(exc).__name__}
    resolved_mark, resolve_detail = resolve_login_name_for_credentials(ak, sk)
    verify_fields = _verification_fields_from_identity(dict(resolve_detail.get("identity_fields") or {}), ak)
    display_mark = resolved_mark or _identity_display_value(verify_fields)
    summary = _identity_summary(resolve_detail, verify_fields)
    if not verify_fields:
        return False, NON_PARTNER_MESSAGE, {"identity": identity, "resolve": resolve_detail, "identity_summary": summary}
    remote_result: dict[str, Any] | None = None
    remote_error: PartnerCustomerLookupError | None = None
    remote_fields = _remote_verification_fields(verify_fields)
    if remote_fields:
        try:
            remote_result = _remote_verify(remote_fields)
        except PartnerCustomerLookupError as exc:
            remote_error = exc
        else:
            if remote_result.get("is_customer"):
                return True, WELCOME_MESSAGE, {
                    "identity": identity,
                    "verify": {**remote_result, "source": remote_result.get("source") or "remote_verify"},
                    "mark": display_mark,
                    "resolve": resolve_detail,
                    "identity_summary": summary,
                }
    try:
        local_result = _local_partner_verify(verify_fields)
    except PartnerCustomerLookupError as exc:
        if remote_error is not None:
            if MISSING_IDENTITY_MESSAGE in str(remote_error or "") or _remote_fingerprint_fields_unsupported(remote_error, remote_fields):
                return False, NON_PARTNER_MESSAGE, {
                    "identity": identity,
                    "mark": display_mark,
                    "resolve": resolve_detail,
                    "identity_summary": summary,
                    "error_type": type(remote_error).__name__,
                    "local_error_type": type(exc).__name__,
                }
            return False, classify_auth_exception(remote_error), {
                "identity": identity,
                "mark": display_mark,
                "resolve": resolve_detail,
                "identity_summary": summary,
                "error_type": type(remote_error).__name__,
                "local_error_type": type(exc).__name__,
            }
        local_result = {"ok": False, "is_customer": False, "source": "ctyun_agent_service", "error_type": type(exc).__name__}
    if local_result.get("is_customer"):
        return True, WELCOME_MESSAGE, {
            "identity": identity,
            "verify": local_result,
            "mark": display_mark,
            "resolve": resolve_detail,
            "identity_summary": summary,
        }
    if not remote_fields and not summary.get("resolved"):
        return False, SERVER_ERROR_MESSAGE, {
            "identity": identity,
            "local_verify": local_result,
            "mark": display_mark,
            "resolve": resolve_detail,
            "identity_summary": summary,
        }
    if remote_error is not None:
        if MISSING_IDENTITY_MESSAGE in str(remote_error or "") or _remote_fingerprint_fields_unsupported(remote_error, remote_fields):
            return False, NON_PARTNER_MESSAGE, {
                "identity": identity,
                "mark": display_mark,
                "resolve": resolve_detail,
                "identity_summary": summary,
                "error_type": type(remote_error).__name__,
                "local_verify": local_result,
            }
        return False, classify_auth_exception(remote_error), {
            "identity": identity,
            "mark": display_mark,
            "resolve": resolve_detail,
            "identity_summary": summary,
            "error_type": type(remote_error).__name__,
            "local_verify": local_result,
        }
    return False, NON_PARTNER_MESSAGE, {
        "identity": identity,
        "verify": remote_result,
        "local_verify": local_result,
        "mark": display_mark,
        "resolve": resolve_detail,
        "identity_summary": summary,
    }
