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)
            if response.status_code == 200:
                # 将响应体转换为字典格式
                result_dict = response.body.to_map()
                logger.info('response success. result:{}'.format(result_dict))
                
                if result_dict['Code'] == 200:
                    risk_level = result_dict['Data']['RiskLevel']
                    if risk_level in ["high", "medium"]:
                        return False, result_dict
                    return True, result_dict
                else:
                    logger.error('Business error. code:{}, message:{}'.format(
                        result_dict['Code'], result_dict['Message']))
                    return False, result_dict
            else:
                error_msg = 'API request failed. status:{}'.format(response.status_code)
                logger.error(error_msg)
                return False, {'error': error_msg}
                
        except Exception as err:
            logger.error(f"Error in security check: {str(err)}")
            return False, {'error': str(err)}
if __name__ == '__main__':
    is_safe, result = Security.security("测试安全边界")
    print(f"安全状态: {is_safe}")
    print(f"详细结果: {result}")