# coding: utf-8
#
# Copyright 2026 祁筱欣
# SPDX-License-Identifier: Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import base64
import json
import os
from typing import Type
from huaweicloudsdkcore.exceptions import exceptions
from huaweicloudsdkmoderation.v3 import *
from huaweicloudsdkmoderation.v3.region.moderation_region import ModerationRegion
from .client import HuaweiCloudClient, ClientClass, RegionClass
class ModerationClientManager(HuaweiCloudClient):
"""内容审核客户端管理器"""
def get_client_class(self) -> Type[ClientClass]:
return ModerationClient
def get_region_class(self) -> Type[RegionClass]:
return ModerationRegion
def get_cache_key(self) -> str:
return "moderation"
# 全局实例
_moderation_manager = ModerationClientManager()
async def check_image_moderation(
url: str = None,
image_base64: str = None,
image_file_path: str = None,
) -> dict:
"""使用华为云内容审核服务检查图片是否包含违规内容。
Args:
url (str, optional): 图片的 URL 地址（需可公开访问）。
image_base64 (str, optional): 图片的 Base64 编码。
image_file_path (str, optional): 本地图片文件路径。
Returns:
dict: 内容审核结果，包含审核结果和相关信息。
Note:
url、image_base64 和 image_file_path 参数三选一。
使用默认的审核配置，无需额外参数。
"""
# 参数验证
provided_count = sum([
1 for param in [url, image_base64, image_file_path]
if param is not None
])
if provided_count == 0:
return {"error": "必须提供 url、image_base64 或 image_file_path 参数之一"}
if provided_count > 1:
return {"error": "url、image_base64 和 image_file_path 参数只能提供一个"}
# 安全验证：image_file_path 路径限制
if image_file_path:
# 规范化路径，防止路径遍历攻击
normalized_path = os.path.normpath(image_file_path)
# 检查是否为绝对路径
if os.path.isabs(normalized_path):
return {"error": "不支持使用绝对路径，请使用相对路径"}
# 检查是否包含路径遍历字符
if ".." in normalized_path or normalized_path.startswith(("/", "~")):
return {"error": "路径包含非法字符或路径遍历尝试"}
# 限制允许的文件扩展名
allowed_extensions = {'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.webp'}
file_ext = os.path.splitext(normalized_path)[1].lower()
if file_ext not in allowed_extensions:
return {"error": f"不支持的文件格式: {file_ext}，仅支持图片文件"}
# 检查文件大小（限制为 10MB）
try:
file_size = os.path.getsize(normalized_path)
if file_size > 10 * 1024 * 1024: # 10MB
return {"error": f"文件过大: {file_size / 1024 / 1024:.2f}MB，最大支持 10MB"}
except OSError:
return {"error": f"无法访问文件: {image_file_path}"}
try:
client = _moderation_manager.get_client()
request = CheckImageModerationRequest()
# 根据输入类型设置请求参数
if url:
request.body = ImageDetectionReq(url=url.strip())
elif image_base64:
request.body = ImageDetectionReq(image=image_base64)
elif image_file_path:
# 读取本地文件并转换为 Base64
try:
with open(image_file_path, "rb") as f:
image_data = f.read()
image_base64_data = base64.b64encode(image_data).decode("utf-8")
request.body = ImageDetectionReq(image=image_base64_data)
except FileNotFoundError:
return {"error": f"文件未找到: {image_file_path}"}
except IOError as e:
return {"error": f"读取文件失败: {str(e)}"}
response = client.check_image_moderation(request) # type: ignore
# 将 Unicode 转义序列解码为实际字符
result = response.to_json_object()
return json.loads(json.dumps(result, ensure_ascii=False))
except exceptions.ClientRequestException as e:
return {
"error": {
"status_code": e.status_code,
"request_id": e.request_id,
"error_code": e.error_code,
"error_msg": e.error_msg,
}
}
