ASJServer/Object/ContentSecurityObject.py

# @Author    : Rocky
# @File      : ContentSecurityObject.py
# @Time      : 2023/8/18 11:02
from alibabacloud_green20220302.client import Client as Green20220302Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_green20220302 import models as green_20220302_models
from alibabacloud_tea_util import models as util_models
from Ansjer.config import LOGGER
from django.conf import settings
ALICLOUD_AK = settings.ALICLOUD_AK
ALICLOUD_SK = settings.ALICLOUD_SK

TEXT_ILLEGAL_LABEL_LIST = ['political_content', 'profanity', 'contraband', 'sexual_content', 'violence',
                           'negative_content', 'religion', 'cyberbullying']
IMAGE_ILLEGAL_LABEL_LIST = ['pornographic_adultContent', 'sexual_suggestiveContent', 'political_politicalFigure_1',
                            'violent_blood', 'contraband_drug']


class ContentSecurity:
    """
    阿里云内容安全服务类
    """
    def __init__(self):
        pass

    @staticmethod
    def create_client() -> Green20220302Client:
        """
        使用AK&SK初始化账号Client
        @return: Client
        @throws Exception
        """
        config = open_api_models.Config(
            # 必填，您的 AccessKey ID,
            access_key_id=ALICLOUD_AK,
            # 必填，您的 AccessKey Secret,
            access_key_secret=ALICLOUD_SK
        )
        # Endpoint 请参考 https://api.aliyun.com/product/Green
        config.endpoint = f'green-cip.cn-shanghai.aliyuncs.com'
        return Green20220302Client(config)

    def text_review(self, service, service_parameters):
        """
        文本审核
        @param service: 审核服务类型
        @param service_parameters: 审核服务参数集
        @return: bool, True: 审核通过, False: 审核不通过
        """
        client = self.create_client()
        text_moderation_request = green_20220302_models.TextModerationRequest(service, service_parameters)
        runtime = util_models.RuntimeOptions()
        res = client.text_moderation_with_options(text_moderation_request, runtime)
        print(res)
        if res.status_code != 200:
            return False
        labels = res.body.data.labels
        labels_list = labels.split(',')
        for label in labels_list:
            if label in TEXT_ILLEGAL_LABEL_LIST:
                return False
            return True

    def image_review(self, service, service_parameters):
        """
        图片审核
        @param service: 审核服务类型
        @param service_parameters: 审核服务参数集
        @return: bool, True: 审核通过, False: 审核不通过
        """
        client = self.create_client()
        image_moderation_request = green_20220302_models.ImageModerationRequest(service, service_parameters)
        runtime = util_models.RuntimeOptions()
        res = client.image_moderation_with_options(image_moderation_request, runtime)
        print(res)
        LOGGER.info('图片审核响应内容:{}'.format(res))
        if res.status_code != 200:
            return False
        # 获取审核结果
        result = res.body
        if result.code != 200:
            return False
        result_list = result.data.result
        # 配到风险标签且置信分高于80时返回False
        for result in result_list:
            if result.label in IMAGE_ILLEGAL_LABEL_LIST:
                if result.confidence >= 80:
                    return False
        return True