commit 4e0c898cefa103e08ac12916d0801b2ab9159d33 Author: xyzm Date: Sun Sep 7 21:48:21 2025 +0800 first commit diff --git a/README.md b/README.md new file mode 100644 index 0000000..518e5f7 --- /dev/null +++ b/README.md @@ -0,0 +1,94 @@ +# 腾讯云人脸识别Home Assistant插件 + +这是一个基于腾讯云人脸识别API的Home Assistant插件，提供了人脸搜索功能。 + +## 功能特性 + +- **人脸搜索**：支持通过图片URL或本地文件路径搜索人脸，返回匹配的人员信息 + +## 安装 + +1. 将`custom_components/tencent_face_recognition`目录复制到Home Assistant的`custom_components`目录下 +2. 重启Home Assistant +3. 在"配置" -> "集成"中点击"+"添加集成 +4. 搜索"腾讯云人脸识别"并点击 +5. 输入您的腾讯云凭据和配置信息 + +## 配置 + +### 必需配置 + +- **Secret ID**：腾讯云API的Secret ID +- **Secret Key**：腾讯云API的Secret Key + +### 可选配置 + +- **人员库ID**：默认使用的人员库ID（默认值：Hass） +- **区域**：腾讯云服务区域（默认值：ap-shanghai） +- **名称**：集成名称（默认值：腾讯云人脸识别） + +## 服务 + +插件提供以下服务，可以通过开发者工具中的"服务"选项卡调用： + +### 人脸搜索 + +**服务名称**：`tencent_face_recognition.face_search` + +**描述**：在人员库中搜索人脸 + +**参数**： + +- `image_url`（可选）：要搜索的图片URL +- `image_path`（可选）：要搜索的本地图片路径 +- `person_group_id`（可选）：要搜索的人员库ID，留空使用默认人员库 +- `max_face_num`（可选，默认1）：最多处理的人脸数量 +- `min_face_size`（可选，默认34）：最小人脸尺寸（像素） +- `max_user_num`（可选，默认5）：最多返回的匹配人员数量 +- `quality_control`（可选，默认1）：是否进行质量控制（0：不控制，1：低质量控制，2：高质量控制） +- `need_rotate_check`（可选，默认1）：是否进行旋转检查（0：不检查，1：检查） +- `face_match_threshold`（可选，默认60.0）：人脸匹配阈值（0-100） + +**示例**： + +``` yaml +action: tencent_face_recognition.face_search +response_variable: face_recognition_result_raw +data: + group_id: Hass + face_match_threshold: 60 + min_face_size: 34 + max_face_num: 10 + max_user_num: 10 + image_path: /config/www/camera/face.jpg +``` + + + +```yaml +service: tencent_face_recognition.face_search +data: + image_url: "https://example.com/face.jpg" + group_id: "Hass" +``` + +## 故障排除 + +### 常见问题 + +1. **配置失败**：请检查腾讯云凭据是否正确，以及是否有足够的权限 +2. **图片处理失败**：请确保图片URL可访问或本地文件路径正确 +3. **人脸检测失败**：请确保图片中包含清晰的人脸，且人脸尺寸足够大 +4. **API调用失败**：请检查腾讯云账户余额是否充足，以及API调用配额是否足够 + +### 日志查看 + +在Home Assistant的"开发者工具" -> "日志"中查看插件日志，可以获取更多错误信息。 + +## 贡献 + +欢迎提交Issue和Pull Request来改进这个插件。 + +## 许可证 + +MIT许可证 \ No newline at end of file diff --git a/__init__.py b/__init__.py new file mode 100644 index 0000000..e1a4b08 --- /dev/null +++ b/__init__.py @@ -0,0 +1,103 @@ +""" +腾讯云人脸识别Home Assistant插件 +""" +import logging +from typing import Dict, Any + +from homeassistant.components.http import StaticPathConfig +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant, ServiceCall +from homeassistant.exceptions import ConfigEntryNotReady +from homeassistant.helpers.entity import Entity +from homeassistant.helpers.entity_platform import AddEntitiesCallback + +from .const import ( + DOMAIN, + PLATFORMS, + CONF_SECRET_ID, + CONF_SECRET_KEY, + CONF_REGION, + CONF_PERSON_GROUP_ID +) +from .tencent_cloud_client import TencentCloudClient +from .face_recognition import FaceRecognition +from .services import async_setup_services, async_unload_services + +_LOGGER = logging.getLogger(__name__) + +# 添加传感器平台 +PLATFORMS.append("sensor") + +async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + """设置配置项""" + _LOGGER.info("设置腾讯云人脸识别插件") + + # 获取配置 + secret_id = entry.data.get(CONF_SECRET_ID) + secret_key = entry.data.get(CONF_SECRET_KEY) + region = entry.data.get(CONF_REGION, "ap-shanghai") + person_group_id = entry.data.get(CONF_PERSON_GROUP_ID, "Hass") + + # 创建腾讯云客户端 + try: + client = TencentCloudClient(secret_id, secret_key, region) + except Exception as ex: + _LOGGER.error("创建腾讯云客户端失败: %s", ex) + raise ConfigEntryNotReady(f"创建腾讯云客户端失败: {ex}") + + # 创建人脸识别实例 + face_recognition = FaceRecognition(hass, client) + + # 存储配置和实例 + hass.data.setdefault(DOMAIN, {})[entry.entry_id] = { + "entry": entry, + "client": client, + "face_recognition": face_recognition, + "secret_id": secret_id, + "secret_key": secret_key, + "region": region, + "person_group_id": person_group_id + } + + # 设置服务 + await async_setup_services(hass) + + # 设置平台 + # 转发配置项到平台 + for platform in PLATFORMS: + hass.async_create_task( + hass.config_entries.async_forward_entry_setups(entry, [platform]) + ) + + # 不再注册前端面板，改为使用配置流管理 + + return True + + +async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + """卸载配置项""" + _LOGGER.info("卸载腾讯云人脸识别插件") + + # 注销自定义面板 + # 注意：在新版本的 Home Assistant 中，面板注册方式已更改 + # 不需要手动注销面板，系统会自动处理 + + # 卸载平台 + unload_ok = True + for platform in PLATFORMS: + result = await hass.config_entries.async_forward_entry_unload(entry, platform) + unload_ok = unload_ok and result + + if unload_ok: + # 卸载服务 + await async_unload_services(hass) + + # 移除数据 + hass.data[DOMAIN].pop(entry.entry_id) + + return unload_ok + + +async def async_remove_entry(hass: HomeAssistant, entry: ConfigEntry) -> None: + """移除配置项""" + _LOGGER.info("移除腾讯云人脸识别插件配置") \ No newline at end of file diff --git a/config_flow.py b/config_flow.py new file mode 100644 index 0000000..b16e0af --- /dev/null +++ b/config_flow.py @@ -0,0 +1,445 @@ +"""腾讯云人脸识别插件配置流程""" + +import logging +import re +from typing import Dict, Any, Optional, List + +import voluptuous as vol + +from homeassistant import config_entries +from homeassistant.core import callback +from homeassistant.const import CONF_NAME +from homeassistant.helpers.selector import selector + +from .const import ( + DOMAIN, + CONF_SECRET_ID, + CONF_SECRET_KEY, + CONF_REGION, + CONF_PERSON_GROUP_ID, + DEFAULT_REGION, + ERROR_INVALID_CONFIG, + ERROR_API_ERROR, +) +from .errors import ( + validate_config, + InvalidConfigError, + AuthenticationError, + NetworkError, + RateLimitError, + log_and_raise_error +) +from .tencent_cloud_client import TencentCloudClient + +_LOGGER = logging.getLogger(__name__) + +# 腾讯云区域列表 +AVAILABLE_REGIONS = [ + "ap-beijing", "ap-shanghai", "ap-guangzhou", "ap-chengdu", "ap-chongqing", "ap-nanjing", "ap-hangzhou" +] + +# Secret ID 格式验证 +SECRET_ID_REGEX = re.compile(r'^AKID[a-zA-Z0-9\-]{36,}$') + +# Secret Key 格式验证 +SECRET_KEY_REGEX = re.compile(r'^[a-zA-Z0-9+/]{20,}$') + +# 人员库ID格式验证 +PERSON_GROUP_ID_REGEX = re.compile(r'^[a-zA-Z0-9\-_]{1,64}$') + + +class TencentFaceRecognitionConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): + """腾讯云人脸识别配置流程""" + + VERSION = 1 + CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL + + def __init__(self): + """初始化配置流程""" + self._client = None + self._config_info = {} + + async def async_step_user(self, user_input: Optional[Dict[str, Any]] = None): + """处理用户配置步骤""" + errors = {} + + if user_input is not None: + # 验证配置 + try: + # 验证输入格式 + self._validate_input_format(user_input, errors) + + if errors: + return self._show_user_form(errors) + + # 基本配置验证 + validate_config(user_input) + + # 尝试创建腾讯云客户端以验证凭据 + validation_result = await self._validate_tencent_cloud_credentials(user_input) + + if not validation_result["success"]: + errors = validation_result["errors"] + return self._show_user_form(errors) + + # 保存配置并创建配置项 + return self.async_create_entry( + title=user_input.get(CONF_NAME, "腾讯云人脸识别"), + data=user_input + ) + + except InvalidConfigError as ex: + _LOGGER.error("配置验证失败: %s", ex) + errors["base"] = str(ex) + except Exception as ex: + _LOGGER.error("配置验证失败: %s", ex, exc_info=True) + errors["base"] = f"配置验证失败: {str(ex)}" + + # 显示配置表单 + return self._show_user_form(errors) + + def _show_user_form(self, errors: Dict[str, str]) -> dict: + """显示用户配置表单""" + return self.async_show_form( + step_id="user", + data_schema=vol.Schema({ + vol.Required(CONF_SECRET_ID): vol.All(str, vol.Length(min=1)), + vol.Required(CONF_SECRET_KEY): vol.All(str, vol.Length(min=1)), + vol.Optional(CONF_REGION, default=DEFAULT_REGION): selector({ + "select": { + "options": [{"label": region, "value": region} for region in AVAILABLE_REGIONS], + "mode": "dropdown" + } + }), + vol.Optional(CONF_PERSON_GROUP_ID, default="Hass"): vol.All( + str, + vol.Length(min=1, max=64) + ), + vol.Optional(CONF_NAME, default="腾讯云人脸识别"): vol.All(str, vol.Length(min=1, max=50)), + }), + errors=errors, + description_placeholders={ + "secret_id_example": "AKIDxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx", + "secret_key_example": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx", + "person_group_id_example": "Hass", + "region_example": "ap-shanghai" + } + ) + + def _validate_input_format(self, user_input: Dict[str, Any], errors: Dict[str, str]) -> None: + """验证输入格式""" + # 验证Secret ID格式 + secret_id = user_input.get(CONF_SECRET_ID, "") + if not SECRET_ID_REGEX.match(secret_id): + errors[CONF_SECRET_ID] = "Secret ID格式不正确，应以AKID开头，后跟36个以上字符" + + # 验证Secret Key格式 + secret_key = user_input.get(CONF_SECRET_KEY, "") + if not SECRET_KEY_REGEX.match(secret_key): + errors[CONF_SECRET_KEY] = "Secret Key格式不正确，应为20个以上字符" + + # 验证区域 + region = user_input.get(CONF_REGION, DEFAULT_REGION) + if region not in AVAILABLE_REGIONS: + errors[CONF_REGION] = f"无效的区域，请从列表中选择有效的区域" + + # 验证人员库ID格式 + person_group_id = user_input.get(CONF_PERSON_GROUP_ID, "Hass") + if not PERSON_GROUP_ID_REGEX.match(person_group_id): + errors[CONF_PERSON_GROUP_ID] = "人员库ID只能包含字母、数字、连字符和下划线，长度1-64个字符" + + # 验证名称 + name = user_input.get(CONF_NAME, "腾讯云人脸识别") + if len(name) > 50: + errors[CONF_NAME] = "名称长度不能超过50个字符" + + async def _validate_tencent_cloud_credentials(self, user_input: Dict[str, Any]) -> Dict[str, Any]: + """验证腾讯云凭据""" + try: + _LOGGER.info("开始验证腾讯云凭据") + + # 创建腾讯云客户端 + self._client = TencentCloudClient( + user_input[CONF_SECRET_ID], + user_input[CONF_SECRET_KEY], + user_input.get(CONF_REGION, DEFAULT_REGION) + ) + + _LOGGER.info("腾讯云客户端创建成功，凭据验证通过") + + return {"success": True} + + except AuthenticationError as ex: + _LOGGER.error("腾讯云认证失败: %s", ex) + return { + "success": False, + "errors": { + CONF_SECRET_ID: "认证失败，请检查Secret ID", + CONF_SECRET_KEY: "认证失败，请检查Secret Key", + "base": "腾讯云认证失败，请检查Secret ID和Secret Key是否正确" + } + } + except NetworkError as ex: + _LOGGER.error("腾讯云网络错误: %s", ex) + return { + "success": False, + "errors": { + CONF_REGION: "网络错误，请检查区域设置和网络连接", + "base": "连接腾讯云失败，请检查网络连接和区域设置" + } + } + except RateLimitError as ex: + _LOGGER.error("腾讯云速率限制错误: %s", ex) + return { + "success": False, + "errors": { + "base": "腾讯云API调用频率过高，请稍后再试" + } + } + except Exception as ex: + _LOGGER.error("腾讯云凭据验证失败: %s", ex, exc_info=True) + return { + "success": False, + "errors": { + "base": f"腾讯云API验证失败: {str(ex)}" + } + } + + @staticmethod + @callback + def async_get_options_flow(config_entry): + """获取选项流程""" + return TencentFaceRecognitionOptionsFlow(config_entry) + + +class TencentFaceRecognitionOptionsFlow(config_entries.OptionsFlow): + """腾讯云人脸识别选项流程""" + + def __init__(self, config_entry): + """初始化选项流程""" + # 存储需要的数据而不是整个 config_entry 实例 + self._entry_id = config_entry.entry_id + self._title = config_entry.title + self._data = dict(config_entry.data) + self._options = dict(config_entry.options) + + async def async_step_init(self, user_input=None): + """管理选项流程""" + if user_input is not None: + action = user_input.get("action") + + if action == "config": + return await self.async_step_config() + elif action == "test_connection": + return await self.async_step_test_connection() + elif action == "reauth": + return await self.async_step_reauth() + + return self.async_create_entry(title="", data=user_input) + + # 显示管理菜单 + return self.async_show_form( + step_id="init", + data_schema=vol.Schema({ + vol.Required("action"): selector({ + "select": { + "options": [ + {"label": "配置设置", "value": "config"}, + {"label": "测试连接", "value": "test_connection"}, + {"label": "重新认证", "value": "reauth"} + ] + } + }) + }), + description_placeholders={ + "description": "请选择要管理的项目" + } + ) + + async def async_step_config(self, user_input=None): + """配置设置步骤""" + errors = {} + + if user_input is not None: + # 验证区域 + region = user_input.get(CONF_REGION) + if region not in AVAILABLE_REGIONS: + errors[CONF_REGION] = f"无效的区域，请从列表中选择有效的区域" + + if not errors: + # 验证新配置是否有效 + try: + # 创建临时客户端进行验证 + temp_client = TencentCloudClient( + self._data.get(CONF_SECRET_ID), + self._data.get(CONF_SECRET_KEY), + region + ) + + _LOGGER.info("配置验证成功") + + # 保存配置 + return self.async_create_entry(title="", data=user_input) + + except AuthenticationError as ex: + _LOGGER.error("腾讯云认证失败: %s", ex) + errors[CONF_REGION] = "认证失败，请检查区域设置" + errors["base"] = "腾讯云认证失败，请检查区域设置" + except NetworkError as ex: + _LOGGER.error("腾讯云网络错误: %s", ex) + errors[CONF_REGION] = "网络错误，请检查区域设置" + errors["base"] = "连接腾讯云失败，请检查网络连接和区域设置" + except Exception as ex: + _LOGGER.error("配置验证失败: %s", ex, exc_info=True) + errors["base"] = f"配置验证失败: {str(ex)}" + + options = { + vol.Optional( + CONF_REGION, + default=self._options.get(CONF_REGION, self._data.get(CONF_REGION, DEFAULT_REGION)) + ): selector({ + "select": { + "options": [{"label": region, "value": region} for region in AVAILABLE_REGIONS], + "mode": "dropdown" + } + }), + vol.Optional( + CONF_PERSON_GROUP_ID, + default=self._options.get(CONF_PERSON_GROUP_ID, self._data.get(CONF_PERSON_GROUP_ID, "Hass")) + ): str, + } + + return self.async_show_form( + step_id="config", + data_schema=vol.Schema(options), + errors=errors, + ) + + async def async_step_test_connection(self, user_input=None): + """测试连接步骤""" + errors = {} + connection_status = "未知" + + if user_input is None: + # 尝试测试连接 + try: + # 使用当前配置创建客户端 + client = TencentCloudClient( + self._data.get(CONF_SECRET_ID), + self._data.get(CONF_SECRET_KEY), + self._data.get(CONF_REGION, DEFAULT_REGION) + ) + + # 使用 GetGroupInfo 测试连接 + group_id = self._data.get(CONF_PERSON_GROUP_ID, "Hass") + + success = await self.hass.async_add_executor_job( + client._test_api_connection, + group_id + ) + + if not success: + raise Exception("连接测试失败，请检查人员库ID是否正确") + + connection_status = "连接成功" + _LOGGER.info("连接测试成功") + + except AuthenticationError as ex: + connection_status = f"认证失败: {str(ex)}" + errors["base"] = "认证失败，请检查Secret ID和Secret Key" + _LOGGER.error("连接测试认证失败: %s", ex) + except NetworkError as ex: + connection_status = f"网络错误: {str(ex)}" + errors["base"] = "网络错误，请检查网络连接和区域设置" + _LOGGER.error("连接测试网络失败: %s", ex) + except Exception as ex: + connection_status = f"测试失败: {str(ex)}" + errors["base"] = f"连接测试失败: {str(ex)}" + _LOGGER.error("连接测试失败: %s", ex, exc_info=True) + + return self.async_show_form( + step_id="test_connection", + data_schema=vol.Schema({}), + errors=errors, + description_placeholders={ + "connection_status": connection_status + } + ) + + async def async_step_reauth(self, user_input=None): + """重新认证步骤""" + errors = {} + + if user_input is not None: + # 验证新凭据 + try: + # 验证输入格式 + temp_errors = {} + self._validate_input_format(user_input, temp_errors) + + if temp_errors: + errors = temp_errors + return self.async_show_form( + step_id="reauth", + data_schema=vol.Schema({ + vol.Required(CONF_SECRET_ID, default=self._data.get(CONF_SECRET_ID, "")): str, + vol.Required(CONF_SECRET_KEY, default=self._data.get(CONF_SECRET_KEY, "")): str, + }), + errors=errors, + ) + + # 创建腾讯云客户端验证新凭据 + new_client = TencentCloudClient( + user_input[CONF_SECRET_ID], + user_input[CONF_SECRET_KEY], + self._data.get(CONF_REGION, DEFAULT_REGION) + ) + + _LOGGER.info("重新认证成功") + + # 更新配置 + new_data = dict(self._data) + new_data.update(user_input) + + # 重新创建配置项 + self.hass.config_entries.async_update_entry( + self.hass.config_entries.async_get_entry(self._entry_id), + data=new_data + ) + + return self.async_create_entry(title="", data={}) + + except AuthenticationError as ex: + _LOGGER.error("重新认证失败: %s", ex) + errors["base"] = "认证失败，请检查Secret ID和Secret Key" + except NetworkError as ex: + _LOGGER.error("重新认证网络错误: %s", ex) + errors["base"] = "网络错误，请检查网络连接和区域设置" + except Exception as ex: + _LOGGER.error("重新认证失败: %s", ex, exc_info=True) + errors["base"] = f"重新认证失败: {str(ex)}" + + return self.async_show_form( + step_id="reauth", + data_schema=vol.Schema({ + vol.Required(CONF_SECRET_ID, default=self._data.get(CONF_SECRET_ID, "")): str, + vol.Required(CONF_SECRET_KEY, default=self._data.get(CONF_SECRET_KEY, "")): str, + }), + errors=errors, + description_placeholders={ + "current_secret_id": self._data.get(CONF_SECRET_ID, ""), + "current_region": self._data.get(CONF_REGION, DEFAULT_REGION) + } + ) + + def _validate_input_format(self, user_input: Dict[str, Any], errors: Dict[str, str]) -> None: + """验证输入格式""" + # 验证Secret ID格式 + secret_id = user_input.get(CONF_SECRET_ID, "") + if not SECRET_ID_REGEX.match(secret_id): + errors[CONF_SECRET_ID] = "Secret ID格式不正确，应以AKID开头，后跟36个以上字符" + + # 验证Secret Key格式 + secret_key = user_input.get(CONF_SECRET_KEY, "") + if not SECRET_KEY_REGEX.match(secret_key): + errors[CONF_SECRET_KEY] = "Secret Key格式不正确，应为20个以上字符" \ No newline at end of file diff --git a/const.py b/const.py new file mode 100644 index 0000000..1f9d3ef --- /dev/null +++ b/const.py @@ -0,0 +1,55 @@ +"""腾讯云人脸识别插件常量定义""" + +DOMAIN = "tencent_face_recognition" +PLATFORMS = [] + +# 配置项键 +CONF_SECRET_ID = "secret_id" +CONF_SECRET_KEY = "secret_key" +CONF_REGION = "region" +CONF_PERSON_GROUP_ID = "person_group_id" + +# 默认值 +DEFAULT_REGION = "ap-shanghai" +DEFAULT_PERSON_GROUP_ID = "Hass" +DEFAULT_MAX_FACE_NUM = 1 +DEFAULT_MIN_FACE_SIZE = 34 +DEFAULT_MAX_USER_NUM = 5 +DEFAULT_QUALITY_CONTROL = 1 +DEFAULT_NEED_ROTATE_CHECK = 1 +DEFAULT_FACE_MATCH_THRESHOLD = 60.0 + +# 服务 +SERVICE_FACE_SEARCH = "face_search" +SERVICE_DETECT_FACE = "detect_face" +SERVICE_GET_FACE_ATTRIBUTES = "get_face_attributes" + +# 服务参数 +ATTR_IMAGE_URL = "image_url" +ATTR_IMAGE_PATH = "image_path" +ATTR_IMAGE_FILE = "image_file" +ATTR_MAX_FACE_NUM = "max_face_num" +ATTR_MIN_FACE_SIZE = "min_face_size" +ATTR_MAX_USER_NUM = "max_user_num" +ATTR_QUALITY_CONTROL = "quality_control" +ATTR_NEED_ROTATE_CHECK = "need_rotate_check" +ATTR_FACE_MATCH_THRESHOLD = "face_match_threshold" + +# 错误消息 +ERROR_INVALID_CONFIG = "无效的配置" +ERROR_API_ERROR = "API调用错误" +ERROR_IMAGE_NOT_FOUND = "图片未找到" +ERROR_FACE_NOT_DETECTED = "未检测到人脸" + +# 状态 +STATE_CONNECTED = "已连接" +STATE_DISCONNECTED = "未连接" +STATE_CONNECTING = "连接中" +STATE_ERROR = "连接错误" +STATE_PARTIALLY_CONNECTED = "部分连接" + +# 日志级别 +LOG_LEVEL_DEBUG = "DEBUG" +LOG_LEVEL_INFO = "INFO" +LOG_LEVEL_WARNING = "WARNING" +LOG_LEVEL_ERROR = "ERROR" \ No newline at end of file diff --git a/errors.py b/errors.py new file mode 100644 index 0000000..78e37ef --- /dev/null +++ b/errors.py @@ -0,0 +1,384 @@ +"""错误处理模块""" + +import logging +from typing import Dict, Any, Optional + +from homeassistant.exceptions import HomeAssistantError + +_LOGGER = logging.getLogger(__name__) + +# 错误消息本地化 +ERROR_MESSAGES = { + "zh_CN": { + # 配置错误 + "missing_config": "缺少必需的配置项: {key}", + "invalid_secret_id": "Secret ID 必须是非空字符串，且应以 AKID 开头", + "invalid_secret_key": "Secret Key 必须是非空字符串", + "invalid_region": "区域必须是非空字符串，且为有效的腾讯云区域", + "invalid_person_group_id": "人员库ID必须是非空字符串，只能包含字母、数字、连字符和下划线", + + # 图片错误 + "image_not_found": "图片未找到: {path}", + "image_too_large": "图片文件过大: {path}, 大小: {size} 字节，最大支持: {max_size}", + "image_format_not_supported": "不支持的图片格式: {format}, 支持的格式: {supported_formats}", + "image_decode_failed": "图片解码失败，请确保图片格式正确", + "image_download_timeout": "下载图片超时: {url}", + "image_download_error": "下载图片失败: {url}, 错误: {error}", + "invalid_image_url": "无效的图片URL: {url}", + "permission_denied": "没有权限读取图片文件: {path}", + + # 人脸检测错误 + "face_not_detected": "未检测到人脸，请确保图片中包含清晰的人脸", + "face_too_small": "人脸太小，请使用包含更大人脸的图片", + "face_quality_poor": "人脸质量较差，请使用更清晰的图片", + + # API错误 + "api_error": "API调用失败: {code} - {message}", + "person_not_found": "人员未找到: {person_id}", + "group_not_found": "人员库未找到: {group_id}", + "face_not_found": "人脸未找到: {face_id}", + + # 网络错误 + "network_error": "网络连接错误，请检查网络连接", + "timeout_error": "请求超时，请稍后重试", + + # 认证错误 + "auth_failure": "认证失败，请检查Secret ID和Secret Key是否正确", + "auth_expired": "认证已过期，请更新凭证", + + # 限制错误 + "rate_limit_exceeded": "API调用频率超限，请稍后重试", + "quota_exceeded": "API调用配额已用完，请等待配额重置或升级套餐", + + # 通用错误 + "unknown_error": "未知错误: {message}", + "invalid_parameter": "无效参数: {parameter}", + }, + "en_US": { + # Configuration errors + "missing_config": "Missing required configuration: {key}", + "invalid_secret_id": "Secret ID must be a non-empty string starting with AKID", + "invalid_secret_key": "Secret Key must be a non-empty string", + "invalid_region": "Region must be a non-empty string and a valid Tencent Cloud region", + "invalid_person_group_id": "Person Group ID must be a non-empty string containing only letters, numbers, hyphens, and underscores", + + # Image errors + "image_not_found": "Image not found: {path}", + "image_too_large": "Image file too large: {path}, size: {size} bytes, maximum supported: {max_size}", + "image_format_not_supported": "Unsupported image format: {format}, supported formats: {supported_formats}", + "image_decode_failed": "Image decode failed, please ensure the image format is correct", + "image_download_timeout": "Image download timeout: {url}", + "image_download_error": "Image download failed: {url}, error: {error}", + "invalid_image_url": "Invalid image URL: {url}", + "permission_denied": "Permission denied when reading image file: {path}", + + # Face detection errors + "face_not_detected": "No face detected, please ensure the image contains clear faces", + "face_too_small": "Face too small, please use an image with larger faces", + "face_quality_poor": "Poor face quality, please use a clearer image", + + # API errors + "api_error": "API call failed: {code} - {message}", + "person_not_found": "Person not found: {person_id}", + "group_not_found": "Person group not found: {group_id}", + "face_not_found": "Face not found: {face_id}", + + # Network errors + "network_error": "Network connection error, please check your network connection", + "timeout_error": "Request timeout, please try again later", + + # Authentication errors + "auth_failure": "Authentication failed, please check if Secret ID and Secret Key are correct", + "auth_expired": "Authentication expired, please update credentials", + + # Limit errors + "rate_limit_exceeded": "API call rate limit exceeded, please try again later", + "quota_exceeded": "API call quota used up, please wait for quota reset or upgrade plan", + + # General errors + "unknown_error": "Unknown error: {message}", + "invalid_parameter": "Invalid parameter: {parameter}", + } +} + +# 错误解决建议 +ERROR_SOLUTIONS = { + "zh_CN": { + "missing_config": "请在配置中添加缺少的配置项", + "invalid_secret_id": "请检查Secret ID格式，应以AKID开头", + "invalid_secret_key": "请检查Secret Key格式，确保输入正确", + "invalid_region": "请选择有效的腾讯云区域，如ap-shanghai", + "invalid_person_group_id": "请使用字母、数字、连字符和下划线组合", + + "image_not_found": "请检查图片路径是否正确", + "image_too_large": "请使用小于10MB的图片，或对图片进行压缩", + "image_format_not_supported": "请将图片转换为JPG、PNG、BMP或GIF格式", + "image_decode_failed": "请确保图片文件未损坏，并尝试重新保存", + "image_download_timeout": "请检查网络连接，或尝试使用其他图片URL", + "image_download_error": "请检查图片URL是否有效，或稍后重试", + "invalid_image_url": "请提供有效的图片URL", + "permission_denied": "请检查文件权限，或使用其他图片", + + "face_not_detected": "请使用包含清晰人脸的图片", + "face_too_small": "请使用人脸更大的图片，或调整拍摄距离", + "face_quality_poor": "请使用光线充足、对焦清晰的图片", + + "api_error": "请稍后重试，如果问题持续，请联系技术支持", + "person_not_found": "请检查人员ID是否正确，或确保人员已添加到人员库", + "group_not_found": "请检查人员库ID是否正确，或创建新的人员库", + "face_not_found": "请检查人脸ID是否正确", + + "network_error": "请检查网络连接，确保网络通畅", + "timeout_error": "请稍后重试，或在网络状况良好时使用", + + "auth_failure": "请检查Secret ID和Secret Key是否正确", + "auth_expired": "请更新腾讯云凭证", + + "rate_limit_exceeded": "请降低调用频率，稍后重试", + "quota_exceeded": "请等待配额重置，或升级腾讯云套餐", + + "unknown_error": "请记录错误信息并联系技术支持", + "invalid_parameter": "请检查API调用参数是否正确", + }, + "en_US": { + "missing_config": "Please add the missing configuration item", + "invalid_secret_id": "Please check the Secret ID format, it should start with AKID", + "invalid_secret_key": "Please check the Secret Key format, ensure it's entered correctly", + "invalid_region": "Please select a valid Tencent Cloud region, such as ap-shanghai", + "invalid_person_group_id": "Please use a combination of letters, numbers, hyphens, and underscores", + + "image_not_found": "Please check if the image path is correct", + "image_too_large": "Please use an image smaller than 10MB, or compress the image", + "image_format_not_supported": "Please convert the image to JPG, PNG, BMP, or GIF format", + "image_decode_failed": "Please ensure the image file is not corrupted and try saving it again", + "image_download_timeout": "Please check your network connection, or try using a different image URL", + "image_download_error": "Please check if the image URL is valid, or try again later", + "invalid_image_url": "Please provide a valid image URL", + "permission_denied": "Please check file permissions, or use a different image", + + "face_not_detected": "Please use an image that contains clear faces", + "face_too_small": "Please use an image with larger faces, or adjust the shooting distance", + "face_quality_poor": "Please use a well-lit and focused image", + + "api_error": "Please try again later, if the problem persists, please contact technical support", + "person_not_found": "Please check if the person ID is correct, or ensure the person has been added to the person group", + "group_not_found": "Please check if the person group ID is correct, or create a new person group", + "face_not_found": "Please check if the face ID is correct", + + "network_error": "Please check your network connection, ensure it's working properly", + "timeout_error": "Please try again later, or use when network conditions are good", + + "auth_failure": "Please check if Secret ID and Secret Key are correct", + "auth_expired": "Please update Tencent Cloud credentials", + + "rate_limit_exceeded": "Please reduce call frequency and try again later", + "quota_exceeded": "Please wait for quota reset, or upgrade Tencent Cloud plan", + + "unknown_error": "Please record the error information and contact technical support", + "invalid_parameter": "Please check if the API call parameters are correct", + } +} + + +class TencentFaceRecognitionError(HomeAssistantError): + """腾讯云人脸识别基础错误类""" + + def __init__( + self, + message: str, + error_key: Optional[str] = None, + error_details: Optional[Dict[str, Any]] = None, + locale: str = "zh_CN" + ): + """初始化错误""" + super().__init__(message) + self.error_key = error_key + self.error_details = error_details or {} + self.locale = locale + + # 获取本地化消息 + self.localized_message = self._get_localized_message() + # 获取解决建议 + self.solution = self._get_solution() + + def _get_localized_message(self) -> str: + """获取本地化错误消息""" + if not self.error_key: + return str(self) + + messages = ERROR_MESSAGES.get(self.locale, ERROR_MESSAGES["zh_CN"]) + template = messages.get(self.error_key, str(self)) + + try: + # 使用错误详情填充模板 + return template.format(**self.error_details) + except (KeyError, ValueError): + # 如果填充失败，返回原始消息 + return str(self) + + def _get_solution(self) -> str: + """获取错误解决建议""" + if not self.error_key: + return "" + + solutions = ERROR_SOLUTIONS.get(self.locale, ERROR_SOLUTIONS["zh_CN"]) + return solutions.get(self.error_key, "") + + +class InvalidConfigError(TencentFaceRecognitionError): + """无效配置错误""" + pass + + +class ApiError(TencentFaceRecognitionError): + """API调用错误""" + pass + + +class ImageNotFoundError(TencentFaceRecognitionError): + """图片未找到错误""" + pass + + +class FaceNotDetectedError(TencentFaceRecognitionError): + """未检测到人脸错误""" + pass + + +class PersonNotFoundError(TencentFaceRecognitionError): + """人员未找到错误""" + pass + + +class PersonGroupNotFoundError(TencentFaceRecognitionError): + """人员库未找到错误""" + pass + + +class FaceNotFoundError(TencentFaceRecognitionError): + """人脸未找到错误""" + pass + + +class NetworkError(TencentFaceRecognitionError): + """网络错误""" + pass + + +class AuthenticationError(TencentFaceRecognitionError): + """认证错误""" + pass + + +class RateLimitError(TencentFaceRecognitionError): + """速率限制错误""" + pass + + +class QuotaExceededError(TencentFaceRecognitionError): + """配额超限错误""" + pass + + +def handle_api_error(error_code: str, error_message: str) -> None: + """处理API错误""" + error_mapping = { + "FailedOperation.ImageDecodeFailed": (ImageNotFoundError, "image_decode_failed"), + "InvalidParameterValue.NoFaceInPhoto": (FaceNotDetectedError, "face_not_detected"), + "FailedOperation.PersonNotFound": (PersonNotFoundError, "person_not_found"), + "FailedOperation.GroupNotFound": (PersonGroupNotFoundError, "group_not_found"), + "FailedOperation.FaceNotFound": (FaceNotFoundError, "face_not_found"), + "AuthFailure": (AuthenticationError, "auth_failure"), + "RequestLimitExceeded": (RateLimitError, "rate_limit_exceeded"), + "RequestQuotaExceeded": (QuotaExceededError, "quota_exceeded"), + "ResourceUnavailable.NetworkError": (NetworkError, "network_error"), + } + + error_info = error_mapping.get(error_code, (ApiError, "api_error")) + error_class = error_info[0] + error_key = error_info[1] + + _LOGGER.error("腾讯云API错误: %s - %s", error_code, error_message) + + # 创建带有本地化消息的错误 + error_details = { + "code": error_code, + "message": error_message + } + + raise error_class( + f"{error_code}: {error_message}", + error_key=error_key, + error_details=error_details + ) + + +def log_and_raise_error(error_class, message: str, details: Dict[str, Any] = None, error_key: Optional[str] = None) -> None: + """记录日志并抛出错误""" + if details: + _LOGGER.error("%s: %s, 详情: %s", error_class.__name__, message, details) + else: + _LOGGER.error("%s: %s", error_class.__name__, message) + + # 创建带有本地化消息的错误 + raise error_class( + message, + error_key=error_key, + error_details=details or {} + ) + + +def validate_config(config: Dict[str, Any]) -> None: + """验证配置""" + required_keys = ["secret_id", "secret_key"] + + for key in required_keys: + if not config.get(key): + log_and_raise_error( + InvalidConfigError, + f"缺少必需的配置项: {key}", + {"key": key}, + "missing_config" + ) + + # 验证Secret ID格式 + secret_id = config.get("secret_id") + if not isinstance(secret_id, str) or len(secret_id.strip()) == 0: + log_and_raise_error( + InvalidConfigError, + "Secret ID必须是非空字符串，且应以 AKID 开头", + {"secret_id": str(secret_id)}, + "invalid_secret_id" + ) + + # 验证Secret Key格式 + secret_key = config.get("secret_key") + if not isinstance(secret_key, str) or len(secret_key.strip()) == 0: + log_and_raise_error( + InvalidConfigError, + "Secret Key必须是非空字符串", + {"secret_key": str(secret_key)}, + "invalid_secret_key" + ) + + # 验证区域格式 + region = config.get("region", "ap-shanghai") + if not isinstance(region, str) or len(region.strip()) == 0: + log_and_raise_error( + InvalidConfigError, + "区域必须是非空字符串，且为有效的腾讯云区域", + {"region": str(region)}, + "invalid_region" + ) + + # 验证人员库ID格式 + person_group_id = config.get("person_group_id", "Hass") + if not isinstance(person_group_id, str) or len(person_group_id.strip()) == 0: + log_and_raise_error( + InvalidConfigError, + "人员库ID必须是非空字符串，只能包含字母、数字、连字符和下划线", + {"person_group_id": str(person_group_id)}, + "invalid_person_group_id" + ) + + _LOGGER.debug("配置验证通过") \ No newline at end of file diff --git a/face_recognition.py b/face_recognition.py new file mode 100644 index 0000000..b448f0c --- /dev/null +++ b/face_recognition.py @@ -0,0 +1,208 @@ +"""人脸识别核心功能""" + +import logging +import json +from typing import Dict, Any, List, Optional + +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import HomeAssistantError + +from .const import ( + ATTR_IMAGE_URL, + ATTR_IMAGE_PATH, + ATTR_MAX_FACE_NUM, + ATTR_MIN_FACE_SIZE, + ATTR_MAX_USER_NUM, + ATTR_QUALITY_CONTROL, + ATTR_NEED_ROTATE_CHECK, + ATTR_FACE_MATCH_THRESHOLD, + ERROR_IMAGE_NOT_FOUND, + ERROR_FACE_NOT_DETECTED, +) +from .tencent_cloud_client import TencentCloudClient + +_LOGGER = logging.getLogger(__name__) + + +class FaceRecognition: + """人脸识别核心功能""" + + def __init__(self, hass: HomeAssistant, client: TencentCloudClient): + """初始化人脸识别功能""" + self.hass = hass + self.client = client + + async def async_search_faces( + self, + image_url: str = None, + image_path: str = None, + image_file: str = None, + group_ids: 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 + ) -> List[Dict[str, Any]]: + """搜索人脸""" + try: + # 验证参数 + self._validate_image_params(image_url, image_path, image_file) + + # 调用腾讯云API搜索人脸 + results = await self.hass.async_add_executor_job( + self.client.search_faces, + image_url, + image_path, + image_file, + group_ids, + max_face_num, + min_face_size, + max_user_num, + quality_control, + need_rotate_check, + face_match_threshold + ) + + # 处理结果 + return self._process_search_results(results) + + except ValueError as ex: + _LOGGER.error("人脸搜索参数错误: %s", ex) + raise HomeAssistantError(str(ex)) + except Exception as ex: + _LOGGER.error("人脸搜索失败: %s", ex) + raise HomeAssistantError(f"人脸搜索失败: {str(ex)}") + + def _validate_image_params(self, image_url: str, image_path: str, image_file: str) -> None: + """验证图片参数""" + if not image_url and not image_path and not image_file: + raise ValueError("必须提供image_url、image_path或image_file") + + def _process_search_results(self, results: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + """处理人脸搜索结果""" + processed_results = [] + for result in results: + processed_result = { + "face_id": result.get("face_id"), + "candidates": [] + } + + for candidate in result.get("candidates", []): + processed_candidate = { + "person_id": candidate.get("person_id"), + "person_name": candidate.get("person_name"), + "score": candidate.get("score"), + "person_tag": candidate.get("person_tag") + } + processed_result["candidates"].append(processed_candidate) + + processed_results.append(processed_result) + + return processed_results + + def _process_detect_results(self, results: List[Dict[str, Any]], include_attributes: bool = False) -> List[Dict[str, Any]]: + """处理人脸检测结果""" + processed_results = [] + for result in results: + face_result = { + "face_id": result.get("face_id", ""), + "x": result.get("x", 0), + "y": result.get("y", 0), + "width": result.get("width", 0), + "height": result.get("height", 0), + "face_rect": result.get("face_rect") + } + + # 如果需要包含属性信息 + if include_attributes: + face_result.update({ + "gender": result.get("gender"), + "age": result.get("age"), + "expression": result.get("expression"), + "beauty": result.get("beauty") + }) + + processed_results.append(face_result) + + return processed_results + + def _handle_api_error(self, operation: str, ex: Exception) -> None: + """处理API错误""" + _LOGGER.error("%s失败: %s", operation, ex) + if isinstance(ex, ValueError): + raise HomeAssistantError(str(ex)) + else: + raise HomeAssistantError(f"{operation}失败: {str(ex)}") + + def _get_config_entry(self): + """获取配置项""" + from homeassistant.config_entries import ConfigEntry + from . import DOMAIN + + for entry in self.hass.config_entries.async_entries(DOMAIN): + return entry + return None + + async def async_get_face_attributes( + self, + image_url: str = None, + image_path: str = None, + image_file: str = None, + max_face_num: int = 1, + need_rotate_check: int = 1 + ) -> List[Dict[str, Any]]: + """获取人脸属性""" + try: + # 验证参数 + self._validate_image_params(image_url, image_path, image_file) + + # 调用腾讯云API获取人脸属性 + results = await self.hass.async_add_executor_job( + self.client.get_face_attributes, + image_url, + image_path, + image_file, + max_face_num, + need_rotate_check + ) + + return results + + except Exception as ex: + self._handle_api_error("获取人脸属性", ex) + + # 移除 _get_face_attributes_sync 方法，因为现在直接使用客户端的 get_face_attributes 方法 + + async def async_detect_faces( + self, + image_url: str = None, + image_path: str = None, + image_file: str = None, + max_face_num: int = 1, + min_face_size: int = 34, + need_rotate_check: int = 1 + ) -> List[Dict[str, Any]]: + """检测人脸""" + try: + # 验证参数 + self._validate_image_params(image_url, image_path, image_file) + + # 调用腾讯云API检测人脸 + results = await self.hass.async_add_executor_job( + self.client.detect_faces, + image_url, + image_path, + image_file, + max_face_num, + min_face_size, + need_rotate_check + ) + + return results + + except Exception as ex: + self._handle_api_error("人脸检测", ex) + + # 移除 _detect_faces_sync 方法，因为现在直接使用客户端的 detect_faces 方法 \ No newline at end of file diff --git a/manifest.json b/manifest.json new file mode 100644 index 0000000..c617f28 --- /dev/null +++ b/manifest.json @@ -0,0 +1,16 @@ +{ + "domain": "tencent_face_recognition", + "name": "腾讯云人脸识别", + "config_flow": true, + "documentation": "https://code.nextrt.com/Hass/tencent_face_recognition", + "issue_tracker": "https://code.nextrt.com/Hass/tencent_face_recognition/issues", + "requirements": [ + "tencentcloud-sdk-python-common", + "tencentcloud-sdk-python-iai" + ], + "dependencies": [], + "codeowners": ["@xyzmos"], + "version": "1.0.0", + "iot_class": "cloud_polling", + "resources": [] +} \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..037943d --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +tencentcloud-sdk-python-common>=3.0.0 +tencentcloud-sdk-python-iai>=3.0.0 \ No newline at end of file diff --git a/sensor.py b/sensor.py new file mode 100644 index 0000000..8ad013a --- /dev/null +++ b/sensor.py @@ -0,0 +1,206 @@ +"""传感器实体""" + +import logging +import base64 +from typing import Dict, Any, Optional +from datetime import datetime, timedelta + +from homeassistant.components.sensor import SensorEntity +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import CONF_NAME +from homeassistant.core import HomeAssistant, callback +from homeassistant.helpers.entity import EntityCategory +from homeassistant.helpers.entity_platform import AddEntitiesCallback +from homeassistant.helpers.event import async_track_time_interval + +from .const import DOMAIN + +_LOGGER = logging.getLogger(__name__) + + +async def async_setup_entry( + hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback +) -> None: + """设置传感器实体""" + _LOGGER.debug("设置腾讯云人脸识别传感器实体") + + # 获取配置项 + name = entry.data.get(CONF_NAME, "腾讯云人脸识别") + + # 创建传感器实体 + async_add_entities( + [ + TencentFaceRecognitionStatusSensor(hass, entry, name), + ] + ) + + # 获取人员列表并创建实体 + client = hass.data[DOMAIN][entry.entry_id]["client"] + person_group_id = entry.data.get("person_group_id", "Hass") + + try: + persons = await hass.async_add_executor_job( + client.get_person_list, person_group_id + ) + + for person in persons: + async_add_entities( + [ + TencentFaceRecognitionPersonSensor(hass, entry, person), + ] + ) + except Exception as ex: + _LOGGER.error("获取人员列表失败: %s", ex) + + +class TencentFaceRecognitionPersonSensor(SensorEntity): + """腾讯云人脸识别人员传感器""" + + _attr_icon = "mdi:account" + + def __init__(self, hass: HomeAssistant, entry: ConfigEntry, person: Dict[str, Any]): + """初始化传感器""" + self.hass = hass + self._entry = entry + self._person = person + self._attr_name = person.get("person_name", "未知人员") + self._attr_unique_id = f"{entry.entry_id}_{person.get('person_id')}" + self._state = "未知" + self._extra_state_attributes = { + "person_id": person.get("person_id"), + "gender": "男" if person.get("gender") == 1 else "女", + "face_ids": person.get("face_ids"), + } + + @property + def state(self) -> str: + """返回传感器状态""" + return self._state + + @property + def extra_state_attributes(self) -> Dict[str, Any]: + """返回传感器属性""" + return self._extra_state_attributes + + +class TencentFaceRecognitionStatusSensor(SensorEntity): + """腾讯云人脸识别状态传感器""" + + _attr_icon = "mdi:face-recognition" + _attr_entity_category = EntityCategory.DIAGNOSTIC + + def __init__(self, hass: HomeAssistant, entry: ConfigEntry, name: str): + """初始化传感器""" + self.hass = hass + self._entry = entry + self._name = name + self._attr_name = f"{name} 状态" + self._attr_unique_id = f"{entry.entry_id}_status" + self._state = "未连接" + self._extra_state_attributes = { + "last_updated": None, + "api_status": "未知", + "client_status": "未知", + "last_test_time": None, + "error_count": 0, + "success_count": 0, + "last_error": None + } + self._client = None + self._last_successful_test = None + + @property + def state(self) -> str: + """返回传感器状态""" + return self._state + + @property + def extra_state_attributes(self) -> Dict[str, Any]: + """返回传感器属性""" + return self._extra_state_attributes + + async def async_added_to_hass(self) -> None: + """当实体添加到Home Assistant时调用""" + # 设置定期更新 + async_track_time_interval( + self.hass, + self._async_update, + timedelta(minutes=5) # 每5分钟更新一次 + ) + + # 立即执行一次更新 + await self._async_update(None) + + async def _async_update(self, _) -> None: + """更新传感器状态""" + await self.async_update() + + async def async_update(self) -> None: + """更新传感器状态""" + try: + # 获取插件实例 + entry_data = self.hass.data[DOMAIN].get(self._entry.entry_id) + if not entry_data: + self._state = "未初始化" + self._extra_state_attributes["api_status"] = "插件未正确初始化" + self._extra_state_attributes["client_status"] = "未初始化" + return + + # 获取客户端实例 + client = entry_data.get("client") + face_recognition = entry_data.get("face_recognition") + + if not client or not face_recognition: + self._state = "未初始化" + self._extra_state_attributes["api_status"] = "客户端未正确初始化" + self._extra_state_attributes["client_status"] = "未初始化" + return + + # 更新客户端状态 + self._client = client + self._extra_state_attributes["client_status"] = "已初始化" + + # 使用内置的测试图片进行连接测试 + await self._test_api_connection(face_recognition) + + except Exception as ex: + _LOGGER.error("更新腾讯云人脸识别状态传感器失败: %s", ex, exc_info=True) + self._state = "连接错误" + self._extra_state_attributes["api_status"] = f"错误: {str(ex)}" + self._extra_state_attributes["last_error"] = str(ex) + self._extra_state_attributes["error_count"] = self._extra_state_attributes.get("error_count", 0) + 1 + + # 更新最后更新时间 + self._extra_state_attributes["last_updated"] = datetime.now().isoformat() + + async def _test_api_connection(self, face_recognition) -> None: + """测试API连接""" + try: + group_id = self._entry.data.get("person_group_id", "Hass") + + # 使用 GetGroupInfo 测试连接 + group_info = await self.hass.async_add_executor_job( + self._client.get_group_info, + group_id + ) + + if group_info: + self._state = "已连接" + self._extra_state_attributes["api_status"] = "正常" + self._extra_state_attributes["success_count"] = self._extra_state_attributes.get("success_count", 0) + 1 + self._last_successful_test = datetime.now() + self._extra_state_attributes["last_test_time"] = self._last_successful_test.isoformat() + + if "last_error" in self._extra_state_attributes: + self._extra_state_attributes["last_error"] = None + else: + self._state = "连接错误" + self._extra_state_attributes["api_status"] = "获取人员库信息失败" + + except Exception as ex: + error_msg = str(ex) + self._state = "连接错误" + self._extra_state_attributes["api_status"] = f"连接失败: {error_msg}" + self._extra_state_attributes["last_error"] = error_msg + self._extra_state_attributes["error_count"] = self._extra_state_attributes.get("error_count", 0) + 1 + _LOGGER.debug("API连接测试失败: %s", error_msg) \ No newline at end of file diff --git a/services.py b/services.py new file mode 100644 index 0000000..c8cd9e4 --- /dev/null +++ b/services.py @@ -0,0 +1,258 @@ +"""服务定义""" + +import logging +from typing import Dict, Any, List, Optional + +import voluptuous as vol + +from homeassistant.core import HomeAssistant, ServiceCall +from homeassistant.helpers import config_validation as cv + +from .const import ( + DOMAIN, + SERVICE_FACE_SEARCH, + SERVICE_DETECT_FACE, + SERVICE_GET_FACE_ATTRIBUTES, + ATTR_IMAGE_URL, + ATTR_IMAGE_FILE, + ATTR_IMAGE_PATH, + ATTR_MAX_FACE_NUM, + ATTR_MIN_FACE_SIZE, + ATTR_MAX_USER_NUM, + ATTR_QUALITY_CONTROL, + ATTR_NEED_ROTATE_CHECK, + ATTR_FACE_MATCH_THRESHOLD, +) + +_LOGGER = logging.getLogger(__name__) + +# 人脸搜索服务参数 +FACE_SEARCH_SCHEMA = vol.Schema({ + vol.Required('group_id'): cv.string, + vol.Optional(ATTR_IMAGE_URL): cv.string, + vol.Optional(ATTR_IMAGE_PATH): cv.string, + vol.Optional(ATTR_IMAGE_FILE): cv.string, + vol.Optional(ATTR_MAX_FACE_NUM, default=1): cv.positive_int, + vol.Optional(ATTR_MIN_FACE_SIZE, default=34): cv.positive_int, + vol.Optional(ATTR_MAX_USER_NUM, default=5): cv.positive_int, + vol.Optional(ATTR_QUALITY_CONTROL, default=1): vol.In([0, 1]), + vol.Optional(ATTR_NEED_ROTATE_CHECK, default=1): vol.In([0, 1]), + vol.Optional(ATTR_FACE_MATCH_THRESHOLD, default=60.0): vol.Range(min=0, max=100), +}) + +# 人脸检测服务参数 +DETECT_FACE_SCHEMA = vol.Schema({ + vol.Optional(ATTR_IMAGE_URL): cv.string, + vol.Optional(ATTR_IMAGE_PATH): cv.string, + vol.Optional(ATTR_IMAGE_FILE): cv.string, + vol.Optional(ATTR_MAX_FACE_NUM, default=1): cv.positive_int, + vol.Optional(ATTR_MIN_FACE_SIZE, default=34): cv.positive_int, + vol.Optional(ATTR_NEED_ROTATE_CHECK, default=1): vol.In([0, 1]), +}) + +# 获取人脸属性服务参数 +GET_FACE_ATTRIBUTES_SCHEMA = vol.Schema({ + vol.Optional(ATTR_IMAGE_URL): cv.string, + vol.Optional(ATTR_IMAGE_PATH): cv.string, + vol.Optional(ATTR_IMAGE_FILE): cv.string, + vol.Optional(ATTR_MAX_FACE_NUM, default=1): cv.positive_int, + vol.Optional(ATTR_NEED_ROTATE_CHECK, default=1): vol.In([0, 1]), +}) + + +async def async_setup_services(hass: HomeAssistant) -> None: + """设置服务""" + _LOGGER.info("设置腾讯云人脸识别服务") + + # 注册人脸搜索服务 + hass.services.async_register( + DOMAIN, + SERVICE_FACE_SEARCH, + async_face_search_service, + schema=FACE_SEARCH_SCHEMA, + supports_response=True + ) + + # 注册人脸检测服务 + hass.services.async_register( + DOMAIN, + SERVICE_DETECT_FACE, + async_detect_face_service, + schema=DETECT_FACE_SCHEMA + ) + + # 注册获取人脸属性服务 + hass.services.async_register( + DOMAIN, + SERVICE_GET_FACE_ATTRIBUTES, + async_get_face_attributes_service, + schema=GET_FACE_ATTRIBUTES_SCHEMA + ) + + +async def async_unload_services(hass: HomeAssistant) -> None: + """卸载服务""" + _LOGGER.info("卸载腾讯云人脸识别服务") + + # 注销所有服务 + hass.services.async_remove(DOMAIN, SERVICE_FACE_SEARCH) + hass.services.async_remove(DOMAIN, SERVICE_DETECT_FACE) + hass.services.async_remove(DOMAIN, SERVICE_GET_FACE_ATTRIBUTES) + + +async def async_face_search_service(call: ServiceCall) -> Dict[str, Any]: + """人脸搜索服务""" + from homeassistant.exceptions import HomeAssistantError + + hass = call.hass + data = call.data + + # 获取配置项 + config_entry = _get_config_entry(hass) + if not config_entry: + raise HomeAssistantError("未找到腾讯云人脸识别配置") + + # 获取插件实例 + entry_data = _get_entry_data(hass) + if not entry_data: + raise HomeAssistantError("腾讯云人脸识别插件未正确初始化") + + face_recognition = entry_data.get("face_recognition") + if not face_recognition: + raise HomeAssistantError("腾讯云人脸识别插件未正确初始化") + + # 调用人脸搜索功能 + try: + group_id = data.get('group_id') + image_url = data.get(ATTR_IMAGE_URL) + image_path = data.get(ATTR_IMAGE_PATH) + image_file = data.get(ATTR_IMAGE_FILE) + max_face_num = data.get(ATTR_MAX_FACE_NUM, 1) + min_face_size = data.get(ATTR_MIN_FACE_SIZE, 34) + max_user_num = data.get(ATTR_MAX_USER_NUM, 5) + quality_control = data.get(ATTR_QUALITY_CONTROL, 1) + need_rotate_check = data.get(ATTR_NEED_ROTATE_CHECK, 1) + face_match_threshold = data.get(ATTR_FACE_MATCH_THRESHOLD, 60.0) + + results = await face_recognition.async_search_faces( + image_url=image_url, + image_path=image_path, + image_file=image_file, + group_ids=[group_id], + 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 + ) + + return {"results": results} + + except Exception as ex: + _LOGGER.error("人脸搜索服务调用失败: %s", ex) + raise HomeAssistantError(f"人脸搜索服务调用失败: {str(ex)}") + + +async def async_detect_face_service(call: ServiceCall) -> Dict[str, Any]: + """人脸检测服务""" + from homeassistant.exceptions import HomeAssistantError + + hass = call.hass + data = call.data + + # 获取配置项 + config_entry = _get_config_entry(hass) + if not config_entry: + raise HomeAssistantError("未找到腾讯云人脸识别配置") + + # 获取插件实例 + entry_data = _get_entry_data(hass) + if not entry_data: + raise HomeAssistantError("腾讯云人脸识别插件未正确初始化") + + face_recognition = entry_data.get("face_recognition") + if not face_recognition: + raise HomeAssistantError("腾讯云人脸识别插件未正确初始化") + + # 调用人脸检测功能 + try: + image_url = data.get(ATTR_IMAGE_URL) + image_path = data.get(ATTR_IMAGE_PATH) + image_file = data.get(ATTR_IMAGE_FILE) + max_face_num = data.get(ATTR_MAX_FACE_NUM, 1) + min_face_size = data.get(ATTR_MIN_FACE_SIZE, 34) + need_rotate_check = data.get(ATTR_NEED_ROTATE_CHECK, 1) + + results = await face_recognition.async_detect_faces( + image_url=image_url, + image_path=image_path, + image_file=image_file, + max_face_num=max_face_num, + min_face_size=min_face_size, + need_rotate_check=need_rotate_check + ) + + return {"results": results} + + except Exception as ex: + _LOGGER.error("人脸检测服务调用失败: %s", ex) + raise HomeAssistantError(f"人脸检测服务调用失败: {str(ex)}") + + +async def async_get_face_attributes_service(call: ServiceCall) -> Dict[str, Any]: + """获取人脸属性服务""" + from homeassistant.exceptions import HomeAssistantError + + hass = call.hass + data = call.data + + # 获取配置项 + config_entry = _get_config_entry(hass) + if not config_entry: + raise HomeAssistantError("未找到腾讯云人脸识别配置") + + # 获取插件实例 + entry_data = _get_entry_data(hass) + if not entry_data: + raise HomeAssistantError("腾讯云人脸识别插件未正确初始化") + + face_recognition = entry_data.get("face_recognition") + if not face_recognition: + raise HomeAssistantError("腾讯云人脸识别插件未正确初始化") + + # 调用获取人脸属性功能 + try: + image_url = data.get(ATTR_IMAGE_URL) + image_path = data.get(ATTR_IMAGE_PATH) + image_file = data.get(ATTR_IMAGE_FILE) + max_face_num = data.get(ATTR_MAX_FACE_NUM, 1) + need_rotate_check = data.get(ATTR_NEED_ROTATE_CHECK, 1) + + results = await face_recognition.async_get_face_attributes( + image_url=image_url, + image_path=image_path, + image_file=image_file, + max_face_num=max_face_num, + need_rotate_check=need_rotate_check + ) + + return {"results": results} + + except Exception as ex: + _LOGGER.error("获取人脸属性服务调用失败: %s", ex) + raise HomeAssistantError(f"获取人脸属性服务调用失败: {str(ex)}") + + +def _get_config_entry(hass: HomeAssistant): + """获取配置项""" + for entry in hass.config_entries.async_entries(DOMAIN): + return entry + return None + +def _get_entry_data(hass: HomeAssistant): + """获取配置项数据""" + for entry_id, data in hass.data.get(DOMAIN, {}).items(): + if isinstance(data, dict) and "face_recognition" in data: + return data + return None \ No newline at end of file diff --git a/services.yaml b/services.yaml new file mode 100644 index 0000000..8974f1f --- /dev/null +++ b/services.yaml @@ -0,0 +1,185 @@ +# 腾讯云人脸识别服务定义 +face_search: + name: 人脸搜索 + description: 在指定的人员库中搜索人脸。 + fields: + group_id: + name: 人员库ID + description: 人员库 ID。 + required: true + example: "Hass" + selector: + text: + image_url: + name: 图片URL + description: 要搜索的人脸图片的 URL。 + example: "https://example.com/image.jpg" + selector: + text: + image_path: + name: 图片路径 + description: Home Assistant 本地可访问的人脸图片路径。 + example: "/config/www/image.jpg" + selector: + text: + image_file: + name: 图片文件 + description: 通过文件上传的人脸图片。 + selector: + text: + max_face_num: + name: 最大人脸数量 + description: 最多处理的人脸数量。 + default: 1 + example: 1 + selector: + number: + min: 1 + max: 10 + step: 1 + mode: box + min_face_size: + name: 最小人脸尺寸 + description: 最小人脸尺寸（像素）。 + default: 34 + example: 34 + selector: + number: + min: 1 + max: 200 + step: 1 + mode: box + max_user_num: + name: 最大返回人员数量 + description: 最多返回的匹配人员数量。 + default: 5 + example: 5 + selector: + number: + min: 1 + max: 20 + step: 1 + mode: box + quality_control: + name: 质量控制 + description: 是否进行质量控制。 + default: true + example: true + selector: + boolean: + need_rotate_check: + name: 旋转检查 + description: 是否进行旋转检查。 + default: true + example: true + selector: + boolean: + face_match_threshold: + name: 人脸匹配阈值 + description: 人脸匹配阈值（0-100）。 + default: 60 + example: 60 + selector: + number: + min: 0 + max: 100 + step: 1 + mode: slider + response: + description: "人脸搜索结果" + fields: + results: + name: "结果" + description: "包含人脸ID和候选人列表的数组" + example: '[{"face_id": "3440118425101275818", "candidates": [{"person_id": "person_1", "person_name": "张三", "score": 99.9, "person_tag": "员工"}]}]' + +detect_face: + name: 人脸检测 + description: 检测图片中的人脸。 + fields: + image_url: + name: 图片URL + description: 要检测的图片URL。 + example: "https://example.com/image.jpg" + selector: + text: + image_path: + name: 图片路径 + description: 要检测的本地图片路径。 + example: "/config/www/image.jpg" + selector: + text: + image_file: + name: 图片文件 + description: 要检测的图片文件。 + selector: + text: + max_face_num: + name: 最大人脸数量 + description: 最多检测的人脸数量。 + default: 1 + example: 1 + selector: + number: + min: 1 + max: 10 + step: 1 + mode: box + min_face_size: + name: 最小人脸尺寸 + description: 最小人脸尺寸（像素）。 + default: 34 + example: 34 + selector: + number: + min: 1 + max: 200 + step: 1 + mode: box + need_rotate_check: + name: 旋转检查 + description: 是否进行旋转检查。 + default: true + example: true + selector: + boolean: + +get_face_attributes: + name: 获取人脸属性 + description: 获取图片中人脸的属性信息。 + fields: + image_url: + name: 图片URL + description: 要分析的图片URL。 + example: "https://example.com/image.jpg" + selector: + text: + image_path: + name: 图片路径 + description: 要分析的本地图片路径。 + example: "/config/www/image.jpg" + selector: + text: + image_file: + name: 图片文件 + description: 要分析的图片文件。 + selector: + text: + max_face_num: + name: 最大人脸数量 + description: 最多分析的人脸数量。 + default: 1 + example: 1 + selector: + number: + min: 1 + max: 10 + step: 1 + mode: box + need_rotate_check: + name: 旋转检查 + description: 是否进行旋转检查。 + default: true + example: true + selector: + boolean: \ No newline at end of file diff --git a/strings.json b/strings.json new file mode 100644 index 0000000..766cd40 --- /dev/null +++ b/strings.json @@ -0,0 +1,144 @@ +{ + "config": { + "step": { + "user": { + "title": "腾讯云人脸识别", + "description": "请输入您的腾讯云人脸识别服务凭据", + "data": { + "secret_id": "Secret ID", + "secret_key": "Secret Key", + "region": "区域", + "person_group_id": "人员组ID", + "name": "名称" + }, + "data_description": { + "secret_id": "腾讯云API的Secret ID", + "secret_key": "腾讯云API的Secret Key", + "region": "腾讯云服务区域，例如：ap-shanghai", + "person_group_id": "人脸库人员组ID，用于存储和搜索人脸", + "name": "集成名称" + } + } + }, + "error": { + "invalid_config": "无效的配置" + }, + "abort": { + "already_configured": "已经配置了腾讯云人脸识别服务" + } + }, + "options": { + "step": { + "init": { + "title": "腾讯云人脸识别选项", + "description": "配置腾讯云人脸识别服务选项", + "data": { + "region": "区域" + }, + "data_description": { + "region": "腾讯云服务区域，例如：ap-shanghai" + } + } + } + }, + "services": { + "face_search": { + "name": "人脸搜索", + "description": "搜索人脸", + "fields": { + "image_url": { + "name": "图片URL", + "description": "要搜索的图片URL" + }, + "image_path": { + "name": "图片路径", + "description": "要搜索的本地图片路径" + }, + "image_file": { + "name": "图片文件", + "description": "要搜索的图片文件（Base64编码）" + }, + "max_face_num": { + "name": "最大人脸数量", + "description": "最多处理的人脸数量" + }, + "min_face_size": { + "name": "最小人脸尺寸", + "description": "最小人脸尺寸（像素）" + }, + "max_user_num": { + "name": "最大返回人员数量", + "description": "最多返回的匹配人员数量" + }, + "quality_control": { + "name": "质量控制", + "description": "是否进行质量控制" + }, + "need_rotate_check": { + "name": "旋转检查", + "description": "是否进行旋转检查" + }, + "face_match_threshold": { + "name": "人脸匹配阈值", + "description": "人脸匹配阈值（0-100）" + } + } + }, + "detect_face": { + "name": "人脸检测", + "description": "检测图片中的人脸", + "fields": { + "image_url": { + "name": "图片URL", + "description": "要检测的图片URL" + }, + "image_path": { + "name": "图片路径", + "description": "要检测的本地图片路径" + }, + "image_file": { + "name": "图片文件", + "description": "要检测的图片文件（Base64编码）" + }, + "max_face_num": { + "name": "最大人脸数量", + "description": "最多检测的人脸数量" + }, + "min_face_size": { + "name": "最小人脸尺寸", + "description": "最小人脸尺寸（像素）" + }, + "need_rotate_check": { + "name": "旋转检查", + "description": "是否进行旋转检查" + } + } + }, + "get_face_attributes": { + "name": "获取人脸属性", + "description": "获取图片中人脸的属性信息", + "fields": { + "image_url": { + "name": "图片URL", + "description": "要分析的图片URL" + }, + "image_path": { + "name": "图片路径", + "description": "要分析的本地图片路径" + }, + "image_file": { + "name": "图片文件", + "description": "要分析的图片文件（Base64编码）" + }, + "max_face_num": { + "name": "最大人脸数量", + "description": "最多分析的人脸数量" + }, + "need_rotate_check": { + "name": "旋转检查", + "description": "是否进行旋转检查" + } + } + } + } +} \ No newline at end of file diff --git a/tencent_cloud_client.py b/tencent_cloud_client.py new file mode 100644 index 0000000..2e34cdf --- /dev/null +++ b/tencent_cloud_client.py @@ -0,0 +1,1016 @@ +"""腾讯云客户端""" + +import base64 +import io +import json +import logging +import os +import random +import re +import requests +import time +import uuid +from typing import Dict, Any, List, Optional, Union, Callable + +from tencentcloud.common import credential +from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException +from tencentcloud.iai.v20200303 import iai_client, models + +from .const import ( + CONF_SECRET_ID, + CONF_SECRET_KEY, + CONF_REGION, +) +from .errors import ( + handle_api_error, + ImageNotFoundError, + FaceNotDetectedError, + log_and_raise_error, + validate_config, + NetworkError, + AuthenticationError, + RateLimitError, +) + +_LOGGER = logging.getLogger(__name__) + +# 日志记录配置 +ENABLE_REQUEST_LOGGING = True # 是否启用请求日志记录 +ENABLE_RESPONSE_LOGGING = True # 是否启用响应日志记录 +ENABLE_PERFORMANCE_LOGGING = True # 是否启用性能日志记录 +LOG_REQUEST_PAYLOAD_MAX_SIZE = 1024 # 请求负载日志记录的最大大小（字节） +LOG_RESPONSE_PAYLOAD_MAX_SIZE = 1024 # 响应负载日志记录的最大大小（字节） + +# API调用重试配置 +MAX_RETRIES = 1 +INITIAL_RETRY_DELAY = 1 # 秒 +MAX_RETRY_DELAY = 30 # 秒 +RETRY_BACKOFF_FACTOR = 2 # 指数退避因子 +RETRY_JITTER = 0.1 # 抖动因子，避免多个客户端同时重试 + +# 错误码分类 +NON_RETRYABLE_ERRORS = { + "AuthFailure", # 认证失败 + "InvalidParameter", # 无效参数 + "ResourceNotFound", # 资源未找到 + "LimitExceeded", # 配额超限 + "RequestLimitExceeded", # 请求频率超限 + "RequestQuotaExceeded", # 请求配额超限 +} + +RATE_LIMIT_ERRORS = { + "RequestLimitExceeded", + "RequestQuotaExceeded", + "LimitExceeded", +} + +NETWORK_ERRORS = { + "ResourceUnavailable.NetworkError", + "ResourceUnavailable.ServiceTimeout", +} + +# 图片处理配置 +MAX_IMAGE_SIZE = 10 * 1024 * 1024 # 10MB +MAX_IMAGE_DIMENSION = 4096 # 最大图片尺寸 +SUPPORTED_IMAGE_FORMATS = [".jpg", ".jpeg", ".png", ".bmp", ".gif"] +IMAGE_COMPRESSION_QUALITY = 85 # 图片压缩质量 +MIN_FACE_SIZE = 34 # 最小人脸尺寸 + + +class RetryConfig: + """重试配置类""" + + def __init__( + self, + max_retries: int = MAX_RETRIES, + initial_delay: float = INITIAL_RETRY_DELAY, + max_delay: float = MAX_RETRY_DELAY, + backoff_factor: float = RETRY_BACKOFF_FACTOR, + jitter: float = RETRY_JITTER + ): + self.max_retries = max_retries + self.initial_delay = initial_delay + self.max_delay = max_delay + self.backoff_factor = backoff_factor + self.jitter = jitter + + +class TencentCloudClient: + """腾讯云客户端""" + + def __init__( + self, + secret_id: str, + secret_key: str, + region: str = "ap-shanghai", + retry_config: Optional[RetryConfig] = None + ): + """初始化腾讯云客户端""" + # 验证配置 + config = { + "secret_id": secret_id, + "secret_key": secret_key, + "region": region + } + validate_config(config) + + 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() + + 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", ex, exc_info=True) + log_and_raise_error( + AuthenticationError, + f"创建腾讯云客户端失败: {str(ex)}", + {"region": self.region, "error_type": type(ex).__name__} + ) + + def _handle_error(self, ex: TencentCloudSDKException) -> None: + """处理API错误""" + _LOGGER.error( + "腾讯云API调用失败: 错误码=%s, 错误消息=%s, 请求ID=%s", + ex.code, ex.message, getattr(ex, 'request_id', '未知') + ) + handle_api_error(ex.code, ex.message) + + def _get_image_base64(self, image_url: str = None, image_path: str = None, image_file: str = None) -> str: + """获取图片的base64编码""" + _LOGGER.debug( + "获取图片 base64: image_url=%s, image_path=%s, image_file is not None: %s", + image_url, image_path, image_file is not None + ) + + # 验证参数 + if not any([image_url, image_path, image_file]): + log_and_raise_error( + ValueError, + "必须提供image_url、image_path或image_file" + ) + + try: + if image_url: + # 如果是URL，下载图片并转换为base64 + _LOGGER.debug("使用图片URL: %s", image_url) + return self._download_image_as_base64(image_url) + elif image_path: + # 如果是本地文件路径，读取文件并转换为base64 + _LOGGER.debug("使用本地图片路径: %s", image_path) + return self._read_local_image_as_base64(image_path) + elif image_file: + # 如果是上传的文件，直接转换为base64 + _LOGGER.debug("使用上传的图片文件") + return self._process_uploaded_image_as_base64(image_file) + except Exception as ex: + _LOGGER.error("处理图片失败: %s", ex, exc_info=True) + log_and_raise_error( + ImageNotFoundError, + f"处理图片失败: {str(ex)}", + {"image_url": image_url, "image_path": image_path, "error_type": type(ex).__name__} + ) + + def _download_image_as_base64(self, image_url: str) -> str: + """下载图片并转换为base64编码""" + try: + _LOGGER.debug("开始下载图片: %s", image_url) + + # 验证URL格式 + if not self._is_valid_url(image_url): + log_and_raise_error( + ValueError, + f"无效的图片URL: {image_url}", + {"url": image_url} + ) + + # 设置请求超时和重试 + session = requests.Session() + session.mount('http://', requests.adapters.HTTPAdapter(max_retries=2)) + session.mount('https://', requests.adapters.HTTPAdapter(max_retries=2)) + + # 设置请求头，模拟浏览器请求 + headers = { + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36" + } + + # 下载图片 + response = session.get(image_url, timeout=10, headers=headers) + response.raise_for_status() + + # 检查内容类型是否为图片 + content_type = response.headers.get('content-type', '').lower() + if not content_type.startswith('image/'): + log_and_raise_error( + ValueError, + f"URL不是有效的图片: {image_url}, 内容类型: {content_type}", + {"url": image_url, "content_type": content_type} + ) + + # 处理图片数据 + image_data = self._process_image_data(response.content) + encoded_image = base64.b64encode(image_data).decode("utf-8") + + _LOGGER.debug("成功下载并编码图片，大小: %d 字节", len(encoded_image)) + return encoded_image + + except requests.exceptions.Timeout: + _LOGGER.error("下载图片超时: %s", image_url) + log_and_raise_error( + NetworkError, + f"下载图片超时: {image_url}", + {"url": image_url, "error": "timeout"} + ) + except requests.exceptions.ConnectionError: + _LOGGER.error("下载图片连接错误: %s", image_url) + log_and_raise_error( + NetworkError, + f"下载图片连接错误: {image_url}", + {"url": image_url, "error": "connection_error"} + ) + except requests.exceptions.RequestException as ex: + _LOGGER.error("下载图片失败: %s, 错误: %s", image_url, ex) + log_and_raise_error( + NetworkError, + f"下载图片失败: {image_url}", + {"url": image_url, "error": str(ex)} + ) + + def _read_local_image_as_base64(self, image_path: str) -> str: + """读取本地图片并转换为base64编码""" + try: + # 检查文件是否存在 + if not os.path.exists(image_path): + _LOGGER.error("图片文件不存在: %s", image_path) + log_and_raise_error( + ImageNotFoundError, + f"图片文件不存在: {image_path}", + {"path": image_path} + ) + + # 检查是否为文件 + if not os.path.isfile(image_path): + _LOGGER.error("路径不是文件: %s", image_path) + log_and_raise_error( + ImageNotFoundError, + f"路径不是文件: {image_path}", + {"path": image_path} + ) + + # 检查文件扩展名 + file_ext = os.path.splitext(image_path)[1].lower() + if file_ext not in SUPPORTED_IMAGE_FORMATS: + _LOGGER.error("不支持的图片格式: %s", file_ext) + log_and_raise_error( + ImageNotFoundError, + f"不支持的图片格式: {file_ext}", + {"path": image_path, "supported_formats": SUPPORTED_IMAGE_FORMATS} + ) + + # 检查文件大小 + file_size = os.path.getsize(image_path) + if file_size > MAX_IMAGE_SIZE: + _LOGGER.error("图片文件过大: %s, 大小: %d 字节", image_path, file_size) + log_and_raise_error( + ImageNotFoundError, + f"图片文件过大: {image_path}", + {"path": image_path, "size": file_size, "max_size": f"{MAX_IMAGE_SIZE // (1024*1024)}MB"} + ) + + # 读取文件内容 + with open(image_path, "rb") as img_file: + image_data = img_file.read() + + # 处理图片数据 + processed_image = self._process_image_data(image_data) + encoded_image = base64.b64encode(processed_image).decode("utf-8") + + _LOGGER.debug("成功读取图片文件，大小: %d 字节", len(encoded_image)) + return encoded_image + + except PermissionError: + _LOGGER.error("没有权限读取图片文件: %s", image_path) + log_and_raise_error( + ImageNotFoundError, + f"没有权限读取图片文件: {image_path}", + {"path": image_path, "error": "permission_denied"} + ) + except Exception as ex: + _LOGGER.error("读取图片文件失败: %s, 错误: %s", image_path, ex) + log_and_raise_error( + ImageNotFoundError, + f"读取图片文件失败: {image_path}", + {"path": image_path, "error": str(ex)} + ) + + def _process_uploaded_image_as_base64(self, image_file: str) -> str: + """处理上传的图片文件并转换为base64编码""" + try: + # image_file 应该是 base64 编码的字符串或文件对象 + if isinstance(image_file, str): + # 检查是否已经是 base64 编码 + if image_file.startswith('data:image'): + # 处理 data URL 格式 + _LOGGER.debug("处理 data URL 格式图片") + try: + header, encoded = image_file.split(',', 1) + _LOGGER.debug("成功解析 data URL 格式图片，头部: %s", header) + + # 解码base64并重新处理 + image_data = base64.b64decode(encoded) + processed_image = self._process_image_data(image_data) + return base64.b64encode(processed_image).decode("utf-8") + except ValueError: + _LOGGER.error("无效的 data URL 格式") + log_and_raise_error( + ImageNotFoundError, + "无效的 data URL 格式", + {"data_url_length": len(image_file)} + ) + else: + # 假设已经是 base64 编码 + _LOGGER.debug("使用 base64 编码图片，长度: %d", len(image_file)) + + # 解码base64并重新处理 + try: + image_data = base64.b64decode(image_file) + processed_image = self._process_image_data(image_data) + return base64.b64encode(processed_image).decode("utf-8") + except Exception as ex: + _LOGGER.error("base64解码失败: %s", ex) + log_and_raise_error( + ImageNotFoundError, + "base64解码失败", + {"error": str(ex), "data_length": len(image_file)} + ) + else: + # 如果是文件对象，读取并转换为 base64 + _LOGGER.debug("处理文件对象") + image_data = image_file.read() + + # 处理图片数据 + processed_image = self._process_image_data(image_data) + encoded_image = base64.b64encode(processed_image).decode("utf-8") + + _LOGGER.debug("成功读取上传的图片文件，大小: %d 字节", len(encoded_image)) + return encoded_image + + except Exception as ex: + _LOGGER.error("处理上传的图片文件失败: %s", ex) + log_and_raise_error( + ImageNotFoundError, + f"处理上传的图片文件失败", + {"error": str(ex), "error_type": type(ex).__name__} + ) + + def search_faces( + self, + image_url: str = None, + image_path: str = None, + image_file: str = None, + group_ids: 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 + ) -> List[Dict[str, Any]]: + """搜索人脸""" + _LOGGER.info( + "开始搜索人脸: image_url=%s, image_path=%s, image_file is not None: %s, max_face_num=%d, min_face_size=%d, max_user_num=%d, quality_control=%d, need_rotate_check=%d, face_match_threshold=%.2f", + image_url, image_path, image_file is not None, max_face_num, min_face_size, max_user_num, quality_control, need_rotate_check, face_match_threshold + ) + + # 验证参数 + if not group_ids: + _LOGGER.warning("未提供人员库ID，搜索可能失败") + + # 定义API调用函数 + def api_call(): + # 生成请求ID + request_id = str(uuid.uuid4()) + start_time = time.time() + + try: + # 获取图片数据 + image = self._get_image_base64(image_url, image_path, image_file) + + # 创建请求 + req = models.SearchFacesRequest() + params = { + "NeedPersonInfo": 1, + "Image": image, + "MaxFaceNum": max_face_num, + "MinFaceSize": min_face_size, + "MaxPersonNum": max_user_num, + "QualityControl": quality_control, + "NeedRotateCheck": need_rotate_check, + "FaceMatchThreshold": face_match_threshold, + "GroupIds": group_ids + } + + # 记录请求日志 + self._log_request("人脸搜索", params) + + req.from_json_string(json.dumps(params)) + + # 发送请求 + _LOGGER.debug("发送人脸搜索请求") + resp = self.client.SearchFaces(req) + + # 计算请求耗时 + duration = time.time() - start_time + + # 记录响应日志 + self._log_response("人脸搜索", request_id, resp, duration) + + # 解析响应 + results = [] + if getattr(resp, "Results", []): + _LOGGER.info("检测到 %d 个人脸", len(resp.Results)) + for result in resp.Results: + face_result = { + "face_id": getattr(result, "FaceId", None), + "candidates": [] + } + + if result.Candidates: + face_id = getattr(result, "FaceId", "未知") + _LOGGER.debug("人脸 %s 有 %d 个候选匹配", face_id, len(result.Candidates)) + for candidate in result.Candidates: + candidate_info = { + "person_id": candidate.PersonId, + "person_name": candidate.PersonName, + "score": candidate.Score, + "person_tag": getattr(candidate, "PersonTag", None) + } + face_result["candidates"].append(candidate_info) + _LOGGER.debug( + "候选匹配: person_id=%s, person_name=%s, score=%.2f, person_tag=%s", + candidate.PersonId, candidate.PersonName, candidate.Score, getattr(candidate, "PersonTag", None) + ) + + results.append(face_result) + else: + _LOGGER.info("未检测到人脸") + + return results + + except Exception as ex: + # 计算请求耗时 + duration = time.time() - start_time + + # 记录错误日志 + self._log_error("人脸搜索", request_id, ex, duration) + + # 重新抛出异常 + raise + + # 执行带重试的API调用 + return self._execute_with_retry(api_call, "人脸搜索") + + def detect_faces( + self, + image_url: str = None, + image_path: str = None, + image_file: str = None, + max_face_num: int = 1, + min_face_size: int = 34, + need_rotate_check: int = 1 + ) -> List[Dict[str, Any]]: + """检测人脸""" + _LOGGER.info( + "开始检测人脸: max_face_num=%d, min_face_size=%d, need_rotate_check=%d", + max_face_num, min_face_size, need_rotate_check + ) + + # 定义API调用函数 + def api_call(): + # 生成请求ID + request_id = str(uuid.uuid4()) + start_time = time.time() + + try: + # 获取图片数据 + image = self._get_image_base64(image_url, image_path, image_file) + + # 创建请求 + req = models.DetectFaceRequest() + params = { + "Image": image, + "MaxFaceNum": max_face_num, + "MinFaceSize": min_face_size, + "NeedRotateCheck": need_rotate_check + } + + # 记录请求日志 + self._log_request("人脸检测", params) + + req.from_json_string(json.dumps(params)) + + # 发送请求 + _LOGGER.debug("发送人脸检测请求") + resp = self.client.DetectFace(req) + + # 计算请求耗时 + duration = time.time() - start_time + + # 记录响应日志 + self._log_response("人脸检测", request_id, resp, duration) + + # 解析响应 + results = [] + if getattr(resp, "FaceInfos", []): + _LOGGER.info("检测到 %d 个人脸", len(resp.FaceInfos)) + for face_info in resp.FaceInfos: + face_result = { + "face_id": getattr(face_info, "FaceId", ""), + "x": getattr(face_info, "X", 0), + "y": getattr(face_info, "Y", 0), + "width": getattr(face_info, "Width", 0), + "height": getattr(face_info, "Height", 0), + "face_rect": getattr(face_info, "FaceRect", None).__dict__ if getattr(face_info, "FaceRect", None) else None + } + results.append(face_result) + else: + _LOGGER.info("未检测到人脸") + + return results + + except Exception as ex: + # 计算请求耗时 + duration = time.time() - start_time + + # 记录错误日志 + self._log_error("人脸检测", request_id, ex, duration) + + # 重新抛出异常 + raise + + # 执行带重试的API调用 + return self._execute_with_retry(api_call, "人脸检测") + + def get_face_attributes( + self, + image_url: str = None, + image_path: str = None, + image_file: str = None, + max_face_num: int = 1, + need_rotate_check: int = 1 + ) -> List[Dict[str, Any]]: + """获取人脸属性""" + _LOGGER.info( + "开始获取人脸属性: max_face_num=%d, need_rotate_check=%d", + max_face_num, need_rotate_check + ) + + # 定义API调用函数 + def api_call(): + # 生成请求ID + request_id = str(uuid.uuid4()) + start_time = time.time() + + try: + # 获取图片数据 + image = self._get_image_base64(image_url, image_path, image_file) + + # 创建请求 + req = models.DetectFaceRequest() + params = { + "Image": image, + "MaxFaceNum": max_face_num, + "NeedRotateCheck": need_rotate_check + } + + # 记录请求日志 + self._log_request("获取人脸属性", params) + + req.from_json_string(json.dumps(params)) + + # 发送请求 + _LOGGER.debug("发送获取人脸属性请求") + resp = self.client.DetectFace(req) + + # 计算请求耗时 + duration = time.time() - start_time + + # 记录响应日志 + self._log_response("获取人脸属性", request_id, resp, duration) + + # 解析响应 + results = [] + if getattr(resp, "FaceInfos", []): + _LOGGER.info("检测到 %d 个人脸", len(resp.FaceInfos)) + for face_info in resp.FaceInfos: + face_result = { + "face_id": getattr(face_info, "FaceId", ""), + "x": getattr(face_info, "X", 0), + "y": getattr(face_info, "Y", 0), + "width": getattr(face_info, "Width", 0), + "height": getattr(face_info, "Height", 0), + "gender": getattr(face_info, "Gender", None), + "age": getattr(face_info, "Age", None), + "expression": getattr(face_info, "Expression", None), + "beauty": getattr(face_info, "Beauty", None), + "face_rect": getattr(face_info, "FaceRect", None).__dict__ if getattr(face_info, "FaceRect", None) else None + } + results.append(face_result) + else: + _LOGGER.info("未检测到人脸") + + return results + + except Exception as ex: + # 计算请求耗时 + duration = time.time() - start_time + + # 记录错误日志 + self._log_error("获取人脸属性", request_id, ex, duration) + + # 重新抛出异常 + raise + + # 执行带重试的API调用 + return self._execute_with_retry(api_call, "获取人脸属性") + + def get_group_info(self, group_id: str) -> Dict[str, Any]: + """获取人员库信息""" + _LOGGER.info("开始获取人员库信息: group_id=%s", group_id) + + def api_call(): + request_id = str(uuid.uuid4()) + start_time = time.time() + try: + req = models.GetGroupInfoRequest() + params = {"GroupId": group_id} + self._log_request("获取人员库信息", params) + req.from_json_string(json.dumps(params)) + + _LOGGER.debug("发送获取人员库信息请求") + resp = self.client.GetGroupInfo(req) + + duration = time.time() - start_time + self._log_response("获取人员库信息", request_id, resp, duration) + + return { + "group_name": getattr(resp, "GroupName", ""), + "group_tag": getattr(resp, "GroupTag", ""), + "face_model_version": getattr(resp, "FaceModelVersion", ""), + "creation_timestamp": getattr(resp, "CreationTimestamp", 0), + } + except Exception as ex: + duration = time.time() - start_time + self._log_error("获取人员库信息", request_id, ex, duration) + raise + + return self._execute_with_retry(api_call, "获取人员库信息") + + def _test_api_connection(self, group_id: str) -> bool: + """测试API连接""" + try: + _LOGGER.info("开始测试API连接: group_id=%s", group_id) + self.get_group_info(group_id) + _LOGGER.info("API连接测试成功") + return True + except Exception as ex: + _LOGGER.error("API连接测试失败: %s", ex, exc_info=True) + return False + + def get_person_list(self, group_id: str, limit: int = 100, offset: int = 0) -> List[Dict[str, Any]]: + """获取人员列表""" + _LOGGER.info("开始获取人员列表: group_id=%s, limit=%d, offset=%d", group_id, limit, offset) + + def api_call(): + request_id = str(uuid.uuid4()) + start_time = time.time() + try: + req = models.GetPersonListRequest() + params = { + "GroupId": group_id, + "Limit": limit, + "Offset": offset + } + self._log_request("获取人员列表", params) + req.from_json_string(json.dumps(params)) + + _LOGGER.debug("发送获取人员列表请求") + resp = self.client.GetPersonList(req) + + duration = time.time() - start_time + self._log_response("获取人员列表", request_id, resp, duration) + + persons = [] + if getattr(resp, "PersonInfos", []): + for person_info in resp.PersonInfos: + persons.append({ + "person_name": getattr(person_info, "PersonName", ""), + "person_id": getattr(person_info, "PersonId", ""), + "gender": getattr(person_info, "Gender", 0), + "face_ids": getattr(person_info, "FaceIds", []), + }) + return persons + except Exception as ex: + duration = time.time() - start_time + self._log_error("获取人员列表", request_id, ex, duration) + raise + + return self._execute_with_retry(api_call, "获取人员列表") + + def _execute_with_retry(self, api_call: Callable, operation_name: str) -> Any: + """执行带重试的API调用""" + last_exception = None + + for attempt in range(self.retry_config.max_retries): + try: + if attempt > 0: + # 计算重试延迟 + delay = self._calculate_retry_delay(attempt) + _LOGGER.info("重试第 %d 次%s，延迟 %.2f 秒", attempt + 1, operation_name, delay) + time.sleep(delay) + + # 执行API调用 + return api_call() + + except TencentCloudSDKException as ex: + last_exception = ex + _LOGGER.warning("%s失败 (尝试 %d/%d): %s", operation_name, attempt + 1, self.retry_config.max_retries, ex) + + # 检查是否应该重试 + if not self._should_retry(ex): + _LOGGER.error("遇到不可重试的错误: %s", ex.code) + self._handle_error(ex) + break + + except Exception as ex: + last_exception = ex + _LOGGER.warning("%s失败 (尝试 %d/%d): %s", operation_name, attempt + 1, self.retry_config.max_retries, ex) + + # 所有重试都失败后，抛出最后一个异常 + if last_exception: + if isinstance(last_exception, TencentCloudSDKException): + self._handle_error(last_exception) + else: + _LOGGER.error("%s最终失败: %s", operation_name, last_exception, exc_info=True) + log_and_raise_error( + RuntimeError, + f"{operation_name}失败: {str(last_exception)}", + {"error": str(last_exception), "error_type": type(last_exception).__name__} + ) + + def _calculate_retry_delay(self, attempt: int) -> float: + """计算重试延迟时间""" + # 指数退避算法 + delay = self.retry_config.initial_delay * ( + self.retry_config.backoff_factor ** (attempt - 1) + ) + + # 限制最大延迟 + delay = min(delay, self.retry_config.max_delay) + + # 添加抖动，避免多个客户端同时重试 + jitter = delay * self.retry_config.jitter + delay = delay + random.uniform(-jitter, jitter) + + # 确保延迟不为负数 + return max(0, delay) + + def _should_retry(self, ex: TencentCloudSDKException) -> bool: + """判断是否应该重试""" + # 如果错误码在不可重试的错误列表中，则不重试 + if ex.code in NON_RETRYABLE_ERRORS: + return False + + # 如果是速率限制错误，但已经是最后一次尝试，则不重试 + if ex.code in RATE_LIMIT_ERRORS: + return True + + # 如果是网络错误，则重试 + if ex.code in NETWORK_ERRORS: + return True + + # 默认情况下，其他错误都重试 + return True + + def _is_valid_url(self, url: str) -> bool: + """验证URL格式是否有效""" + url_pattern = re.compile( + r'^https?://' # http:// 或 https:// + r'(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+[A-Z]{2,6}\.?|' # 域名 + r'localhost|' # localhost + r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})' # IP地址 + r'(?::\d+)?' # 可选端口 + r'(?:/?|[/?]\S+)$', re.IGNORECASE) + return url_pattern.match(url) is not None + + def _process_image_data(self, image_data: bytes) -> bytes: + """处理图片数据，包括压缩和格式转换""" + try: + # 检查图片大小 + if len(image_data) > MAX_IMAGE_SIZE: + _LOGGER.warning("图片大小超过限制，尝试压缩: %d 字节", len(image_data)) + image_data = self._compress_image(image_data) + + # 尝试使用PIL处理图片 + try: + from PIL import Image + + # 将字节数据转换为PIL图像对象 + img = Image.open(io.BytesIO(image_data)) + + # 检查图片尺寸 + width, height = img.size + if width > MAX_IMAGE_DIMENSION or height > MAX_IMAGE_DIMENSION: + _LOGGER.warning("图片尺寸过大，尝试缩放: %dx%d", width, height) + + # 计算缩放比例 + scale = min(MAX_IMAGE_DIMENSION / width, MAX_IMAGE_DIMENSION / height) + new_width = int(width * scale) + new_height = int(height * scale) + + # 缩放图片 + img = img.resize((new_width, new_height), Image.LANCZOS) + + # 将处理后的图片转换回字节数据 + output = io.BytesIO() + img.save(output, format='JPEG', quality=IMAGE_COMPRESSION_QUALITY) + image_data = output.getvalue() + + _LOGGER.info("图片缩放完成: %dx%d -> %dx%d", width, height, new_width, new_height) + + # 如果图片不是JPEG格式，转换为JPEG以减少大小 + if img.format != 'JPEG': + _LOGGER.debug("转换图片格式为JPEG: %s", img.format) + output = io.BytesIO() + if img.mode == 'RGBA': + img = img.convert('RGB') + img.save(output, format='JPEG', quality=IMAGE_COMPRESSION_QUALITY) + image_data = output.getvalue() + + except ImportError: + _LOGGER.warning("PIL库未安装，无法进行图片处理") + except Exception as ex: + _LOGGER.warning("图片处理失败: %s", ex) + + return image_data + + except Exception as ex: + _LOGGER.error("处理图片数据失败: %s", ex, exc_info=True) + # 如果处理失败，返回原始数据 + return image_data + + def _compress_image(self, image_data: bytes) -> bytes: + """压缩图片数据""" + try: + from PIL import Image + + # 将字节数据转换为PIL图像对象 + img = Image.open(io.BytesIO(image_data)) + + # 计算压缩比例 + original_size = len(image_data) + target_size = MAX_IMAGE_SIZE * 0.8 # 目标大小为最大限制的80% + quality = IMAGE_COMPRESSION_QUALITY + + # 逐步降低质量直到满足大小要求 + while quality > 30 and original_size > target_size: + output = io.BytesIO() + img.save(output, format='JPEG', quality=quality) + compressed_data = output.getvalue() + + if len(compressed_data) <= target_size: + break + + quality -= 5 + original_size = len(compressed_data) + + _LOGGER.info("图片压缩完成: %d -> %d 字节 (质量: %d)", + len(image_data), len(compressed_data), quality) + + return compressed_data + + except ImportError: + _LOGGER.warning("PIL库未安装，无法进行图片压缩") + return image_data + except Exception as ex: + _LOGGER.error("图片压缩失败: %s", ex, exc_info=True) + return image_data + + def _log_request(self, operation: str, params: Dict[str, Any]) -> None: + """记录API请求日志""" + if not ENABLE_REQUEST_LOGGING: + return + + # 生成请求ID + request_id = str(uuid.uuid4()) + + # 记录请求基本信息 + _LOGGER.info( + "API请求: operation=%s, request_id=%s, params_count=%d", + operation, request_id, len(params) + ) + + # 记录请求参数（截断过长的参数） + if ENABLE_PERFORMANCE_LOGGING: + safe_params = {} + for key, value in params.items(): + if key == "Image": + # 图片参数只记录长度，不记录内容 + safe_params[key] = f"" + else: + safe_params[key] = value + + params_str = json.dumps(safe_params, ensure_ascii=False) + if len(params_str) > LOG_REQUEST_PAYLOAD_MAX_SIZE: + params_str = params_str[:LOG_REQUEST_PAYLOAD_MAX_SIZE] + "...[truncated]" + + _LOGGER.debug( + "API请求详情: operation=%s, request_id=%s, params=%s", + operation, request_id, params_str + ) + + def _log_response(self, operation: str, request_id: str, response: Any, duration: float) -> None: + """记录API响应日志""" + if not ENABLE_RESPONSE_LOGGING: + return + + # 记录响应基本信息 + _LOGGER.info( + "API响应: operation=%s, request_id=%s, duration=%.3fs, response_type=%s", + operation, request_id, duration, type(response).__name__ + ) + + # 记录性能指标 + if ENABLE_PERFORMANCE_LOGGING: + if duration > 5.0: + _LOGGER.warning( + "API响应缓慢: operation=%s, request_id=%s, duration=%.3fs", + operation, request_id, duration + ) + elif duration > 2.0: + _LOGGER.info( + "API响应较慢: operation=%s, request_id=%s, duration=%.3fs", + operation, request_id, duration + ) + + # 记录响应详情（截断过长的响应） + try: + if hasattr(response, "__dict__"): + # 如果是对象，尝试转换为字典 + response_dict = {} + for attr in dir(response): + if not attr.startswith("_") and not callable(getattr(response, attr)): + response_dict[attr] = getattr(response, attr) + + response_str = json.dumps(response_dict, ensure_ascii=False, default=str) + else: + # 如果不是对象，直接转换为字符串 + response_str = str(response) + + if len(response_str) > LOG_RESPONSE_PAYLOAD_MAX_SIZE: + response_str = response_str[:LOG_RESPONSE_PAYLOAD_MAX_SIZE] + "...[truncated]" + + _LOGGER.debug( + "API响应详情: operation=%s, request_id=%s, response=%s", + operation, request_id, response_str + ) + except Exception as ex: + _LOGGER.debug( + "无法记录API响应详情: operation=%s, request_id=%s, error=%s", + operation, request_id, ex + ) + + def _log_error(self, operation: str, request_id: str, error: Exception, duration: float) -> None: + """记录API错误日志""" + error_type = type(error).__name__ + error_msg = str(error) + + # 记录错误基本信息 + _LOGGER.error( + "API错误: operation=%s, request_id=%s, duration=%.3fs, error_type=%s, error_msg=%s", + operation, request_id, duration, error_type, error_msg + ) + + # 记录错误详情 + if isinstance(error, TencentCloudSDKException): + _LOGGER.error( + "腾讯云API错误详情: operation=%s, request_id=%s, code=%s, message=%s, request_id=%s", + operation, request_id, getattr(error, "code", "未知"), + getattr(error, "message", "未知"), + getattr(error, "request_id", "未知") + ) + + # 记录堆栈跟踪 + if _LOGGER.isEnabledFor(logging.DEBUG): + _LOGGER.debug( + "API错误堆栈跟踪: operation=%s, request_id=%s", + operation, request_id, exc_info=True + ) \ No newline at end of file