from __future__ import annotations

import hashlib
import hmac
import json
import secrets
from datetime import datetime, timezone
from typing import Any
from urllib.parse import quote, urlencode, urljoin, urlparse

import aiohttp

from mode.verification.base import RedirectSession, VerificationProviderError, VerificationResult

DEFAULT_API_BASE_URL = "https://server.face.ly-y.cn/"
DEFAULT_H5_BASE_URL = "https://face.ly-y.cn/"
TASK_CREATE_PATH = "/api/merchant/verify/tasks"
TASK_LOOKUP_PATH = "/api/merchant/verify/tasks/lookup"
TASK_QUERY_PATH = "/api/merchant/verify/tasks/{task_no}"
TERMINAL_FAILED_STATUSES = {"failed", "canceled", "cancelled", "expired"}


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


def _first_config_value(config: Any, *names: str) -> str:
    for name in names:
        value = _clean_text(getattr(config, name, None))
        if value:
            return value
    return ""


def _mask_id_card(value: str | None) -> str | None:
    if not value:
        return None
    text = str(value)
    if len(text) <= 8:
        return "*" * len(text)
    return f"{text[:6]}{'*' * max(0, len(text) - 10)}{text[-4:]}"


def _mask_name(value: str | None) -> str | None:
    if not value:
        return None
    text = str(value)
    if len(text) <= 1:
        return "*"
    return f"{text[0]}{'*' * (len(text) - 1)}"


def _sha256_hex(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def _utc_timestamp() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")


def _float_or_none(value: Any) -> float | None:
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def _normalize_api_base_url(value: str) -> str:
    parsed = urlparse(value)
    if parsed.hostname == "face.ly-y.cn":
        return DEFAULT_API_BASE_URL
    return value


class LeafSmH5FaceProvider:
    name = "leafsm_h5_face"
    supported_modes = ("face_redirect",)

    def __init__(self, config: Any):
        self.app_id = _first_config_value(
            config,
            "LEAFSM_H5_FACE_APP_ID",
            "LEAFSM_APP_ID",
            "APP_ID",
        )
        self.app_secret = _first_config_value(
            config,
            "LEAFSM_H5_FACE_APP_SECRET",
            "LEAFSM_APP_SECRET",
            "APP_SECRET",
            "SECRET_KEY",
        )
        self.base_url = _first_config_value(
            config,
            "LEAFSM_H5_FACE_API_BASE_URL",
            "LEAFSM_H5_FACE_BASE_URL",
            "LEAFSM_BASE_URL",
            "BASE_URL",
        ) or DEFAULT_API_BASE_URL
        self.h5_base_url = _first_config_value(
            config,
            "LEAFSM_H5_FACE_H5_BASE_URL",
            "LEAFSM_H5_BASE_URL",
            "H5_BASE_URL",
        ) or DEFAULT_H5_BASE_URL
        self.timeout = aiohttp.ClientTimeout(
            total=float(getattr(config, "LEAFSM_H5_FACE_TIMEOUT", 20) or 20),
        )

        if not self.app_id or not self.app_secret:
            raise VerificationProviderError("LeafSM H5 人脸核身插件未配置 AppId 或 AppSecret", status_code=400)
        if not self.base_url.lower().startswith(("http://", "https://")):
            raise VerificationProviderError("LeafSM H5 人脸核身 API Base URL 必须是 http(s) 地址", status_code=400)
        if not self.h5_base_url.lower().startswith(("http://", "https://")):
            raise VerificationProviderError("LeafSM H5 人脸核身 H5 Base URL 必须是 http(s) 地址", status_code=400)
        self.base_url = _normalize_api_base_url(self.base_url).rstrip("/") + "/"
        self.h5_base_url = self.h5_base_url.rstrip("/") + "/"

    def _url(self, path: str) -> str:
        return urljoin(self.base_url, path.lstrip("/"))

    def _h5_url(self, path: str) -> str:
        return urljoin(self.h5_base_url, path.lstrip("/"))

    def _headers(self, method: str, path: str, body_bytes: bytes) -> dict[str, str]:
        body_sha256 = _sha256_hex(body_bytes)
        timestamp = _utc_timestamp()
        nonce = secrets.token_urlsafe(18)
        sign_text = "\n".join(
            [
                method.upper(),
                path,
                timestamp,
                nonce,
                body_sha256,
            ]
        )
        signature = hmac.new(
            self.app_secret.encode("utf-8"),
            sign_text.encode("utf-8"),
            hashlib.sha256,
        ).hexdigest()
        return {
            "Content-Type": "application/json",
            "X-App-Id": self.app_id,
            "X-Timestamp": timestamp,
            "X-Nonce": nonce,
            "X-Body-Sha256": body_sha256,
            "X-Signature": signature,
        }

    async def _request_json(
        self,
        method: str,
        path: str,
        payload: dict[str, Any] | None = None,
        query: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        body_bytes = b""
        if payload is not None:
            body_bytes = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
        headers = self._headers(method, path, body_bytes)
        kwargs: dict[str, Any] = {"headers": headers}
        if payload is not None:
            kwargs["data"] = body_bytes
        url = self._url(path)
        if query:
            url = f"{url}?{urlencode(query)}"

        try:
            async with aiohttp.ClientSession(timeout=self.timeout) as session:
                async with session.request(method, url, **kwargs) as response:
                    text = await response.text()
                    try:
                        body = json.loads(text) if text else {}
                    except json.JSONDecodeError as exc:
                        raise VerificationProviderError(
                            f"LeafSM H5 人脸核身接口返回非 JSON 响应（HTTP {response.status}）",
                            status_code=502,
                            raw={"http_status": response.status, "text": text[:500]},
                        ) from exc
                    if not (200 <= response.status < 300):
                        raise self._api_error(response.status, body)
        except VerificationProviderError:
            raise
        except aiohttp.ClientError as exc:
            raise VerificationProviderError(f"请求 LeafSM H5 人脸核身接口失败：{exc}", status_code=502) from exc

        if not isinstance(body, dict):
            raise VerificationProviderError("LeafSM H5 人脸核身接口返回格式异常", status_code=502, raw=body)
        return body

    def _api_error(self, http_status: int, body: dict[str, Any]) -> VerificationProviderError:
        error = body.get("error") if isinstance(body.get("error"), dict) else {}
        code = _clean_text(body.get("code") or error.get("code") or body.get("detail"))
        message = _clean_text(body.get("message") or body.get("msg") or error.get("message") or body.get("detail"))
        if code == "TWO_FACTOR_MISMATCH" or http_status == 422:
            message = message or "姓名和身份证号二要素预校验未通过"
            status_code = 422
        elif code in {"INVALID_SIGNATURE", "REPLAY_REQUEST"} or http_status == 401:
            message = message or "LeafSM API Key 签名校验失败，请检查 AppId、AppSecret、服务器时间和签名算法"
            status_code = 400
        elif code == "API_TYPE_NOT_ALLOWED":
            message = message or "当前应用未开通 h5_face 能力"
            status_code = 400
        elif code == "INSUFFICIENT_CREDIT":
            message = message or "LeafSM 账户额度不足"
            status_code = 429
        elif code == "TASK_NOT_FOUND" or http_status == 404:
            message = message or "LeafSM H5 人脸核身任务不存在"
            status_code = 404
        else:
            message = message or f"LeafSM H5 人脸核身接口错误（HTTP {http_status}）"
            status_code = 502
        return VerificationProviderError(message, status_code=status_code, raw=self._safe_raw(body))

    def _safe_raw(self, body: dict[str, Any]) -> dict[str, Any]:
        raw = dict(body)
        task = raw.get("task")
        if isinstance(task, dict):
            task = dict(task)
            if task.get("card_no"):
                task["card_no"] = _mask_id_card(str(task.get("card_no")))
            if task.get("real_name"):
                task["real_name"] = _mask_name(str(task.get("real_name")))
            raw["task"] = task
        return raw

    async def verify_two_factor(self, *, real_name: str, id_card_no: str) -> VerificationResult:
        raise VerificationProviderError("LeafSM H5 人脸核身插件不支持单独二要素核验，请使用 H5 跳转核身", status_code=400)

    async def verify_three_factor(
        self,
        *,
        real_name: str,
        id_card_no: str,
        phone: str,
    ) -> VerificationResult:
        raise VerificationProviderError("LeafSM H5 人脸核身插件不支持三要素核验", status_code=400)

    async def verify_face_api(
        self,
        *,
        real_name: str,
        id_card_no: str,
        image_base64: str,
        image_type: str = "BASE64",
    ) -> VerificationResult:
        raise VerificationProviderError("LeafSM H5 人脸核身插件不支持本地图片扫脸，请使用 H5 跳转核身", status_code=400)

    async def create_redirect_session(
        self,
        *,
        real_name: str,
        id_card_no: str,
        return_url: str,
        notify_url: str | None = None,
    ) -> RedirectSession:
        payload: dict[str, Any] = {
            "type": "h5_face",
            "real_name": real_name,
            "card_no": id_card_no,
            "return_url": return_url,
        }
        if notify_url:
            payload["notify_url"] = notify_url

        body = await self._request_json("POST", TASK_CREATE_PATH, payload)
        task = body.get("task") if isinstance(body.get("task"), dict) else {}
        task_no = _clean_text(body.get("task_no") or task.get("task_no"))
        task_id = _clean_text(body.get("task_id") or task.get("id"))
        verify_url = _clean_text(body.get("verify_url"))
        if not task_no:
            raise VerificationProviderError("LeafSM H5 人脸核身未返回 task_no", status_code=502, raw=self._safe_raw(body))
        if not verify_url:
            raise VerificationProviderError("LeafSM H5 人脸核身未返回 verify_url", status_code=502, raw=self._safe_raw(body))

        try:
            expires_in = int(body.get("expires_in") or 7200)
        except (TypeError, ValueError):
            expires_in = 7200

        return RedirectSession(
            url=self._h5_url(verify_url) if verify_url.startswith("/") else verify_url,
            external_id=task_id or task_no,
            expires_in=expires_in,
            raw={
                "task_no": task_no,
                "task_id": task_id or None,
                "expires_in": expires_in,
                "task": {
                    "status": task.get("status"),
                    "type": task.get("type"),
                    "motions": task.get("motions"),
                    "card_no_masked": task.get("card_no_masked"),
                    "real_name_masked": task.get("real_name_masked"),
                },
            },
        )

    async def get_redirect_result(
        self,
        *,
        external_id: str,
        real_name: str,
        id_card_no: str,
    ) -> VerificationResult:
        task_ref = _clean_text(external_id)
        if not task_ref:
            raise VerificationProviderError("LeafSM H5 人脸核身查询缺少任务编号", status_code=400)

        try:
            body = await self._request_json("GET", TASK_LOOKUP_PATH, query={"task_id": task_ref})
        except VerificationProviderError as exc:
            if exc.status_code != 404:
                raise
            task_no = quote(task_ref, safe="")
            body = await self._request_json("GET", TASK_QUERY_PATH.format(task_no=task_no))
        task = body.get("task") if isinstance(body.get("task"), dict) else {}
        order = body.get("order") if isinstance(body.get("order"), dict) else {}
        liveness = body.get("liveness_result") if isinstance(body.get("liveness_result"), dict) else {}
        task_status = _clean_text(task.get("status")).lower()
        order_status = _clean_text(order.get("status")).lower()
        matched = order.get("matched")
        score = _float_or_none(liveness.get("hack_score") or order.get("score"))
        raw = self._safe_raw(body)

        if task_status == "completed":
            passed = matched is True or str(matched).lower() == "true" or order_status == "completed"
            return VerificationResult(
                passed=passed,
                mode="face_redirect",
                provider=self.name,
                status="verified" if passed else "rejected",
                reason="" if passed else "人脸核身未通过",
                score=score,
                external_id=_clean_text(task.get("task_no")) or external_id,
                raw=raw,
            )

        if task_status in TERMINAL_FAILED_STATUSES:
            reason = _clean_text(task.get("reason") or order.get("reason") or body.get("message") or body.get("msg"))
            return VerificationResult(
                passed=False,
                mode="face_redirect",
                provider=self.name,
                status="rejected",
                reason=reason or f"人脸核身未通过：{task_status}",
                score=score,
                external_id=_clean_text(task.get("task_no")) or external_id,
                raw=raw,
            )

        return VerificationResult(
            passed=False,
            mode="face_redirect",
            provider=self.name,
            status="processing",
            reason="用户尚未完成 H5 人脸核身",
            score=score,
            external_id=_clean_text(task.get("task_no")) or external_id,
            raw=raw,
        )


def create_provider(config: Any) -> LeafSmH5FaceProvider:
    return LeafSmH5FaceProvider(config)
