"""实名认证服务层。

认证能力由 backend/mode/verification 下的插件提供。当前内置百度智能云人脸
实名认证 V3 插件，业务层只处理状态、记录、材料保留和购买拦截。
"""
from __future__ import annotations

import base64
import logging
import secrets
import uuid
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any
from urllib.parse import urlencode, urlparse, urlunparse

from fastapi import HTTPException
from sqlalchemy import desc, func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import settings
from app.core import sensitive
from app.redis_client import k, redis_client
from app.models.balance_log import BalanceChangeType, BalanceLog
from app.models.product import Product
from app.models.product_group import ProductGroup
from app.models.user import IdCardVerifiedStatus, User
from app.models.user_verification_quota import UserVerificationQuota
from app.models.user_verification_record import UserVerificationRecord
from app.services import notification_plugin_service, provider_loader, system_setting_service
from mode.verification.base import VerificationProviderError, VerificationResult

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

ID_CARD_VERIFIED_STATUS = IdCardVerifiedStatus
SEND_COOLDOWN = 60
CODE_TTL_SECONDS = 10 * 60
VERIFY_MAX_ATTEMPTS = 5
VERIFICATION_LIMIT_PER_DAY = 5
VERIFICATION_QUOTA_PURCHASE_MAX = 100
FACE_IMAGE_MAX_BYTES = 2 * 1024 * 1024
FACE_IMAGE_CONTENT_TYPES = {
    "image/jpeg": ".jpg",
    "image/jpg": ".jpg",
    "image/png": ".png",
    "image/bmp": ".bmp",
}


def _code_key(purpose: str, target: str) -> str:
    return k("verification_code", purpose.strip().lower(), target.strip().lower())


def _cooldown_key(purpose: str, target: str) -> str:
    return k("verification_code_cooldown", purpose.strip().lower(), target.strip().lower())


def _attempt_key(purpose: str, target: str) -> str:
    return k("verification_code_attempts", purpose.strip().lower(), target.strip().lower())


def generate_code() -> str:
    return f"{secrets.randbelow(1_000_000):06d}"


async def can_send(purpose: str, target: str) -> int:
    ttl = await redis_client.ttl(_cooldown_key(purpose, target))
    return int(ttl) if ttl and ttl > 0 else 0


async def save_code(purpose: str, target: str, code: str) -> None:
    await redis_client.set(_code_key(purpose, target), str(code), ex=CODE_TTL_SECONDS)
    await redis_client.set(_cooldown_key(purpose, target), "1", ex=SEND_COOLDOWN)
    await redis_client.delete(_attempt_key(purpose, target))


async def verify_code(purpose: str, target: str, code: str) -> bool:
    code_key = _code_key(purpose, target)
    attempts_key = _attempt_key(purpose, target)
    stored = await redis_client.get(code_key)
    if not stored:
        return False
    if str(stored) == str(code).strip():
        await redis_client.delete(code_key, attempts_key)
        return True
    attempts = await redis_client.incr(attempts_key)
    if attempts == 1:
        await redis_client.expire(attempts_key, CODE_TTL_SECONDS)
    if attempts >= VERIFY_MAX_ATTEMPTS:
        await redis_client.delete(code_key, attempts_key)
    return False


def _now() -> datetime:
    return datetime.now(timezone.utc)


def _mask_id_card(value: str | None) -> str | None:
    return sensitive.mask_id_card(value)


def _mask_name(value: str | None) -> str | None:
    return sensitive.mask_name(value)


def _normalize_id_card(value: str) -> str:
    return value.strip().upper()


def _plain_identity(value: str | None) -> str | None:
    return sensitive.decrypt_optional(value)


def _stored_identity(value: str | None) -> str | None:
    return sensitive.encrypt_optional(value)


def _id_card_hash(value: str | None) -> str | None:
    return sensitive.deterministic_hash(value)


def _verification_url() -> str:
    return "/settings?tab=verification"


async def _daily_attempt_count(db: AsyncSession, user_id: int) -> int:
    today = _now().replace(hour=0, minute=0, second=0, microsecond=0)
    count = await db.scalar(
        select(func.count())
        .select_from(UserVerificationRecord)
        .where(
            UserVerificationRecord.user_id == user_id,
            UserVerificationRecord.created_at >= today,
        )
    )
    return int(count or 0)


async def _ensure_attempt_allowed(db: AsyncSession, user: User) -> None:
    if user.id_card_verified == ID_CARD_VERIFIED_STATUS.VERIFIED.value:
        raise HTTPException(status_code=400, detail="您已完成实名认证")
    if await _daily_attempt_count(db, user.id) >= VERIFICATION_LIMIT_PER_DAY:
        raise HTTPException(status_code=429, detail=f"每天最多提交 {VERIFICATION_LIMIT_PER_DAY} 次实名认证")


async def _ensure_identity_not_bound(db: AsyncSession, user: User, id_card_no: str) -> None:
    id_hash = _id_card_hash(id_card_no)
    existing = await db.scalar(
        select(User.id).where(
            or_(
                User.id_card_no_hash == id_hash,
                User.id_card_no == id_card_no,
            ),
            User.id != user.id,
        )
    )
    if existing:
        raise HTTPException(status_code=400, detail="该身份证号已绑定其他账号")


def _quota_config(config: Any) -> dict[str, Any]:
    free_limit = max(0, int(getattr(config, "FREE_QUOTA_PER_USER", 1) or 0))
    price_cents = max(0, int(getattr(config, "QUOTA_PRICE_CENTS", 350) or 0))
    purchase_enabled = bool(getattr(config, "QUOTA_PURCHASE_ENABLED", True))
    return {
        "free_quota_per_user": free_limit,
        "purchase_enabled": purchase_enabled,
        "price_cents": price_cents,
    }


async def _get_or_create_quota(
    db: AsyncSession,
    user_id: int,
    *,
    for_update: bool = False,
) -> UserVerificationQuota:
    stmt = select(UserVerificationQuota).where(UserVerificationQuota.user_id == user_id)
    if for_update:
        stmt = stmt.with_for_update()
    result = await db.execute(stmt)
    quota = result.scalar_one_or_none()
    if quota:
        return quota

    quota = UserVerificationQuota(user_id=user_id)
    db.add(quota)
    await db.flush()
    if for_update:
        result = await db.execute(
            select(UserVerificationQuota)
            .where(UserVerificationQuota.user_id == user_id)
            .with_for_update()
        )
        quota = result.scalar_one()
    return quota


def _quota_summary_from_row(
    quota: UserVerificationQuota,
    *,
    user_balance: int,
    config: Any,
) -> dict[str, Any]:
    quota_cfg = _quota_config(config)
    free_total = int(quota_cfg["free_quota_per_user"])
    free_used = max(0, int(quota.free_used or 0))
    paid_remaining = max(0, int(quota.paid_remaining or 0))
    free_remaining = max(0, free_total - free_used)
    remaining_total = free_remaining + paid_remaining
    price_cents = int(quota_cfg["price_cents"])
    return {
        **quota_cfg,
        "free_used": free_used,
        "free_remaining": free_remaining,
        "paid_remaining": paid_remaining,
        "paid_purchased": max(0, int(quota.paid_purchased or 0)),
        "paid_used": max(0, int(quota.paid_used or 0)),
        "remaining_total": remaining_total,
        "balance_cents": max(0, int(user_balance or 0)),
        "can_submit": remaining_total > 0,
        "can_purchase": bool(quota_cfg["purchase_enabled"]) and price_cents >= 0,
    }


async def get_verification_quota(db: AsyncSession, user: User, config: Any | None = None) -> dict[str, Any]:
    if config is None:
        config = await system_setting_service.get_verification_provider_config(db)
    quota = await _get_or_create_quota(db, user.id)
    return _quota_summary_from_row(quota, user_balance=int(user.balance or 0), config=config)


def _quota_required_error(summary: dict[str, Any]) -> HTTPException:
    price_cents = int(summary.get("price_cents") or 0)
    detail = {
        "code": "VERIFICATION_QUOTA_REQUIRED",
        "message": (
            f"实名认证免费次数已用完，请先购买实名次数（¥{price_cents / 100:.2f}/次）"
            if summary.get("purchase_enabled")
            else "实名认证免费次数已用完，请联系管理员"
        ),
        "quota": summary,
    }
    return HTTPException(status_code=402, detail=detail)


def _attach_quota_to_record(record: UserVerificationRecord, quota: dict[str, Any]) -> None:
    provider_response = record.provider_response if isinstance(record.provider_response, dict) else {}
    provider_response = dict(provider_response)
    provider_response["quota"] = quota
    record.provider_response = provider_response


async def _ensure_verification_quota_available(db: AsyncSession, user: User, config: Any) -> dict[str, Any]:
    summary = await get_verification_quota(db, user, config)
    if int(summary["remaining_total"]) <= 0:
        raise _quota_required_error(summary)
    return summary


async def _consume_verification_quota(db: AsyncSession, user: User, config: Any) -> dict[str, Any]:
    quota = await _get_or_create_quota(db, user.id, for_update=True)
    quota_cfg = _quota_config(config)
    free_total = int(quota_cfg["free_quota_per_user"])
    if int(quota.free_used or 0) < free_total:
        quota.free_used = int(quota.free_used or 0) + 1
        source = "free"
    elif int(quota.paid_remaining or 0) > 0:
        quota.paid_remaining = int(quota.paid_remaining or 0) - 1
        quota.paid_used = int(quota.paid_used or 0) + 1
        source = "paid"
    else:
        summary = _quota_summary_from_row(quota, user_balance=int(user.balance or 0), config=config)
        raise _quota_required_error(summary)

    await db.flush()
    summary = _quota_summary_from_row(quota, user_balance=int(user.balance or 0), config=config)
    return {"source": source, **summary}


async def purchase_verification_quota(db: AsyncSession, user: User, *, count: int) -> dict[str, Any]:
    count = int(count or 0)
    if count < 1 or count > VERIFICATION_QUOTA_PURCHASE_MAX:
        raise HTTPException(status_code=400, detail=f"购买次数必须在 1 到 {VERIFICATION_QUOTA_PURCHASE_MAX} 次之间")

    config = await system_setting_service.get_verification_provider_config(db)
    quota_cfg = _quota_config(config)
    if not bool(quota_cfg["purchase_enabled"]):
        raise HTTPException(status_code=400, detail="实名认证次数购买暂未开放")

    price_cents = int(quota_cfg["price_cents"])
    amount = price_cents * count
    user_result = await db.execute(select(User).where(User.id == user.id).with_for_update())
    locked_user = user_result.scalar_one_or_none()
    if not locked_user:
        raise HTTPException(status_code=404, detail="用户不存在")
    if amount > 0 and int(locked_user.balance or 0) < amount:
        raise HTTPException(
            status_code=402,
            detail={
                "code": "INSUFFICIENT_BALANCE",
                "message": f"余额不足，需要 ¥{amount / 100:.2f}，当前余额 ¥{int(locked_user.balance or 0) / 100:.2f}",
                "required_cents": amount,
                "balance_cents": int(locked_user.balance or 0),
            },
        )

    quota = await _get_or_create_quota(db, locked_user.id, for_update=True)
    if amount > 0:
        locked_user.balance = int(locked_user.balance or 0) - amount
        db.add(
            BalanceLog(
                user_id=locked_user.id,
                change_type=BalanceChangeType.ORDER_PAY,
                delta=-amount,
                balance_after=locked_user.balance,
                ref_no=f"verification_quota:{uuid.uuid4().hex[:20]}",
                remark=f"购买实名认证次数 {count} 次",
            )
        )
    quota.paid_remaining = int(quota.paid_remaining or 0) + count
    quota.paid_purchased = int(quota.paid_purchased or 0) + count
    await db.flush()
    return {
        "purchased_count": count,
        "amount_cents": amount,
        "balance_cents": int(locked_user.balance or 0),
        "quota": _quota_summary_from_row(quota, user_balance=int(locked_user.balance or 0), config=config),
    }


async def _get_provider(db: AsyncSession):
    config = await system_setting_service.get_verification_provider_config(db)
    if not bool(getattr(config, "verification_enabled", False)):
        raise HTTPException(status_code=400, detail="实名认证插件未启用，请联系管理员")
    try:
        return provider_loader.create_verification_provider(config), config
    except VerificationProviderError as exc:
        raise HTTPException(status_code=exc.status_code, detail=exc.message) from exc
    except Exception as exc:
        log.exception("create verification provider failed")
        raise HTTPException(status_code=500, detail=f"实名认证插件加载失败：{exc}") from exc


def _origin(url: str | None) -> str | None:
    if not url:
        return None
    parsed = urlparse(str(url).strip())
    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
        return None
    return f"{parsed.scheme}://{parsed.netloc}".lower()


def _join_url(base: str, path: str, query: dict[str, str] | None = None) -> str:
    parsed = urlparse(base.rstrip("/"))
    joined = parsed._replace(path=path, params="", query=urlencode(query or {}), fragment="")
    return urlunparse(joined)


def _allowed_return_origins(frontend_url: str) -> set[str]:
    origins = {_origin(frontend_url), _origin(settings.FRONTEND_URL)}
    for item in settings.CORS_ORIGINS or []:
        origins.add(_origin(item))
    return {item for item in origins if item}


def _safe_return_url(raw_url: str | None, *, frontend_url: str) -> str:
    default_url = _join_url(frontend_url or settings.FRONTEND_URL, "/settings", {"tab": "verification"})
    if not raw_url:
        return default_url
    candidate = str(raw_url).strip()
    if _origin(candidate) not in _allowed_return_origins(frontend_url):
        raise HTTPException(status_code=400, detail="verification return_url origin is not allowed")
    return candidate


def _safe_notify_url(raw_url: str | None, *, backend_url: str, provider_name: str) -> str | None:
    base = (backend_url or settings.PUBLIC_BACKEND_URL or "").strip()
    if not base:
        return None
    generated = _join_url(base, f"/api/v1/verification/notify/{provider_name}")
    if not raw_url:
        return generated
    candidate = str(raw_url).strip()
    if _origin(candidate) != _origin(base):
        raise HTTPException(status_code=400, detail="verification notify_url must use backend origin")
    return candidate


async def _get_provider_for_record(db: AsyncSession, record: UserVerificationRecord):
    try:
        provider, config = await _get_provider(db)
        if getattr(provider, "name", "verification") == record.provider:
            return provider, config
    except HTTPException:
        pass

    for plugin in notification_plugin_service.list_plugins("verification"):
        if str(plugin.get("id")) != str(record.provider):
            continue
        config = await system_setting_service.get_verification_provider_config(
            db,
            overrides={
                "verification.enabled": True,
                "verification.provider": str(plugin.get("provider")),
            },
        )
        try:
            return provider_loader.create_verification_provider(config), config
        except VerificationProviderError as exc:
            raise HTTPException(status_code=exc.status_code, detail=exc.message) from exc
        except Exception as exc:
            raise HTTPException(status_code=500, detail=f"verification provider load failed: {exc}") from exc

    raise HTTPException(status_code=400, detail="verification provider for this record is unavailable")


def _provider_public_options(provider: Any, config: Any) -> dict[str, Any]:
    supported_modes = [
        item
        for item in list(getattr(provider, "supported_modes", ()))
        if item != "face_api"
    ]
    default_mode = str(getattr(config, "DEFAULT_MODE", "two_factor") or "two_factor")
    if default_mode == "face_api":
        default_mode = "face_redirect" if "face_redirect" in supported_modes else "two_factor"
    if default_mode not in supported_modes and supported_modes:
        default_mode = supported_modes[0]
    return {
        "enabled": True,
        "provider": getattr(provider, "name", "verification"),
        "supported_modes": supported_modes,
        "default_mode": default_mode,
        "material_retention_days": int(getattr(config, "RETENTION_DAYS", 3) or 3),
    }


async def get_user_verification_status(db: AsyncSession, user: User) -> dict[str, Any]:
    config = await system_setting_service.get_verification_provider_config(db)
    return {
        "status": user.id_card_verified,
        "is_verified": user.id_card_verified == ID_CARD_VERIFIED_STATUS.VERIFIED.value,
        "real_name": _mask_name(user.real_name),
        "id_card_no": _mask_id_card(user.id_card_no),
        "verified_at": user.id_card_verified_at.isoformat() if user.id_card_verified_at else None,
        "reject_reason": user.id_card_verify_remark,
        "quota": await get_verification_quota(db, user, config),
    }


async def get_verification_options(db: AsyncSession) -> dict[str, Any]:
    config = await system_setting_service.get_verification_provider_config(db)
    if not bool(getattr(config, "verification_enabled", False)):
        return {
            "enabled": False,
            "provider": None,
            "supported_modes": [],
            "default_mode": str(getattr(config, "DEFAULT_MODE", "two_factor") or "two_factor"),
            "material_retention_days": int(getattr(config, "RETENTION_DAYS", 3) or 3),
            "quota": _quota_config(config),
        }
    provider, config = await _get_provider(db)
    data = _provider_public_options(provider, config)
    data["quota"] = _quota_config(config)
    return data


async def _create_record(
    db: AsyncSession,
    user: User,
    *,
    real_name: str,
    id_card_no: str,
    provider_name: str,
    mode: str,
    face_image_url: str | None = None,
    material_expires_at: datetime | None = None,
    provider_task_id: str | None = None,
    provider_response: dict[str, Any] | None = None,
    status: str = "pending",
) -> UserVerificationRecord:
    stored_real_name = _stored_identity(real_name)
    stored_id_card_no = _stored_identity(id_card_no)
    record = UserVerificationRecord(
        user_id=user.id,
        real_name=stored_real_name,
        id_card_no=stored_id_card_no,
        id_card_no_hash=_id_card_hash(id_card_no),
        id_card_face_image=face_image_url,
        provider=provider_name,
        mode=mode,
        provider_task_id=provider_task_id,
        status=status,
        provider_response=provider_response,
        material_expires_at=material_expires_at,
        material_cleaned=False,
    )
    db.add(record)
    user.id_card_verified = ID_CARD_VERIFIED_STATUS.PENDING.value
    user.id_card_verify_remark = None
    if face_image_url:
        user.id_card_face_image = face_image_url
    await db.flush()
    await db.refresh(record)
    return record


async def _apply_provider_result(
    db: AsyncSession,
    user: User,
    record: UserVerificationRecord,
    result: VerificationResult,
) -> dict[str, Any]:
    previous_response = record.provider_response if isinstance(record.provider_response, dict) else {}
    previous_quota = previous_response.get("quota")
    record.provider_task_id = result.external_id
    record.provider_response = {
        "mode": result.mode,
        "provider": result.provider,
        "status": result.status,
        "passed": result.passed,
        "score": result.score,
        "reason": result.reason,
        "raw": result.raw,
    }
    if previous_quota:
        record.provider_response["quota"] = previous_quota

    if result.status in {"pending", "processing"}:
        record.status = "processing"
        record.reject_reason = None
        record.rejected_at = None
        user.id_card_verified = ID_CARD_VERIFIED_STATUS.PENDING.value
        user.id_card_verify_remark = None
        await db.flush()
        return {
            "record_id": record.id,
            "passed": False,
            "status": record.status,
            "mode": record.mode,
            "provider": record.provider,
            "score": result.score,
            "verified_at": None,
            "reject_reason": None,
            "reason": result.reason,
        }

    if result.passed:
        verified_at = _now()
        record.status = "verified"
        record.verified_at = verified_at
        record.rejected_at = None
        record.reject_reason = None
        user.id_card_verified = ID_CARD_VERIFIED_STATUS.VERIFIED.value
        user.id_card_verified_at = verified_at
        user.id_card_verify_remark = None
        user.real_name = record.real_name
        user.id_card_no = record.id_card_no
        user.id_card_no_hash = record.id_card_no_hash
    else:
        reason = result.reason or "实名认证未通过"
        record.status = "rejected"
        record.rejected_at = _now()
        record.reject_reason = reason
        user.id_card_verified = ID_CARD_VERIFIED_STATUS.REJECTED.value
        user.id_card_verify_remark = reason

    await db.flush()
    return {
        "record_id": record.id,
        "passed": result.passed,
        "status": record.status,
        "mode": record.mode,
        "provider": record.provider,
        "score": result.score,
        "verified_at": record.verified_at.isoformat() if record.verified_at else None,
        "reject_reason": record.reject_reason,
    }


async def _mark_provider_error(
    db: AsyncSession,
    user: User,
    record: UserVerificationRecord,
    exc: VerificationProviderError,
) -> None:
    reason = f"第三方接口错误：{exc.message}"
    previous_response = record.provider_response if isinstance(record.provider_response, dict) else {}
    record.status = "rejected"
    record.rejected_at = _now()
    record.reject_reason = reason
    record.provider_response = {"error": exc.message, "code": exc.code, "raw": exc.raw}
    if previous_response.get("quota"):
        record.provider_response["quota"] = previous_response["quota"]
    user.id_card_verified = ID_CARD_VERIFIED_STATUS.REJECTED.value
    user.id_card_verify_remark = reason
    await db.flush()


async def submit_two_factor(
    db: AsyncSession,
    user: User,
    *,
    real_name: str,
    id_card_no: str,
) -> dict[str, Any]:
    real_name = real_name.strip()
    id_card_no = _normalize_id_card(id_card_no)
    await _ensure_attempt_allowed(db, user)
    await _ensure_identity_not_bound(db, user, id_card_no)
    provider, config = await _get_provider(db)
    await _ensure_verification_quota_available(db, user, config)
    if "two_factor" not in getattr(provider, "supported_modes", ()):
        raise HTTPException(status_code=400, detail="当前实名认证插件不支持二要素核验")

    record = await _create_record(
        db,
        user,
        real_name=real_name,
        id_card_no=id_card_no,
        provider_name=getattr(provider, "name", "verification"),
        mode="two_factor",
    )
    quota = await _consume_verification_quota(db, user, config)
    _attach_quota_to_record(record, quota)
    try:
        result = await provider.verify_two_factor(real_name=real_name, id_card_no=id_card_no)
    except VerificationProviderError as exc:
        await _mark_provider_error(db, user, record, exc)
        raise HTTPException(status_code=exc.status_code, detail=record.reject_reason) from exc
    summary = await _apply_provider_result(db, user, record, result)
    summary["quota"] = quota
    await db.flush()
    return summary


async def submit_three_factor(
    db: AsyncSession,
    user: User,
    *,
    real_name: str,
    id_card_no: str,
    phone: str,
) -> dict[str, Any]:
    real_name = real_name.strip()
    id_card_no = _normalize_id_card(id_card_no)
    phone = phone.strip()
    await _ensure_attempt_allowed(db, user)
    await _ensure_identity_not_bound(db, user, id_card_no)
    provider, config = await _get_provider(db)
    await _ensure_verification_quota_available(db, user, config)
    if "three_factor" not in getattr(provider, "supported_modes", ()):
        raise HTTPException(status_code=400, detail="当前实名认证插件不支持三要素核验")

    record = await _create_record(
        db,
        user,
        real_name=real_name,
        id_card_no=id_card_no,
        provider_name=getattr(provider, "name", "verification"),
        mode="three_factor",
    )
    quota = await _consume_verification_quota(db, user, config)
    _attach_quota_to_record(record, quota)
    try:
        result = await provider.verify_three_factor(
            real_name=real_name,
            id_card_no=id_card_no,
            phone=phone,
        )
    except VerificationProviderError as exc:
        await _mark_provider_error(db, user, record, exc)
        raise HTTPException(status_code=exc.status_code, detail=record.reject_reason) from exc
    summary = await _apply_provider_result(db, user, record, result)
    summary["quota"] = quota
    await db.flush()
    return summary


def _strip_data_url_base64(value: str) -> str:
    text = value.strip()
    if "," in text and text.lower().startswith("data:"):
        return text.split(",", 1)[1].strip()
    return text


def _extension_for_upload(filename: str | None, content_type: str | None) -> str:
    if content_type in FACE_IMAGE_CONTENT_TYPES:
        return FACE_IMAGE_CONTENT_TYPES[content_type]
    suffix = Path(filename or "").suffix.lower()
    return suffix if suffix in {".jpg", ".jpeg", ".png", ".bmp"} else ".jpg"


async def submit_face_api(
    db: AsyncSession,
    user: User,
    *,
    real_name: str,
    id_card_no: str,
    image_bytes: bytes,
    filename: str | None,
    content_type: str | None,
) -> dict[str, Any]:
    raise HTTPException(status_code=400, detail="本地拍摄/上传扫脸已关闭，请使用第三方 H5 跳转核身")

    real_name = real_name.strip()
    id_card_no = _normalize_id_card(id_card_no)
    await _ensure_attempt_allowed(db, user)
    await _ensure_identity_not_bound(db, user, id_card_no)
    provider, config = await _get_provider(db)
    await _ensure_verification_quota_available(db, user, config)

    if content_type not in FACE_IMAGE_CONTENT_TYPES:
        raise HTTPException(status_code=400, detail="扫脸图片仅支持 JPG、JPEG、PNG、BMP")
    if not image_bytes:
        raise HTTPException(status_code=400, detail="请上传扫脸图片")
    if len(image_bytes) > FACE_IMAGE_MAX_BYTES:
        raise HTTPException(status_code=400, detail="扫脸图片不能超过 2MB")

    if "face_api" not in getattr(provider, "supported_modes", ()):
        raise HTTPException(status_code=400, detail="当前实名认证插件不支持扫脸 API 核验")

    retention_days = max(1, int(getattr(config, "RETENTION_DAYS", 3) or 3))
    expires_at = _now() + timedelta(days=retention_days)
    ext = _extension_for_upload(filename, content_type)
    upload_dir = settings.MEDIA_ROOT_PATH / "verifications" / str(user.id)
    upload_dir.mkdir(parents=True, exist_ok=True)
    name = f"face_{uuid.uuid4().hex}{ext}"
    path = upload_dir / name
    path.write_bytes(image_bytes)
    face_image_url = f"{settings.MEDIA_URL}/verifications/{user.id}/{name}"

    record = await _create_record(
        db,
        user,
        real_name=real_name,
        id_card_no=id_card_no,
        provider_name=getattr(provider, "name", "verification"),
        mode="face_api",
        face_image_url=face_image_url,
        material_expires_at=expires_at,
    )
    quota = await _consume_verification_quota(db, user, config)
    _attach_quota_to_record(record, quota)

    image_base64 = _strip_data_url_base64(base64.b64encode(image_bytes).decode("ascii"))
    try:
        result = await provider.verify_face_api(
            real_name=real_name,
            id_card_no=id_card_no,
            image_base64=image_base64,
            image_type="BASE64",
        )
    except VerificationProviderError as exc:
        await _mark_provider_error(db, user, record, exc)
        raise HTTPException(status_code=exc.status_code, detail=record.reject_reason) from exc
    summary = await _apply_provider_result(db, user, record, result)
    summary["quota"] = quota
    await db.flush()
    return summary


async def create_redirect_session(
    db: AsyncSession,
    user: User,
    *,
    real_name: str,
    id_card_no: str,
    return_url: str | None = None,
    notify_url: str | None = None,
) -> dict[str, Any]:
    real_name = real_name.strip()
    id_card_no = _normalize_id_card(id_card_no)
    await _ensure_attempt_allowed(db, user)
    await _ensure_identity_not_bound(db, user, id_card_no)
    provider, config = await _get_provider(db)
    await _ensure_verification_quota_available(db, user, config)
    if "face_redirect" not in getattr(provider, "supported_modes", ()):
        raise HTTPException(status_code=400, detail="current verification plugin does not support H5 redirect")
    business_settings = await system_setting_service.get_business_settings(db)
    provider_name = getattr(provider, "name", "verification")
    safe_return_url = _safe_return_url(return_url, frontend_url=business_settings.frontend_url)
    safe_notify_url = _safe_notify_url(
        notify_url,
        backend_url=business_settings.backend_url,
        provider_name=provider_name,
    )
    retention_days = max(1, int(getattr(config, "RETENTION_DAYS", 3) or 3))
    expires_at = _now() + timedelta(days=retention_days)
    record = await _create_record(
        db,
        user,
        real_name=real_name,
        id_card_no=id_card_no,
        provider_name=provider_name,
        mode="face_redirect",
        material_expires_at=expires_at,
        provider_response={
            "mode": "face_redirect",
            "provider": provider_name,
            "status": "creating",
            "return_url": safe_return_url,
            "notify_url": safe_notify_url,
        },
        status="processing",
    )
    quota = await _consume_verification_quota(db, user, config)
    _attach_quota_to_record(record, quota)
    try:
        session = await provider.create_redirect_session(
            real_name=real_name,
            id_card_no=id_card_no,
            return_url=safe_return_url,
            notify_url=safe_notify_url,
        )
    except VerificationProviderError as exc:
        await _mark_provider_error(db, user, record, exc)
        raise HTTPException(status_code=exc.status_code, detail=exc.message) from exc

    record.provider_task_id = session.external_id
    record.provider_response = {
        "mode": "face_redirect",
        "provider": provider_name,
        "status": "processing",
        "return_url": safe_return_url,
        "notify_url": safe_notify_url,
        "session": {
            "external_id": session.external_id,
            "expires_in": session.expires_in,
            "request_id": session.raw.get("RequestId") if isinstance(session.raw, dict) else None,
            "task_id": session.raw.get("task_id") if isinstance(session.raw, dict) else None,
            "task_no": session.raw.get("task_no") if isinstance(session.raw, dict) else None,
        },
        "quota": quota,
    }
    await db.flush()
    return {
        "record_id": record.id,
        "url": session.url,
        "qr_payload": session.url,
        "external_id": session.external_id,
        "expires_in": session.expires_in,
        "quota": quota,
    }


def _record_result_summary(record: UserVerificationRecord) -> dict[str, Any]:
    provider_response = record.provider_response if isinstance(record.provider_response, dict) else {}
    return {
        "record_id": record.id,
        "passed": record.status == "verified",
        "status": record.status,
        "mode": record.mode,
        "provider": record.provider,
        "score": provider_response.get("score"),
        "verified_at": record.verified_at.isoformat() if record.verified_at else None,
        "reject_reason": record.reject_reason,
        "reason": provider_response.get("reason"),
    }


def _record_session_value(record: UserVerificationRecord, key: str) -> str | None:
    provider_response = record.provider_response if isinstance(record.provider_response, dict) else {}
    session = provider_response.get("session") if isinstance(provider_response.get("session"), dict) else {}
    value = session.get(key)
    if value in (None, ""):
        return None
    return str(value).strip()


async def query_redirect_result(
    db: AsyncSession,
    user: User,
    *,
    external_id: str | None = None,
    record_id: int | None = None,
) -> dict[str, Any]:
    external_id = (external_id or "").strip()
    stmt = select(UserVerificationRecord).where(
        UserVerificationRecord.user_id == user.id,
        UserVerificationRecord.mode == "face_redirect",
    )
    if external_id:
        stmt = stmt.where(UserVerificationRecord.provider_task_id == external_id)
    elif record_id:
        stmt = stmt.where(UserVerificationRecord.id == record_id)
    else:
        stmt = stmt.order_by(desc(UserVerificationRecord.created_at)).limit(1)

    result = await db.execute(stmt)
    record = result.scalar_one_or_none()
    if not record and external_id:
        fallback = await db.execute(
            select(UserVerificationRecord)
            .where(
                UserVerificationRecord.user_id == user.id,
                UserVerificationRecord.mode == "face_redirect",
                UserVerificationRecord.status == "processing",
            )
            .order_by(desc(UserVerificationRecord.created_at))
            .limit(20)
        )
        for item in fallback.scalars().all():
            if external_id in {_record_session_value(item, "task_id"), _record_session_value(item, "task_no")}:
                record = item
                break
    if not record:
        raise HTTPException(status_code=404, detail="未找到待查询的跳转实名认证记录")

    if record.status in {"verified", "rejected"}:
        return _record_result_summary(record)
    if not record.provider_task_id:
        raise HTTPException(status_code=400, detail="该实名认证记录缺少第三方会话编号")

    provider, _config = await _get_provider_for_record(db, record)

    getter = getattr(provider, "get_redirect_result", None)
    if not callable(getter):
        raise HTTPException(status_code=400, detail="当前实名认证插件不支持查询跳转核身结果")

    try:
        verification_result = await getter(
            external_id=record.provider_task_id,
            real_name=_plain_identity(record.real_name) or "",
            id_card_no=_plain_identity(record.id_card_no) or "",
        )
    except VerificationProviderError as exc:
        raise HTTPException(status_code=exc.status_code, detail=exc.message) from exc
    return await _apply_provider_result(db, user, record, verification_result)


def _flatten_payload(payload: Any) -> dict[str, Any]:
    if not isinstance(payload, dict):
        return {}
    flat: dict[str, Any] = {}
    for key, value in payload.items():
        if isinstance(value, dict):
            nested = _flatten_payload(value)
            flat.update(nested)
        elif isinstance(value, list) and len(value) == 1:
            flat[str(key)] = value[0]
        else:
            flat[str(key)] = value
    return flat


def _extract_external_id(payload: dict[str, Any]) -> str | None:
    flat = _flatten_payload(payload)
    for key in (
        "external_id",
        "biz_token",
        "BizToken",
        "bizToken",
        "token",
        "Token",
        "task_no",
        "TaskNo",
        "taskNo",
        "task_id",
        "TaskId",
        "taskId",
        "order_no",
        "OrderNo",
    ):
        value = flat.get(key)
        if value not in (None, ""):
            return str(value).strip()
    return None


async def handle_redirect_notify(
    db: AsyncSession,
    *,
    provider_name: str,
    payload: dict[str, Any],
) -> dict[str, Any]:
    external_id = _extract_external_id(payload)
    if not external_id:
        return {"accepted": True, "updated": False, "reason": "external_id_missing"}

    result = await db.execute(
        select(UserVerificationRecord).where(
            UserVerificationRecord.provider == provider_name,
            UserVerificationRecord.provider_task_id == external_id,
            UserVerificationRecord.mode == "face_redirect",
        )
    )
    record = result.scalar_one_or_none()
    if not record:
        fallback = await db.execute(
            select(UserVerificationRecord)
            .where(
                UserVerificationRecord.provider == provider_name,
                UserVerificationRecord.mode == "face_redirect",
                UserVerificationRecord.status == "processing",
            )
            .order_by(desc(UserVerificationRecord.created_at))
            .limit(50)
        )
        for item in fallback.scalars().all():
            if external_id in {_record_session_value(item, "task_id"), _record_session_value(item, "task_no")}:
                record = item
                break
    if not record:
        return {"accepted": True, "updated": False, "reason": "record_not_found"}
    if record.status in {"verified", "rejected"}:
        return {"accepted": True, "updated": False, "status": record.status}

    user = await db.get(User, record.user_id)
    if not user:
        return {"accepted": True, "updated": False, "reason": "user_not_found"}

    provider, _config = await _get_provider_for_record(db, record)
    getter = getattr(provider, "get_redirect_result", None)
    if not callable(getter):
        return {"accepted": True, "updated": False, "reason": "query_not_supported"}

    try:
        verification_result = await getter(
            external_id=record.provider_task_id,
            real_name=_plain_identity(record.real_name) or "",
            id_card_no=_plain_identity(record.id_card_no) or "",
        )
    except VerificationProviderError as exc:
        log.warning("verification notify query failed: provider=%s external_id=%s error=%s", provider_name, external_id, exc.message)
        return {"accepted": True, "updated": False, "reason": exc.message}

    summary = await _apply_provider_result(db, user, record, verification_result)
    return {"accepted": True, "updated": True, "result": summary}


async def poll_pending_redirect_records(db: AsyncSession, *, limit: int = 50) -> dict[str, int]:
    result = await db.execute(
        select(UserVerificationRecord)
        .where(
            UserVerificationRecord.mode == "face_redirect",
            UserVerificationRecord.status == "processing",
            UserVerificationRecord.provider_task_id.is_not(None),
        )
        .order_by(UserVerificationRecord.id.asc())
        .limit(limit)
    )
    records = list(result.scalars().all())
    checked = 0
    updated = 0
    failed = 0
    for record in records:
        user = await db.get(User, record.user_id)
        if not user:
            failed += 1
            continue
        try:
            provider, _config = await _get_provider_for_record(db, record)
            getter = getattr(provider, "get_redirect_result", None)
            if not callable(getter):
                failed += 1
                continue
            verification_result = await getter(
                external_id=record.provider_task_id or "",
                real_name=_plain_identity(record.real_name) or "",
                id_card_no=_plain_identity(record.id_card_no) or "",
            )
            before = record.status
            await _apply_provider_result(db, user, record, verification_result)
            checked += 1
            if record.status != before:
                updated += 1
        except Exception as exc:  # noqa: BLE001
            failed += 1
            log.warning("pending redirect verification poll failed: record=%s error=%s", record.id, exc)
    return {"checked": checked, "updated": updated, "failed": failed}


def is_user_verified(user: User) -> bool:
    return user.id_card_verified == ID_CARD_VERIFIED_STATUS.VERIFIED.value


def require_verified(user: User) -> None:
    if not is_user_verified(user):
        raise HTTPException(
            status_code=403,
            detail={
                "code": "REAL_NAME_REQUIRED",
                "message": "请先完成实名认证",
                "verification_url": _verification_url(),
            },
        )


def require_verified_for_products(
    user: User,
    products: dict[int, Product],
    groups: dict[int, ProductGroup],
) -> None:
    required_names: list[str] = []
    for product in products.values():
        group = groups.get(int(product.group_id)) if product.group_id else None
        if bool(getattr(product, "requires_real_name", False)) or bool(getattr(group, "requires_real_name", False)):
            required_names.append(product.name)

    if not required_names or is_user_verified(user):
        return

    raise HTTPException(
        status_code=403,
        detail={
            "code": "REAL_NAME_REQUIRED",
            "message": "购买该产品需要先完成实名认证",
            "products": required_names,
            "verification_url": _verification_url(),
        },
    )


async def list_verification_records(
    db: AsyncSession,
    *,
    status: str | None = None,
    user_id: int | None = None,
    page: int = 1,
    per_page: int = 20,
) -> dict[str, Any]:
    stmt = select(UserVerificationRecord).order_by(desc(UserVerificationRecord.created_at))
    count_stmt = select(UserVerificationRecord)

    if status:
        stmt = stmt.where(UserVerificationRecord.status == status)
        count_stmt = count_stmt.where(UserVerificationRecord.status == status)
    if user_id:
        stmt = stmt.where(UserVerificationRecord.user_id == user_id)
        count_stmt = count_stmt.where(UserVerificationRecord.user_id == user_id)

    total = await db.scalar(select(func.count()).select_from(count_stmt.subquery()))
    offset = max(0, page - 1) * per_page
    result = await db.execute(stmt.offset(offset).limit(per_page))
    items = result.scalars().all()

    return {
        "list": [
            {
                "id": r.id,
                "user_id": r.user_id,
                "real_name": _mask_name(r.real_name),
                "id_card_no": _mask_id_card(r.id_card_no),
                "provider": r.provider,
                "mode": r.mode,
                "status": r.status,
                "provider_task_id": r.provider_task_id,
                "score": (r.provider_response or {}).get("score") if isinstance(r.provider_response, dict) else None,
                "verified_at": r.verified_at.isoformat() if r.verified_at else None,
                "rejected_at": r.rejected_at.isoformat() if r.rejected_at else None,
                "reject_reason": r.reject_reason,
                "material_expires_at": r.material_expires_at.isoformat() if r.material_expires_at else None,
                "material_cleaned": r.material_cleaned,
                "created_at": r.created_at.isoformat() if r.created_at else None,
            }
            for r in items
        ],
        "meta": {
            "total": total or 0,
            "page": page,
            "per_page": per_page,
            "total_page": ((total or 0) + per_page - 1) // per_page if (total or 0) > 0 else 0,
        },
    }


async def manual_verify_user(
    db: AsyncSession,
    user_id: int,
    *,
    verified: bool,
    reason: str | None = None,
) -> dict[str, Any]:
    user = await db.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")

    result = await db.execute(
        select(UserVerificationRecord)
        .where(UserVerificationRecord.user_id == user_id)
        .order_by(desc(UserVerificationRecord.created_at))
        .limit(1)
    )
    record = result.scalar_one_or_none()

    if verified:
        user.id_card_verified = ID_CARD_VERIFIED_STATUS.VERIFIED.value
        user.id_card_verified_at = _now()
        user.id_card_verify_remark = None
        if record:
            user.real_name = record.real_name or user.real_name
            user.id_card_no = record.id_card_no or user.id_card_no
            user.id_card_no_hash = record.id_card_no_hash or user.id_card_no_hash
            record.status = "verified"
            record.verified_at = user.id_card_verified_at
            record.reject_reason = None
            record.rejected_at = None
    else:
        user.id_card_verified = ID_CARD_VERIFIED_STATUS.REJECTED.value
        user.id_card_verify_remark = reason or "人工审核未通过"
        if record:
            record.status = "rejected"
            record.rejected_at = _now()
            record.reject_reason = user.id_card_verify_remark

    await db.flush()
    return {
        "user_id": user_id,
        "status": user.id_card_verified,
        "verified_at": user.id_card_verified_at.isoformat() if user.id_card_verified_at else None,
    }


async def check_user_verified(db: AsyncSession, user_id: int) -> bool:
    user = await db.get(User, user_id)
    return bool(user and is_user_verified(user))
