"""百度智能云人脸核身客户端（H5方案）。

基于文档：https://cloud.baidu.com/doc/FACE/s/Ikrycq2k2

标准 H5 人脸核身流程：
1. 获取 Access Token（OAuth2）
2. 获取 session_code：`POST /rest/2.0/face/v1/faceliveness/sessioncode`
3. 生成 H5 认证 URL：`https://face.baidu.com/face-api/?token={token}&session_code={code}&...`
4. 用户完成 H5 活体检测
5. 查询结果：`POST /rest/2.0/face/v1/faceliveness/result`

Token 自动缓存至 Redis，过期前自动刷新。
"""
from __future__ import annotations

import logging
from typing import Any

import aiohttp

from app.config import settings
from app.redis_client import k, redis_client

log = logging.getLogger("qycw.baidu_face")

TOKEN_CACHE_KEY = k("baidu_face", "access_token")
TOKEN_TTL_SECONDS = 25 * 24 * 60 * 60  # 约30天有效期，缓存25天

BAIDU_OAUTH_URL = "https://aip.baidubce.com/oauth/2.0/token"
BAIDU_H5_SESSIONCODE_URL = "https://aip.baidubce.com/rest/2.0/face/v1/faceliveness/sessioncode"
BAIDU_H5_RESULT_URL = "https://aip.baidubce.com/rest/2.0/face/v1/faceliveness/result"
BAIDU_H5_PAGE_URL = "https://face.baidu.com/face-api/"


class BaiduFaceError(Exception):
    """百度智能云人脸核身异常。"""

    def __init__(self, message: str, *, code: int | None = None, raw: Any = None):
        super().__init__(message)
        self.message = message
        self.code = code
        self.raw = raw


class BaiduFaceClient:
    """百度智能云 H5 人脸核身客户端。

    使用示例：
        client = BaiduFaceClient()
        # 获取 H5 认证 URL
        session_code = await client.get_session_code()
        h5_url = client.build_h5_url(session_code, redirect_url="https://your-site.com/verify/callback")
        # 用户完成 H5 后查询结果
        result = await client.get_result(session_code)
    """

    def __init__(
        self,
        *,
        api_key: str | None = None,
        secret_key: str | None = None,
        app_id: str | None = None,
        timeout: float = 20.0,
    ):
        self.api_key = api_key or settings.BAIDU_FACE_API_KEY
        self.secret_key = secret_key or settings.BAIDU_FACE_SECRET_KEY
        self.app_id = app_id or settings.BAIDU_FACE_APP_ID
        self.timeout = aiohttp.ClientTimeout(total=timeout)

        if not self.api_key or not self.secret_key:
            raise BaiduFaceError(
                "百度智能云人脸核身未配置：缺少 BAIDU_FACE_API_KEY / BAIDU_FACE_SECRET_KEY"
            )

    # ── Token 管理 ───────────────────────────────────────────────────────────

    async def _get_access_token(self) -> str:
        """从 Redis 缓存获取 access_token，过期则重新获取。"""
        cached = await redis_client.get(TOKEN_CACHE_KEY)
        if cached:
            return cached

        token = await self._fetch_access_token()
        await redis_client.set(TOKEN_CACHE_KEY, token, ex=TOKEN_TTL_SECONDS)
        return token

    async def _fetch_access_token(self) -> str:
        """向百度 OAuth 接口获取新的 access_token。"""
        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 resp:
                    body = await resp.json()
        except aiohttp.ClientError as exc:
            raise BaiduFaceError(f"获取百度AccessToken失败：网络错误 {exc}") from exc

        if "error" in body:
            raise BaiduFaceError(
                f"获取百度AccessToken失败：{body.get('error_description')}",
                code=body.get("error_code"),
                raw=body,
            )

        token = str(body.get("access_token") or "").strip()
        if not token:
            raise BaiduFaceError("百度返回的 access_token 为空", raw=body)

        log.info("baidu face access_token refreshed, expires_in=%s", body.get("expires_in"))
        return token

    async def invalidate_token(self) -> None:
        """清除缓存的 token（用于错误后强制刷新）。"""
        await redis_client.delete(TOKEN_CACHE_KEY)

    # ── 底层请求 ────────────────────────────────────────────────────────────

    async def _request(
        self,
        url: str,
        payload: dict[str, Any],
        *,
        retry_on_token_error: bool = True,
    ) -> dict[str, Any]:
        """带 token 的 POST 请求，支持 token 过期自动刷新重试。"""
        token = await self._get_access_token()
        full_url = f"{url}?access_token={token}"
        headers = {"Content-Type": "application/x-www-form-urlencoded"}

        # 百度 H5 API 使用 form-data 而非 JSON
        form_data = aiohttp.FormData()
        for key, value in payload.items():
            form_data.add_field(key, str(value))

        async def _do() -> dict[str, Any]:
            async with aiohttp.ClientSession(timeout=self.timeout) as session:
                async with session.post(full_url, data=form_data, headers=headers) as resp:
                    return await resp.json()

        body = await _do()

        # 百度错误码：110 表示 token 过期/无效
        if retry_on_token_error and body.get("error_code") == 110:
            log.warning("baidu face token expired, refreshing...")
            await self.invalidate_token()
            token = await self._get_access_token()
            full_url = f"{url}?access_token={token}"
            body = await _do()

        return body

    def _raise_on_error(self, body: dict[str, Any], label: str) -> None:
        """检查百度返回的错误码，非零则抛出异常。"""
        code = body.get("error_code")
        if code and int(code) != 0:
            msg = body.get("error_msg") or body.get("error_desc") or f"百度返回错误码 {code}"
            raise BaiduFaceError(f"{label}失败：{msg}", code=int(code), raw=body)

    # ── H5 人脸核身 API ───────────────────────────────────────────────────────

    async def get_session_code(
        self,
        *,
        type_id: int = 0,
        version_num: str = "1.0.0",
    ) -> str:
        """获取 H5 人脸核身 session_code。

        Args:
            type_id: 认证类型：0=随机动作（推荐），1=静默活体
            version_num: 版本号，固定 "1.0.0"

        Returns:
            session_code: 用于构建 H5 认证 URL 的唯一码
        """
        payload = {
            "app_id": self.app_id,
            "type_id": type_id,
            "version_num": version_num,
        }
        body = await self._request(BAIDU_H5_SESSIONCODE_URL, payload)
        self._raise_on_error(body, "获取session_code")

        result = body.get("result", {})
        session_code = str(result.get("session_code") or "").strip()
        if not session_code:
            raise BaiduFaceError("百度返回的 session_code 为空", raw=body)

        return session_code

    def build_h5_url(
        self,
        session_code: str,
        *,
        redirect_url: str,
        qualities: str = "1.0",
        liveness_check: str = "1",
        certificate_num: str | None = None,
        name: str | None = None,
    ) -> str:
        """构建 H5 人脸核身认证页面 URL。

        参数说明（参考百度文档）：
        - token: access_token
        - session_code: 上一步获取的码
        - redirect_url: 认证完成后跳转的回调地址
        - qualities: 视频质量，默认 1.0
        - liveness_check: 是否进行活体检测，1=是
        - certificate_num: 身份证号（可选，传入则与识别结果比对）
        - name: 姓名（可选，传入则与识别结果比对）
        """
        token = self._get_access_token_sync()
        params = {
            "token": token,
            "session_code": session_code,
            "redirect_url": redirect_url,
            "qualities": qualities,
            "liveness_check": liveness_check,
        }
        if certificate_num:
            params["certificate_num"] = certificate_num
        if name:
            params["name"] = name

        query = "&".join(f"{k}={v}" for k, v in params.items())
        return f"{BAIDU_H5_PAGE_URL}?{query}"

    def _get_access_token_sync(self) -> str:
        """同步获取 access_token（用于 build_h5_url 非 async 场景）。
        
        警告：此函数仅在确信 Redis 中已缓存 token 时使用。
        如果 token 未缓存，会返回空字符串，调用方应改为 async 方式。
        """
        import asyncio
        try:
            loop = asyncio.get_event_loop()
            if loop.is_running():
                # 在运行中的事件循环中，使用 run_coroutine_threadsafe
                import concurrent.futures
                with concurrent.futures.ThreadPoolExecutor() as executor:
                    future = executor.submit(asyncio.run, self._get_access_token())
                    return future.result(timeout=10)
            return loop.run_until_complete(self._get_access_token())
        except Exception:
            log.warning("sync token fetch failed, H5 URL may be incomplete")
            return ""

    async def get_result(self, session_code: str) -> dict[str, Any]:
        """查询 H5 人脸核身认证结果。

        返回字段说明（参考百度文档）：
        - log_id: 日志ID
        - result: 详细结果
          - face_liveness: 活体分数
          - score: 人脸比对分数
          - realname: 识别到的姓名
          - certificate_num: 识别到的身份证号
          - risk_level: 风险等级
          - risk_tag: 风险标签
        - error_code / error_msg: 错误信息
        """
        payload = {
            "app_id": self.app_id,
            "session_code": session_code,
        }
        body = await self._request(BAIDU_H5_RESULT_URL, payload)
        self._raise_on_error(body, "查询认证结果")
        return body.get("result", {})

    # ── 综合实名认证（推荐）──────────────────────────────────────────────────────

    async def create_h5_verification(
        self,
        *,
        real_name: str,
        id_card_no: str,
        redirect_url: str,
    ) -> dict[str, Any]:
        """创建完整的 H5 实名认证任务。

        返回：
        {
            "session_code": str,
            "h5_url": str,
        }
        """
        session_code = await self.get_session_code()
        h5_url = self.build_h5_url(
            session_code,
            redirect_url=redirect_url,
            certificate_num=id_card_no,
            name=real_name,
        )
        return {
            "session_code": session_code,
            "h5_url": h5_url,
        }

    async def verify_h5_result(
        self,
        session_code: str,
        *,
        expected_name: str,
        expected_id_card_no: str,
    ) -> dict[str, Any]:
        """验证 H5 认证结果，并与预期的姓名+身份证号比对。

        返回：
        {
            "passed": bool,
            "score": float,
            "liveness_score": float,
            "ocr_name": str,
            "ocr_id_card_no": str,
            "reason": str,
        }
        """
        result = await self.get_result(session_code)

        # 解析结果字段
        ocr_name = result.get("realname", "")
        ocr_id = result.get("certificate_num", "")
        score = result.get("score", 0)
        liveness = result.get("face_liveness", 0)
        risk_level = result.get("risk_level", "")
        risk_tag = result.get("risk_tag", [])

        # 风险检查
        if risk_level and risk_level != "1":
            return {"passed": False, "reason": f"风险检测未通过：风险等级 {risk_level}, 标签 {risk_tag}"}

        # 活体检测
        if liveness < 0.05:
            return {"passed": False, "reason": f"活体检测未通过，分数：{liveness:.4f}"}

        # 人脸比对
        if score < 80:
            return {"passed": False, "reason": f"人脸比对未通过，相似度：{score:.2f}"}

        # 姓名+身份证号比对（宽松匹配，去除空格）
        if expected_name.strip() != ocr_name.strip():
            return {"passed": False, "reason": f"姓名不匹配：OCR识别为【{ocr_name}】"}
        if expected_id_card_no.strip().upper() != ocr_id.strip().upper():
            return {"passed": False, "reason": f"身份证号不匹配：OCR识别为【{ocr_id}】"}

        return {
            "passed": True,
            "score": score,
            "liveness_score": liveness,
            "ocr_name": ocr_name,
            "ocr_id_card_no": ocr_id,
            "reason": "",
        }


# ── 辅助工具 ────────────────────────────────────────────────────────────────

async def test_baidu_connection() -> tuple[bool, str]:
    """测试百度智能云人脸核身连通性。"""
    try:
        client = BaiduFaceClient(timeout=8.0)
        await client._fetch_access_token()
        return True, "百度智能云连接成功"
    except BaiduFaceError as exc:
        return False, exc.message
    except Exception as exc:
        return False, f"未知错误：{exc}"
