from __future__ import annotations

import hashlib
import logging
from typing import Any

import aiohttp

from app.redis_client import k, redis_client
from mode.verification.base import RedirectSession, VerificationProviderError, VerificationResult

log = logging.getLogger("qycw.verification.baidu_face_v3")

BAIDU_OAUTH_URL = "https://aip.baidubce.com/oauth/2.0/token"
BAIDU_IDMATCH_URL = "https://aip.baidubce.com/rest/2.0/face/v3/person/idmatch"
BAIDU_FACE_VERIFY_URL = "https://aip.baidubce.com/rest/2.0/face/v3/person/verify"
TOKEN_TTL_SECONDS = 25 * 24 * 60 * 60
TOKEN_ERROR_CODES = {110, 111}
CONTROL_VALUES = {"NONE", "LOW", "NORMAL", "HIGH"}


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


def _control(value: Any, default: str) -> str:
    text = _clean_text(value).upper() or default
    return text if text in CONTROL_VALUES else default


class BaiduFaceV3Provider:
    name = "baidu_face_v3"
    supported_modes = ("two_factor",)

    def __init__(self, config: Any):
        self.api_key = _clean_text(
            getattr(config, "ACCESS_KEY", None)
            or getattr(config, "API_KEY", None)
            or getattr(config, "BAIDU_FACE_API_KEY", None)
        )
        self.secret_key = _clean_text(
            getattr(config, "SECRET_KEY", None)
            or getattr(config, "BAIDU_FACE_SECRET_KEY", None)
        )
        self.score_threshold = float(getattr(config, "SCORE_THRESHOLD", 80) or 80)
        self.quality_control = _control(getattr(config, "QUALITY_CONTROL", "LOW"), "LOW")
        self.liveness_control = _control(getattr(config, "LIVENESS_CONTROL", "NORMAL"), "NORMAL")
        self.spoofing_control = _control(getattr(config, "SPOOFING_CONTROL", "NORMAL"), "NORMAL")
        self.timeout = aiohttp.ClientTimeout(total=float(getattr(config, "TIMEOUT", 20) or 20))

        if not self.api_key or not self.secret_key:
            raise VerificationProviderError("百度实名认证插件未配置 Access Key / Secret Key", status_code=400)

        cache_hash = hashlib.sha256(self.api_key.encode("utf-8")).hexdigest()[:16]
        self.token_cache_key = k("verification", "baidu_face_v3", cache_hash, "token")

    async def _fetch_access_token(self) -> str:
        params = {
            "grant_type": "client_credentials",
            "client_id": self.api_key,
            "client_secret": self.secret_key,
        }
        try:
            async with aiohttp.ClientSession(timeout=self.timeout) as session:
                async with session.post(BAIDU_OAUTH_URL, data=params) as response:
                    body = await response.json(content_type=None)
        except aiohttp.ClientError as exc:
            raise VerificationProviderError(f"获取百度 AccessToken 失败：{exc}", status_code=502) from exc

        if body.get("error"):
            raise VerificationProviderError(
                f"获取百度 AccessToken 失败：{body.get('error_description') or body.get('error')}",
                status_code=502,
                raw=body,
            )
        token = _clean_text(body.get("access_token"))
        if not token:
            raise VerificationProviderError("百度返回的 access_token 为空", status_code=502, raw=body)
        return token

    async def _get_access_token(self) -> str:
        cached = await redis_client.get(self.token_cache_key)
        if cached:
            return str(cached)
        token = await self._fetch_access_token()
        await redis_client.set(self.token_cache_key, token, ex=TOKEN_TTL_SECONDS)
        return token

    async def _invalidate_token(self) -> None:
        await redis_client.delete(self.token_cache_key)

    async def _post_json(
        self,
        url: str,
        payload: dict[str, Any],
        *,
        retry_on_token_error: bool = True,
    ) -> dict[str, Any]:
        token = await self._get_access_token()
        request_url = f"{url}?access_token={token}"
        try:
            async with aiohttp.ClientSession(timeout=self.timeout) as session:
                async with session.post(request_url, json=payload) as response:
                    body = await response.json(content_type=None)
        except aiohttp.ClientError as exc:
            raise VerificationProviderError(f"请求百度实名认证接口失败：{exc}", status_code=502) from exc

        try:
            error_code = int(body.get("error_code") or 0)
        except (TypeError, ValueError):
            error_code = 0

        if retry_on_token_error and error_code in TOKEN_ERROR_CODES:
            await self._invalidate_token()
            return await self._post_json(url, payload, retry_on_token_error=False)
        return body

    def _business_result(
        self,
        *,
        body: dict[str, Any],
        mode: str,
        passed: bool,
        reason: str,
        score: float | None = None,
    ) -> VerificationResult:
        return VerificationResult(
            passed=passed,
            mode=mode,
            provider=self.name,
            status="verified" if passed else "rejected",
            reason=reason,
            score=score,
            external_id=str(body.get("log_id")) if body.get("log_id") is not None else None,
            raw=body,
        )

    async def verify_two_factor(self, *, real_name: str, id_card_no: str) -> VerificationResult:
        payload = {
            "id_card_number": id_card_no,
            "name": real_name,
        }
        body = await self._post_json(BAIDU_IDMATCH_URL, payload)
        try:
            error_code = int(body.get("error_code") or 0)
        except (TypeError, ValueError):
            error_code = -1
        if error_code == 0:
            return self._business_result(body=body, mode="two_factor", passed=True, reason="")
        reason = _clean_text(body.get("error_msg")) or f"百度实名认证未通过（{error_code}）"
        return self._business_result(body=body, mode="two_factor", passed=False, reason=reason)

    async def verify_three_factor(
        self,
        *,
        real_name: str,
        id_card_no: str,
        phone: str,
    ) -> VerificationResult:
        raise VerificationProviderError("百度人脸实名认证 V3 插件暂未提供三要素接口", status_code=400)

    async def verify_face_api(
        self,
        *,
        real_name: str,
        id_card_no: str,
        image_base64: str,
        image_type: str = "BASE64",
    ) -> VerificationResult:
        payload = {
            "image": image_base64,
            "image_type": image_type,
            "id_card_number": id_card_no,
            "name": real_name,
            "quality_control": self.quality_control,
            "liveness_control": self.liveness_control,
            "spoofing_control": self.spoofing_control,
        }
        body = await self._post_json(BAIDU_FACE_VERIFY_URL, payload)
        try:
            error_code = int(body.get("error_code") or 0)
        except (TypeError, ValueError):
            error_code = -1
        if error_code != 0:
            reason = _clean_text(body.get("error_msg")) or f"百度扫脸认证未通过（{error_code}）"
            return self._business_result(body=body, mode="face_api", passed=False, reason=reason)

        result = body.get("result") if isinstance(body.get("result"), dict) else {}
        try:
            score = float(result.get("score") or 0)
        except (TypeError, ValueError):
            score = 0.0
        passed = score >= self.score_threshold
        reason = "" if passed else f"人脸比对分数 {score:.2f} 低于阈值 {self.score_threshold:.2f}"
        return self._business_result(
            body=body,
            mode="face_api",
            passed=passed,
            reason=reason,
            score=score,
        )

    async def create_redirect_session(
        self,
        *,
        real_name: str,
        id_card_no: str,
        return_url: str,
        notify_url: str | None = None,
    ) -> RedirectSession:
        raise VerificationProviderError("百度人脸实名认证 V3 插件暂不支持第三方页面跳转核身", status_code=400)


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