From 18a6a8ca9be9e8858b87700ee46d2823588483f8 Mon Sep 17 00:00:00 2001 From: Hermes Date: Tue, 7 Jul 2026 02:33:38 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E5=85=A8=E9=9D=A2=E9=87=8D?= =?UTF-8?q?=E6=9E=84=E8=85=BE=E8=AE=AF=E4=BA=91=E4=BA=BA=E8=84=B8=E8=AF=86?= =?UTF-8?q?=E5=88=AB=E6=8F=92=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 拆分 tencent_cloud_client.py（1311行）为 api_operations.py / image_processor.py / retry.py - 更新 HA 废弃 API: SupportsResponse.OPTIONAL, async_unload_platforms, native_value, 移除 CONNECTION_CLASS - 新增动态人员传感器管理（自动增/删） - 新增 detect_face 服务定义（services.yaml） - 移除 secret_key 持久化存储，增强安全性 - 完善错误映射前缀匹配 + 异常类型安全退避 - 修复 Coordinator 双重人员 ID 跟踪死代码 - 添加 _sanitize_error 脱敏，ImageCache LRU 缓存 - manifest.json 版本 2.1.0 + Pillow 依赖 --- .fix_instructions.md | 54 ++ __init__.py | 88 ++- api_operations.py | 618 +++++++++++++++++ config_flow.py | 61 +- const.py | 6 + errors.py | 70 +- face_recognition.py | 9 +- image_processor.py | 358 ++++++++++ manifest.json | 5 +- requirements.txt | 1 + retry.py | 168 +++++ sensor.py | 221 +++--- services.py | 18 +- services.yaml | 68 ++ tencent_cloud_client.py | 1429 +++++++-------------------------------- 15 files changed, 1829 insertions(+), 1345 deletions(-) create mode 100644 .fix_instructions.md create mode 100644 api_operations.py create mode 100644 image_processor.py create mode 100644 retry.py diff --git a/.fix_instructions.md b/.fix_instructions.md new file mode 100644 index 0000000..69096ad --- /dev/null +++ b/.fix_instructions.md @@ -0,0 +1,54 @@ +你是一名Home Assistant自定义集成开发专家。请全面审查并修复 /workspace/tencent_face_recognition/ 下的腾讯云人脸识别插件。 + +## 项目结构 +- manifest.json: 插件清单 +- const.py: 常量和默认值 +- __init__.py: 入口，async_setup_entry/async_unload_entry +- config_flow.py: 配置流程 +- errors.py: 错误处理 +- face_recognition.py: 人脸识别核心功能 +- sensor.py: 传感器实体 +- services.py: 服务定义 +- tencent_cloud_client.py: 腾讯云API客户端(1311行) +- strings.json: 字符串翻译 + +## 已知问题需要修复 + +### 1. tencent_cloud_client.py +- 文件太大(1311行)，拆分为多个文件 +- 将 get_group_info, get_person_list_all, verify_credentials 等方法提取到独立模块 +- 图片压缩/验证逻辑太复杂，简化 +- 重试逻辑可复用，简化 + +### 2. __init__.py +- async_forward_entry_setups 在新版HA中需要确认API +- ConfigEntryNotReady 使用是否正确 +- 缺少异常处理边界 + +### 3. sensor.py +- CoordinatorEntity 使用是否正确 +- 传感器属性更新机制 +- 人员删除后传感器需要正确处理 + +### 4. services.py +- supports_response 在最新HA中的兼容性 +- 服务注册模式是否正确 + +### 5. config_flow.py +- CONNECTION_CLASS 已废弃，需要更新 +- 配置验证流程优化 + +### 6. const.py +- 添加 missing constants +- 更新默认值 + +## 修复原则 +1. 保持向后兼容性 +2. 使用最新的 Home Assistant 2025+ API +3. 使用最新的腾讯云 IAI SDK API +4. 完善错误处理 +5. 优化代码结构，合理拆分大文件 +6. 所有文件保持 UTF-8 编码 +7. 更新 manifest.json 版本号到 2.1.0 + +请开始修复。每次修改一个文件，确保语法正确。 diff --git a/__init__.py b/__init__.py index 15c64e9..aca8e1c 100644 --- a/__init__.py +++ b/__init__.py @@ -1,44 +1,60 @@ """ -腾讯云人脸识别Home Assistant插件 +腾讯云人脸识别 Home Assistant 插件 """ + +from __future__ import annotations + import logging -from typing import Dict, Any +from typing import Any, Dict from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant from homeassistant.exceptions import ConfigEntryNotReady from .const import ( - DOMAIN, - PLATFORMS, + CONF_PERSON_GROUP_ID, + CONF_REGION, CONF_SECRET_ID, CONF_SECRET_KEY, - CONF_REGION, - CONF_PERSON_GROUP_ID, + DOMAIN, + PLATFORMS, get_entry_value, ) -from .tencent_cloud_client import TencentCloudClient +from .errors import AuthenticationError, NetworkError from .face_recognition import FaceRecognition from .services import async_setup_services, async_unload_services +from .tencent_cloud_client import TencentCloudClient _LOGGER = logging.getLogger(__name__) async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: - """设置配置项""" - _LOGGER.info("设置腾讯云人脸识别插件") + """设置配置项。""" + _LOGGER.info("设置腾讯云人脸识别插件: %s", entry.entry_id) secret_id = entry.data.get(CONF_SECRET_ID) secret_key = entry.data.get(CONF_SECRET_KEY) region = get_entry_value(entry, CONF_REGION, "ap-shanghai") person_group_id = get_entry_value(entry, CONF_PERSON_GROUP_ID, "Hass") + # 1. 创建客户端 try: client = TencentCloudClient(secret_id, secret_key, region) - except Exception as ex: - _LOGGER.error("创建腾讯云客户端失败") - raise ConfigEntryNotReady(f"创建腾讯云客户端失败: {ex}") + except Exception as ex: # noqa: BLE001 + _LOGGER.error("创建腾讯云客户端失败: %s", ex) + raise ConfigEntryNotReady(f"创建腾讯云客户端失败: {ex}") from ex + # 2. 验证凭据（失败则让 HA 稍后重试 setup） + try: + await hass.async_add_executor_job(client.verify_credentials) + except (AuthenticationError, NetworkError) as ex: + _LOGGER.error("腾讯云凭据验证失败: %s", ex) + await hass.async_add_executor_job(client.close) + raise ConfigEntryNotReady(f"腾讯云凭据验证失败: {ex}") from ex + except Exception as ex: # noqa: BLE001 + _LOGGER.warning("腾讯云凭据验证异常（将继续加载）: %s", ex) + + # 3. 构建核心功能与存储 face_recognition = FaceRecognition(hass, client) hass.data.setdefault(DOMAIN, {})[entry.entry_id] = { @@ -46,35 +62,59 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: "client": client, "face_recognition": face_recognition, "secret_id": secret_id, - "secret_key": secret_key, "region": region, - "person_group_id": person_group_id + "person_group_id": person_group_id, } + # 4. 注册服务 await async_setup_services(hass) + # 5. 前向设置平台（sensor） await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) + # 6. 配置项更新时自动重新加载平台 + entry.async_on_unload(entry.add_update_listener(_async_update_listener)) + + _LOGGER.info("腾讯云人脸识别插件设置完成") return True +async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None: + """配置项选项变更时重新加载。""" + _LOGGER.debug("配置项变更，重新加载: %s", entry.entry_id) + await hass.config_entries.async_reload(entry.entry_id) + + async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: - """卸载配置项""" - _LOGGER.info("卸载腾讯云人脸识别插件") + """卸载配置项。""" + _LOGGER.info("卸载腾讯云人脸识别插件: %s", entry.entry_id) - unload_ok = await hass.config_entries.async_forward_entry_unload(entry, "sensor") + # 使用统一的卸载 API（兼容新版 HA） + unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS) - if unload_ok: - await async_unload_services(hass) + entry_data: Dict[str, Any] = hass.data.get(DOMAIN, {}).pop(entry.entry_id, None) - entry_data = hass.data[DOMAIN].pop(entry.entry_id, None) - if entry_data and "client" in entry_data: - client = entry_data["client"] + if entry_data and "client" in entry_data: + client: TencentCloudClient = entry_data["client"] + try: await hass.async_add_executor_job(client.close) + except Exception as ex: # noqa: BLE001 + _LOGGER.warning("关闭腾讯云客户端时出错: %s", ex) + + # 若已无该域的配置项，则卸载服务 + if not hass.data.get(DOMAIN): + await async_unload_services(hass) return unload_ok async def async_remove_entry(hass: HomeAssistant, entry: ConfigEntry) -> None: - """移除配置项""" - _LOGGER.info("移除腾讯云人脸识别插件配置") + """移除配置项。""" + _LOGGER.info("移除腾讯云人脸识别插件配置: %s", entry.entry_id) + # 客户端资源已在 unload 时释放；这里仅做兜底清理 + data = hass.data.get(DOMAIN, {}).pop(entry.entry_id, None) + if data and "client" in data: + try: + await hass.async_add_executor_job(data["client"].close) + except Exception: # noqa: BLE001 + pass diff --git a/api_operations.py b/api_operations.py new file mode 100644 index 0000000..8d91815 --- /dev/null +++ b/api_operations.py @@ -0,0 +1,618 @@ +"""API 操作实现。 + +将原 ``tencent_cloud_client.py`` 中各 API 调用方法（搜索/检测/属性/创建人员/ +删除人员/注册人脸/删除人脸/获取库信息/获取人员列表等）抽离为独立函数， +由 ``TencentCloudClient`` 委托调用。每个函数返回标准化的结果字典。 + +腾讯云 IAI SDK (``tencentcloud.iai.v20200303``) 的若干字段名已更新： +- ``NeedRotateCheck`` 已废弃，使用 ``NeedRotateDetection``。 +- ``CreateFaceRequest`` 使用 ``Images``（列表）而非 ``Image``。 +- ``Candidate`` 没有 ``PersonTag`` 字段（标签在 ``PersonGroupInfos`` 中）。 +- ``FaceInfo`` 的属性位于 ``FaceAttributesInfo`` / ``FaceQualityInfo`` 子结构。 +- ``GetGroupInfoResponse`` 的库标签字段是 ``Tag`` 而非 ``GroupTag``。 +- ``CreateFaceResponse`` 的成功人脸 ID 列表是 ``SucFaceIds`` 而非 ``FaceIds``。 +""" + +from __future__ import annotations + +import json +import logging +import time +import uuid +from typing import Any, Dict, List, Optional + +from tencentcloud.iai.v20200303 import models + +from .retry import RetryConfig, execute_with_retry + +_LOGGER = logging.getLogger(__name__) + +# 是否记录请求/响应载荷详情（调试用） +ENABLE_REQUEST_LOGGING = False +ENABLE_RESPONSE_LOGGING = False +ENABLE_PERFORMANCE_LOGGING = True +LOG_PAYLOAD_MAX_SIZE = 1024 + + +def _log_request(operation: str, params: Dict[str, Any], request_id: str) -> None: + if not ENABLE_REQUEST_LOGGING: + return + safe_params = {} + for key, value in params.items(): + if key in ("Image", "Images", "Url"): + safe_params[key] = f"" + else: + safe_params[key] = value + try: + params_str = json.dumps(safe_params, ensure_ascii=False, default=str) + except (TypeError, ValueError): + params_str = str(safe_params) + if len(params_str) > LOG_PAYLOAD_MAX_SIZE: + params_str = params_str[:LOG_PAYLOAD_MAX_SIZE] + "...[truncated]" + _LOGGER.debug("API请求: op=%s id=%s params=%s", operation, request_id, params_str) + + +def _log_response(operation: str, request_id: str, response: Any, duration: float) -> None: + if ENABLE_PERFORMANCE_LOGGING: + if duration > 5.0: + _LOGGER.warning("API响应缓慢: op=%s id=%s duration=%.3fs", operation, request_id, duration) + elif duration > 2.0: + _LOGGER.info("API响应较慢: op=%s id=%s duration=%.3fs", operation, request_id, duration) + if not ENABLE_RESPONSE_LOGGING: + return + try: + resp_dict = {} + for attr in dir(response): + if attr.startswith("_") or callable(getattr(response, attr)): + continue + resp_dict[attr] = getattr(response, attr) + resp_str = json.dumps(resp_dict, ensure_ascii=False, default=str) + except Exception: # noqa: BLE001 + resp_str = str(response) + if len(resp_str) > LOG_PAYLOAD_MAX_SIZE: + resp_str = resp_str[:LOG_PAYLOAD_MAX_SIZE] + "...[truncated]" + _LOGGER.debug("API响应: op=%s id=%s response=%s", operation, request_id, resp_str) + + +def _log_error(operation: str, request_id: str, error: Exception, duration: float, sanitize=None) -> None: + sanitize = sanitize or (lambda s: s) + _LOGGER.error( + "API错误: op=%s id=%s duration=%.3fs error_type=%s msg=%s", + operation, request_id, duration, type(error).__name__, + sanitize(str(error)), + ) + # SDK 异常的 code 在 TencentCloudSDKException 上 + code = getattr(error, "code", None) + if code: + _LOGGER.error("腾讯云API错误详情: op=%s id=%s code=%s", operation, request_id, code) + if _LOGGER.isEnabledFor(logging.DEBUG): + _LOGGER.debug("API错误堆栈: op=%s id=%s", operation, request_id, exc_info=True) + + +def _build_error_result(ex: Exception, **extra: Any) -> Dict[str, Any]: + """构造失败结果字典。""" + from tencentcloud.common.exception.tencent_cloud_sdk_exception import ( + TencentCloudSDKException, + ) + result: Dict[str, Any] = {"success": False} + result.update(extra) + if isinstance(ex, TencentCloudSDKException): + result["error_code"] = getattr(ex, "code", None) + result["error_message"] = getattr(ex, "message", str(ex)) + else: + result["error_code"] = None + result["error_message"] = str(ex) + result["error"] = str(ex) + return result + + +def _run( + client, + operation_name: str, + build_request, + send, + parse_response, + *, + retry_config: Optional[RetryConfig] = None, + sanitize=None, + success_extra: Optional[Dict[str, Any]] = None, +) -> Dict[str, Any]: + """统一的 API 调用包装：构建请求 -> 发送 -> 解析响应，带重试。 + + - ``build_request()`` 返回 ``(models.XxxRequest, params_dict)``。 + - ``send(req)`` 调用 client 的实际 SDK 方法，返回 response。 + - ``parse_response(resp)`` 返回业务结果字典（成功部分）。 + """ + + def api_call() -> Dict[str, Any]: + request_id = str(uuid.uuid4()) + start = time.time() + try: + req, params = build_request() + _log_request(operation_name, params, request_id) + resp = send(req) + duration = time.time() - start + _log_response(operation_name, request_id, resp, duration) + result = {"success": True} + if success_extra: + result.update(success_extra) + result.update(parse_response(resp)) + return result + except Exception as ex: # noqa: BLE001 + duration = time.time() - start + _log_error(operation_name, request_id, ex, duration, sanitize) + # 抛出交由 execute_with_retry 决定是否重试 + raise + + return execute_with_retry( + api_call, operation_name, config=retry_config, sanitize_error=sanitize + ) + + +# --- 各操作实现 ----------------------------------------------------------- + + +def search_faces( + client, + *, + image_base64: str, + group_ids: List[str], + max_face_num: int = 1, + min_face_size: int = 34, + max_user_num: int = 5, + quality_control: int = 1, + need_rotate_check: int = 1, + face_match_threshold: float = 60.0, + retry_config: Optional[RetryConfig] = None, + sanitize=None, +) -> Dict[str, Any]: + """搜索人脸。""" + + def build_request(): + req = models.SearchFacesRequest() + params = { + "NeedPersonInfo": 1, + "Image": image_base64, + "MaxFaceNum": max_face_num, + "MinFaceSize": min_face_size, + "MaxPersonNum": max_user_num, + "QualityControl": quality_control, + "NeedRotateDetection": need_rotate_check, + "FaceMatchThreshold": face_match_threshold, + "GroupIds": group_ids, + } + req.from_json_string(json.dumps(params)) + return req, params + + def parse_response(resp): + results: List[Dict[str, Any]] = [] + for result in getattr(resp, "Results", []) or []: + face_result: Dict[str, Any] = { + "face_id": None, + "candidates": [], + "face_rect": _face_rect_dict(getattr(result, "FaceRect", None)), + } + for candidate in getattr(result, "Candidates", []) or []: + face_result["candidates"].append({ + "person_id": getattr(candidate, "PersonId", None), + "person_name": getattr(candidate, "PersonName", None), + "score": getattr(candidate, "Score", None), + "face_id": getattr(candidate, "FaceId", None), + "gender": getattr(candidate, "Gender", None), + "person_group_infos": _person_group_infos( + getattr(candidate, "PersonGroupInfos", None) + ), + }) + results.append(face_result) + return {"faces": results, "face_count": len(results)} + + return _run( + client, "人脸搜索", build_request, + lambda req: client.SearchFaces(req), parse_response, + retry_config=retry_config, sanitize=sanitize, + ) + + +def detect_faces( + client, + *, + image_base64: str, + max_face_num: int = 1, + min_face_size: int = 34, + need_rotate_check: int = 1, + retry_config: Optional[RetryConfig] = None, + sanitize=None, +) -> Dict[str, Any]: + """检测人脸（不返回属性）。""" + + def build_request(): + req = models.DetectFaceRequest() + params = { + "Image": image_base64, + "MaxFaceNum": max_face_num, + "MinFaceSize": min_face_size, + "NeedRotateDetection": need_rotate_check, + } + req.from_json_string(json.dumps(params)) + return req, params + + def parse_response(resp): + results = [] + for info in getattr(resp, "FaceInfos", []) or []: + results.append({ + "x": getattr(info, "X", 0), + "y": getattr(info, "Y", 0), + "width": getattr(info, "Width", 0), + "height": getattr(info, "Height", 0), + }) + return {"faces": results, "face_count": len(results), + "image_width": getattr(resp, "ImageWidth", 0), + "image_height": getattr(resp, "ImageHeight", 0)} + + return _run( + client, "人脸检测", build_request, + lambda req: client.DetectFace(req), parse_response, + retry_config=retry_config, sanitize=sanitize, + ) + + +def get_face_attributes( + client, + *, + image_base64: str, + max_face_num: int = 1, + need_rotate_check: int = 1, + retry_config: Optional[RetryConfig] = None, + sanitize=None, +) -> Dict[str, Any]: + """获取人脸属性。""" + + def build_request(): + req = models.DetectFaceRequest() + params = { + "Image": image_base64, + "MaxFaceNum": max_face_num, + "NeedRotateDetection": need_rotate_check, + "NeedFaceAttributes": 1, + "NeedQualityDetection": 1, + } + req.from_json_string(json.dumps(params)) + return req, params + + def parse_response(resp): + results = [] + for info in getattr(resp, "FaceInfos", []) or []: + attrs = getattr(info, "FaceAttributesInfo", None) + quality = getattr(info, "FaceQualityInfo", None) + results.append({ + "x": getattr(info, "X", 0), + "y": getattr(info, "Y", 0), + "width": getattr(info, "Width", 0), + "height": getattr(info, "Height", 0), + "gender": getattr(attrs, "Gender", None) if attrs else None, + "age": getattr(attrs, "Age", None) if attrs else None, + "expression": getattr(attrs, "Expression", None) if attrs else None, + "beauty": getattr(attrs, "Beauty", None) if attrs else None, + "glass": getattr(attrs, "Glass", None) if attrs else None, + "pitch": getattr(attrs, "Pitch", None) if attrs else None, + "yaw": getattr(attrs, "Yaw", None) if attrs else None, + "roll": getattr(attrs, "Roll", None) if attrs else None, + "eye_open": getattr(attrs, "EyeOpen", None) if attrs else None, + "quality_score": getattr(quality, "Score", None) if quality else None, + "quality_brightness": getattr(quality, "Brightness", None) if quality else None, + "quality_sharpness": getattr(quality, "Sharpness", None) if quality else None, + }) + return {"faces": results, "face_count": len(results), + "image_width": getattr(resp, "ImageWidth", 0), + "image_height": getattr(resp, "ImageHeight", 0)} + + return _run( + client, "获取人脸属性", build_request, + lambda req: client.DetectFace(req), parse_response, + retry_config=retry_config, sanitize=sanitize, + ) + + +def create_person( + client, + *, + person_id: str, + person_name: str, + group_id: str, + image_base64: Optional[str] = None, + gender: Optional[int] = None, + person_tag: Optional[str] = None, + quality_control: int = 1, + need_rotate_check: int = 1, + retry_config: Optional[RetryConfig] = None, + sanitize=None, +) -> Dict[str, Any]: + """创建人员。""" + + def build_request(): + req = models.CreatePersonRequest() + params: Dict[str, Any] = { + "PersonId": person_id, + "PersonName": person_name, + "GroupId": group_id, + "QualityControl": quality_control, + "NeedRotateDetection": need_rotate_check, + } + if image_base64: + params["Image"] = image_base64 + if gender is not None: + params["Gender"] = gender + # ``PersonTag`` 字段在 SDK 中已废弃，人员备注需通过 + # ``PersonExDescriptionInfos``（外部描述列表）存储。 + if person_tag is not None: + params["PersonExDescriptionInfos"] = [ + { + "PersonExDescriptionIndex": 0, + "PersonExDescription": person_tag, + } + ] + req.from_json_string(json.dumps(params)) + return req, params + + def parse_response(resp): + return { + "person_id": getattr(resp, "PersonId", person_id), + "face_id": getattr(resp, "FaceId", ""), + "face_rect": _face_rect_dict(getattr(resp, "FaceRect", None)), + "face_model_version": getattr(resp, "FaceModelVersion", ""), + } + + return _run( + client, "创建人员", build_request, + lambda req: client.CreatePerson(req), parse_response, + retry_config=retry_config, sanitize=sanitize, + ) + + +def delete_person( + client, + *, + person_id: str, + retry_config: Optional[RetryConfig] = None, + sanitize=None, +) -> Dict[str, Any]: + """删除人员。""" + + def build_request(): + req = models.DeletePersonRequest() + params = {"PersonId": person_id} + req.from_json_string(json.dumps(params)) + return req, params + + return _run( + client, "删除人员", build_request, + lambda req: client.DeletePerson(req), lambda resp: {"person_id": person_id}, + retry_config=retry_config, sanitize=sanitize, + ) + + +def create_face( + client, + *, + person_id: str, + image_base64: str, + quality_control: int = 1, + need_rotate_check: int = 1, + retry_config: Optional[RetryConfig] = None, + sanitize=None, +) -> Dict[str, Any]: + """为人员注册人脸。 + + ``CreateFaceRequest`` 接受 ``Images``（列表），单张图片包装为单元素列表。 + """ + + def build_request(): + req = models.CreateFaceRequest() + params = { + "PersonId": person_id, + "Images": [image_base64], + "QualityControl": quality_control, + "NeedRotateDetection": need_rotate_check, + } + req.from_json_string(json.dumps(params)) + return req, params + + def parse_response(resp): + return { + "person_id": person_id, + "face_ids": list(getattr(resp, "SucFaceIds", []) or []), + "face_rects": [_face_rect_dict(r) for r in (getattr(resp, "SucFaceRects", []) or [])], + "suc_face_num": getattr(resp, "SucFaceNum", 0), + "ret_code": getattr(resp, "RetCode", None), + "face_model_version": getattr(resp, "FaceModelVersion", ""), + } + + return _run( + client, "注册人脸", build_request, + lambda req: client.CreateFace(req), parse_response, + retry_config=retry_config, sanitize=sanitize, + ) + + +def delete_face( + client, + *, + person_id: str, + face_id: str, + retry_config: Optional[RetryConfig] = None, + sanitize=None, +) -> Dict[str, Any]: + """删除人脸。``DeleteFaceRequest`` 接受 ``FaceIds``（列表）。""" + + def build_request(): + req = models.DeleteFaceRequest() + params = {"PersonId": person_id, "FaceIds": [face_id]} + req.from_json_string(json.dumps(params)) + return req, params + + return _run( + client, "删除人脸", build_request, + lambda req: client.DeleteFace(req), + lambda resp: {"person_id": person_id, "face_id": face_id, + "suc_indexes": list(getattr(resp, "SucIndexes", []) or [])}, + retry_config=retry_config, sanitize=sanitize, + ) + + +def get_group_info( + client, + *, + group_id: str, + retry_config: Optional[RetryConfig] = None, + sanitize=None, +) -> Dict[str, Any]: + """获取人员库信息。""" + + def build_request(): + req = models.GetGroupInfoRequest() + params = {"GroupId": group_id} + req.from_json_string(json.dumps(params)) + return req, params + + def parse_response(resp): + return { + "group_id": getattr(resp, "GroupId", group_id), + "group_name": getattr(resp, "GroupName", ""), + # SDK 字段名为 Tag（库标签），原代码误用 GroupTag + "group_tag": getattr(resp, "Tag", ""), + "face_model_version": getattr(resp, "FaceModelVersion", ""), + "creation_timestamp": getattr(resp, "CreationTimestamp", 0), + } + + return _run( + client, "获取人员库信息", build_request, + lambda req: client.GetGroupInfo(req), parse_response, + retry_config=retry_config, sanitize=sanitize, + ) + + +def get_person_list( + client, + *, + group_id: str, + limit: int = 100, + offset: int = 0, + retry_config: Optional[RetryConfig] = None, + sanitize=None, +) -> Dict[str, Any]: + """获取人员列表（单页）。""" + + def build_request(): + req = models.GetPersonListRequest() + params = {"GroupId": group_id, "Limit": limit, "Offset": offset} + req.from_json_string(json.dumps(params)) + return req, params + + def parse_response(resp): + persons = [] + for info in getattr(resp, "PersonInfos", []) or []: + persons.append({ + "person_id": getattr(info, "PersonId", ""), + "person_name": getattr(info, "PersonName", ""), + "gender": getattr(info, "Gender", 0), + "face_ids": list(getattr(info, "FaceIds", []) or []), + "creation_timestamp": getattr(info, "CreationTimestamp", 0), + }) + return { + "persons": persons, + "person_num": getattr(resp, "PersonNum", len(persons)), + "face_model_version": getattr(resp, "FaceModelVersion", ""), + } + + return _run( + client, "获取人员列表", build_request, + lambda req: client.GetPersonList(req), parse_response, + retry_config=retry_config, sanitize=sanitize, + ) + + +def get_person_list_all( + client, + *, + group_id: str, + limit: int = 100, + retry_config: Optional[RetryConfig] = None, + sanitize=None, +) -> Dict[str, Any]: + """获取人员列表（自动分页获取全部）。""" + all_persons: List[Dict[str, Any]] = [] + offset = 0 + face_model_version = "" + while True: + page = get_person_list( + client, group_id=group_id, limit=limit, offset=offset, + retry_config=retry_config, sanitize=sanitize, + ) + if not page.get("success", False): + return page + persons = page.get("persons", []) + all_persons.extend(persons) + face_model_version = page.get("face_model_version", face_model_version) + # 单页不足 limit 视为末页 + if len(persons) < limit: + break + offset += limit + # 防御：person_num 已达上限则停止 + person_num = page.get("person_num", 0) + if person_num and len(all_persons) >= person_num: + break + + return { + "success": True, + "persons": all_persons, + "person_count": len(all_persons), + "face_model_version": face_model_version, + "error": None, + "error_code": None, + "error_message": None, + } + + +def verify_credentials(client, retry_config: Optional[RetryConfig] = None) -> bool: + """验证凭据是否有效，通过调用 GetGroupList 轻量测试。 + + 成功返回 True；失败抛出对应的 ``TencentFaceRecognitionError``。 + """ + + def api_call() -> bool: + req = models.GetGroupListRequest() + req.from_json_string(json.dumps({"Limit": 1, "Offset": 0})) + client.GetGroupList(req) + return True + + execute_with_retry(api_call, "凭据验证", config=retry_config) + _LOGGER.info("凭据验证成功") + return True + + +# --- 辅助 ----------------------------------------------------------------- + + +def _face_rect_dict(face_rect) -> Optional[Dict[str, Any]]: + """将 SDK 的 FaceRect 对象转为字典。""" + if face_rect is None: + return None + return { + "x": getattr(face_rect, "X", 0), + "y": getattr(face_rect, "Y", 0), + "width": getattr(face_rect, "Width", 0), + "height": getattr(face_rect, "Height", 0), + } + + +def _person_group_infos(infos) -> List[Dict[str, Any]]: + """将 Candidate.PersonGroupInfos 转为列表字典。""" + if not infos: + return [] + result = [] + for info in infos: + result.append({ + "group_id": getattr(info, "GroupId", ""), + "person_ex_descriptions": list(getattr(info, "PersonExDescriptions", []) or []), + }) + return result diff --git a/config_flow.py b/config_flow.py index e0b151b..b4bca0d 100644 --- a/config_flow.py +++ b/config_flow.py @@ -20,12 +20,11 @@ from .const import ( DEFAULT_REGION, ) from .errors import ( - validate_config, - InvalidConfigError, AuthenticationError, + InvalidConfigError, NetworkError, RateLimitError, - log_and_raise_error + validate_config, ) from .tencent_cloud_client import TencentCloudClient @@ -36,7 +35,9 @@ class TencentFaceRecognitionConfigFlow(config_entries.ConfigFlow, domain=DOMAIN) """腾讯云人脸识别配置流程""" VERSION = 1 - CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL + + # CONNECTION_CLASS 已在 HA 2024.7 废弃，连接类型现由 manifest.json 的 + # iot_class 字段声明（本项目为 cloud_polling）。 AVAILABLE_REGIONS = [ "ap-beijing", "ap-shanghai", "ap-guangzhou", "ap-chengdu", @@ -65,6 +66,12 @@ class TencentFaceRecognitionConfigFlow(config_entries.ConfigFlow, domain=DOMAIN) validate_config(user_input) + # 防止重复配置：以 secret_id 作为唯一标识 + await self.async_set_unique_id( + f"{DOMAIN}_{user_input.get(CONF_SECRET_ID)}" + ) + self._abort_if_unique_id_configured() + validation_result = await self._validate_tencent_cloud_credentials(user_input) if not validation_result["success"]: @@ -137,6 +144,7 @@ class TencentFaceRecognitionConfigFlow(config_entries.ConfigFlow, domain=DOMAIN) async def _validate_tencent_cloud_credentials(self, user_input: Dict[str, Any]) -> Dict[str, Any]: """验证腾讯云凭据 - 通过真实API调用验证""" + client = None try: _LOGGER.info("开始验证腾讯云凭据") @@ -186,6 +194,9 @@ class TencentFaceRecognitionConfigFlow(config_entries.ConfigFlow, domain=DOMAIN) "base": f"腾讯云API验证失败: {str(ex)}" } } + finally: + if client is not None: + await self.hass.async_add_executor_job(client.close) @staticmethod @callback @@ -252,6 +263,7 @@ class TencentFaceRecognitionOptionsFlow(config_entries.OptionsFlow): errors[CONF_PERSON_GROUP_ID] = "人员库ID只能包含字母、数字、连字符和下划线，长度1-64个字符" if not errors: + client = None try: client = TencentCloudClient( self._data.get(CONF_SECRET_ID), @@ -287,6 +299,9 @@ class TencentFaceRecognitionOptionsFlow(config_entries.OptionsFlow): except Exception as ex: _LOGGER.error("配置验证失败: %s", ex) errors["base"] = f"配置验证失败: {str(ex)}" + finally: + if client is not None: + await self.hass.async_add_executor_job(client.close) options = { vol.Optional( @@ -316,6 +331,7 @@ class TencentFaceRecognitionOptionsFlow(config_entries.OptionsFlow): connection_status = "未知" if user_input is None: + client = None try: client = TencentCloudClient( self._data.get(CONF_SECRET_ID), @@ -324,7 +340,7 @@ class TencentFaceRecognitionOptionsFlow(config_entries.OptionsFlow): ) success = await self.hass.async_add_executor_job( - client._test_api_connection, + client.test_api_connection, self._data.get(CONF_PERSON_GROUP_ID, "Hass") ) @@ -346,6 +362,9 @@ class TencentFaceRecognitionOptionsFlow(config_entries.OptionsFlow): connection_status = f"测试失败: {str(ex)}" errors["base"] = f"连接测试失败: {str(ex)}" _LOGGER.error("连接测试失败: %s", ex) + finally: + if client is not None: + await self.hass.async_add_executor_job(client.close) return self.async_show_form( step_id="test_connection", @@ -361,21 +380,22 @@ class TencentFaceRecognitionOptionsFlow(config_entries.OptionsFlow): errors = {} if user_input is not None: + temp_errors = {} + self._validate_input_format(user_input, temp_errors) + + if temp_errors: + errors = temp_errors + return self.async_show_form( + step_id="reauth", + data_schema=vol.Schema({ + vol.Required(CONF_SECRET_ID, default=self._data.get(CONF_SECRET_ID, "")): str, + vol.Required(CONF_SECRET_KEY, default=self._data.get(CONF_SECRET_KEY, "")): str, + }), + errors=errors, + ) + + new_client = None try: - temp_errors = {} - self._validate_input_format(user_input, temp_errors) - - if temp_errors: - errors = temp_errors - return self.async_show_form( - step_id="reauth", - data_schema=vol.Schema({ - vol.Required(CONF_SECRET_ID, default=self._data.get(CONF_SECRET_ID, "")): str, - vol.Required(CONF_SECRET_KEY, default=self._data.get(CONF_SECRET_KEY, "")): str, - }), - errors=errors, - ) - new_client = TencentCloudClient( user_input[CONF_SECRET_ID], user_input[CONF_SECRET_KEY], @@ -405,6 +425,9 @@ class TencentFaceRecognitionOptionsFlow(config_entries.OptionsFlow): except Exception as ex: _LOGGER.error("重新认证失败: %s", ex) errors["base"] = f"重新认证失败: {str(ex)}" + finally: + if new_client is not None: + await self.hass.async_add_executor_job(new_client.close) return self.async_show_form( step_id="reauth", diff --git a/const.py b/const.py index f3d46f6..5ebd587 100644 --- a/const.py +++ b/const.py @@ -7,6 +7,7 @@ CONF_SECRET_ID = "secret_id" CONF_SECRET_KEY = "secret_key" CONF_REGION = "region" CONF_PERSON_GROUP_ID = "person_group_id" +CONF_SCAN_INTERVAL = "scan_interval" DEFAULT_REGION = "ap-shanghai" DEFAULT_PERSON_GROUP_ID = "Hass" @@ -16,6 +17,7 @@ DEFAULT_MAX_USER_NUM = 5 DEFAULT_QUALITY_CONTROL = 1 DEFAULT_NEED_ROTATE_CHECK = 1 DEFAULT_FACE_MATCH_THRESHOLD = 60.0 +DEFAULT_SCAN_INTERVAL = 300 # seconds (5 minutes) SERVICE_FACE_SEARCH = "face_search" SERVICE_DETECT_FACE = "detect_face" @@ -45,6 +47,10 @@ ATTR_PERSON_TAG = "person_tag" EVENT_FACE_DETECTED = "face_detected" +# Sensor identifiers +SENSOR_STATUS = "status" +SENSOR_PERSON = "person" + ERROR_INVALID_CONFIG = "无效的配置" ERROR_API_ERROR = "API调用错误" ERROR_IMAGE_NOT_FOUND = "图片未找到" diff --git a/errors.py b/errors.py index 6e901d4..eb8cba8 100644 --- a/errors.py +++ b/errors.py @@ -258,22 +258,51 @@ class QuotaExceededError(TencentFaceRecognitionError): def handle_api_error(error_code: str, error_message: str) -> None: - """处理API错误""" + """处理API错误。 + + 先尝试精确匹配，再用前缀匹配，最后回退到通用 ``ApiError``。 + """ error_mapping = { "FailedOperation.ImageDecodeFailed": (ImageNotFoundError, "image_decode_failed"), + "FailedOperation.ImageSizeTooLarge": (ImageNotFoundError, "image_too_large"), "InvalidParameterValue.NoFaceInPhoto": (FaceNotDetectedError, "face_not_detected"), + "FailedOperation.NoFaceInPhoto": (FaceNotDetectedError, "face_not_detected"), + "FailedOperation.FaceSizeTooSmall": (FaceNotDetectedError, "face_too_small"), "FailedOperation.PersonNotFound": (PersonNotFoundError, "person_not_found"), "FailedOperation.GroupNotFound": (PersonGroupNotFoundError, "group_not_found"), "FailedOperation.FaceNotFound": (FaceNotFoundError, "face_not_found"), - "AuthFailure": (AuthenticationError, "auth_failure"), + "FailedOperation.ConflictInstance": (ApiError, "invalid_parameter"), + "AuthFailure.SignatureFailure": (AuthenticationError, "auth_failure"), + "AuthFailure.SecretIdNotFound": (AuthenticationError, "auth_failure"), "RequestLimitExceeded": (RateLimitError, "rate_limit_exceeded"), "RequestQuotaExceeded": (QuotaExceededError, "quota_exceeded"), "ResourceUnavailable.NetworkError": (NetworkError, "network_error"), } - error_info = error_mapping.get(error_code, (ApiError, "api_error")) - error_class = error_info[0] - error_key = error_info[1] + error_class = None + error_key = None + + if error_code in error_mapping: + error_class, error_key = error_mapping[error_code] + else: + for prefix in ("AuthFailure", "InvalidParameter", "ResourceNotFound", + "LimitExceeded", "FailedOperation"): + if error_code.startswith(prefix): + _LOGGER.debug("按前缀匹配错误码: %s -> %s", error_code, prefix) + if prefix == "AuthFailure": + error_class, error_key = AuthenticationError, "auth_failure" + elif prefix == "InvalidParameter": + error_class, error_key = ApiError, "invalid_parameter" + elif prefix == "ResourceNotFound": + error_class, error_key = ApiError, "api_error" + elif prefix == "LimitExceeded": + error_class, error_key = RateLimitError, "rate_limit_exceeded" + else: + error_class, error_key = ApiError, "api_error" + break + + if error_class is None: + error_class, error_key = ApiError, "api_error" _LOGGER.error("腾讯云API错误: %s - %s", error_code, error_message) @@ -290,17 +319,28 @@ 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", error_class.__name__, message) - else: - _LOGGER.error("%s: %s", error_class.__name__, message) + """记录日志并抛出错误。 - raise error_class( - message, - error_key=error_key, - error_details=details or {} - ) + ``error_class`` 必须是 ``TencentFaceRecognitionError``（或 ``HomeAssistantError``） + 的子类，以保证上层 ``except HomeAssistantError`` 能正确捕获。 + 如果传入的 ``error_class`` 无法构造为带 ``error_key``/``error_details`` 的自定义错误 + （例如 ``ValueError``），则退化为抛出 ``ApiError``，避免异常类型不可控。 + """ + _LOGGER.error("%s: %s", error_class.__name__, message) + + try: + raise error_class( + message, + error_key=error_key, + error_details=details or {} + ) + except TypeError: + # error_class 不支持 error_key/error_details 关键字参数 + raise ApiError( + message, + error_key=error_key, + error_details=details or {} + ) from None def validate_config(config: Dict[str, Any]) -> None: diff --git a/face_recognition.py b/face_recognition.py index 19ebe92..847d055 100644 --- a/face_recognition.py +++ b/face_recognition.py @@ -53,7 +53,6 @@ class FaceRecognition: 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: try: from homeassistant.components.camera import async_get_image @@ -151,11 +150,13 @@ class FaceRecognition: return result def _process_search_results(self, results: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + """规范化搜索结果字段（保留客户端返回的全部信息）。""" processed_results = [] for result in results: processed_result = { "face_id": result.get("face_id"), - "candidates": [] + "face_rect": result.get("face_rect"), + "candidates": [], } for candidate in result.get("candidates", []): @@ -163,7 +164,9 @@ class FaceRecognition: "person_id": candidate.get("person_id"), "person_name": candidate.get("person_name"), "score": candidate.get("score"), - "person_tag": candidate.get("person_tag") + "face_id": candidate.get("face_id"), + "gender": candidate.get("gender"), + "person_group_infos": candidate.get("person_group_infos", []), } processed_result["candidates"].append(processed_candidate) diff --git a/image_processor.py b/image_processor.py new file mode 100644 index 0000000..9b89382 --- /dev/null +++ b/image_processor.py @@ -0,0 +1,358 @@ +"""图片处理模块。 + +负责从 URL / 本地路径 / 上传文件 / data URL 获取并编码图片为 base64， +并在需要时压缩、缩放、转码为 JPEG。原 ``tencent_cloud_client.py`` 中 +分散的图片逻辑在此集中、简化。 +""" + +from __future__ import annotations + +import base64 +import io +import logging +import os +import re +from functools import lru_cache +from typing import Optional + +import requests + +from .errors import ImageNotFoundError, NetworkError, log_and_raise_error + +_LOGGER = logging.getLogger(__name__) + +# 图片限制 +MAX_IMAGE_SIZE = 10 * 1024 * 1024 # 10MB +MAX_IMAGE_DIMENSION = 4096 +SUPPORTED_IMAGE_FORMATS = (".jpg", ".jpeg", ".png", ".bmp", ".gif") +IMAGE_COMPRESSION_QUALITY = 85 +COMPRESS_TARGET_RATIO = 0.8 # 压缩目标 = MAX_IMAGE_SIZE * 0.8 + +_URL_PATTERN = re.compile( + 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, +) + + +def is_valid_url(url: str) -> bool: + """验证 URL 格式是否有效。""" + return bool(_URL_PATTERN.match(url or "")) + + +def _pil_available() -> bool: + try: + import PIL # noqa: F401 + return True + except ImportError: + return False + + +def _open_image(data: bytes): + """打开图片，返回 PIL Image 对象；PIL 不可用时抛出 ImageNotFoundError。""" + try: + from PIL import Image + except ImportError as ex: + _LOGGER.warning( + "PIL库(Pillow)未安装，无法进行图片处理。建议安装: pip install Pillow>=10.0.0" + ) + raise ImageNotFoundError( + "Pillow 未安装，无法处理图片", + error_key="image_decode_failed", + error_details={"reason": "pillow_missing"}, + ) from ex + return Image.open(io.BytesIO(data)) + + +def process_image_data(image_data: bytes) -> bytes: + """处理图片数据：尺寸校验、缩放、格式转换、压缩。 + + PIL 不可用时直接返回原始数据（但仍受调用方的尺寸约束）。 + """ + if not image_data: + return image_data + + # 超过限制先压缩 + if len(image_data) > MAX_IMAGE_SIZE: + _LOGGER.warning("图片大小超过限制，尝试压缩: %d 字节", len(image_data)) + + if not _pil_available(): + # 无法压缩时只能按原样返回，由服务端兜底校验 + return image_data + + try: + img = _open_image(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_size = (int(width * scale), int(height * scale)) + img = img.resize(new_size, _get_resample_filter()) + _LOGGER.info( + "图片缩放完成: %dx%d -> %dx%d", + width, height, new_size[0], new_size[1], + ) + + # 统一转 JPEG + output = io.BytesIO() + if img.mode in ("RGBA", "LA", "P"): + img = img.convert("RGB") + img.save(output, format="JPEG", quality=IMAGE_COMPRESSION_QUALITY, optimize=True) + result = output.getvalue() + + # 仍然过大则进一步压缩 + if len(result) > MAX_IMAGE_SIZE: + result = _compress_to_fit(img, MAX_IMAGE_SIZE * COMPRESS_TARGET_RATIO) + + _LOGGER.debug("图片处理完成: %d -> %d 字节", len(image_data), len(result)) + return result + + except ImageNotFoundError: + raise + except Exception as ex: # noqa: BLE001 + _LOGGER.warning("图片处理失败，使用原始数据: %s", ex) + return image_data + + +def _get_resample_filter(): + from PIL import Image + # LANCZOS 在不同 Pillow 版本中名称一致 + return Image.LANCZOS + + +def _compress_to_fit(img, target_size: int) -> bytes: + """逐步降低质量以将图片压缩到目标大小。""" + quality = IMAGE_COMPRESSION_QUALITY + last_data = b"" + while quality >= 30: + output = io.BytesIO() + out = img + if out.mode in ("RGBA", "LA", "P"): + out = out.convert("RGB") + out.save(output, format="JPEG", quality=quality, optimize=True) + data = output.getvalue() + last_data = data + if len(data) <= target_size: + _LOGGER.info( + "图片压缩完成，最终质量: %d, 大小: %d 字节", quality, len(data) + ) + return data + quality -= 5 + _LOGGER.warning("图片压缩已达最低质量，仍超过目标大小: %d 字节", len(last_data)) + return last_data + + +def download_image_as_base64(image_url: str, session: Optional[requests.Session] = None) -> str: + """下载图片并转换为 base64 编码。""" + _LOGGER.debug("开始下载图片: %s", image_url) + + if not is_valid_url(image_url): + log_and_raise_error( + ImageNotFoundError, + f"无效的图片URL: {image_url}", + {"url": image_url}, + ) + + sess = session or requests + headers = {"User-Agent": "Mozilla/5.0 (HomeAssistant TencentFaceRecognition)"} + + try: + response = sess.get(image_url, timeout=15, headers=headers) + response.raise_for_status() + except requests.exceptions.Timeout as ex: + log_and_raise_error( + NetworkError, + f"下载图片超时: {image_url}", + {"url": image_url, "error": "timeout"}, + ) + raise # log_and_raise_error 抛出，此行仅为类型提示 + except requests.exceptions.ConnectionError as ex: + log_and_raise_error( + NetworkError, + f"下载图片连接错误: {image_url}", + {"url": image_url, "error": "connection_error"}, + ) + raise + except requests.exceptions.RequestException as ex: + log_and_raise_error( + NetworkError, + f"下载图片失败: {image_url}", + {"url": image_url, "error": str(ex)}, + ) + raise + + content_type = response.headers.get("content-type", "").lower() + if content_type and not content_type.startswith("image/"): + log_and_raise_error( + ImageNotFoundError, + f"URL不是有效的图片: {image_url}, 内容类型: {content_type}", + {"url": image_url, "content_type": content_type}, + ) + + image_data = process_image_data(response.content) + return base64.b64encode(image_data).decode("utf-8") + + +def read_local_image_as_base64(image_path: str) -> str: + """读取本地图片并转换为 base64 编码。""" + if not os.path.exists(image_path): + log_and_raise_error( + ImageNotFoundError, + f"图片文件不存在: {image_path}", + {"path": image_path}, + ) + if not os.path.isfile(image_path): + log_and_raise_error( + ImageNotFoundError, + f"路径不是文件: {image_path}", + {"path": image_path}, + ) + + file_ext = os.path.splitext(image_path)[1].lower() + if file_ext not in SUPPORTED_IMAGE_FORMATS: + log_and_raise_error( + ImageNotFoundError, + f"不支持的图片格式: {file_ext}", + {"path": image_path, "supported_formats": list(SUPPORTED_IMAGE_FORMATS)}, + ) + + file_size = os.path.getsize(image_path) + if file_size > MAX_IMAGE_SIZE: + log_and_raise_error( + ImageNotFoundError, + f"图片文件过大: {image_path}", + { + "path": image_path, + "size": file_size, + "max_size": f"{MAX_IMAGE_SIZE // (1024 * 1024)}MB", + }, + ) + + try: + with open(image_path, "rb") as img_file: + image_data = img_file.read() + except PermissionError: + log_and_raise_error( + ImageNotFoundError, + f"没有权限读取图片文件: {image_path}", + {"path": image_path, "error": "permission_denied"}, + ) + raise + + processed = process_image_data(image_data) + return base64.b64encode(processed).decode("utf-8") + + +def encode_uploaded_image_as_base64(image_file) -> str: + """处理上传的图片（data URL / base64 字符串 / 文件对象）并返回 base64。""" + if isinstance(image_file, str): + if image_file.startswith("data:image"): + _LOGGER.debug("处理 data URL 格式图片") + try: + _, encoded = image_file.split(",", 1) + image_data = base64.b64decode(encoded) + except ValueError: + log_and_raise_error( + ImageNotFoundError, + "无效的 data URL 格式", + {"data_url_length": len(image_file)}, + ) + raise + processed = process_image_data(image_data) + return base64.b64encode(processed).decode("utf-8") + + # 视为 base64 字符串 + _LOGGER.debug("使用 base64 编码图片，长度: %d", len(image_file)) + try: + image_data = base64.b64decode(image_file) + except Exception as ex: # noqa: BLE001 + log_and_raise_error( + ImageNotFoundError, + "base64 解码失败", + {"error": str(ex), "data_length": len(image_file)}, + ) + raise + processed = process_image_data(image_data) + return base64.b64encode(processed).decode("utf-8") + + # 文件对象 + _LOGGER.debug("处理文件对象") + image_data = image_file.read() + processed = process_image_data(image_data) + return base64.b64encode(processed).decode("utf-8") + + +def get_image_base64( + image_url: Optional[str] = None, + image_path: Optional[str] = None, + image_file=None, + session: Optional[requests.Session] = None, +) -> str: + """根据提供的来源获取图片 base64。 + + 三者任选其一；优先级 url > path > file。 + """ + if not any([image_url, image_path, image_file]): + log_and_raise_error( + ImageNotFoundError, + "必须提供 image_url、image_path 或 image_file", + {}, + ) + + if image_url: + return download_image_as_base64(image_url, session=session) + if image_path: + return read_local_image_as_base64(image_path) + return encode_uploaded_image_as_base64(image_file) + + +class ImageCache: + """简单的 LRU 图片缓存，避免短时间内重复下载/读取同一图片。 + + 下载使用的 ``requests.Session`` 通过 ``download_fn`` 注入，不参与缓存键。 + """ + + def __init__( + self, + maxsize: int = 32, + download_fn=None, + read_fn=None, + ) -> None: + self._download = lru_cache(maxsize=maxsize)( + download_fn or download_image_as_base64 + ) + self._read_local = lru_cache(maxsize=maxsize)( + read_fn or read_local_image_as_base64 + ) + + def get_url(self, image_url: str) -> str: + return self._download(image_url) + + def get_path(self, image_path: str) -> str: + return self._read_local(image_path) + + def get(self, image_url=None, image_path=None, image_file=None) -> str: + if image_url: + return self.get_url(image_url) + if image_path: + return self.get_path(image_path) + if image_file: + return encode_uploaded_image_as_base64(image_file) + log_and_raise_error( + ImageNotFoundError, + "必须提供 image_url、image_path 或 image_file", + {}, + ) + + def clear(self) -> None: + self._download.cache_clear() + self._read_local.cache_clear() diff --git a/manifest.json b/manifest.json index 0f26673..110a91c 100644 --- a/manifest.json +++ b/manifest.json @@ -6,11 +6,12 @@ "issue_tracker": "https://code.nextrt.com/Hass/tencent_face_recognition/issues", "requirements": [ "tencentcloud-sdk-python-common>=3.0.0,<4.0.0", - "tencentcloud-sdk-python-iai>=3.0.0,<4.0.0" + "tencentcloud-sdk-python-iai>=3.0.0,<4.0.0", + "Pillow>=10.0.0" ], "dependencies": [], "codeowners": ["@xyzmos"], - "version": "2.0.0", + "version": "2.1.0", "iot_class": "cloud_polling", "resources": [] } diff --git a/requirements.txt b/requirements.txt index 6346c12..7a1f30f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,3 @@ tencentcloud-sdk-python-common>=3.0.0,<4.0.0 tencentcloud-sdk-python-iai>=3.0.0,<4.0.0 +Pillow>=10.0.0 diff --git a/retry.py b/retry.py new file mode 100644 index 0000000..0d3bfde --- /dev/null +++ b/retry.py @@ -0,0 +1,168 @@ +"""重试与执行辅助模块。 + +将原 ``tencent_cloud_client.py`` 中的重试逻辑抽离出来，便于复用与测试。 +""" + +from __future__ import annotations + +import logging +import random +import time +from typing import Any, Callable, Optional + +from tencentcloud.common.exception.tencent_cloud_sdk_exception import ( + TencentCloudSDKException, +) + +from .errors import handle_api_error, log_and_raise_error, ApiError + +_LOGGER = logging.getLogger(__name__) + +# 重试默认参数 +MAX_ATTEMPTS = 3 +INITIAL_RETRY_DELAY = 1.0 +MAX_RETRY_DELAY = 30.0 +RETRY_BACKOFF_FACTOR = 2.0 +RETRY_JITTER = 0.1 + +# 不可重试的错误码（业务错误，重试无意义） +NON_RETRYABLE_ERRORS = frozenset({ + "AuthFailure", + "InvalidParameter", + "InvalidParameterValue", + "ResourceNotFound", + "ResourceNotFoundInUse", + "FailedOperation.PersonExist", + "FailedOperation.PersonNotExist", + "FailedOperation.FaceExist", + "FailedOperation.FaceNotExist", + "FailedOperation.GroupExist", + "FailedOperation.GroupNotExist", +}) + +# 速率限制类错误码（可重试） +RATE_LIMIT_ERRORS = frozenset({ + "RequestLimitExceeded", + "RequestQuotaExceeded", + "LimitExceeded", +}) + +# 网络/服务端临时错误码（可重试） +NETWORK_ERRORS = frozenset({ + "ResourceUnavailable.NetworkError", + "ResourceUnavailable.ServiceTimeout", + "InternalError", + "FailedOperation.RequestTimeout", +}) + + +class RetryConfig: + """重试配置。""" + + def __init__( + self, + 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, + ) -> None: + self.max_attempts = max(max_attempts, 1) + self.initial_delay = max(initial_delay, 0.0) + self.max_delay = max(max_delay, self.initial_delay) + self.backoff_factor = max(backoff_factor, 1.0) + self.jitter = max(0.0, min(jitter, 1.0)) + + +def calculate_retry_delay(attempt: int, config: RetryConfig) -> float: + """计算第 ``attempt`` 次重试（从 1 开始）的延迟时间。""" + delay = config.initial_delay * (config.backoff_factor ** (attempt - 1)) + delay = min(delay, config.max_delay) + jitter = delay * config.jitter + delay += random.uniform(-jitter, jitter) + return max(0.0, delay) + + +def should_retry(exception: TencentCloudSDKException) -> bool: + """判断 SDK 异常是否可重试。""" + code = getattr(exception, "code", "") or "" + + # 精确匹配不可重试 + if code in NON_RETRYABLE_ERRORS: + return False + + # 精确匹配可重试 + if code in RATE_LIMIT_ERRORS or code in NETWORK_ERRORS: + return True + + # 前缀匹配：AuthFailure.* / InvalidParameter.* / ResourceNotFound.* 不可重试 + for prefix in ("AuthFailure", "InvalidParameter", "ResourceNotFound"): + if code.startswith(prefix): + return False + + # 其它未知错误默认可重试（服务端临时问题居多） + return True + + +def execute_with_retry( + api_call: Callable[[], Any], + operation_name: str, + config: Optional[RetryConfig] = None, + sanitize_error: Optional[Callable[[str], str]] = None, +) -> Any: + """执行带重试的 API 调用。 + + ``api_call`` 应返回业务结果字典（成功）或抛出 ``TencentCloudSDKException`` + /其它异常。当 ``api_call`` 内部已经把异常转化为 ``{success: False}`` 结果时， + 本函数不会重试——重试仅针对抛出的异常。 + + Args: + api_call: 实际的 API 调用函数。 + operation_name: 操作名称，用于日志。 + config: 重试配置，默认 ``RetryConfig()``。 + sanitize_error: 可选的错误信息脱敏函数。 + """ + cfg = config or RetryConfig() + sanitize = sanitize_error or (lambda s: s) + last_exception: Optional[Exception] = None + + for attempt in range(cfg.max_attempts): + try: + if attempt > 0: + delay = calculate_retry_delay(attempt, cfg) + _LOGGER.info( + "重试第 %d 次 %s，延迟 %.2f 秒", attempt, operation_name, delay + ) + time.sleep(delay) + return api_call() + except TencentCloudSDKException as ex: + last_exception = ex + _LOGGER.warning( + "%s 失败 (尝试 %d/%d): %s", + operation_name, attempt + 1, cfg.max_attempts, ex.code, + ) + if not should_retry(ex): + _LOGGER.error("遇到不可重试的错误: %s", ex.code) + break + except Exception as ex: # noqa: BLE001 — 顶层兜底 + last_exception = ex + _LOGGER.warning( + "%s 失败 (尝试 %d/%d): %s", + operation_name, attempt + 1, cfg.max_attempts, sanitize(str(ex)), + ) + + # 所有尝试均失败 + if isinstance(last_exception, TencentCloudSDKException): + handle_api_error(last_exception.code, last_exception.message) + # handle_api_error 总会抛出，这里保险返回 + raise ApiError(f"{operation_name} 失败: {last_exception.code}") + + if last_exception is not None: + log_and_raise_error( + ApiError, + f"{operation_name} 失败: {last_exception}", + {"error": str(last_exception), "error_type": type(last_exception).__name__}, + ) + + # 理论不可达 + raise ApiError(f"{operation_name} 失败: 未知错误") diff --git a/sensor.py b/sensor.py index 62e5271..e593da9 100644 --- a/sensor.py +++ b/sensor.py @@ -1,8 +1,18 @@ -"""传感器实体""" +"""传感器实体。 + +提供两类传感器： +- ``TencentFaceRecognitionStatusSensor``：诊断类状态传感器，反映连接状态与库信息。 +- ``TencentFaceRecognitionPersonSensor``：每个人员库成员一个传感器。 + +人员传感器基于 ``DataUpdateCoordinator`` 的数据动态创建/移除：当人员从 +人员库中删除后，对应传感器在下次刷新时被标记为不可用并移除。 +""" + +from __future__ import annotations import logging -from typing import Dict, Any, Optional, List -from datetime import datetime, timedelta +from datetime import timedelta +from typing import Any, Dict, Optional from homeassistant.components.sensor import SensorEntity from homeassistant.config_entries import ConfigEntry @@ -17,31 +27,40 @@ from homeassistant.helpers.update_coordinator import ( ) from .const import ( - DOMAIN, CONF_PERSON_GROUP_ID, DEFAULT_PERSON_GROUP_ID, + DEFAULT_SCAN_INTERVAL, + DOMAIN, + SENSOR_PERSON, + SENSOR_STATUS, get_entry_value, ) _LOGGER = logging.getLogger(__name__) _LOGGER_POLLING = _LOGGER.getChild("polling") +# 性别枚举（与腾讯云 SDK 一致：0=女, 1=男） +_GENDER_MAP = {0: "女", 1: "男", None: "未知"} + class TencentFaceCoordinator(DataUpdateCoordinator): - """腾讯云人脸识别数据更新协调器""" + """腾讯云人脸识别数据更新协调器。""" - def __init__(self, hass: HomeAssistant, entry: ConfigEntry, client): + def __init__(self, hass: HomeAssistant, entry: ConfigEntry, client) -> None: super().__init__( hass, _LOGGER_POLLING, name="Tencent Face Recognition", - update_interval=timedelta(minutes=5), + update_interval=timedelta(seconds=DEFAULT_SCAN_INTERVAL), ) self.client = client self.entry = entry + # 动态传感器跟踪集合 + self._entity_person_ids: set[str] = set() + self._person_sensors: dict[str, TencentFaceRecognitionPersonSensor] = {} - async def _async_update_data(self): - """获取最新数据""" + async def _async_update_data(self) -> Dict[str, Any]: + """获取最新的人员库与成员数据。""" try: group_id = get_entry_value( self.entry, CONF_PERSON_GROUP_ID, DEFAULT_PERSON_GROUP_ID @@ -55,118 +74,166 @@ class TencentFaceCoordinator(DataUpdateCoordinator): persons_result = await self.hass.async_add_executor_job( self.client.get_person_list_all, group_id ) - persons = persons_result.get("persons", []) if persons_result.get("success", False) else [] + if not persons_result.get("success", False): + raise UpdateFailed( + f"获取人员列表失败: {persons_result.get('error_message', '未知错误')}" + ) + persons = persons_result.get("persons", []) return { "group_info": group_info, "persons": persons, "person_count": len(persons), + "face_model_version": persons_result.get( + "face_model_version", + group_info.get("face_model_version", ""), + ), } - except Exception as ex: - _LOGGER_POLLING.debug("数据更新失败: %s", ex) + except UpdateFailed: + raise + except Exception as ex: # noqa: BLE001 + _LOGGER_POLLING.warning("数据更新失败: %s", ex) raise UpdateFailed(f"数据更新失败: {ex}") from ex async def async_setup_entry( hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback ) -> None: - """设置传感器实体""" + """设置传感器实体。""" _LOGGER.debug("设置腾讯云人脸识别传感器实体") client = hass.data[DOMAIN][entry.entry_id]["client"] name = entry.data.get(CONF_NAME, "腾讯云人脸识别") coordinator = TencentFaceCoordinator(hass, entry, client) + hass.data[DOMAIN][entry.entry_id]["coordinator"] = coordinator await coordinator.async_config_entry_first_refresh() - hass.data[DOMAIN][entry.entry_id]["coordinator"] = coordinator - - entities = [ + entities: list[SensorEntity] = [ TencentFaceRecognitionStatusSensor(coordinator, entry, name), ] - for person in coordinator.data.get("persons", []): - entities.append( - TencentFaceRecognitionPersonSensor(coordinator, entry, person) - ) + current_persons = {p["person_id"]: p for p in coordinator.data.get("persons", []) if p.get("person_id")} + for person in current_persons.values(): + entities.append(TencentFaceRecognitionPersonSensor(coordinator, entry, person)) async_add_entities(entities) + # 注册协调器更新监听，动态增减人员传感器 + @callback + def _async_update_person_sensors() -> None: + """根据协调器数据动态添加/移除人员传感器。""" + if coordinator.data is None: + return + + new_persons = { + p["person_id"]: p + for p in coordinator.data.get("persons", []) + if p.get("person_id") + } + + # 新增的人员 -> 创建传感器 + to_add = [ + TencentFaceRecognitionPersonSensor(coordinator, entry, person) + for pid, person in new_persons.items() + if pid not in coordinator._entity_person_ids + ] + if to_add: + async_add_entities(to_add) + for sensor in to_add: + coordinator._entity_person_ids.add(sensor._person_id) + # _person_sensors 已在传感器 __init__ 中注册 + + # 已删除的人员 -> 移除对应传感器 + for pid in list(coordinator._entity_person_ids): + if pid not in new_persons: + sensor = coordinator._person_sensors.get(pid) + if sensor is not None: + sensor.async_remove() + coordinator._entity_person_ids.discard(pid) + coordinator._person_sensors.pop(pid, None) + + # 补录初始已知人员 ID 集合（_person_sensors 已在传感器 __init__ 中填充） + coordinator._entity_person_ids = set(current_persons.keys()) + + entry.async_on_unload(coordinator.async_add_listener(_async_update_person_sensors)) + class TencentFaceRecognitionPersonSensor(CoordinatorEntity, SensorEntity): - """腾讯云人脸识别人员传感器""" + """腾讯云人脸识别人员传感器。 + + 状态为人员名称；当人员从库中被删除后，传感器变为不可用并自动移除。 + """ _attr_icon = "mdi:account" _attr_has_entity_name = True + _attr_translation_key = "person_sensor" - def __init__(self, coordinator, entry: ConfigEntry, person: Dict[str, Any]): - """初始化传感器""" + def __init__(self, coordinator, entry: ConfigEntry, person: Dict[str, Any]) -> None: super().__init__(coordinator) self._entry = entry - 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" + self._person_id: Optional[str] = person.get("person_id") + self._attr_name = person.get("person_name", "未知人员") + self._attr_unique_id = f"{entry.entry_id}_{SENSOR_PERSON}_{self._person_id}" + # 注册到协调器的传感器映射，供动态删除使用 + coordinator._person_sensors[self._person_id] = self - @property - def state(self) -> str: - """返回传感器状态""" + def _current_person(self) -> Optional[Dict[str, Any]]: if self.coordinator.data is None: - return "未知" + return None for person in self.coordinator.data.get("persons", []): if person.get("person_id") == self._person_id: - return person.get("person_name", "未知") - return "已删除" + return person + return None + + @property + def native_value(self) -> str: + """返回传感器状态（人员名称）。""" + person = self._current_person() + if person is None: + return "已删除" + return person.get("person_name", "未知") @property def extra_state_attributes(self) -> Dict[str, Any]: - """返回传感器属性""" - 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": "已删除"} + """返回传感器属性。""" + person = self._current_person() + if person is None: + return {"person_id": self._person_id, "status": "已删除"} + face_ids = person.get("face_ids", []) or [] + return { + "person_id": person.get("person_id"), + "gender": _GENDER_MAP.get(person.get("gender"), "未知"), + "face_ids": face_ids, + "face_count": len(face_ids), + "creation_timestamp": person.get("creation_timestamp"), + } @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 + """人员存在时才可用。""" + return self._current_person() is not None 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, coordinator, entry: ConfigEntry, name: str): - """初始化传感器""" + def __init__(self, coordinator, entry: ConfigEntry, name: str) -> None: super().__init__(coordinator) self._entry = entry self._name = name self._attr_name = f"{name} 状态" - self._attr_unique_id = f"{entry.entry_id}_status" + self._attr_unique_id = f"{entry.entry_id}_{SENSOR_STATUS}" @property - def state(self) -> str: - """返回传感器状态""" + def native_value(self) -> str: + """返回连接状态。""" if self.coordinator.data is None: return "未连接" if self.coordinator.last_update_success: @@ -175,30 +242,28 @@ class TencentFaceRecognitionStatusSensor(CoordinatorEntity, SensorEntity): @property def extra_state_attributes(self) -> Dict[str, Any]: - """返回传感器属性""" - attrs = { - "last_updated": ( - self.coordinator.data.get("last_updated") - if self.coordinator.data - else None - ), + """返回状态传感器属性。""" + data = self.coordinator.data or {} + group_info = data.get("group_info", {}) if isinstance(data.get("group_info"), dict) else {} + + attrs: Dict[str, Any] = { "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, + "person_count": data.get("person_count", 0), + "face_model_version": data.get("face_model_version", "") + or group_info.get("face_model_version", ""), + "last_exception": str(self.coordinator.last_exception) if self.coordinator.last_exception else None, + "last_update_success": self.coordinator.last_update_success, } - if self.coordinator.data and self.coordinator.data.get("group_info"): - group_info = self.coordinator.data["group_info"] + if group_info: + attrs["group_id"] = group_info.get("group_id", "") attrs["group_name"] = group_info.get("group_name", "") - attrs["face_model_version"] = group_info.get("face_model_version", "") + attrs["group_tag"] = group_info.get("group_tag", "") + attrs["creation_timestamp"] = group_info.get("creation_timestamp", 0) return attrs @property def available(self) -> bool: - """状态传感器始终可用""" + """状态传感器始终可用。""" return True diff --git a/services.py b/services.py index c32ceb5..5f0a82f 100644 --- a/services.py +++ b/services.py @@ -8,6 +8,7 @@ import voluptuous as vol from homeassistant.core import HomeAssistant, ServiceCall from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers import config_validation as cv +from homeassistant.helpers.service import SupportsResponse from .const import ( DOMAIN, @@ -125,7 +126,7 @@ async def async_setup_services(hass: HomeAssistant) -> None: SERVICE_FACE_SEARCH, async_face_search_service, schema=FACE_SEARCH_SCHEMA, - supports_response=True + supports_response=SupportsResponse.OPTIONAL, ) hass.services.async_register( @@ -133,7 +134,7 @@ async def async_setup_services(hass: HomeAssistant) -> None: SERVICE_DETECT_FACE, async_detect_face_service, schema=DETECT_FACE_SCHEMA, - supports_response=True + supports_response=SupportsResponse.OPTIONAL, ) hass.services.async_register( @@ -141,7 +142,7 @@ async def async_setup_services(hass: HomeAssistant) -> None: SERVICE_GET_FACE_ATTRIBUTES, async_get_face_attributes_service, schema=GET_FACE_ATTRIBUTES_SCHEMA, - supports_response=True + supports_response=SupportsResponse.OPTIONAL, ) hass.services.async_register( @@ -149,7 +150,7 @@ async def async_setup_services(hass: HomeAssistant) -> None: SERVICE_CREATE_PERSON, async_create_person_service, schema=CREATE_PERSON_SCHEMA, - supports_response=True + supports_response=SupportsResponse.OPTIONAL, ) hass.services.async_register( @@ -157,7 +158,7 @@ async def async_setup_services(hass: HomeAssistant) -> None: SERVICE_DELETE_PERSON, async_delete_person_service, schema=DELETE_PERSON_SCHEMA, - supports_response=True + supports_response=SupportsResponse.OPTIONAL, ) hass.services.async_register( @@ -165,7 +166,7 @@ async def async_setup_services(hass: HomeAssistant) -> None: SERVICE_CREATE_FACE, async_create_face_service, schema=CREATE_FACE_SCHEMA, - supports_response=True + supports_response=SupportsResponse.OPTIONAL, ) hass.services.async_register( @@ -173,7 +174,7 @@ async def async_setup_services(hass: HomeAssistant) -> None: SERVICE_DELETE_FACE, async_delete_face_service, schema=DELETE_FACE_SCHEMA, - supports_response=True + supports_response=SupportsResponse.OPTIONAL, ) @@ -274,7 +275,8 @@ async def async_face_search_service(call: ServiceCall) -> Dict[str, Any]: 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_FACE_ID: candidate.get("face_id"), + "gender": candidate.get("gender"), ATTR_GROUP_ID: group_id, ATTR_CAMERA_ENTITY_ID: camera_entity_id, }) diff --git a/services.yaml b/services.yaml index 6cf6f75..efdb703 100644 --- a/services.yaml +++ b/services.yaml @@ -84,6 +84,74 @@ face_search: - label: "是" value: "1" +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: 要检测的人脸图片文件（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: + name: 最大人脸数量 + description: 最多检测的人脸数量。 + default: 1 + example: 1 + selector: + number: + min: 1 + max: 10 + step: 1 + mode: box + min_face_size: + name: 最小人脸尺寸 + description: 最小人脸尺寸（像素）。 + default: 34 + example: 34 + selector: + number: + min: 1 + max: 200 + step: 1 + mode: box + need_rotate_check: + name: 旋转检查 + description: 是否进行旋转检查（0=否，1=是）。 + default: 1 + example: 1 + selector: + select: + options: + - label: "否" + value: "0" + - label: "是" + value: "1" + get_face_attributes: name: 获取人脸属性 description: 获取图片中人脸的属性信息。 diff --git a/tencent_cloud_client.py b/tencent_cloud_client.py index 9546b82..c4f9099 100644 --- a/tencent_cloud_client.py +++ b/tencent_cloud_client.py @@ -1,132 +1,91 @@ -"""腾讯云客户端""" +"""腾讯云人脸识别客户端（精简门面）。 + +原 1311 行的客户端已拆分为： +- ``retry.py``: 重试配置与执行器 +- ``image_processor.py``: 图片获取/压缩/编码 +- ``api_operations.py``: 各 API 操作的实现（搜索/检测/属性/人员/人脸/库信息/列表） + +本文件仅保留：凭据创建、HTTP Session 管理、错误脱敏、以及对外暴露的 +``TencentCloudClient`` 门面，委托给上述模块。 +""" + +from __future__ import annotations -import base64 -import io -import json import logging -import os -import random -import re +from typing import Any, Dict, List, Optional + import requests -import time -import uuid -from functools import lru_cache -from typing import Dict, Any, List, Optional, Union, Callable - from tencentcloud.common import credential -from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException -from tencentcloud.iai.v20200303 import iai_client, models - -from .const import ( - CONF_SECRET_ID, - CONF_SECRET_KEY, - CONF_REGION, +from tencentcloud.common.exception.tencent_cloud_sdk_exception import ( + TencentCloudSDKException, ) +from tencentcloud.iai.v20200303 import iai_client + +from . import api_operations from .errors import ( - handle_api_error, - ImageNotFoundError, - FaceNotDetectedError, + ApiError, + AuthenticationError, log_and_raise_error, validate_config, - NetworkError, - AuthenticationError, - RateLimitError, ) +from .image_processor import ImageCache, get_image_base64 +from .retry import RetryConfig +from .const import CONF_REGION, CONF_SECRET_ID, CONF_SECRET_KEY _LOGGER = logging.getLogger(__name__) -_LOGGER_POLLING = _LOGGER.getChild("polling") -_LOGGER_POLLING.setLevel(logging.DEBUG) - -ENABLE_REQUEST_LOGGING = False -ENABLE_RESPONSE_LOGGING = False -ENABLE_PERFORMANCE_LOGGING = True -LOG_REQUEST_PAYLOAD_MAX_SIZE = 1024 -LOG_RESPONSE_PAYLOAD_MAX_SIZE = 1024 - -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", -} - -RATE_LIMIT_ERRORS = { - "RequestLimitExceeded", - "RequestQuotaExceeded", - "LimitExceeded", -} - -NETWORK_ERRORS = { - "ResourceUnavailable.NetworkError", - "ResourceUnavailable.ServiceTimeout", -} - -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 - - -class RetryConfig: - """重试配置类""" - - def __init__( - self, - 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_attempts = max_attempts - self.initial_delay = initial_delay - self.max_delay = max_delay - self.backoff_factor = backoff_factor - self.jitter = jitter class TencentCloudClient: - """腾讯云客户端""" + """腾讯云人脸识别客户端门面。""" def __init__( self, secret_id: str, secret_key: str, region: str = "ap-shanghai", - retry_config: Optional[RetryConfig] = None - ): - config = { - "secret_id": secret_id, - "secret_key": secret_key, - "region": region - } - validate_config(config) + retry_config: Optional[RetryConfig] = None, + ) -> None: + validate_config({ + CONF_SECRET_ID: secret_id, + CONF_SECRET_KEY: secret_key, + CONF_REGION: region, + }) self.secret_id = secret_id self.secret_key = secret_key self.region = region self.retry_config = retry_config or RetryConfig() self.client = self._create_client() + + # HTTP 连接池化 self._session = requests.Session() - self._session.mount('http://', requests.adapters.HTTPAdapter(max_retries=2)) - self._session.mount('https://', requests.adapters.HTTPAdapter(max_retries=2)) + self._session.mount( + "http://", requests.adapters.HTTPAdapter(max_retries=2) + ) + self._session.mount( + "https://", requests.adapters.HTTPAdapter(max_retries=2) + ) + + # 图片缓存（url/path），通过 session 下载 + self._image_cache = ImageCache( + maxsize=32, + download_fn=lambda url: get_image_base64( + image_url=url, session=self._session + ), + read_fn=None, + ) + + # --- 资源管理 -------------------------------------------------------- def close(self) -> None: - """释放资源""" + """释放 HTTP Session 等资源。""" if self._session: self._session.close() + self._image_cache.clear() + + # --- 客户端创建 ------------------------------------------------------- def _create_client(self) -> iai_client.IaiClient: - """创建腾讯云人脸识别客户端""" try: _LOGGER.debug("创建腾讯云客户端，区域: %s", self.region) cred = credential.Credential(self.secret_id, self.secret_key) @@ -138,1174 +97,252 @@ class TencentCloudClient: log_and_raise_error( AuthenticationError, "创建腾讯云客户端失败", - {"region": self.region, "error_type": type(ex).__name__} + {"region": self.region, "error_type": type(ex).__name__}, ) + # --- 凭据验证 --------------------------------------------------------- + def verify_credentials(self) -> bool: - """验证凭据是否有效，通过调用 GetGroupList 测试""" + """验证凭据（轻量 GetGroupList）。失败抛出对应错误。""" + return api_operations.verify_credentials( + self.client, retry_config=self.retry_config + ) + + def verify_credentials_lightweight(self) -> bool: + """测试 API 连接（轻量验证）。失败返回 False，不抛异常。""" try: - req = models.GetGroupListRequest() - params = {"Limit": 1, "Offset": 0} - req.from_json_string(json.dumps(params)) - self.client.GetGroupList(req) - _LOGGER.info("凭据验证成功") + self.verify_credentials() return True - except TencentCloudSDKException as ex: - _LOGGER.error("凭据验证失败: %s", ex.code) - handle_api_error(ex.code, ex.message) + except Exception as ex: # noqa: BLE001 + _LOGGER.error("API连接测试失败: %s", self._sanitize_error(str(ex))) + return False - def _handle_error(self, ex: TencentCloudSDKException) -> None: - """处理API错误""" - _LOGGER.error( - "腾讯云API调用失败: 错误码=%s, 请求ID=%s", - ex.code, getattr(ex, 'request_id', '未知') - ) - handle_api_error(ex.code, ex.message) + def test_api_connection(self, group_id: str) -> bool: + """测试与指定人员库的连接。失败返回 False。""" + try: + _LOGGER.info("开始测试API连接: group_id=%s", group_id) + result = self.get_group_info(group_id) + if result.get("success", False): + _LOGGER.info("API连接测试成功") + return True + _LOGGER.error("API连接测试失败: %s", result.get("error_message")) + return False + except Exception as ex: # noqa: BLE001 + _LOGGER.error("API连接测试失败: %s", self._sanitize_error(str(ex))) + return False - 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 + # 兼容旧调用名 + _test_api_connection = test_api_connection - @lru_cache(maxsize=32) - def _cached_download_image_as_base64(self, image_url: str) -> str: - return self._download_image_as_base64_uncached(image_url) + # --- 图片 ------------------------------------------------------------ - @lru_cache(maxsize=32) - def _cached_read_local_image_as_base64(self, image_path: str) -> str: - return self._read_local_image_as_base64_uncached(image_path) - - 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 + def _get_image_base64( + self, + image_url: Optional[str] = None, + image_path: Optional[str] = None, + image_file: Optional[Any] = None, + ) -> str: + """获取图片 base64（带 url/path 缓存）。""" + if image_url: + return self._image_cache.get_url(image_url) + if image_path: + return self._image_cache.get_path(image_path) + if image_file: + return self._image_cache.get(image_file=image_file) + log_and_raise_error( + ApiError, + "必须提供 image_url、image_path 或 image_file", + {"reason": "no_image_source"}, ) - if not any([image_url, image_path, image_file]): - log_and_raise_error( - ValueError, - "必须提供image_url、image_path或image_file" - ) - - try: - if image_url: - _LOGGER.debug("使用图片URL: %s", image_url) - return self._cached_download_image_as_base64(image_url) - elif image_path: - _LOGGER.debug("使用本地图片路径: %s", image_path) - return self._cached_read_local_image_as_base64(image_path) - elif image_file: - _LOGGER.debug("使用上传的图片文件") - return self._process_uploaded_image_as_base64(image_file) - except Exception as ex: - _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_uncached(self, image_url: str) -> str: - """下载图片并转换为base64编码""" - try: - _LOGGER.debug("开始下载图片: %s", image_url) - - if not self._is_valid_url(image_url): - log_and_raise_error( - ValueError, - f"无效的图片URL: {image_url}", - {"url": image_url} - ) - - session = self._session - - headers = { - "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( - ValueError, - f"URL不是有效的图片: {image_url}, 内容类型: {content_type}", - {"url": image_url, "content_type": content_type} - ) - - image_data = self._process_image_data(response.content) - encoded_image = base64.b64encode(image_data).decode("utf-8") - - _LOGGER.debug("成功下载并编码图片，大小: %d 字节", len(encoded_image)) - return encoded_image - - except requests.exceptions.Timeout: - _LOGGER.error("下载图片超时: %s", image_url) - log_and_raise_error( - NetworkError, - f"下载图片超时: {image_url}", - {"url": image_url, "error": "timeout"} - ) - except requests.exceptions.ConnectionError: - _LOGGER.error("下载图片连接错误: %s", image_url) - log_and_raise_error( - NetworkError, - f"下载图片连接错误: {image_url}", - {"url": image_url, "error": "connection_error"} - ) - except requests.exceptions.RequestException as ex: - _LOGGER.error("下载图片失败: %s", image_url) - log_and_raise_error( - NetworkError, - f"下载图片失败: {image_url}", - {"url": image_url, "error": str(ex)} - ) - - def _read_local_image_as_base64_uncached(self, image_path: str) -> str: - """读取本地图片并转换为base64编码""" - try: - if not os.path.exists(image_path): - _LOGGER.error("图片文件不存在: %s", image_path) - log_and_raise_error( - ImageNotFoundError, - f"图片文件不存在: {image_path}", - {"path": image_path} - ) - - if not os.path.isfile(image_path): - _LOGGER.error("路径不是文件: %s", image_path) - log_and_raise_error( - ImageNotFoundError, - f"路径不是文件: {image_path}", - {"path": image_path} - ) - - file_ext = os.path.splitext(image_path)[1].lower() - if file_ext not in SUPPORTED_IMAGE_FORMATS: - _LOGGER.error("不支持的图片格式: %s", file_ext) - log_and_raise_error( - ImageNotFoundError, - f"不支持的图片格式: {file_ext}", - {"path": image_path, "supported_formats": SUPPORTED_IMAGE_FORMATS} - ) - - file_size = os.path.getsize(image_path) - if file_size > MAX_IMAGE_SIZE: - _LOGGER.error("图片文件过大: %s, 大小: %d 字节", image_path, file_size) - log_and_raise_error( - ImageNotFoundError, - f"图片文件过大: {image_path}", - {"path": image_path, "size": file_size, "max_size": f"{MAX_IMAGE_SIZE // (1024*1024)}MB"} - ) - - with open(image_path, "rb") as img_file: - image_data = img_file.read() - - processed_image = self._process_image_data(image_data) - encoded_image = base64.b64encode(processed_image).decode("utf-8") - - _LOGGER.debug("成功读取图片文件，大小: %d 字节", len(encoded_image)) - return encoded_image - - except PermissionError: - _LOGGER.error("没有权限读取图片文件: %s", image_path) - log_and_raise_error( - ImageNotFoundError, - f"没有权限读取图片文件: {image_path}", - {"path": image_path, "error": "permission_denied"} - ) - except ImageNotFoundError: - raise - except Exception as 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: - if isinstance(image_file, str): - if image_file.startswith('data:image'): - _LOGGER.debug("处理 data URL 格式图片") - try: - header, encoded = image_file.split(',', 1) - image_data = base64.b64decode(encoded) - processed_image = self._process_image_data(image_data) - return base64.b64encode(processed_image).decode("utf-8") - except ValueError: - log_and_raise_error( - ImageNotFoundError, - "无效的 data URL 格式", - {"data_url_length": len(image_file)} - ) - else: - _LOGGER.debug("使用 base64 编码图片，长度: %d", len(image_file)) - 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: - _LOGGER.error("base64解码失败: %s", ex) - log_and_raise_error( - ImageNotFoundError, - "base64解码失败", - {"error": str(ex), "data_length": len(image_file)} - ) - else: - _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, - "处理上传的图片文件失败", - {"error": str(ex), "error_type": type(ex).__name__} - ) + # --- 业务 API（委托 api_operations） ---------------------------------- def search_faces( self, - image_url: str = None, - image_path: str = None, - image_file: str = None, - image_base64: str = None, - group_ids: List[str] = None, + image_url: Optional[str] = None, + image_path: Optional[str] = None, + image_file: Optional[Any] = None, + image_base64: Optional[str] = None, + group_ids: Optional[List[str]] = None, max_face_num: int = 1, min_face_size: int = 34, max_user_num: int = 5, quality_control: int = 1, need_rotate_check: int = 1, - face_match_threshold: float = 60.0 + face_match_threshold: float = 60.0, ) -> Dict[str, Any]: - """搜索人脸""" - _LOGGER.info( - "开始搜索人脸: 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，搜索可能失败") - 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.SearchFacesRequest() - params = { - "NeedPersonInfo": 1, - "Image": image_base64, - "MaxFaceNum": max_face_num, - "MinFaceSize": min_face_size, - "MaxPersonNum": max_user_num, - "QualityControl": quality_control, - "NeedRotateCheck": need_rotate_check, - "FaceMatchThreshold": face_match_threshold, - "GroupIds": group_ids - } - - self._log_request("人脸搜索", params, 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)) - for result in resp.Results: - face_result = { - "face_id": getattr(result, "FaceId", None), - "candidates": [] - } - - if result.Candidates: - for candidate in result.Candidates: - candidate_info = { - "person_id": candidate.PersonId, - "person_name": candidate.PersonName, - "score": candidate.Score, - "person_tag": getattr(candidate, "PersonTag", None) - } - face_result["candidates"].append(candidate_info) - - results.append(face_result) - else: - _LOGGER.info("未检测到人脸") - - return { - "success": True, - "faces": results, - "error": None, - "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) - - if isinstance(ex, TencentCloudSDKException): - error_code = getattr(ex, "code", None) - error_message = getattr(ex, "message", str(ex)) - - return { - "success": False, - "faces": [], - "error": str(ex), - "error_code": error_code, - "error_message": error_message - } - - return self._execute_with_retry(api_call, "人脸搜索") + if not group_ids: + _LOGGER.warning("未提供人员库ID，搜索可能失败") + return api_operations.search_faces( + self.client, + image_base64=image_base64, + group_ids=group_ids or [], + max_face_num=max_face_num, + min_face_size=min_face_size, + max_user_num=max_user_num, + quality_control=quality_control, + need_rotate_check=need_rotate_check, + face_match_threshold=face_match_threshold, + retry_config=self.retry_config, + sanitize=self._sanitize_error, + ) def detect_faces( self, - image_url: str = None, - image_path: str = None, - image_file: str = None, - image_base64: str = None, + image_url: Optional[str] = None, + image_path: Optional[str] = None, + image_file: Optional[Any] = None, + image_base64: Optional[str] = None, max_face_num: int = 1, min_face_size: int = 34, - need_rotate_check: int = 1 + need_rotate_check: int = 1, ) -> Dict[str, Any]: - """检测人脸""" - _LOGGER.info( - "开始检测人脸: max_face_num=%d, min_face_size=%d, need_rotate_check=%d", - max_face_num, min_face_size, need_rotate_check - ) - 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.DetectFaceRequest() - params = { - "Image": image_base64, - "MaxFaceNum": max_face_num, - "MinFaceSize": min_face_size, - "NeedRotateCheck": need_rotate_check - } - - 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)) - for face_info in resp.FaceInfos: - face_result = { - "face_id": getattr(face_info, "FaceId", ""), - "x": getattr(face_info, "X", 0), - "y": getattr(face_info, "Y", 0), - "width": getattr(face_info, "Width", 0), - "height": getattr(face_info, "Height", 0), - "face_rect": getattr(face_info, "FaceRect", None).__dict__ if getattr(face_info, "FaceRect", None) else None - } - results.append(face_result) - else: - _LOGGER.info("未检测到人脸") - - return { - "success": True, - "faces": results, - "error": None, - "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) - - if isinstance(ex, TencentCloudSDKException): - error_code = getattr(ex, "code", None) - error_message = getattr(ex, "message", str(ex)) - - return { - "success": False, - "faces": [], - "error": str(ex), - "error_code": error_code, - "error_message": error_message - } - - return self._execute_with_retry(api_call, "人脸检测") + return api_operations.detect_faces( + self.client, + image_base64=image_base64, + max_face_num=max_face_num, + min_face_size=min_face_size, + need_rotate_check=need_rotate_check, + retry_config=self.retry_config, + sanitize=self._sanitize_error, + ) def get_face_attributes( self, - image_url: str = None, - image_path: str = None, - image_file: str = None, - image_base64: str = None, + image_url: Optional[str] = None, + image_path: Optional[str] = None, + image_file: Optional[Any] = None, + image_base64: Optional[str] = None, max_face_num: int = 1, - need_rotate_check: int = 1 + need_rotate_check: int = 1, ) -> Dict[str, Any]: - """获取人脸属性""" - _LOGGER.info( - "开始获取人脸属性: max_face_num=%d, need_rotate_check=%d", - max_face_num, need_rotate_check - ) - 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.DetectFaceRequest() - params = { - "Image": image_base64, - "MaxFaceNum": max_face_num, - "NeedRotateCheck": need_rotate_check, - "NeedFaceAttributes": 1, - "NeedQualityDetection": 1, - } - - 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)) - for face_info in resp.FaceInfos: - face_result = { - "face_id": getattr(face_info, "FaceId", ""), - "x": getattr(face_info, "X", 0), - "y": getattr(face_info, "Y", 0), - "width": getattr(face_info, "Width", 0), - "height": getattr(face_info, "Height", 0), - "gender": getattr(face_info, "Gender", None), - "age": getattr(face_info, "Age", None), - "expression": getattr(face_info, "Expression", None), - "beauty": getattr(face_info, "Beauty", None), - "face_rect": getattr(face_info, "FaceRect", None).__dict__ if getattr(face_info, "FaceRect", None) else None - } - results.append(face_result) - else: - _LOGGER.info("未检测到人脸") - - return { - "success": True, - "faces": results, - "error": None, - "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) - - if isinstance(ex, TencentCloudSDKException): - error_code = getattr(ex, "code", None) - error_message = getattr(ex, "message", str(ex)) - - return { - "success": False, - "faces": [], - "error": str(ex), - "error_code": error_code, - "error_message": error_message - } - - return self._execute_with_retry(api_call, "获取人脸属性") + return api_operations.get_face_attributes( + self.client, + image_base64=image_base64, + max_face_num=max_face_num, + need_rotate_check=need_rotate_check, + retry_config=self.retry_config, + sanitize=self._sanitize_error, + ) def create_person( self, person_id: str, person_name: str, group_id: str, - image_url: str = None, - image_path: str = None, - image_file: str = None, - image_base64: str = None, - gender: int = None, - person_tag: str = None, + image_url: Optional[str] = None, + image_path: Optional[str] = None, + image_file: Optional[Any] = None, + image_base64: Optional[str] = None, + gender: Optional[int] = None, + person_tag: Optional[str] = None, quality_control: int = 1, need_rotate_check: int = 1, ) -> Dict[str, Any]: - """创建人员""" - _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) - - error_code = None - error_message = str(ex) - - if isinstance(ex, TencentCloudSDKException): - error_code = getattr(ex, "code", None) - error_message = getattr(ex, "message", str(ex)) - - return { - "success": False, - "person_id": person_id, - "face_id": "", - "face_rect": "", - "error": str(ex), - "error_code": error_code, - "error_message": error_message - } - - return self._execute_with_retry(api_call, "创建人员") + return api_operations.create_person( + self.client, + person_id=person_id, + person_name=person_name, + group_id=group_id, + image_base64=image_base64, + gender=gender, + person_tag=person_tag, + quality_control=quality_control, + need_rotate_check=need_rotate_check, + retry_config=self.retry_config, + sanitize=self._sanitize_error, + ) def delete_person(self, person_id: str) -> Dict[str, Any]: - """删除人员""" - _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) - - error_code = None - error_message = str(ex) - - if isinstance(ex, TencentCloudSDKException): - error_code = getattr(ex, "code", None) - error_message = getattr(ex, "message", str(ex)) - - return { - "success": False, - "person_id": person_id, - "error": str(ex), - "error_code": error_code, - "error_message": error_message - } - - return self._execute_with_retry(api_call, "删除人员") + return api_operations.delete_person( + self.client, + person_id=person_id, + retry_config=self.retry_config, + sanitize=self._sanitize_error, + ) def create_face( self, person_id: str, - image_url: str = None, - image_path: str = None, - image_file: str = None, - image_base64: str = None, + image_url: Optional[str] = None, + image_path: Optional[str] = None, + image_file: Optional[Any] = None, + image_base64: Optional[str] = None, quality_control: int = 1, need_rotate_check: int = 1, ) -> Dict[str, Any]: - """为人员注册人脸""" - _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) - - error_code = None - error_message = str(ex) - - if isinstance(ex, TencentCloudSDKException): - error_code = getattr(ex, "code", None) - error_message = getattr(ex, "message", str(ex)) - - return { - "success": False, - "person_id": person_id, - "face_ids": [], - "face_rects": [], - "error": str(ex), - "error_code": error_code, - "error_message": error_message - } - - return self._execute_with_retry(api_call, "注册人脸") + return api_operations.create_face( + self.client, + person_id=person_id, + image_base64=image_base64, + quality_control=quality_control, + need_rotate_check=need_rotate_check, + retry_config=self.retry_config, + sanitize=self._sanitize_error, + ) def delete_face(self, person_id: str, face_id: str) -> Dict[str, Any]: - """删除人脸""" - _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) - - error_code = None - error_message = str(ex) - - if isinstance(ex, TencentCloudSDKException): - error_code = getattr(ex, "code", None) - error_message = getattr(ex, "message", str(ex)) - - return { - "success": False, - "person_id": person_id, - "face_id": face_id, - "error": str(ex), - "error_code": error_code, - "error_message": error_message - } - - return self._execute_with_retry(api_call, "删除人脸") + return api_operations.delete_face( + self.client, + person_id=person_id, + face_id=face_id, + retry_config=self.retry_config, + sanitize=self._sanitize_error, + ) def get_group_info(self, group_id: str) -> Dict[str, Any]: - """获取人员库信息""" - _LOGGER.info("开始获取人员库信息: group_id=%s", group_id) + return api_operations.get_group_info( + self.client, + group_id=group_id, + retry_config=self.retry_config, + sanitize=self._sanitize_error, + ) - def api_call(): - request_id = str(uuid.uuid4()) - start_time = time.time() - try: - req = models.GetGroupInfoRequest() - params = {"GroupId": group_id} - 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 { - "success": True, - "group_name": getattr(resp, "GroupName", ""), - "group_tag": getattr(resp, "GroupTag", ""), - "face_model_version": getattr(resp, "FaceModelVersion", ""), - "creation_timestamp": getattr(resp, "CreationTimestamp", 0), - } - except Exception as ex: - duration = time.time() - start_time - self._log_error("获取人员库信息", request_id, ex, duration) - - error_code = None - error_message = str(ex) - - if isinstance(ex, TencentCloudSDKException): - error_code = getattr(ex, "code", None) - error_message = getattr(ex, "message", str(ex)) - - return { - "success": False, - "group_name": "", - "group_tag": "", - "face_model_version": "", - "creation_timestamp": 0, - "error": str(ex), - "error_code": error_code, - "error_message": error_message - } - - 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: - _LOGGER.info("开始测试API连接: group_id=%s", group_id) - self.get_group_info(group_id) - _LOGGER.info("API连接测试成功") - return True - except Exception as ex: - _LOGGER.error("API连接测试失败: %s", self._sanitize_error(str(ex))) - return False - - def get_person_list(self, group_id: str, limit: int = 100, offset: int = 0) -> Dict[str, Any]: - """获取人员列表""" - _LOGGER.info("开始获取人员列表: group_id=%s, limit=%d, offset=%d", group_id, limit, offset) - - def api_call(): - request_id = str(uuid.uuid4()) - start_time = time.time() - try: - req = models.GetPersonListRequest() - params = { - "GroupId": group_id, - "Limit": limit, - "Offset": offset - } - self._log_request("获取人员列表", params, 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: - persons.append({ - "person_name": getattr(person_info, "PersonName", ""), - "person_id": getattr(person_info, "PersonId", ""), - "gender": getattr(person_info, "Gender", 0), - "face_ids": getattr(person_info, "FaceIds", []), - }) - return { - "success": True, - "persons": persons, - "error": None, - "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) - - if isinstance(ex, TencentCloudSDKException): - error_code = getattr(ex, "code", None) - error_message = getattr(ex, "message", str(ex)) - - return { - "success": False, - "persons": [], - "error": str(ex), - "error_code": error_code, - "error_message": error_message - } - - return self._execute_with_retry(api_call, "获取人员列表") + def get_person_list( + self, group_id: str, limit: int = 100, offset: int = 0 + ) -> Dict[str, Any]: + return api_operations.get_person_list( + self.client, + group_id=group_id, + limit=limit, + offset=offset, + retry_config=self.retry_config, + sanitize=self._sanitize_error, + ) def get_person_list_all(self, group_id: str) -> Dict[str, Any]: - """获取人员列表（自动分页获取全部）""" - all_persons = [] - offset = 0 - limit = 100 - while True: - result = self.get_person_list(group_id, limit=limit, offset=offset) - if not result.get("success", False): - return result - persons = result.get("persons", []) - all_persons.extend(persons) - if len(persons) < limit: - break - offset += limit - return { - "success": True, - "persons": all_persons, - "error": None, - "error_code": None, - "error_message": None - } - - def _execute_with_retry(self, api_call: Callable, operation_name: str) -> Any: - """执行带重试的API调用，max_attempts 为总尝试次数（1次初始 + N次重试）""" - last_exception = None - - 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, operation_name, delay) - time.sleep(delay) - - return api_call() - - except TencentCloudSDKException as ex: - last_exception = 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_attempts, ex) - - if last_exception: - if isinstance(last_exception, TencentCloudSDKException): - self._handle_error(last_exception) - else: - _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) + return api_operations.get_person_list_all( + self.client, + group_id=group_id, + retry_config=self.retry_config, + sanitize=self._sanitize_error, ) - 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?://' - 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) - - try: - from PIL import Image - - 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) - - if img.format != 'JPEG': - _LOGGER.debug("转换图片格式为JPEG: %s", img.format) - output = io.BytesIO() - if img.mode == 'RGBA': - img = img.convert('RGB') - img.save(output, format='JPEG', quality=IMAGE_COMPRESSION_QUALITY) - image_data = output.getvalue() - - except ImportError: - _LOGGER.warning("PIL库(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) - return image_data - - def _compress_image(self, image_data: bytes) -> bytes: - """压缩图片数据""" - try: - from PIL import Image - - img = Image.open(io.BytesIO(image_data)) - - original_size = len(image_data) - 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库(Pillow)未安装，无法进行图片压缩。建议安装: pip install Pillow>=10.0.0") - return image_data - except Exception as ex: - _LOGGER.error("图片压缩失败: %s", ex) - return image_data - - def _log_request(self, operation: str, params: Dict[str, Any], request_id: str) -> None: - """记录API请求日志""" - if not ENABLE_REQUEST_LOGGING: - return - - _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 _sanitize_error(self, error_msg: str) -> str: + """脱敏错误信息中的凭据。""" + if not isinstance(error_msg, str): + return str(error_msg) + if self.secret_id and self.secret_id in error_msg: + error_msg = error_msg.replace( + self.secret_id, f"{self.secret_id[:8]}***" ) + if self.secret_key and self.secret_key in error_msg: + error_msg = error_msg.replace(self.secret_key, "***") + return error_msg - def _log_response(self, operation: str, request_id: str, response: Any, duration: float) -> None: - """记录API响应日志""" - if not ENABLE_RESPONSE_LOGGING: - return - - _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( - "API响应缓慢: operation=%s, request_id=%s, duration=%.3fs", - operation, request_id, duration - ) - elif duration > 2.0: - _LOGGER.info( - "API响应较慢: operation=%s, request_id=%s, duration=%.3fs", - operation, request_id, duration - ) - - try: - if hasattr(response, "__dict__"): - response_dict = {} - for attr in dir(response): - if not attr.startswith("_") and not callable(getattr(response, attr)): - response_dict[attr] = getattr(response, attr) - - response_str = json.dumps(response_dict, ensure_ascii=False, default=str) - else: - response_str = str(response) - - if len(response_str) > LOG_RESPONSE_PAYLOAD_MAX_SIZE: - response_str = response_str[:LOG_RESPONSE_PAYLOAD_MAX_SIZE] + "...[truncated]" - - _LOGGER.debug( - "API响应详情: operation=%s, request_id=%s, response=%s", - operation, request_id, response_str - ) - except Exception as ex: - _LOGGER.debug( - "无法记录API响应详情: operation=%s, request_id=%s, error=%s", - operation, request_id, ex - ) - - def _log_error(self, operation: str, request_id: str, error: Exception, duration: float) -> None: - """记录API错误日志""" - error_type = type(error).__name__ - error_msg = self._sanitize_error(str(error)) - - _LOGGER.error( - "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", - 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 - ) + # 兼容旧代码：暴露 TencentCloudSDKException 别名 + @staticmethod + def is_sdk_exception(ex: Exception) -> bool: + return isinstance(ex, TencentCloudSDKException)