Hermes
18a6a8ca9b
- 拆分 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 依赖
359 lines
12 KiB
Python
359 lines
12 KiB
Python
"""图片处理模块。
|
||
|
||
负责从 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()
|