tencent_face_recognition/tencent_cloud_client.py

"""腾讯云人脸识别客户端（精简门面）。

原 1311 行的客户端已拆分为：
- ``retry.py``: 重试配置与执行器
- ``image_processor.py``: 图片获取/压缩/编码
- ``api_operations.py``: 各 API 操作的实现（搜索/检测/属性/人员/人脸/库信息/列表）

本文件仅保留：凭据创建、HTTP Session 管理、错误脱敏、以及对外暴露的
``TencentCloudClient`` 门面，委托给上述模块。
"""

from __future__ import annotations

import logging
from typing import Any, Dict, List, Optional

import requests
from tencentcloud.common import credential
from tencentcloud.common.exception.tencent_cloud_sdk_exception import (
    TencentCloudSDKException,
)
from tencentcloud.iai.v20200303 import iai_client

from . import api_operations
from .errors import (
    ApiError,
    AuthenticationError,
    log_and_raise_error,
    validate_config,
)
from .image_processor import ImageCache, get_image_base64
from .retry import RetryConfig
from .const import CONF_REGION, CONF_SECRET_ID, CONF_SECRET_KEY

_LOGGER = logging.getLogger(__name__)


class TencentCloudClient:
    """腾讯云人脸识别客户端门面。"""

    def __init__(
        self,
        secret_id: str,
        secret_key: str,
        region: str = "ap-shanghai",
        retry_config: Optional[RetryConfig] = None,
    ) -> None:
        validate_config({
            CONF_SECRET_ID: secret_id,
            CONF_SECRET_KEY: secret_key,
            CONF_REGION: region,
        })

        self.secret_id = secret_id
        self.secret_key = secret_key
        self.region = region
        self.retry_config = retry_config or RetryConfig()
        self.client = self._create_client()

        # HTTP 连接池化
        self._session = requests.Session()
        self._session.mount(
            "http://", requests.adapters.HTTPAdapter(max_retries=2)
        )
        self._session.mount(
            "https://", requests.adapters.HTTPAdapter(max_retries=2)
        )

        # 图片缓存（url/path），通过 session 下载
        self._image_cache = ImageCache(
            maxsize=32,
            download_fn=lambda url: get_image_base64(
                image_url=url, session=self._session
            ),
            read_fn=None,
        )

    # --- 资源管理 --------------------------------------------------------

    def close(self) -> None:
        """释放 HTTP Session 等资源。"""
        if self._session:
            self._session.close()
        self._image_cache.clear()

    # --- 客户端创建 -------------------------------------------------------

    def _create_client(self) -> iai_client.IaiClient:
        try:
            _LOGGER.debug("创建腾讯云客户端，区域: %s", self.region)
            cred = credential.Credential(self.secret_id, self.secret_key)
            client = iai_client.IaiClient(cred, self.region)
            _LOGGER.debug("腾讯云客户端创建成功")
            return client
        except Exception as ex:
            _LOGGER.error("创建腾讯云客户端失败: %s", self._sanitize_error(str(ex)))
            log_and_raise_error(
                AuthenticationError,
                "创建腾讯云客户端失败",
                {"region": self.region, "error_type": type(ex).__name__},
            )

    # --- 凭据验证 ---------------------------------------------------------

    def verify_credentials(self) -> bool:
        """验证凭据（轻量 GetGroupList）。失败抛出对应错误。"""
        return api_operations.verify_credentials(
            self.client, retry_config=self.retry_config
        )

    def verify_credentials_lightweight(self) -> bool:
        """测试 API 连接（轻量验证）。失败返回 False，不抛异常。"""
        try:
            self.verify_credentials()
            return True
        except Exception as ex:  # noqa: BLE001
            _LOGGER.error("API连接测试失败: %s", self._sanitize_error(str(ex)))
            return False

    def test_api_connection(self, group_id: str) -> bool:
        """测试与指定人员库的连接。失败返回 False。"""
        try:
            _LOGGER.info("开始测试API连接: group_id=%s", group_id)
            result = self.get_group_info(group_id)
            if result.get("success", False):
                _LOGGER.info("API连接测试成功")
                return True
            _LOGGER.error("API连接测试失败: %s", result.get("error_message"))
            return False
        except Exception as ex:  # noqa: BLE001
            _LOGGER.error("API连接测试失败: %s", self._sanitize_error(str(ex)))
            return False

    # 兼容旧调用名
    _test_api_connection = test_api_connection

    # --- 图片 ------------------------------------------------------------

    def _get_image_base64(
        self,
        image_url: Optional[str] = None,
        image_path: Optional[str] = None,
        image_file: Optional[Any] = None,
    ) -> str:
        """获取图片 base64（带 url/path 缓存）。"""
        if image_url:
            return self._image_cache.get_url(image_url)
        if image_path:
            return self._image_cache.get_path(image_path)
        if image_file:
            return self._image_cache.get(image_file=image_file)
        log_and_raise_error(
            ApiError,
            "必须提供 image_url、image_path 或 image_file",
            {"reason": "no_image_source"},
        )

    # --- 业务 API（委托 api_operations） ----------------------------------

    def search_faces(
        self,
        image_url: Optional[str] = None,
        image_path: Optional[str] = None,
        image_file: Optional[Any] = None,
        image_base64: Optional[str] = None,
        group_ids: Optional[List[str]] = None,
        max_face_num: int = 1,
        min_face_size: int = 34,
        max_user_num: int = 5,
        quality_control: int = 1,
        need_rotate_check: int = 1,
        face_match_threshold: float = 60.0,
    ) -> Dict[str, Any]:
        if image_base64 is None:
            image_base64 = self._get_image_base64(image_url, image_path, image_file)
        if not group_ids:
            _LOGGER.warning("未提供人员库ID，搜索可能失败")
        return api_operations.search_faces(
            self.client,
            image_base64=image_base64,
            group_ids=group_ids or [],
            max_face_num=max_face_num,
            min_face_size=min_face_size,
            max_user_num=max_user_num,
            quality_control=quality_control,
            need_rotate_check=need_rotate_check,
            face_match_threshold=face_match_threshold,
            retry_config=self.retry_config,
            sanitize=self._sanitize_error,
        )

    def detect_faces(
        self,
        image_url: Optional[str] = None,
        image_path: Optional[str] = None,
        image_file: Optional[Any] = None,
        image_base64: Optional[str] = None,
        max_face_num: int = 1,
        min_face_size: int = 34,
        need_rotate_check: int = 1,
    ) -> Dict[str, Any]:
        if image_base64 is None:
            image_base64 = self._get_image_base64(image_url, image_path, image_file)
        return api_operations.detect_faces(
            self.client,
            image_base64=image_base64,
            max_face_num=max_face_num,
            min_face_size=min_face_size,
            need_rotate_check=need_rotate_check,
            retry_config=self.retry_config,
            sanitize=self._sanitize_error,
        )

    def get_face_attributes(
        self,
        image_url: Optional[str] = None,
        image_path: Optional[str] = None,
        image_file: Optional[Any] = None,
        image_base64: Optional[str] = None,
        max_face_num: int = 1,
        need_rotate_check: int = 1,
    ) -> Dict[str, Any]:
        if image_base64 is None:
            image_base64 = self._get_image_base64(image_url, image_path, image_file)
        return api_operations.get_face_attributes(
            self.client,
            image_base64=image_base64,
            max_face_num=max_face_num,
            need_rotate_check=need_rotate_check,
            retry_config=self.retry_config,
            sanitize=self._sanitize_error,
        )

    def create_person(
        self,
        person_id: str,
        person_name: str,
        group_id: str,
        image_url: Optional[str] = None,
        image_path: Optional[str] = None,
        image_file: Optional[Any] = None,
        image_base64: Optional[str] = None,
        gender: Optional[int] = None,
        person_tag: Optional[str] = None,
        quality_control: int = 1,
        need_rotate_check: int = 1,
    ) -> Dict[str, Any]:
        if image_base64 is None and any([image_url, image_path, image_file]):
            image_base64 = self._get_image_base64(image_url, image_path, image_file)
        return api_operations.create_person(
            self.client,
            person_id=person_id,
            person_name=person_name,
            group_id=group_id,
            image_base64=image_base64,
            gender=gender,
            person_tag=person_tag,
            quality_control=quality_control,
            need_rotate_check=need_rotate_check,
            retry_config=self.retry_config,
            sanitize=self._sanitize_error,
        )

    def delete_person(self, person_id: str) -> Dict[str, Any]:
        return api_operations.delete_person(
            self.client,
            person_id=person_id,
            retry_config=self.retry_config,
            sanitize=self._sanitize_error,
        )

    def create_face(
        self,
        person_id: str,
        image_url: Optional[str] = None,
        image_path: Optional[str] = None,
        image_file: Optional[Any] = None,
        image_base64: Optional[str] = None,
        quality_control: int = 1,
        need_rotate_check: int = 1,
    ) -> Dict[str, Any]:
        if image_base64 is None:
            image_base64 = self._get_image_base64(image_url, image_path, image_file)
        return api_operations.create_face(
            self.client,
            person_id=person_id,
            image_base64=image_base64,
            quality_control=quality_control,
            need_rotate_check=need_rotate_check,
            retry_config=self.retry_config,
            sanitize=self._sanitize_error,
        )

    def delete_face(self, person_id: str, face_id: str) -> Dict[str, Any]:
        return api_operations.delete_face(
            self.client,
            person_id=person_id,
            face_id=face_id,
            retry_config=self.retry_config,
            sanitize=self._sanitize_error,
        )

    def get_group_info(self, group_id: str) -> Dict[str, Any]:
        return api_operations.get_group_info(
            self.client,
            group_id=group_id,
            retry_config=self.retry_config,
            sanitize=self._sanitize_error,
        )

    def get_person_list(
        self, group_id: str, limit: int = 100, offset: int = 0
    ) -> Dict[str, Any]:
        return api_operations.get_person_list(
            self.client,
            group_id=group_id,
            limit=limit,
            offset=offset,
            retry_config=self.retry_config,
            sanitize=self._sanitize_error,
        )

    def get_person_list_all(self, group_id: str) -> Dict[str, Any]:
        return api_operations.get_person_list_all(
            self.client,
            group_id=group_id,
            retry_config=self.retry_config,
            sanitize=self._sanitize_error,
        )

    # --- 辅助 ------------------------------------------------------------

    def _sanitize_error(self, error_msg: str) -> str:
        """脱敏错误信息中的凭据。"""
        if not isinstance(error_msg, str):
            return str(error_msg)
        if self.secret_id and self.secret_id in error_msg:
            error_msg = error_msg.replace(
                self.secret_id, f"{self.secret_id[:8]}***"
            )
        if self.secret_key and self.secret_key in error_msg:
            error_msg = error_msg.replace(self.secret_key, "***")
        return error_msg

    # 兼容旧代码：暴露 TencentCloudSDKException 别名
    @staticmethod
    def is_sdk_exception(ex: Exception) -> bool:
        return isinstance(ex, TencentCloudSDKException)