From d081a51aa0fd765e05f2c0a5f5201bc4e5062f5b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AD=90=E5=A2=A8?= Date: Fri, 1 May 2026 23:14:39 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=8D=87=E7=BA=A7=E8=87=B3v2.0.0?= =?UTF-8?q?=EF=BC=8C=E6=96=B0=E5=A2=9E=E4=BA=BA=E5=91=98/=E4=BA=BA?= =?UTF-8?q?=E8=84=B8=E7=AE=A1=E7=90=86=E6=9C=8D=E5=8A=A1=E4=B8=8E=E6=91=84?= =?UTF-8?q?=E5=83=8F=E5=A4=B4=E6=94=AF=E6=8C=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 create_person、delete_person、create_face、delete_face 服务 - 支持通过摄像头实体获取图片 (camera_entity_id 参数) - 添加配置选项支持，可通过 Options Flow 修改配置 - 传感器采用 CoordinatorEntity 模式，优化数据更新机制 - 改进重试逻辑与错误处理 - 重构代码结构，清理冗余代码 --- __init__.py | 66 ++-- config_flow.py | 213 +++++------ const.py | 31 +- errors.py | 143 +++----- face_recognition.py | 257 +++++++++---- manifest.json | 8 +- requirements.txt | 4 +- sensor.py | 282 ++++++++------- services.py | 390 +++++++++++++++----- services.yaml | 326 ++++++++++++----- strings.json | 217 ++++++++++- tencent_cloud_client.py | 774 +++++++++++++++++++++++----------------- 12 files changed, 1711 insertions(+), 1000 deletions(-) diff --git a/__init__.py b/__init__.py index e1a4b08..78adaa2 100644 --- a/__init__.py +++ b/__init__.py @@ -4,12 +4,9 @@ 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.core import HomeAssistant from homeassistant.exceptions import ConfigEntryNotReady -from homeassistant.helpers.entity import Entity -from homeassistant.helpers.entity_platform import AddEntitiesCallback from .const import ( DOMAIN, @@ -17,7 +14,8 @@ from .const import ( CONF_SECRET_ID, CONF_SECRET_KEY, CONF_REGION, - CONF_PERSON_GROUP_ID + CONF_PERSON_GROUP_ID, + get_entry_value, ) from .tencent_cloud_client import TencentCloudClient from .face_recognition import FaceRecognition @@ -25,30 +23,24 @@ 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") - - # 创建腾讯云客户端 + region = get_entry_value(entry, CONF_REGION, "ap-shanghai") + person_group_id = get_entry_value(entry, CONF_PERSON_GROUP_ID, "Hass") + try: client = TencentCloudClient(secret_id, secret_key, region) except Exception as ex: - _LOGGER.error("创建腾讯云客户端失败: %s", ex) + _LOGGER.error("创建腾讯云客户端失败") raise ConfigEntryNotReady(f"创建腾讯云客户端失败: {ex}") - - # 创建人脸识别实例 + face_recognition = FaceRecognition(hass, client) - - # 存储配置和实例 + hass.data.setdefault(DOMAIN, {})[entry.entry_id] = { "entry": entry, "client": client, @@ -58,46 +50,28 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: "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]) - ) - - # 不再注册前端面板，改为使用配置流管理 - + + await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) + 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 - + + unload_ok = await hass.config_entries.async_forward_entry_unload(entry, "sensor") + 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 + _LOGGER.info("移除腾讯云人脸识别插件配置") diff --git a/config_flow.py b/config_flow.py index b16e0af..e0b151b 100644 --- a/config_flow.py +++ b/config_flow.py @@ -18,8 +18,6 @@ from .const import ( CONF_REGION, CONF_PERSON_GROUP_ID, DEFAULT_REGION, - ERROR_INVALID_CONFIG, - ERROR_API_ERROR, ) from .errors import ( validate_config, @@ -33,20 +31,6 @@ 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): """腾讯云人脸识别配置流程""" @@ -54,6 +38,15 @@ class TencentFaceRecognitionConfigFlow(config_entries.ConfigFlow, domain=DOMAIN) VERSION = 1 CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL + AVAILABLE_REGIONS = [ + "ap-beijing", "ap-shanghai", "ap-guangzhou", "ap-chengdu", + "ap-chongqing", "ap-nanjing", "ap-hangzhou" + ] + + SECRET_ID_REGEX = re.compile(r'^AKID[a-zA-Z0-9\-]{36,}$') + SECRET_KEY_REGEX = re.compile(r'^[a-zA-Z0-9+/]{20,}$') + PERSON_GROUP_ID_REGEX = re.compile(r'^[a-zA-Z0-9\-_]{1,64}$') + def __init__(self): """初始化配置流程""" self._client = None @@ -64,38 +57,32 @@ class TencentFaceRecognitionConfigFlow(config_entries.ConfigFlow, domain=DOMAIN) 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) + _LOGGER.error("配置验证失败: %s", ex) errors["base"] = f"配置验证失败: {str(ex)}" - # 显示配置表单 return self._show_user_form(errors) def _show_user_form(self, errors: Dict[str, str]) -> dict: @@ -107,7 +94,7 @@ class TencentFaceRecognitionConfigFlow(config_entries.ConfigFlow, domain=DOMAIN) 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], + "options": [{"label": region, "value": region} for region in self.AVAILABLE_REGIONS], "mode": "dropdown" } }), @@ -128,47 +115,42 @@ class TencentFaceRecognitionConfigFlow(config_entries.ConfigFlow, domain=DOMAIN) 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): + if not self.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): + if not self.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格式 + if region not in self.AVAILABLE_REGIONS: + errors[CONF_REGION] = "无效的区域，请从列表中选择有效的区域" + person_group_id = user_input.get(CONF_PERSON_GROUP_ID, "Hass") - if not PERSON_GROUP_ID_REGEX.match(person_group_id): + if not self.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]: - """验证腾讯云凭据""" + """验证腾讯云凭据 - 通过真实API调用验证""" try: _LOGGER.info("开始验证腾讯云凭据") - - # 创建腾讯云客户端 - self._client = TencentCloudClient( + + client = TencentCloudClient( user_input[CONF_SECRET_ID], user_input[CONF_SECRET_KEY], user_input.get(CONF_REGION, DEFAULT_REGION) ) - - _LOGGER.info("腾讯云客户端创建成功，凭据验证通过") - + + await self.hass.async_add_executor_job(client.verify_credentials) + + _LOGGER.info("腾讯云凭据验证通过") return {"success": True} - + except AuthenticationError as ex: _LOGGER.error("腾讯云认证失败: %s", ex) return { @@ -197,7 +179,7 @@ class TencentFaceRecognitionConfigFlow(config_entries.ConfigFlow, domain=DOMAIN) } } except Exception as ex: - _LOGGER.error("腾讯云凭据验证失败: %s", ex, exc_info=True) + _LOGGER.error("腾讯云凭据验证失败: %s", ex) return { "success": False, "errors": { @@ -215,9 +197,10 @@ class TencentFaceRecognitionConfigFlow(config_entries.ConfigFlow, domain=DOMAIN) class TencentFaceRecognitionOptionsFlow(config_entries.OptionsFlow): """腾讯云人脸识别选项流程""" + AVAILABLE_REGIONS = TencentFaceRecognitionConfigFlow.AVAILABLE_REGIONS + 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) @@ -227,17 +210,16 @@ class TencentFaceRecognitionOptionsFlow(config_entries.OptionsFlow): """管理选项流程""" 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({ @@ -259,28 +241,41 @@ class TencentFaceRecognitionOptionsFlow(config_entries.OptionsFlow): 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 region not in self.AVAILABLE_REGIONS: + errors[CONF_REGION] = "无效的区域，请从列表中选择有效的区域" + + person_group_id = user_input.get(CONF_PERSON_GROUP_ID, "") + if person_group_id and not re.match(r'^[a-zA-Z0-9\-_]{1,64}$', person_group_id): + errors[CONF_PERSON_GROUP_ID] = "人员库ID只能包含字母、数字、连字符和下划线，长度1-64个字符" + if not errors: - # 验证新配置是否有效 try: - # 创建临时客户端进行验证 - temp_client = TencentCloudClient( + client = TencentCloudClient( self._data.get(CONF_SECRET_ID), self._data.get(CONF_SECRET_KEY), region ) - + + await self.hass.async_add_executor_job(client.verify_credentials) + _LOGGER.info("配置验证成功") - - # 保存配置 - return self.async_create_entry(title="", data=user_input) - + + new_data = dict(self._data) + new_data[CONF_REGION] = region + if person_group_id: + new_data[CONF_PERSON_GROUP_ID] = person_group_id + + entry = self.hass.config_entries.async_get_entry(self._entry_id) + self.hass.config_entries.async_update_entry(entry, data=new_data) + + return self.async_create_entry(title="", data={ + CONF_REGION: region, + CONF_PERSON_GROUP_ID: person_group_id, + }) + except AuthenticationError as ex: _LOGGER.error("腾讯云认证失败: %s", ex) errors[CONF_REGION] = "认证失败，请检查区域设置" @@ -290,7 +285,7 @@ class TencentFaceRecognitionOptionsFlow(config_entries.OptionsFlow): errors[CONF_REGION] = "网络错误，请检查区域设置" errors["base"] = "连接腾讯云失败，请检查网络连接和区域设置" except Exception as ex: - _LOGGER.error("配置验证失败: %s", ex, exc_info=True) + _LOGGER.error("配置验证失败: %s", ex) errors["base"] = f"配置验证失败: {str(ex)}" options = { @@ -299,7 +294,7 @@ class TencentFaceRecognitionOptionsFlow(config_entries.OptionsFlow): 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], + "options": [{"label": region, "value": region} for region in self.AVAILABLE_REGIONS], "mode": "dropdown" } }), @@ -314,36 +309,31 @@ class TencentFaceRecognitionOptionsFlow(config_entries.OptionsFlow): 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 + self._data.get(CONF_PERSON_GROUP_ID, "Hass") ) - + 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" @@ -355,8 +345,8 @@ class TencentFaceRecognitionOptionsFlow(config_entries.OptionsFlow): except Exception as ex: connection_status = f"测试失败: {str(ex)}" errors["base"] = f"连接测试失败: {str(ex)}" - _LOGGER.error("连接测试失败: %s", ex, exc_info=True) - + _LOGGER.error("连接测试失败: %s", ex) + return self.async_show_form( step_id="test_connection", data_schema=vol.Schema({}), @@ -365,18 +355,16 @@ class TencentFaceRecognitionOptionsFlow(config_entries.OptionsFlow): "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( @@ -387,28 +375,27 @@ class TencentFaceRecognitionOptionsFlow(config_entries.OptionsFlow): }), errors=errors, ) - - # 创建腾讯云客户端验证新凭据 + new_client = TencentCloudClient( user_input[CONF_SECRET_ID], user_input[CONF_SECRET_KEY], self._data.get(CONF_REGION, DEFAULT_REGION) ) - + + await self.hass.async_add_executor_job(new_client.verify_credentials) + _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 - ) - + + entry = self.hass.config_entries.async_get_entry(self._entry_id) + self.hass.config_entries.async_update_entry(entry, data=new_data) + + await self.hass.config_entries.async_reload(entry.entry_id) + return self.async_create_entry(title="", data={}) - + except AuthenticationError as ex: _LOGGER.error("重新认证失败: %s", ex) errors["base"] = "认证失败，请检查Secret ID和Secret Key" @@ -416,9 +403,9 @@ class TencentFaceRecognitionOptionsFlow(config_entries.OptionsFlow): _LOGGER.error("重新认证网络错误: %s", ex) errors["base"] = "网络错误，请检查网络连接和区域设置" except Exception as ex: - _LOGGER.error("重新认证失败: %s", ex, exc_info=True) + _LOGGER.error("重新认证失败: %s", ex) errors["base"] = f"重新认证失败: {str(ex)}" - + return self.async_show_form( step_id="reauth", data_schema=vol.Schema({ @@ -427,19 +414,17 @@ class TencentFaceRecognitionOptionsFlow(config_entries.OptionsFlow): }), errors=errors, description_placeholders={ - "current_secret_id": self._data.get(CONF_SECRET_ID, ""), + "current_secret_id": self._data.get(CONF_SECRET_ID, "")[:8] + "***" if self._data.get(CONF_SECRET_ID) else "", "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): + if not TencentFaceRecognitionConfigFlow.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 + if not TencentFaceRecognitionConfigFlow.SECRET_KEY_REGEX.match(secret_key): + errors[CONF_SECRET_KEY] = "Secret Key格式不正确，应为20个以上字符" diff --git a/const.py b/const.py index 1f9d3ef..f3d46f6 100644 --- a/const.py +++ b/const.py @@ -1,15 +1,13 @@ """腾讯云人脸识别插件常量定义""" DOMAIN = "tencent_face_recognition" -PLATFORMS = [] +PLATFORMS = ["sensor"] -# 配置项键 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 @@ -19,37 +17,46 @@ 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" +SERVICE_CREATE_PERSON = "create_person" +SERVICE_DELETE_PERSON = "delete_person" +SERVICE_CREATE_FACE = "create_face" +SERVICE_DELETE_FACE = "delete_face" -# 服务参数 ATTR_IMAGE_URL = "image_url" ATTR_IMAGE_PATH = "image_path" ATTR_IMAGE_FILE = "image_file" +ATTR_CAMERA_ENTITY_ID = "camera_entity_id" +ATTR_CONFIG_ENTRY_ID = "config_entry_id" +ATTR_GROUP_ID = "group_id" 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" +ATTR_PERSON_ID = "person_id" +ATTR_PERSON_NAME = "person_name" +ATTR_GENDER = "gender" +ATTR_FACE_ID = "face_id" +ATTR_PERSON_TAG = "person_tag" + +EVENT_FACE_DETECTED = "face_detected" -# 错误消息 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 + +def get_entry_value(entry, key, default=None): + """Get a config value from entry.options with fallback to entry.data.""" + return entry.options.get(key, entry.data.get(key, default)) diff --git a/errors.py b/errors.py index 78e37ef..6e901d4 100644 --- a/errors.py +++ b/errors.py @@ -5,19 +5,18 @@ from typing import Dict, Any, Optional from homeassistant.exceptions import HomeAssistantError +from .const import CONF_SECRET_ID, CONF_SECRET_KEY, CONF_REGION, CONF_PERSON_GROUP_ID + _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}", @@ -26,43 +25,35 @@ ERROR_MESSAGES = { "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}", @@ -71,37 +62,30 @@ ERROR_MESSAGES = { "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": "请在配置中添加缺少的配置项", @@ -109,7 +93,7 @@ ERROR_SOLUTIONS = { "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格式", @@ -118,25 +102,25 @@ ERROR_SOLUTIONS = { "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调用参数是否正确", }, @@ -146,7 +130,7 @@ ERROR_SOLUTIONS = { "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", @@ -155,25 +139,25 @@ ERROR_SOLUTIONS = { "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", } @@ -182,7 +166,7 @@ ERROR_SOLUTIONS = { class TencentFaceRecognitionError(HomeAssistantError): """腾讯云人脸识别基础错误类""" - + def __init__( self, message: str, @@ -190,37 +174,30 @@ class TencentFaceRecognitionError(HomeAssistantError): 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, "") @@ -293,19 +270,18 @@ def handle_api_error(error_code: str, error_message: str) -> None: "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, @@ -316,11 +292,10 @@ def handle_api_error(error_code: str, error_message: str) -> None: 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) + _LOGGER.error("%s: %s", error_class.__name__, message) else: _LOGGER.error("%s: %s", error_class.__name__, message) - - # 创建带有本地化消息的错误 + raise error_class( message, error_key=error_key, @@ -330,8 +305,8 @@ def log_and_raise_error(error_class, message: str, details: Dict[str, Any] = Non def validate_config(config: Dict[str, Any]) -> None: """验证配置""" - required_keys = ["secret_id", "secret_key"] - + required_keys = [CONF_SECRET_ID, CONF_SECRET_KEY] + for key in required_keys: if not config.get(key): log_and_raise_error( @@ -340,29 +315,26 @@ def validate_config(config: Dict[str, Any]) -> None: {"key": key}, "missing_config" ) - - # 验证Secret ID格式 - secret_id = config.get("secret_id") + + secret_id = config.get(CONF_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)}, + {"secret_id": "***"}, "invalid_secret_id" ) - - # 验证Secret Key格式 - secret_key = config.get("secret_key") + + secret_key = config.get(CONF_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)}, + {"secret_key": "***"}, "invalid_secret_key" ) - - # 验证区域格式 - region = config.get("region", "ap-shanghai") + + region = config.get(CONF_REGION, "ap-shanghai") if not isinstance(region, str) or len(region.strip()) == 0: log_and_raise_error( InvalidConfigError, @@ -370,9 +342,8 @@ def validate_config(config: Dict[str, Any]) -> None: {"region": str(region)}, "invalid_region" ) - - # 验证人员库ID格式 - person_group_id = config.get("person_group_id", "Hass") + + person_group_id = config.get(CONF_PERSON_GROUP_ID, "Hass") if not isinstance(person_group_id, str) or len(person_group_id.strip()) == 0: log_and_raise_error( InvalidConfigError, @@ -380,5 +351,5 @@ def validate_config(config: Dict[str, Any]) -> None: {"person_group_id": str(person_group_id)}, "invalid_person_group_id" ) - - _LOGGER.debug("配置验证通过") \ No newline at end of file + + _LOGGER.debug("配置验证通过") diff --git a/face_recognition.py b/face_recognition.py index 1239057..942db47 100644 --- a/face_recognition.py +++ b/face_recognition.py @@ -1,7 +1,7 @@ """人脸识别核心功能""" +import base64 import logging -import json from typing import Dict, Any, List, Optional from homeassistant.core import HomeAssistant @@ -10,14 +10,19 @@ from homeassistant.exceptions import HomeAssistantError from .const import ( ATTR_IMAGE_URL, ATTR_IMAGE_PATH, + ATTR_CAMERA_ENTITY_ID, 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, + ATTR_PERSON_ID, + ATTR_PERSON_NAME, + ATTR_GENDER, + ATTR_FACE_ID, + ATTR_PERSON_TAG, + ATTR_GROUP_ID, ) from .tencent_cloud_client import TencentCloudClient @@ -32,11 +37,50 @@ class FaceRecognition: self.hass = hass self.client = client + async def _get_image_base64_from_source( + self, + image_url: str = None, + image_path: str = None, + image_file: str = None, + camera_entity_id: str = None, + ) -> str: + """从各种来源获取图片 base64 编码""" + if camera_entity_id: + return await self._get_camera_image_base64(camera_entity_id) + if image_url or image_path or image_file: + return await self.hass.async_add_executor_job( + self.client._get_image_base64, image_url, image_path, image_file + ) + raise ValueError("必须提供image_url、image_path、image_file或camera_entity_id") + + async def _get_camera_image_base64(self, camera_entity_id: str) -> str: + """从摄像头实体获取图片 base64""" + try: + from homeassistant.components.camera import async_get_image + image = await async_get_image(self.hass, camera_entity_id) + return base64.b64encode(image.content).decode("utf-8") + except ImportError: + raise HomeAssistantError("摄像头组件不可用，无法获取图片") + except Exception as ex: + raise HomeAssistantError(f"获取摄像头图片失败: {ex}") + + def _validate_image_params( + self, + image_url: str = None, + image_path: str = None, + image_file: str = None, + camera_entity_id: str = None, + ) -> None: + """验证图片参数""" + if not any([image_url, image_path, image_file, camera_entity_id]): + raise ValueError("必须提供image_url、image_path、image_file或camera_entity_id") + async def async_search_faces( self, image_url: str = None, image_path: str = None, image_file: str = None, + camera_entity_id: str = None, group_ids: List[str] = None, max_face_num: int = 1, min_face_size: int = 34, @@ -47,15 +91,15 @@ class FaceRecognition: ) -> Dict[str, Any]: """搜索人脸""" try: - # 验证参数 - self._validate_image_params(image_url, image_path, image_file) + self._validate_image_params(image_url, image_path, image_file, camera_entity_id) + + image_base64 = await self._get_image_base64_from_source( + image_url, image_path, image_file, camera_entity_id + ) - # 调用腾讯云API搜索人脸 result = await self.hass.async_add_executor_job( self.client.search_faces, - image_url, - image_path, - image_file, + None, None, None, image_base64, group_ids, max_face_num, min_face_size, @@ -65,7 +109,6 @@ class FaceRecognition: face_match_threshold ) - # 处理结果 if result.get("success", False): processed_faces = self._process_search_results(result.get("faces", [])) return { @@ -76,7 +119,6 @@ class FaceRecognition: "error_message": None } else: - # 返回错误信息 return { "success": False, "faces": [], @@ -94,6 +136,8 @@ class FaceRecognition: "error_code": "invalid_parameter", "error_message": str(ex) } + except HomeAssistantError: + raise except Exception as ex: _LOGGER.error("人脸搜索失败: %s", ex) return { @@ -104,11 +148,6 @@ class FaceRecognition: "error_message": 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 = [] @@ -130,74 +169,31 @@ class FaceRecognition: 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, + camera_entity_id: str = None, max_face_num: int = 1, need_rotate_check: int = 1 ) -> Dict[str, Any]: """获取人脸属性""" try: - # 验证参数 - self._validate_image_params(image_url, image_path, image_file) + self._validate_image_params(image_url, image_path, image_file, camera_entity_id) + + image_base64 = await self._get_image_base64_from_source( + image_url, image_path, image_file, camera_entity_id + ) - # 调用腾讯云API获取人脸属性 result = await self.hass.async_add_executor_job( self.client.get_face_attributes, - image_url, - image_path, - image_file, + None, None, None, image_base64, max_face_num, need_rotate_check ) - # 处理结果 if result.get("success", False): return { "success": True, @@ -207,7 +203,6 @@ class FaceRecognition: "error_message": None } else: - # 返回错误信息 return { "success": False, "faces": [], @@ -216,6 +211,8 @@ class FaceRecognition: "error_message": result.get("error_message") } + except HomeAssistantError: + raise except Exception as ex: _LOGGER.error("获取人脸属性失败: %s", ex) return { @@ -226,34 +223,32 @@ class FaceRecognition: "error_message": str(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, + camera_entity_id: str = None, max_face_num: int = 1, min_face_size: int = 34, need_rotate_check: int = 1 ) -> Dict[str, Any]: """检测人脸""" try: - # 验证参数 - self._validate_image_params(image_url, image_path, image_file) + self._validate_image_params(image_url, image_path, image_file, camera_entity_id) + + image_base64 = await self._get_image_base64_from_source( + image_url, image_path, image_file, camera_entity_id + ) - # 调用腾讯云API检测人脸 result = await self.hass.async_add_executor_job( self.client.detect_faces, - image_url, - image_path, - image_file, + None, None, None, image_base64, max_face_num, min_face_size, need_rotate_check ) - # 处理结果 if result.get("success", False): return { "success": True, @@ -263,7 +258,6 @@ class FaceRecognition: "error_message": None } else: - # 返回错误信息 return { "success": False, "faces": [], @@ -272,6 +266,8 @@ class FaceRecognition: "error_message": result.get("error_message") } + except HomeAssistantError: + raise except Exception as ex: _LOGGER.error("人脸检测失败: %s", ex) return { @@ -282,4 +278,109 @@ class FaceRecognition: "error_message": str(ex) } - # 移除 _detect_faces_sync 方法，因为现在直接使用客户端的 detect_faces 方法 \ No newline at end of file + async def async_create_person( + self, + person_id: str, + person_name: str, + group_id: str, + image_url: str = None, + image_path: str = None, + image_file: str = None, + camera_entity_id: str = None, + gender: int = None, + person_tag: str = None, + quality_control: int = 1, + need_rotate_check: int = 1, + ) -> Dict[str, Any]: + """创建人员""" + try: + image_base64 = None + if any([image_url, image_path, image_file, camera_entity_id]): + image_base64 = await self._get_image_base64_from_source( + image_url, image_path, image_file, camera_entity_id + ) + + result = await self.hass.async_add_executor_job( + self.client.create_person, + person_id, person_name, group_id, + None, None, None, image_base64, + gender, person_tag, + quality_control, need_rotate_check, + ) + return result + except HomeAssistantError: + raise + except Exception as ex: + _LOGGER.error("创建人员失败: %s", ex) + return { + "success": False, + "error": str(ex), + "error_code": "unknown_error", + "error_message": str(ex) + } + + async def async_delete_person(self, person_id: str) -> Dict[str, Any]: + """删除人员""" + try: + result = await self.hass.async_add_executor_job( + self.client.delete_person, person_id + ) + return result + except Exception as ex: + _LOGGER.error("删除人员失败: %s", ex) + return { + "success": False, + "error": str(ex), + "error_code": "unknown_error", + "error_message": str(ex) + } + + async def async_create_face( + self, + person_id: str, + image_url: str = None, + image_path: str = None, + image_file: str = None, + camera_entity_id: str = None, + quality_control: int = 1, + need_rotate_check: int = 1, + ) -> Dict[str, Any]: + """为人员注册人脸""" + try: + image_base64 = await self._get_image_base64_from_source( + image_url, image_path, image_file, camera_entity_id + ) + + result = await self.hass.async_add_executor_job( + self.client.create_face, + person_id, + None, None, None, image_base64, + quality_control, need_rotate_check, + ) + return result + except HomeAssistantError: + raise + except Exception as ex: + _LOGGER.error("注册人脸失败: %s", ex) + return { + "success": False, + "error": str(ex), + "error_code": "unknown_error", + "error_message": str(ex) + } + + async def async_delete_face(self, person_id: str, face_id: str) -> Dict[str, Any]: + """删除人脸""" + try: + result = await self.hass.async_add_executor_job( + self.client.delete_face, person_id, face_id + ) + return result + except Exception as ex: + _LOGGER.error("删除人脸失败: %s", ex) + return { + "success": False, + "error": str(ex), + "error_code": "unknown_error", + "error_message": str(ex) + } diff --git a/manifest.json b/manifest.json index c617f28..0f26673 100644 --- a/manifest.json +++ b/manifest.json @@ -5,12 +5,12 @@ "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" + "tencentcloud-sdk-python-common>=3.0.0,<4.0.0", + "tencentcloud-sdk-python-iai>=3.0.0,<4.0.0" ], "dependencies": [], "codeowners": ["@xyzmos"], - "version": "1.0.0", + "version": "2.0.0", "iot_class": "cloud_polling", "resources": [] -} \ No newline at end of file +} diff --git a/requirements.txt b/requirements.txt index 037943d..6346c12 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,2 @@ -tencentcloud-sdk-python-common>=3.0.0 -tencentcloud-sdk-python-iai>=3.0.0 \ No newline at end of file +tencentcloud-sdk-python-common>=3.0.0,<4.0.0 +tencentcloud-sdk-python-iai>=3.0.0,<4.0.0 diff --git a/sensor.py b/sensor.py index 8ad013a..f7c6600 100644 --- a/sensor.py +++ b/sensor.py @@ -1,8 +1,7 @@ """传感器实体""" import logging -import base64 -from typing import Dict, Any, Optional +from typing import Dict, Any, Optional, List from datetime import datetime, timedelta from homeassistant.components.sensor import SensorEntity @@ -11,11 +10,59 @@ 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 homeassistant.helpers.update_coordinator import ( + CoordinatorEntity, + DataUpdateCoordinator, + UpdateFailed, +) -from .const import DOMAIN +from .const import ( + DOMAIN, + CONF_PERSON_GROUP_ID, + DEFAULT_PERSON_GROUP_ID, + get_entry_value, +) _LOGGER = logging.getLogger(__name__) +_LOGGER_POLLING = _LOGGER.getChild("polling") + + +class TencentFaceCoordinator(DataUpdateCoordinator): + """腾讯云人脸识别数据更新协调器""" + + def __init__(self, hass: HomeAssistant, entry: ConfigEntry, client): + super().__init__( + hass, + _LOGGER_POLLING, + name="Tencent Face Recognition", + update_interval=timedelta(minutes=5), + ) + self.client = client + self.entry = entry + + async def _async_update_data(self): + """获取最新数据""" + try: + group_id = get_entry_value( + self.entry, CONF_PERSON_GROUP_ID, DEFAULT_PERSON_GROUP_ID + ) + + group_info = await self.hass.async_add_executor_job( + self.client.get_group_info, group_id + ) + + persons = await self.hass.async_add_executor_job( + self.client.get_person_list_all, group_id + ) + + return { + "group_info": group_info, + "persons": persons, + "person_count": len(persons), + } + except Exception as ex: + _LOGGER_POLLING.debug("数据更新失败: %s", ex) + raise UpdateFailed(f"数据更新失败: {ex}") from ex async def async_setup_entry( @@ -23,184 +70,133 @@ async def async_setup_entry( ) -> 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 + name = entry.data.get(CONF_NAME, "腾讯云人脸识别") + + coordinator = TencentFaceCoordinator(hass, entry, client) + + await coordinator.async_config_entry_first_refresh() + + hass.data[DOMAIN][entry.entry_id]["coordinator"] = coordinator + + entities = [ + TencentFaceRecognitionStatusSensor(coordinator, entry, name), + ] + + for person in coordinator.data.get("persons", []): + entities.append( + TencentFaceRecognitionPersonSensor(coordinator, entry, person) ) - - for person in persons: - async_add_entities( - [ - TencentFaceRecognitionPersonSensor(hass, entry, person), - ] - ) - except Exception as ex: - _LOGGER.error("获取人员列表失败: %s", ex) + + async_add_entities(entities) -class TencentFaceRecognitionPersonSensor(SensorEntity): +class TencentFaceRecognitionPersonSensor(CoordinatorEntity, SensorEntity): """腾讯云人脸识别人员传感器""" _attr_icon = "mdi:account" + _attr_has_entity_name = True - def __init__(self, hass: HomeAssistant, entry: ConfigEntry, person: Dict[str, Any]): + def __init__(self, coordinator, entry: ConfigEntry, person: Dict[str, Any]): """初始化传感器""" - self.hass = hass + super().__init__(coordinator) 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"), - } + self._person_id = person.get("person_id") + person_name = person.get("person_name", "未知人员") + self._attr_name = person_name + self._attr_unique_id = f"{entry.entry_id}_{self._person_id}" + self._attr_translation_key = "person_sensor" @property def state(self) -> str: """返回传感器状态""" - return self._state + if self.coordinator.data is None: + return "未知" + for person in self.coordinator.data.get("persons", []): + if person.get("person_id") == self._person_id: + return person.get("person_name", "未知") + return "已删除" @property def extra_state_attributes(self) -> Dict[str, Any]: """返回传感器属性""" - return self._extra_state_attributes + if self.coordinator.data is None: + return {"person_id": self._person_id} + + for person in self.coordinator.data.get("persons", []): + if person.get("person_id") == self._person_id: + return { + "person_id": person.get("person_id"), + "gender": "男" if person.get("gender") == 1 else "女", + "face_ids": person.get("face_ids"), + "face_count": len(person.get("face_ids", [])), + } + + return {"person_id": self._person_id, "status": "已删除"} + + @property + def available(self) -> bool: + """传感器是否可用""" + if self.coordinator.data is None: + return False + for person in self.coordinator.data.get("persons", []): + if person.get("person_id") == self._person_id: + return True + return False -class TencentFaceRecognitionStatusSensor(SensorEntity): +class TencentFaceRecognitionStatusSensor(CoordinatorEntity, SensorEntity): """腾讯云人脸识别状态传感器""" _attr_icon = "mdi:face-recognition" _attr_entity_category = EntityCategory.DIAGNOSTIC + _attr_has_entity_name = True + _attr_translation_key = "status_sensor" - def __init__(self, hass: HomeAssistant, entry: ConfigEntry, name: str): + def __init__(self, coordinator, entry: ConfigEntry, name: str): """初始化传感器""" - self.hass = hass + super().__init__(coordinator) 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 + if self.coordinator.data is None: + return "未连接" + if self.coordinator.last_update_success: + return "已连接" + return "连接错误" @property def extra_state_attributes(self) -> Dict[str, Any]: """返回传感器属性""" - return self._extra_state_attributes + attrs = { + "last_updated": ( + self.coordinator.data.get("last_updated") + if self.coordinator.data + else None + ), + "api_status": "正常" if self.coordinator.last_update_success else "错误", + "person_count": ( + self.coordinator.data.get("person_count", 0) + if self.coordinator.data + else 0 + ), + "last_error": None, + } - 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) + if self.coordinator.data and self.coordinator.data.get("group_info"): + group_info = self.coordinator.data["group_info"] + attrs["group_name"] = group_info.get("group_name", "") + attrs["face_model_version"] = group_info.get("face_model_version", "") - async def _async_update(self, _) -> None: - """更新传感器状态""" - await self.async_update() + return attrs - 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 + @property + def available(self) -> bool: + """状态传感器始终可用""" + return True diff --git a/services.py b/services.py index b97d87a..c39ad3f 100644 --- a/services.py +++ b/services.py @@ -6,6 +6,7 @@ from typing import Dict, Any, List, Optional import voluptuous as vol from homeassistant.core import HomeAssistant, ServiceCall +from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers import config_validation as cv from .const import ( @@ -13,25 +14,39 @@ from .const import ( SERVICE_FACE_SEARCH, SERVICE_DETECT_FACE, SERVICE_GET_FACE_ATTRIBUTES, + SERVICE_CREATE_PERSON, + SERVICE_DELETE_PERSON, + SERVICE_CREATE_FACE, + SERVICE_DELETE_FACE, ATTR_IMAGE_URL, ATTR_IMAGE_FILE, ATTR_IMAGE_PATH, + ATTR_CAMERA_ENTITY_ID, + ATTR_CONFIG_ENTRY_ID, + ATTR_GROUP_ID, ATTR_MAX_FACE_NUM, ATTR_MIN_FACE_SIZE, ATTR_MAX_USER_NUM, ATTR_QUALITY_CONTROL, ATTR_NEED_ROTATE_CHECK, ATTR_FACE_MATCH_THRESHOLD, + ATTR_PERSON_ID, + ATTR_PERSON_NAME, + ATTR_GENDER, + ATTR_FACE_ID, + ATTR_PERSON_TAG, + EVENT_FACE_DETECTED, ) _LOGGER = logging.getLogger(__name__) -# 人脸搜索服务参数 FACE_SEARCH_SCHEMA = vol.Schema({ - vol.Required('group_id'): cv.string, + vol.Required(ATTR_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_CAMERA_ENTITY_ID): cv.string, + vol.Optional(ATTR_CONFIG_ENTRY_ID): 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, @@ -40,31 +55,72 @@ FACE_SEARCH_SCHEMA = vol.Schema({ 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_CAMERA_ENTITY_ID): cv.string, + vol.Optional(ATTR_CONFIG_ENTRY_ID): 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_CAMERA_ENTITY_ID): cv.string, + vol.Optional(ATTR_CONFIG_ENTRY_ID): 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]), }) +CREATE_PERSON_SCHEMA = vol.Schema({ + vol.Required(ATTR_PERSON_ID): cv.string, + vol.Required(ATTR_PERSON_NAME): cv.string, + vol.Required(ATTR_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_CAMERA_ENTITY_ID): cv.string, + vol.Optional(ATTR_CONFIG_ENTRY_ID): cv.string, + vol.Optional(ATTR_GENDER): vol.In([0, 1]), + vol.Optional(ATTR_PERSON_TAG): cv.string, + vol.Optional(ATTR_QUALITY_CONTROL, default=1): vol.In([0, 1]), + vol.Optional(ATTR_NEED_ROTATE_CHECK, default=1): vol.In([0, 1]), +}) + +DELETE_PERSON_SCHEMA = vol.Schema({ + vol.Required(ATTR_PERSON_ID): cv.string, + vol.Optional(ATTR_CONFIG_ENTRY_ID): cv.string, +}) + +CREATE_FACE_SCHEMA = vol.Schema({ + vol.Required(ATTR_PERSON_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_CAMERA_ENTITY_ID): cv.string, + vol.Optional(ATTR_CONFIG_ENTRY_ID): cv.string, + vol.Optional(ATTR_QUALITY_CONTROL, default=1): vol.In([0, 1]), + vol.Optional(ATTR_NEED_ROTATE_CHECK, default=1): vol.In([0, 1]), +}) + +DELETE_FACE_SCHEMA = vol.Schema({ + vol.Required(ATTR_PERSON_ID): cv.string, + vol.Required(ATTR_FACE_ID): cv.string, + vol.Optional(ATTR_CONFIG_ENTRY_ID): cv.string, +}) + async def async_setup_services(hass: HomeAssistant) -> None: """设置服务""" + if hass.services.has_service(DOMAIN, SERVICE_FACE_SEARCH): + return + _LOGGER.info("设置腾讯云人脸识别服务") - # 注册人脸搜索服务 hass.services.async_register( DOMAIN, SERVICE_FACE_SEARCH, @@ -73,60 +129,108 @@ async def async_setup_services(hass: HomeAssistant) -> None: supports_response=True ) - # 注册人脸检测服务 hass.services.async_register( DOMAIN, SERVICE_DETECT_FACE, async_detect_face_service, - schema=DETECT_FACE_SCHEMA + schema=DETECT_FACE_SCHEMA, + supports_response=True ) - # 注册获取人脸属性服务 hass.services.async_register( DOMAIN, SERVICE_GET_FACE_ATTRIBUTES, async_get_face_attributes_service, - schema=GET_FACE_ATTRIBUTES_SCHEMA + schema=GET_FACE_ATTRIBUTES_SCHEMA, + supports_response=True + ) + + hass.services.async_register( + DOMAIN, + SERVICE_CREATE_PERSON, + async_create_person_service, + schema=CREATE_PERSON_SCHEMA, + supports_response=True + ) + + hass.services.async_register( + DOMAIN, + SERVICE_DELETE_PERSON, + async_delete_person_service, + schema=DELETE_PERSON_SCHEMA, + supports_response=True + ) + + hass.services.async_register( + DOMAIN, + SERVICE_CREATE_FACE, + async_create_face_service, + schema=CREATE_FACE_SCHEMA, + supports_response=True + ) + + hass.services.async_register( + DOMAIN, + SERVICE_DELETE_FACE, + async_delete_face_service, + schema=DELETE_FACE_SCHEMA, + supports_response=True ) async def async_unload_services(hass: HomeAssistant) -> None: """卸载服务""" + if any(hass.config_entries.async_entries(DOMAIN)): + return + _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) + for service_name in [ + SERVICE_FACE_SEARCH, + SERVICE_DETECT_FACE, + SERVICE_GET_FACE_ATTRIBUTES, + SERVICE_CREATE_PERSON, + SERVICE_DELETE_PERSON, + SERVICE_CREATE_FACE, + SERVICE_DELETE_FACE, + ]: + hass.services.async_remove(DOMAIN, service_name) + + +def _get_entry_data(hass: HomeAssistant, config_entry_id: str = None): + """获取配置项数据，支持多配置条目""" + domain_data = hass.data.get(DOMAIN, {}) + if config_entry_id: + data = domain_data.get(config_entry_id) + if data and isinstance(data, dict) and "face_recognition" in data: + return data + return None + + for entry_id, data in domain_data.items(): + if isinstance(data, dict) and "face_recognition" in data: + return data + return None 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: + entry_data = _get_entry_data(hass, data.get(ATTR_CONFIG_ENTRY_ID)) + if not entry_data: 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') + group_id = data.get(ATTR_GROUP_ID) image_url = data.get(ATTR_IMAGE_URL) image_path = data.get(ATTR_IMAGE_PATH) image_file = data.get(ATTR_IMAGE_FILE) + camera_entity_id = data.get(ATTR_CAMERA_ENTITY_ID) 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) @@ -138,6 +242,7 @@ async def async_face_search_service(call: ServiceCall) -> Dict[str, Any]: image_url=image_url, image_path=image_path, image_file=image_file, + camera_entity_id=camera_entity_id, group_ids=[group_id], max_face_num=max_face_num, min_face_size=min_face_size, @@ -147,47 +252,45 @@ async def async_face_search_service(call: ServiceCall) -> Dict[str, Any]: face_match_threshold=face_match_threshold ) - # 直接返回结果，包含状态和错误信息 + if result.get("success", False): + for face in result.get("faces", []): + for candidate in face.get("candidates", []): + hass.bus.async_fire(EVENT_FACE_DETECTED, { + ATTR_PERSON_ID: candidate.get("person_id"), + ATTR_PERSON_NAME: candidate.get("person_name"), + "score": candidate.get("score"), + ATTR_PERSON_TAG: candidate.get("person_tag"), + ATTR_GROUP_ID: group_id, + ATTR_CAMERA_ENTITY_ID: camera_entity_id, + }) + return result + except HomeAssistantError: + raise except Exception as ex: _LOGGER.error("人脸搜索服务调用失败: %s", ex) - # 返回错误信息，而不是抛出异常 - return { - "success": False, - "faces": [], - "error": str(ex), - "error_code": "service_error", - "error_message": str(ex) - } + raise HomeAssistantError(f"人脸搜索服务调用失败: {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: + entry_data = _get_entry_data(hass, data.get(ATTR_CONFIG_ENTRY_ID)) + if not entry_data: 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) + camera_entity_id = data.get(ATTR_CAMERA_ENTITY_ID) 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) @@ -196,52 +299,43 @@ async def async_detect_face_service(call: ServiceCall) -> Dict[str, Any]: image_url=image_url, image_path=image_path, image_file=image_file, + camera_entity_id=camera_entity_id, max_face_num=max_face_num, min_face_size=min_face_size, need_rotate_check=need_rotate_check ) - # 直接返回结果，包含状态和错误信息 + if not result.get("success", False): + error_msg = result.get("error_message", result.get("error", "未知错误")) + raise HomeAssistantError(f"人脸检测失败: {error_msg}") + return result + except HomeAssistantError: + raise except Exception as ex: _LOGGER.error("人脸检测服务调用失败: %s", ex) - # 返回错误信息，而不是抛出异常 - return { - "success": False, - "faces": [], - "error": str(ex), - "error_code": "service_error", - "error_message": str(ex) - } + raise HomeAssistantError(f"人脸检测服务调用失败: {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: + entry_data = _get_entry_data(hass, data.get(ATTR_CONFIG_ENTRY_ID)) + if not entry_data: 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) + camera_entity_id = data.get(ATTR_CAMERA_ENTITY_ID) max_face_num = data.get(ATTR_MAX_FACE_NUM, 1) need_rotate_check = data.get(ATTR_NEED_ROTATE_CHECK, 1) @@ -249,34 +343,160 @@ async def async_get_face_attributes_service(call: ServiceCall) -> Dict[str, Any] image_url=image_url, image_path=image_path, image_file=image_file, + camera_entity_id=camera_entity_id, max_face_num=max_face_num, need_rotate_check=need_rotate_check ) - # 直接返回结果，包含状态和错误信息 + if not result.get("success", False): + error_msg = result.get("error_message", result.get("error", "未知错误")) + raise HomeAssistantError(f"获取人脸属性失败: {error_msg}") + return result + except HomeAssistantError: + raise except Exception as ex: _LOGGER.error("获取人脸属性服务调用失败: %s", ex) - # 返回错误信息，而不是抛出异常 - return { - "success": False, - "faces": [], - "error": str(ex), - "error_code": "service_error", - "error_message": str(ex) - } + raise HomeAssistantError(f"获取人脸属性服务调用失败: {ex}") -def _get_config_entry(hass: HomeAssistant): - """获取配置项""" - for entry in hass.config_entries.async_entries(DOMAIN): - return entry - return None +async def async_create_person_service(call: ServiceCall) -> Dict[str, Any]: + """创建人员服务""" + hass = call.hass + data = call.data -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 + entry_data = _get_entry_data(hass, data.get(ATTR_CONFIG_ENTRY_ID)) + if not entry_data: + raise HomeAssistantError("未找到腾讯云人脸识别配置") + + face_recognition = entry_data.get("face_recognition") + if not face_recognition: + raise HomeAssistantError("腾讯云人脸识别插件未正确初始化") + + try: + result = await face_recognition.async_create_person( + person_id=data.get(ATTR_PERSON_ID), + person_name=data.get(ATTR_PERSON_NAME), + group_id=data.get(ATTR_GROUP_ID), + image_url=data.get(ATTR_IMAGE_URL), + image_path=data.get(ATTR_IMAGE_PATH), + image_file=data.get(ATTR_IMAGE_FILE), + camera_entity_id=data.get(ATTR_CAMERA_ENTITY_ID), + gender=data.get(ATTR_GENDER), + person_tag=data.get(ATTR_PERSON_TAG), + quality_control=data.get(ATTR_QUALITY_CONTROL, 1), + need_rotate_check=data.get(ATTR_NEED_ROTATE_CHECK, 1), + ) + + if not result.get("success", False): + error_msg = result.get("error_message", result.get("error", "未知错误")) + raise HomeAssistantError(f"创建人员失败: {error_msg}") + + return result + + except HomeAssistantError: + raise + except Exception as ex: + _LOGGER.error("创建人员服务调用失败: %s", ex) + raise HomeAssistantError(f"创建人员服务调用失败: {ex}") + + +async def async_delete_person_service(call: ServiceCall) -> Dict[str, Any]: + """删除人员服务""" + hass = call.hass + data = call.data + + entry_data = _get_entry_data(hass, data.get(ATTR_CONFIG_ENTRY_ID)) + if not entry_data: + raise HomeAssistantError("未找到腾讯云人脸识别配置") + + face_recognition = entry_data.get("face_recognition") + if not face_recognition: + raise HomeAssistantError("腾讯云人脸识别插件未正确初始化") + + try: + result = await face_recognition.async_delete_person( + person_id=data.get(ATTR_PERSON_ID), + ) + + if not result.get("success", False): + error_msg = result.get("error_message", result.get("error", "未知错误")) + raise HomeAssistantError(f"删除人员失败: {error_msg}") + + return result + + except HomeAssistantError: + raise + except Exception as ex: + _LOGGER.error("删除人员服务调用失败: %s", ex) + raise HomeAssistantError(f"删除人员服务调用失败: {ex}") + + +async def async_create_face_service(call: ServiceCall) -> Dict[str, Any]: + """注册人脸服务""" + hass = call.hass + data = call.data + + entry_data = _get_entry_data(hass, data.get(ATTR_CONFIG_ENTRY_ID)) + if not entry_data: + raise HomeAssistantError("未找到腾讯云人脸识别配置") + + face_recognition = entry_data.get("face_recognition") + if not face_recognition: + raise HomeAssistantError("腾讯云人脸识别插件未正确初始化") + + try: + result = await face_recognition.async_create_face( + person_id=data.get(ATTR_PERSON_ID), + image_url=data.get(ATTR_IMAGE_URL), + image_path=data.get(ATTR_IMAGE_PATH), + image_file=data.get(ATTR_IMAGE_FILE), + camera_entity_id=data.get(ATTR_CAMERA_ENTITY_ID), + quality_control=data.get(ATTR_QUALITY_CONTROL, 1), + need_rotate_check=data.get(ATTR_NEED_ROTATE_CHECK, 1), + ) + + if not result.get("success", False): + error_msg = result.get("error_message", result.get("error", "未知错误")) + raise HomeAssistantError(f"注册人脸失败: {error_msg}") + + return result + + except HomeAssistantError: + raise + except Exception as ex: + _LOGGER.error("注册人脸服务调用失败: %s", ex) + raise HomeAssistantError(f"注册人脸服务调用失败: {ex}") + + +async def async_delete_face_service(call: ServiceCall) -> Dict[str, Any]: + """删除人脸服务""" + hass = call.hass + data = call.data + + entry_data = _get_entry_data(hass, data.get(ATTR_CONFIG_ENTRY_ID)) + if not entry_data: + raise HomeAssistantError("未找到腾讯云人脸识别配置") + + face_recognition = entry_data.get("face_recognition") + if not face_recognition: + raise HomeAssistantError("腾讯云人脸识别插件未正确初始化") + + try: + result = await face_recognition.async_delete_face( + person_id=data.get(ATTR_PERSON_ID), + face_id=data.get(ATTR_FACE_ID), + ) + + if not result.get("success", False): + error_msg = result.get("error_message", result.get("error", "未知错误")) + raise HomeAssistantError(f"删除人脸失败: {error_msg}") + + return result + + except HomeAssistantError: + raise + except Exception as ex: + _LOGGER.error("删除人脸服务调用失败: %s", ex) + raise HomeAssistantError(f"删除人脸服务调用失败: {ex}") diff --git a/services.yaml b/services.yaml index 27e31ac..6cf6f75 100644 --- a/services.yaml +++ b/services.yaml @@ -1,4 +1,3 @@ -# 腾讯云人脸识别服务定义 face_search: name: 人脸搜索 description: 在指定的人员库中搜索人脸。 @@ -24,7 +23,19 @@ face_search: text: image_file: name: 图片文件 - description: 通过文件上传的人脸图片。 + description: 通过文件上传的人脸图片（Base64编码）。 + selector: + text: + camera_entity_id: + name: 摄像头实体 + description: 用于获取图片的摄像头实体ID。 + example: "camera.front_door" + selector: + entity: + domain: camera + config_entry_id: + name: 配置项ID + description: 使用的配置项ID（多配置时指定）。 selector: text: max_face_num: @@ -62,90 +73,16 @@ face_search: mode: box quality_control: name: 质量控制 - description: 是否进行质量控制。 - default: true - example: 1 - selector: - number: - min: 0 - max: 1 - mode: box - 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: 最多检测的人脸数量。 + description: 是否进行质量控制（0=否，1=是）。 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: + select: + options: + - label: "否" + value: "0" + - label: "是" + value: "1" get_face_attributes: name: 获取人脸属性 @@ -165,7 +102,19 @@ get_face_attributes: text: image_file: name: 图片文件 - description: 要分析的图片文件。 + description: 要分析的图片文件（Base64编码）。 + selector: + text: + camera_entity_id: + name: 摄像头实体 + description: 用于获取图片的摄像头实体ID。 + example: "camera.front_door" + selector: + entity: + domain: camera + config_entry_id: + name: 配置项ID + description: 使用的配置项ID（多配置时指定）。 selector: text: max_face_num: @@ -181,8 +130,209 @@ get_face_attributes: mode: box need_rotate_check: name: 旋转检查 - description: 是否进行旋转检查。 - default: true - example: true + description: 是否进行旋转检查（0=否，1=是）。 + default: 1 + example: 1 selector: - boolean: \ No newline at end of file + select: + options: + - label: "否" + value: "0" + - label: "是" + value: "1" + +create_person: + name: 创建人员 + description: 在人员库中创建新人员。 + fields: + person_id: + name: 人员ID + description: 人员的唯一标识符。 + required: true + example: "person_001" + selector: + text: + person_name: + name: 人员名称 + description: 人员名称。 + required: true + example: "张三" + selector: + text: + group_id: + name: 人员库ID + description: 要添加到的人员库ID。 + required: true + example: "Hass" + selector: + text: + image_url: + name: 图片URL + description: 人员人脸图片URL。 + example: "https://example.com/face.jpg" + selector: + text: + image_path: + name: 图片路径 + description: 人员人脸本地图片路径。 + example: "/config/www/face.jpg" + selector: + text: + image_file: + name: 图片文件 + description: 人员人脸图片文件（Base64编码）。 + selector: + text: + camera_entity_id: + name: 摄像头实体 + description: 用于获取人脸图片的摄像头实体ID。 + example: "camera.front_door" + selector: + entity: + domain: camera + config_entry_id: + name: 配置项ID + description: 使用的配置项ID（多配置时指定）。 + selector: + text: + gender: + name: 性别 + description: 性别（0=女，1=男）。 + selector: + select: + options: + - label: "女" + value: "0" + - label: "男" + value: "1" + person_tag: + name: 人员标签 + description: 人员备注标签。 + selector: + text: + quality_control: + name: 质量控制 + description: 是否进行质量控制（0=否，1=是）。 + default: 1 + selector: + select: + options: + - label: "否" + value: "0" + - label: "是" + value: "1" + need_rotate_check: + name: 旋转检查 + description: 是否进行旋转检查（0=否，1=是）。 + default: 1 + selector: + select: + options: + - label: "否" + value: "0" + - label: "是" + value: "1" + +delete_person: + name: 删除人员 + description: 从人员库中删除人员。 + fields: + person_id: + name: 人员ID + description: 要删除的人员ID。 + required: true + example: "person_001" + selector: + text: + config_entry_id: + name: 配置项ID + description: 使用的配置项ID（多配置时指定）。 + selector: + text: + +create_face: + name: 注册人脸 + description: 为已有人员添加新的人脸照片。 + fields: + person_id: + name: 人员ID + description: 要添加人脸的人员ID。 + required: true + example: "person_001" + selector: + text: + image_url: + name: 图片URL + description: 人脸图片URL。 + example: "https://example.com/face.jpg" + selector: + text: + image_path: + name: 图片路径 + description: 人脸本地图片路径。 + example: "/config/www/face.jpg" + selector: + text: + image_file: + name: 图片文件 + description: 人脸图片文件（Base64编码）。 + selector: + text: + camera_entity_id: + name: 摄像头实体 + description: 用于获取人脸图片的摄像头实体ID。 + example: "camera.front_door" + selector: + entity: + domain: camera + config_entry_id: + name: 配置项ID + description: 使用的配置项ID（多配置时指定）。 + selector: + text: + quality_control: + name: 质量控制 + description: 是否进行质量控制（0=否，1=是）。 + default: 1 + selector: + select: + options: + - label: "否" + value: "0" + - label: "是" + value: "1" + need_rotate_check: + name: 旋转检查 + description: 是否进行旋转检查（0=否，1=是）。 + default: 1 + selector: + select: + options: + - label: "否" + value: "0" + - label: "是" + value: "1" + +delete_face: + name: 删除人脸 + description: 删除指定人员的人脸。 + fields: + person_id: + name: 人员ID + description: 人员ID。 + required: true + example: "person_001" + selector: + text: + face_id: + name: 人脸ID + description: 要删除的人脸ID。 + required: true + example: "3440118425101275818" + selector: + text: + config_entry_id: + name: 配置项ID + description: 使用的配置项ID（多配置时指定）。 + selector: + text: diff --git a/strings.json b/strings.json index 766cd40..d282c08 100644 --- a/strings.json +++ b/strings.json @@ -12,7 +12,7 @@ "name": "名称" }, "data_description": { - "secret_id": "腾讯云API的Secret ID", + "secret_id": "腾讯云API的Secret ID，以AKID开头", "secret_key": "腾讯云API的Secret Key", "region": "腾讯云服务区域，例如：ap-shanghai", "person_group_id": "人脸库人员组ID，用于存储和搜索人脸", @@ -21,7 +21,10 @@ } }, "error": { - "invalid_config": "无效的配置" + "invalid_config": "无效的配置", + "auth_failure": "认证失败，请检查Secret ID和Secret Key", + "network_error": "网络错误，请检查网络连接和区域设置", + "rate_limit_exceeded": "API调用频率过高，请稍后再试" }, "abort": { "already_configured": "已经配置了腾讯云人脸识别服务" @@ -31,21 +34,65 @@ "step": { "init": { "title": "腾讯云人脸识别选项", - "description": "配置腾讯云人脸识别服务选项", + "description": "请选择要管理的项目", "data": { - "region": "区域" + "action": "操作" + } + }, + "config": { + "title": "配置设置", + "description": "修改腾讯云人脸识别的配置设置", + "data": { + "region": "区域", + "person_group_id": "人员组ID" }, "data_description": { - "region": "腾讯云服务区域，例如：ap-shanghai" + "region": "腾讯云服务区域", + "person_group_id": "人脸库人员组ID" } + }, + "test_connection": { + "title": "测试连接", + "description": "测试腾讯云API连接状态\n连接状态: {connection_status}" + }, + "reauth": { + "title": "重新认证", + "description": "更新腾讯云API凭据\n当前Secret ID: {current_secret_id}\n当前区域: {current_region}", + "data": { + "secret_id": "Secret ID", + "secret_key": "Secret Key" + }, + "data_description": { + "secret_id": "新的腾讯云API Secret ID", + "secret_key": "新的腾讯云API Secret Key" + } + } + }, + "error": { + "auth_failure": "认证失败，请检查Secret ID和Secret Key", + "network_error": "网络错误，请检查网络连接和区域设置", + "invalid_config": "配置验证失败" + } + }, + "entity": { + "sensor": { + "status_sensor": { + "name": "状态" + }, + "person_sensor": { + "name": "人员" } } }, "services": { "face_search": { "name": "人脸搜索", - "description": "搜索人脸", + "description": "在指定的人员库中搜索人脸", "fields": { + "group_id": { + "name": "人员库ID", + "description": "要搜索的人员库ID" + }, "image_url": { "name": "图片URL", "description": "要搜索的图片URL" @@ -58,6 +105,14 @@ "name": "图片文件", "description": "要搜索的图片文件（Base64编码）" }, + "camera_entity_id": { + "name": "摄像头实体", + "description": "用于获取图片的摄像头实体ID" + }, + "config_entry_id": { + "name": "配置项ID", + "description": "使用的配置项ID（多配置时指定）" + }, "max_face_num": { "name": "最大人脸数量", "description": "最多处理的人脸数量" @@ -72,11 +127,11 @@ }, "quality_control": { "name": "质量控制", - "description": "是否进行质量控制" + "description": "是否进行质量控制（0=否，1=是）" }, "need_rotate_check": { "name": "旋转检查", - "description": "是否进行旋转检查" + "description": "是否进行旋转检查（0=否，1=是）" }, "face_match_threshold": { "name": "人脸匹配阈值", @@ -100,6 +155,14 @@ "name": "图片文件", "description": "要检测的图片文件（Base64编码）" }, + "camera_entity_id": { + "name": "摄像头实体", + "description": "用于获取图片的摄像头实体ID" + }, + "config_entry_id": { + "name": "配置项ID", + "description": "使用的配置项ID（多配置时指定）" + }, "max_face_num": { "name": "最大人脸数量", "description": "最多检测的人脸数量" @@ -110,7 +173,7 @@ }, "need_rotate_check": { "name": "旋转检查", - "description": "是否进行旋转检查" + "description": "是否进行旋转检查（0=否，1=是）" } } }, @@ -130,15 +193,147 @@ "name": "图片文件", "description": "要分析的图片文件（Base64编码）" }, + "camera_entity_id": { + "name": "摄像头实体", + "description": "用于获取图片的摄像头实体ID" + }, + "config_entry_id": { + "name": "配置项ID", + "description": "使用的配置项ID（多配置时指定）" + }, "max_face_num": { "name": "最大人脸数量", "description": "最多分析的人脸数量" }, "need_rotate_check": { "name": "旋转检查", - "description": "是否进行旋转检查" + "description": "是否进行旋转检查（0=否，1=是）" + } + } + }, + "create_person": { + "name": "创建人员", + "description": "在人员库中创建新人员", + "fields": { + "person_id": { + "name": "人员ID", + "description": "人员的唯一标识符" + }, + "person_name": { + "name": "人员名称", + "description": "人员名称" + }, + "group_id": { + "name": "人员库ID", + "description": "要添加到的人员库ID" + }, + "image_url": { + "name": "图片URL", + "description": "人员人脸图片URL" + }, + "image_path": { + "name": "图片路径", + "description": "人员人脸本地图片路径" + }, + "image_file": { + "name": "图片文件", + "description": "人员人脸图片文件（Base64编码）" + }, + "camera_entity_id": { + "name": "摄像头实体", + "description": "用于获取人脸图片的摄像头实体ID" + }, + "config_entry_id": { + "name": "配置项ID", + "description": "使用的配置项ID（多配置时指定）" + }, + "gender": { + "name": "性别", + "description": "性别（0=女，1=男）" + }, + "person_tag": { + "name": "人员标签", + "description": "人员备注标签" + }, + "quality_control": { + "name": "质量控制", + "description": "是否进行质量控制（0=否，1=是）" + }, + "need_rotate_check": { + "name": "旋转检查", + "description": "是否进行旋转检查（0=否，1=是）" + } + } + }, + "delete_person": { + "name": "删除人员", + "description": "从人员库中删除人员", + "fields": { + "person_id": { + "name": "人员ID", + "description": "要删除的人员ID" + }, + "config_entry_id": { + "name": "配置项ID", + "description": "使用的配置项ID（多配置时指定）" + } + } + }, + "create_face": { + "name": "注册人脸", + "description": "为已有人员添加新的人脸照片", + "fields": { + "person_id": { + "name": "人员ID", + "description": "要添加人脸的人员ID" + }, + "image_url": { + "name": "图片URL", + "description": "人脸图片URL" + }, + "image_path": { + "name": "图片路径", + "description": "人脸本地图片路径" + }, + "image_file": { + "name": "图片文件", + "description": "人脸图片文件（Base64编码）" + }, + "camera_entity_id": { + "name": "摄像头实体", + "description": "用于获取人脸图片的摄像头实体ID" + }, + "config_entry_id": { + "name": "配置项ID", + "description": "使用的配置项ID（多配置时指定）" + }, + "quality_control": { + "name": "质量控制", + "description": "是否进行质量控制（0=否，1=是）" + }, + "need_rotate_check": { + "name": "旋转检查", + "description": "是否进行旋转检查（0=否，1=是）" + } + } + }, + "delete_face": { + "name": "删除人脸", + "description": "删除指定人员的人脸", + "fields": { + "person_id": { + "name": "人员ID", + "description": "人员ID" + }, + "face_id": { + "name": "人脸ID", + "description": "要删除的人脸ID" + }, + "config_entry_id": { + "name": "配置项ID", + "description": "使用的配置项ID（多配置时指定）" } } } } -} \ No newline at end of file +} diff --git a/tencent_cloud_client.py b/tencent_cloud_client.py index 16fa535..dd7ea3a 100644 --- a/tencent_cloud_client.py +++ b/tencent_cloud_client.py @@ -33,29 +33,28 @@ from .errors import ( ) _LOGGER = logging.getLogger(__name__) +_LOGGER_POLLING = _LOGGER.getChild("polling") +_LOGGER_POLLING.setLevel(logging.DEBUG) -# 日志记录配置 -ENABLE_REQUEST_LOGGING = True # 是否启用请求日志记录 -ENABLE_RESPONSE_LOGGING = True # 是否启用响应日志记录 -ENABLE_PERFORMANCE_LOGGING = True # 是否启用性能日志记录 -LOG_REQUEST_PAYLOAD_MAX_SIZE = 1024 # 请求负载日志记录的最大大小（字节） -LOG_RESPONSE_PAYLOAD_MAX_SIZE = 1024 # 响应负载日志记录的最大大小（字节） +ENABLE_REQUEST_LOGGING = False +ENABLE_RESPONSE_LOGGING = False +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 # 抖动因子，避免多个客户端同时重试 +MAX_ATTEMPTS = 3 +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", # 请求配额超限 + "AuthFailure", + "InvalidParameter", + "ResourceNotFound", + "LimitExceeded", + "RequestLimitExceeded", + "RequestQuotaExceeded", } RATE_LIMIT_ERRORS = { @@ -69,26 +68,25 @@ NETWORK_ERRORS = { "ResourceUnavailable.ServiceTimeout", } -# 图片处理配置 -MAX_IMAGE_SIZE = 10 * 1024 * 1024 # 10MB -MAX_IMAGE_DIMENSION = 4096 # 最大图片尺寸 +MAX_IMAGE_SIZE = 10 * 1024 * 1024 +MAX_IMAGE_DIMENSION = 4096 SUPPORTED_IMAGE_FORMATS = [".jpg", ".jpeg", ".png", ".bmp", ".gif"] -IMAGE_COMPRESSION_QUALITY = 85 # 图片压缩质量 -MIN_FACE_SIZE = 34 # 最小人脸尺寸 +IMAGE_COMPRESSION_QUALITY = 85 +MIN_FACE_SIZE = 34 class RetryConfig: """重试配置类""" - + def __init__( self, - max_retries: int = MAX_RETRIES, + max_attempts: int = MAX_ATTEMPTS, 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.max_attempts = max_attempts self.initial_delay = initial_delay self.max_delay = max_delay self.backoff_factor = backoff_factor @@ -105,15 +103,13 @@ class TencentCloudClient: 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 @@ -129,84 +125,96 @@ class TencentCloudClient: _LOGGER.debug("腾讯云客户端创建成功") return client except Exception as ex: - _LOGGER.error("创建腾讯云客户端失败: %s", ex, exc_info=True) + _LOGGER.error("创建腾讯云客户端失败: %s", self._sanitize_error(str(ex))) log_and_raise_error( AuthenticationError, - f"创建腾讯云客户端失败: {str(ex)}", + "创建腾讯云客户端失败", {"region": self.region, "error_type": type(ex).__name__} ) + def verify_credentials(self) -> bool: + """验证凭据是否有效，通过调用 GetGroupList 测试""" + try: + req = models.GetGroupListRequest() + params = {"Limit": 1, "Offset": 0} + req.from_json_string(json.dumps(params)) + self.client.GetGroupList(req) + _LOGGER.info("凭据验证成功") + return True + except TencentCloudSDKException as ex: + _LOGGER.error("凭据验证失败: %s", ex.code) + handle_api_error(ex.code, ex.message) + def _handle_error(self, ex: TencentCloudSDKException) -> None: """处理API错误""" _LOGGER.error( - "腾讯云API调用失败: 错误码=%s, 错误消息=%s, 请求ID=%s", - ex.code, ex.message, getattr(ex, 'request_id', '未知') + "腾讯云API调用失败: 错误码=%s, 请求ID=%s", + ex.code, getattr(ex, 'request_id', '未知') ) handle_api_error(ex.code, ex.message) + def _sanitize_error(self, error_msg: str) -> str: + """脱敏错误信息中的凭据""" + 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 + 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) + _LOGGER.error("处理图片失败: %s", ex) 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" + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/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( @@ -214,14 +222,13 @@ class TencentCloudClient: 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( @@ -237,17 +244,16 @@ class TencentCloudClient: {"url": image_url, "error": "connection_error"} ) except requests.exceptions.RequestException as ex: - _LOGGER.error("下载图片失败: %s, 错误: %s", image_url, ex) + _LOGGER.error("下载图片失败: %s", image_url) 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( @@ -255,8 +261,7 @@ class TencentCloudClient: f"图片文件不存在: {image_path}", {"path": image_path} ) - - # 检查是否为文件 + if not os.path.isfile(image_path): _LOGGER.error("路径不是文件: %s", image_path) log_and_raise_error( @@ -264,8 +269,7 @@ class TencentCloudClient: 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) @@ -274,8 +278,7 @@ class TencentCloudClient: 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) @@ -284,18 +287,16 @@ class TencentCloudClient: 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( @@ -303,45 +304,50 @@ class TencentCloudClient: f"没有权限读取图片文件: {image_path}", {"path": image_path, "error": "permission_denied"} ) + except ImageNotFoundError: + raise except Exception as ex: - _LOGGER.error("读取图片文件失败: %s, 错误: %s", image_path, ex) + _LOGGER.error("读取图片文件失败: %s", image_path) 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) + if len(image_data) <= MAX_IMAGE_SIZE: + try: + from PIL import Image + img = Image.open(io.BytesIO(image_data)) + if (img.size[0] <= MAX_IMAGE_DIMENSION + and img.size[1] <= MAX_IMAGE_DIMENSION + and img.format == 'JPEG'): + return image_file + except ImportError: + pass + except Exception: + pass + processed_image = self._process_image_data(image_data) return base64.b64encode(processed_image).decode("utf-8") except Exception as ex: @@ -352,22 +358,20 @@ class TencentCloudClient: {"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 ImageNotFoundError: + raise except Exception as ex: _LOGGER.error("处理上传的图片文件失败: %s", ex) log_and_raise_error( ImageNotFoundError, - f"处理上传的图片文件失败", + "处理上传的图片文件失败", {"error": str(ex), "error_type": type(ex).__name__} ) @@ -376,6 +380,7 @@ class TencentCloudClient: image_url: str = None, image_path: str = None, image_file: str = None, + image_base64: str = None, group_ids: List[str] = None, max_face_num: int = 1, min_face_size: int = 34, @@ -386,29 +391,25 @@ class TencentCloudClient: ) -> 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 + "开始搜索人脸: max_face_num=%d, min_face_size=%d, max_user_num=%d, quality_control=%d, need_rotate_check=%d, face_match_threshold=%.2f", + 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调用函数 + + if image_base64 is None: + image_base64 = self._get_image_base64(image_url, image_path, image_file) + 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, + "Image": image_base64, "MaxFaceNum": max_face_num, "MinFaceSize": min_face_size, "MaxPersonNum": max_user_num, @@ -417,23 +418,18 @@ class TencentCloudClient: "FaceMatchThreshold": face_match_threshold, "GroupIds": group_ids } - - # 记录请求日志 - self._log_request("人脸搜索", params) - + + self._log_request("人脸搜索", params, request_id) + 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)) @@ -442,10 +438,8 @@ class TencentCloudClient: "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, @@ -454,15 +448,11 @@ class TencentCloudClient: "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 { "success": True, "faces": results, @@ -470,23 +460,18 @@ class TencentCloudClient: "error_code": None, "error_message": None } - + except Exception as ex: - # 计算请求耗时 duration = time.time() - start_time - - # 记录错误日志 self._log_error("人脸搜索", request_id, ex, duration) - - # 不抛出异常，而是返回包含错误信息的结构化结果 + error_code = None error_message = str(ex) - - # 如果是腾讯云API异常，提取错误码和消息 + if isinstance(ex, TencentCloudSDKException): error_code = getattr(ex, "code", None) error_message = getattr(ex, "message", str(ex)) - + return { "success": False, "faces": [], @@ -494,8 +479,7 @@ class TencentCloudClient: "error_code": error_code, "error_message": error_message } - - # 执行带重试的API调用 + return self._execute_with_retry(api_call, "人脸搜索") def detect_faces( @@ -503,6 +487,7 @@ class TencentCloudClient: image_url: str = None, image_path: str = None, image_file: str = None, + image_base64: str = None, max_face_num: int = 1, min_face_size: int = 34, need_rotate_check: int = 1 @@ -512,42 +497,33 @@ class TencentCloudClient: "开始检测人脸: max_face_num=%d, min_face_size=%d, need_rotate_check=%d", max_face_num, min_face_size, need_rotate_check ) - - # 定义API调用函数 + + if image_base64 is None: + image_base64 = self._get_image_base64(image_url, image_path, image_file) + 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, + "Image": image_base64, "MaxFaceNum": max_face_num, "MinFaceSize": min_face_size, "NeedRotateCheck": need_rotate_check } - - # 记录请求日志 - self._log_request("人脸检测", params) - + + self._log_request("人脸检测", params, request_id) + 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)) @@ -563,7 +539,7 @@ class TencentCloudClient: results.append(face_result) else: _LOGGER.info("未检测到人脸") - + return { "success": True, "faces": results, @@ -571,23 +547,18 @@ class TencentCloudClient: "error_code": None, "error_message": None } - + except Exception as ex: - # 计算请求耗时 duration = time.time() - start_time - - # 记录错误日志 self._log_error("人脸检测", request_id, ex, duration) - - # 不抛出异常，而是返回包含错误信息的结构化结果 + error_code = None error_message = str(ex) - - # 如果是腾讯云API异常，提取错误码和消息 + if isinstance(ex, TencentCloudSDKException): error_code = getattr(ex, "code", None) error_message = getattr(ex, "message", str(ex)) - + return { "success": False, "faces": [], @@ -595,8 +566,7 @@ class TencentCloudClient: "error_code": error_code, "error_message": error_message } - - # 执行带重试的API调用 + return self._execute_with_retry(api_call, "人脸检测") def get_face_attributes( @@ -604,6 +574,7 @@ class TencentCloudClient: image_url: str = None, image_path: str = None, image_file: str = None, + image_base64: str = None, max_face_num: int = 1, need_rotate_check: int = 1 ) -> Dict[str, Any]: @@ -612,41 +583,34 @@ class TencentCloudClient: "开始获取人脸属性: max_face_num=%d, need_rotate_check=%d", max_face_num, need_rotate_check ) - - # 定义API调用函数 + + if image_base64 is None: + image_base64 = self._get_image_base64(image_url, image_path, image_file) + 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, + "Image": image_base64, "MaxFaceNum": max_face_num, - "NeedRotateCheck": need_rotate_check + "NeedRotateCheck": need_rotate_check, + "NeedFaceAttributes": 1, + "NeedQualityDetection": 1, } - - # 记录请求日志 - self._log_request("获取人脸属性", params) - + + self._log_request("获取人脸属性", params, request_id) + 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)) @@ -666,7 +630,7 @@ class TencentCloudClient: results.append(face_result) else: _LOGGER.info("未检测到人脸") - + return { "success": True, "faces": results, @@ -674,23 +638,18 @@ class TencentCloudClient: "error_code": None, "error_message": None } - + except Exception as ex: - # 计算请求耗时 duration = time.time() - start_time - - # 记录错误日志 self._log_error("获取人脸属性", request_id, ex, duration) - - # 不抛出异常，而是返回包含错误信息的结构化结果 + error_code = None error_message = str(ex) - - # 如果是腾讯云API异常，提取错误码和消息 + if isinstance(ex, TencentCloudSDKException): error_code = getattr(ex, "code", None) error_message = getattr(ex, "message", str(ex)) - + return { "success": False, "faces": [], @@ -698,10 +657,176 @@ class TencentCloudClient: "error_code": error_code, "error_message": error_message } - - # 执行带重试的API调用 + return self._execute_with_retry(api_call, "获取人脸属性") - + + def create_person( + self, + person_id: str, + person_name: str, + group_id: str, + image_url: str = None, + image_path: str = None, + image_file: str = None, + image_base64: str = None, + gender: int = None, + person_tag: str = None, + quality_control: int = 1, + need_rotate_check: int = 1, + ) -> Dict[str, Any]: + """创建人员""" + _LOGGER.info("开始创建人员: person_id=%s, group_id=%s", person_id, group_id) + + 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) + + def api_call(): + request_id = str(uuid.uuid4()) + start_time = time.time() + try: + req = models.CreatePersonRequest() + params = { + "PersonId": person_id, + "PersonName": person_name, + "GroupId": group_id, + "QualityControl": quality_control, + "NeedRotateCheck": need_rotate_check, + } + if image_base64: + params["Image"] = image_base64 + if gender is not None: + params["Gender"] = gender + if person_tag is not None: + params["PersonTag"] = person_tag + + self._log_request("创建人员", params, request_id) + req.from_json_string(json.dumps(params)) + + _LOGGER.debug("发送创建人员请求") + resp = self.client.CreatePerson(req) + + duration = time.time() - start_time + self._log_response("创建人员", request_id, resp, duration) + + return { + "success": True, + "person_id": getattr(resp, "PersonId", person_id), + "face_id": getattr(resp, "FaceId", ""), + "face_rect": str(getattr(resp, "FaceRect", "")), + } + 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 delete_person(self, person_id: str) -> Dict[str, Any]: + """删除人员""" + _LOGGER.info("开始删除人员: person_id=%s", person_id) + + def api_call(): + request_id = str(uuid.uuid4()) + start_time = time.time() + try: + req = models.DeletePersonRequest() + params = {"PersonId": person_id} + self._log_request("删除人员", params, request_id) + req.from_json_string(json.dumps(params)) + + _LOGGER.debug("发送删除人员请求") + resp = self.client.DeletePerson(req) + + duration = time.time() - start_time + self._log_response("删除人员", request_id, resp, duration) + + return {"success": True, "person_id": person_id} + 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 create_face( + self, + person_id: str, + image_url: str = None, + image_path: str = None, + image_file: str = None, + image_base64: str = None, + quality_control: int = 1, + need_rotate_check: int = 1, + ) -> Dict[str, Any]: + """为人员注册人脸""" + _LOGGER.info("开始注册人脸: person_id=%s", person_id) + + if image_base64 is None: + image_base64 = self._get_image_base64(image_url, image_path, image_file) + + def api_call(): + request_id = str(uuid.uuid4()) + start_time = time.time() + try: + req = models.CreateFaceRequest() + params = { + "PersonId": person_id, + "Image": image_base64, + "QualityControl": quality_control, + "NeedRotateCheck": need_rotate_check, + } + self._log_request("注册人脸", params, request_id) + req.from_json_string(json.dumps(params)) + + _LOGGER.debug("发送注册人脸请求") + resp = self.client.CreateFace(req) + + duration = time.time() - start_time + self._log_response("注册人脸", request_id, resp, duration) + + return { + "success": True, + "person_id": person_id, + "face_ids": getattr(resp, "FaceIds", []), + "face_rects": [str(r) for r in getattr(resp, "FaceRects", [])], + } + 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 delete_face(self, person_id: str, face_id: str) -> Dict[str, Any]: + """删除人脸""" + _LOGGER.info("开始删除人脸: person_id=%s, face_id=%s", person_id, face_id) + + def api_call(): + request_id = str(uuid.uuid4()) + start_time = time.time() + try: + req = models.DeleteFaceRequest() + params = { + "PersonId": person_id, + "FaceId": face_id, + } + self._log_request("删除人脸", params, request_id) + req.from_json_string(json.dumps(params)) + + _LOGGER.debug("发送删除人脸请求") + resp = self.client.DeleteFace(req) + + duration = time.time() - start_time + self._log_response("删除人脸", request_id, resp, duration) + + return {"success": True, "person_id": person_id, "face_id": face_id} + 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 get_group_info(self, group_id: str) -> Dict[str, Any]: """获取人员库信息""" _LOGGER.info("开始获取人员库信息: group_id=%s", group_id) @@ -712,15 +837,15 @@ class TencentCloudClient: try: req = models.GetGroupInfoRequest() params = {"GroupId": group_id} - self._log_request("获取人员库信息", params) + self._log_request("获取人员库信息", params, request_id) 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", ""), @@ -731,9 +856,23 @@ class TencentCloudClient: duration = time.time() - start_time self._log_error("获取人员库信息", request_id, ex, duration) raise - + return self._execute_with_retry(api_call, "获取人员库信息") + def verify_credentials_lightweight(self) -> bool: + """测试API连接，使用 GetGroupList 轻量验证凭据""" + try: + _LOGGER.info("开始测试API连接（轻量验证）") + req = models.GetGroupListRequest() + params = {"Limit": 1, "Offset": 0} + req.from_json_string(json.dumps(params)) + self.client.GetGroupList(req) + _LOGGER.info("API连接测试成功") + return True + except Exception as ex: + _LOGGER.error("API连接测试失败: %s", self._sanitize_error(str(ex))) + return False + def _test_api_connection(self, group_id: str) -> bool: """测试API连接""" try: @@ -742,7 +881,7 @@ class TencentCloudClient: _LOGGER.info("API连接测试成功") return True except Exception as ex: - _LOGGER.error("API连接测试失败: %s", ex, exc_info=True) + _LOGGER.error("API连接测试失败: %s", self._sanitize_error(str(ex))) return False def get_person_list(self, group_id: str, limit: int = 100, offset: int = 0) -> List[Dict[str, Any]]: @@ -759,15 +898,15 @@ class TencentCloudClient: "Limit": limit, "Offset": offset } - self._log_request("获取人员列表", params) + self._log_request("获取人员列表", params, request_id) 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: @@ -782,131 +921,124 @@ class TencentCloudClient: duration = time.time() - start_time self._log_error("获取人员列表", request_id, ex, duration) raise - + return self._execute_with_retry(api_call, "获取人员列表") + def get_person_list_all(self, group_id: str) -> List[Dict[str, Any]]: + """获取人员列表（自动分页获取全部）""" + all_persons = [] + offset = 0 + limit = 100 + while True: + persons = self.get_person_list(group_id, limit=limit, offset=offset) + all_persons.extend(persons) + if len(persons) < limit: + break + offset += limit + return all_persons + def _execute_with_retry(self, api_call: Callable, operation_name: str) -> Any: - """执行带重试的API调用""" + """执行带重试的API调用，max_attempts 为总尝试次数（1次初始 + N次重试）""" last_exception = None - - for attempt in range(self.retry_config.max_retries): + + for attempt in range(self.retry_config.max_attempts): try: if attempt > 0: - # 计算重试延迟 delay = self._calculate_retry_delay(attempt) - _LOGGER.info("重试第 %d 次%s，延迟 %.2f 秒", attempt + 1, operation_name, delay) + _LOGGER.info("重试第 %d 次%s，延迟 %.2f 秒", attempt, 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) - - # 检查是否应该重试 + _LOGGER.warning("%s失败 (尝试 %d/%d): %s", operation_name, attempt + 1, self.retry_config.max_attempts, ex.code) + 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) - - # 所有重试都失败后，抛出最后一个异常 + _LOGGER.warning("%s失败 (尝试 %d/%d): %s", operation_name, attempt + 1, self.retry_config.max_attempts, 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) + _LOGGER.error("%s最终失败: %s", operation_name, last_exception) 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'^https?://' + r'(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+[A-Z]{2,6}\.?|' + r'localhost|' + r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})' + 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() @@ -914,101 +1046,89 @@ class TencentCloudClient: img = img.convert('RGB') img.save(output, format='JPEG', quality=IMAGE_COMPRESSION_QUALITY) image_data = output.getvalue() - + except ImportError: - _LOGGER.warning("PIL库未安装，无法进行图片处理") + _LOGGER.warning("PIL库(Pillow)未安装，无法进行图片处理。建议安装: pip install Pillow>=10.0.0") except Exception as ex: _LOGGER.warning("图片处理失败: %s", ex) - + return image_data - + except Exception as ex: - _LOGGER.error("处理图片数据失败: %s", ex, exc_info=True) - # 如果处理失败，返回原始数据 + _LOGGER.error("处理图片数据失败: %s", ex) 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% + target_size = MAX_IMAGE_SIZE * 0.8 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库未安装，无法进行图片压缩") + _LOGGER.warning("PIL库(Pillow)未安装，无法进行图片压缩。建议安装: pip install Pillow>=10.0.0") return image_data except Exception as ex: - _LOGGER.error("图片压缩失败: %s", ex, exc_info=True) + _LOGGER.error("图片压缩失败: %s", ex) return image_data - - def _log_request(self, operation: str, params: Dict[str, Any]) -> None: + + def _log_request(self, operation: str, params: Dict[str, Any], request_id: str) -> None: """记录API请求日志""" if not ENABLE_REQUEST_LOGGING: return - - # 生成请求ID - request_id = str(uuid.uuid4()) - - # 记录请求基本信息 - _LOGGER.info( + + _LOGGER.debug( "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( + + _LOGGER.debug( "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( @@ -1020,24 +1140,21 @@ class TencentCloudClient: "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 @@ -1047,30 +1164,25 @@ class TencentCloudClient: "无法记录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) - - # 记录错误基本信息 + error_msg = self._sanitize_error(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 + "API错误: operation=%s, request_id=%s, duration=%.3fs, error_type=%s", + operation, request_id, duration, error_type ) - - # 记录错误详情 + 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", "未知") + "腾讯云API错误详情: operation=%s, request_id=%s, code=%s", + operation, request_id, getattr(error, "code", "未知") ) - - # 记录堆栈跟踪 + 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 + )