video-insight/utils/aliyun_security.py

# coding=utf-8
# python version >= 3.6
from alibabacloud_green20220302.client import Client
from alibabacloud_green20220302 import models
from alibabacloud_tea_openapi.models import Config
import json
from loguru import logger

config = Config(
    # 阿里云账号AccessKey拥有所有API的访问权限，建议您使用RAM用户进行API访问或日常运维。
    # 强烈建议不要把AccessKey ID和AccessKey Secret保存到工程代码里，否则可能导致AccessKey泄露，威胁您账号下所有资源的安全。
    # 常见获取环境变量方式：
    # contentsecurity@1894469520484605.onaliyun.com
    access_key_id='LTAI5tEv9SCKZFpTshJpi618',
    access_key_secret='7N13xskcYEJzgiRLcYO1ngEVFmTq6Z',
    # 连接超时时间 单位毫秒(ms)
    connect_timeout=10000,
    # 读超时时间 单位毫秒(ms)
    read_timeout=3000,
    region_id='cn-hangzhou',
    #endpoint='green-cip-vpc.cn-hangzhou.aliyuncs.com'
    endpoint='green-cip.cn-hangzhou.aliyuncs.com'
)

clt = Client(config)

class Security:
    @classmethod
    def security(cls, content):
        """
        内容安全审核方法
        返回: (是否安全, 结果字典/字符串)
        """
        serviceParameters = {
            'content': content
        }
        textModerationPlusRequest = models.TextModerationPlusRequest(
            service='llm_response_moderation_01',
            service_parameters=json.dumps(serviceParameters)
        )

        try:
            response = clt.text_moderation_plus(textModerationPlusRequest)
            
            # HTTP请求成功
            if response.status_code == 200:
                try:
                    # 将响应体转换为字典格式
                    result_dict = response.body.to_map()
                    logger.info(f'审核成功. 结果:{result_dict}')
                    
                    # 业务状态码检查
                    if result_dict.get('Code') == 200:
                        risk_level = result_dict.get('Data', {}).get('RiskLevel', '').lower()
                        
                        # 风险等级判断
                        if risk_level in ["high", "medium"]:
                            return False, json.dumps(result_dict, ensure_ascii=False)
                        return True, json.dumps(result_dict, ensure_ascii=False)
                    else:
                        # 业务错误
                        error_msg = f"业务错误. 代码:{result_dict.get('Code')}, 消息:{result_dict.get('Message', '无')}"
                        logger.error(error_msg)
                        return False, error_msg
                        
                except Exception as parse_error:
                    # 响应解析错误
                    error_msg = f"响应解析错误: {str(parse_error)}"
                    logger.error(error_msg)
                    return False, error_msg
                    
            else:
                # HTTP请求失败
                error_msg = f"API请求失败. 状态码:{response.status_code}, 响应:{str(response)}"
                logger.error(error_msg)
                return False, error_msg
                
        except Exception as e:
            # 全局异常处理
            error_msg = f"内容审核异常: {str(e)}"
            logger.exception(error_msg)
            return False, error_msg
if __name__ == '__main__':
    is_safe, result = Security.security("测试安全边界")
    print(f"安全状态: {is_safe}")
    print(f"详细结果: {result}")