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 = response.body
                logger.info('response success. result:{}'.format(result))
                if(result.code == 200){
                    if(result.Data.RiskLevel == "high" 
                        || result.Data.RiskLevel == "medium" ) {
                        return false,result
                    }
                }
                return true,result 
            else:
                print('response not success. status:{} ,result:{}'.format(response.status_code, response))
        except Exception as err:
            print(err)
if __name__ == '__main__':
    Security.security("测试安全边界")