如何使用PythonSDK图片审核接口_内容安全(Content Moderation)-阿里云帮助中心

本文介绍了如何使用Python SDK图片审核接口，检测图片中是否包含风险内容。

功能描述

图片审核支持同步检测和异步检测两种方式。

同步检测实时返回检测结果。关于参数的详细信息，请参见同步检测。
异步检测需要您轮询结果或者通过callback回调通知获取检测结果。关于参数的详细信息，请参见异步检测。

前提条件

安装Python依赖。关于安装Python依赖的具体操作，请参见安装Python依赖。
说明
请一定按照安装Python依赖页面中的版本安装，否则会导致调用失败。
下载并在项目工程中引入Extension.Uploader工具类。

（推荐）同步图片检测

接口	描述	支持的地域
ImageSyncScanRequest	提交图片同步检测任务，对图片进行多个风险场景的识别，包括色情、暴恐涉政、广告、二维码、不良场景、Logo（商标台标）识别。	cn-shanghai：华东2（上海） cn-beijing：华北2（北京） cn-shenzhen：华南1（深圳） ap-southeast-1：新加坡

示例代码

传图片URL进行检测

#coding=utf-8
# 以下代码用于调用图片同步检测接口并实时返回检测结果。
from aliyunsdkcore import client
from aliyunsdkgreen.request.v20180509 import ImageSyncScanRequest
import json
import uuid

# 注意：此处实例化的client尽可能重复使用，提升检测性能。避免重复建立连接。
# 常见获取环境变量方式：
# 获取RAM用户AccessKey ID：os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID']
# 获取RAM用户AccessKey Secret：os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
clt = client.AcsClient("建议从环境变量中获取RAM用户AccessKey ID", "建议从环境变量中获取RAM用户AccessKey Secret", "cn-shanghai")
# 每次请求时需要新建request，请勿复用request对象。
request = ImageSyncScanRequest.ImageSyncScanRequest()
request.set_accept_format('JSON')
task = {
    "dataId": str(uuid.uuid1()),
    "url": "https://www.aliyun.com/test1.jpg"
}

# 设置待检测的图片，一张图片对应一个检测任务。
# 多张图片同时检测时，处理时间由最后一张处理完的图片决定。
# 通常情况下批量检测的平均响应时间比单任务检测长，一次批量提交的图片数越多，响应时间被拉长的概率越高。
# 代码中以单张图片检测作为示例，如果需要批量检测多张图片，请自行构建多个任务。
# 一次请求中可以检测多张图片，每张图片支持检测多个风险场景，计费按照单图片单场景检测叠加计算。
# 例如，检测2张图片，场景传递porn、terrorism，则计费按照2张图片鉴黄和2张图片暴恐检测计算。
request.set_content(json.dumps({"tasks": [task], "scenes": ["porn"]}))
response = clt.do_action_with_exception(request)
print(response)
result = json.loads(response)
if 200 == result["code"]:
    taskResults = result["data"]
    for taskResult in taskResults:
        if 200 == taskResult["code"]:
            sceneResults = taskResult["results"]
            for sceneResult in sceneResults:
                scene = sceneResult["scene"]
                suggestion = sceneResult["suggestion"]
                # 根据scene和suggestion设置后续操作。
                # 根据不同的suggestion结果做业务上的不同处理。例如，将违规数据删除等。
                print(suggestion)
                print(scene)

传本地图片文件进行检测

# coding=utf-8
# 以下代码用于调用图片同步检测接口并实时返回检测结果。
from aliyunsdkcore import client
from aliyunsdkgreen.request.v20180509 import ImageSyncScanRequest
from aliyunsdkgreenextension.request.extension import ClientUploader
from aliyunsdkgreenextension.request.extension import HttpContentHelper
import json
import uuid
import sys

# 设置编码规则，以便支持本地中文路径。
# Python2中添加以下内容，Python3无需添加。
if sys.version_info[0] == 2:
    reload(sys)
    sys.setdefaultencoding('utf-8')

# 注意：此处实例化的client尽可能重复使用，提升检测性能。避免重复建立连接。
# 常见获取环境变量方式：
# 获取RAM用户AccessKey ID：os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID']
# 获取RAM用户AccessKey Secret：os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
clt = client.AcsClient("建议从环境变量中获取RAM用户AccessKey ID", "建议从环境变量中获取RAM用户AccessKey Secret", "cn-shanghai")
# 每次请求时需要新建request，请勿复用request对象。
request = ImageSyncScanRequest.ImageSyncScanRequest()
request.set_accept_format('JSON')

# 请修改成您的本地文件路径。
# 上传本地文件到服务端。
uploader = ClientUploader.getImageClientUploader(clt)
url = uploader.uploadFile('d:/test/test.jpg')

# 获取上传后的URL进行检测。
task = {"dataId": str(uuid.uuid1()),
         "url":url
        }

# 设置待检测的图片，一张图片对应一个检测任务。
# 多张图片同时检测时，处理时间由最后一张处理完的图片决定。
# 通常情况下批量检测的平均响应时间比单任务检测长，一次批量提交的图片数越多，响应时间被拉长的概率越高。
# 代码中以单张图片检测作为示例，如果需要批量检测多张图片，请自行构建多个任务。
# 一次请求中可以检测多张图片，每张图片支持检测多个风险场景，计费按照单图片单场景检测叠加计算。
# 例如，检测2张图片，场景传递porn、terrorism，则计费按照2张图片鉴黄和2张图片暴恐检测计算。
request.set_content(HttpContentHelper.toValue({"tasks": [task], "scenes": ["porn"]}))
response = clt.do_action_with_exception(request)
print(response)
result = json.loads(response)
if 200 == result["code"]:
    taskResults = result["data"]
    for taskResult in taskResults:
        if (200 == taskResult["code"]):
            sceneResults = taskResult["results"]
            for sceneResult in sceneResults:
                # 根据scene和suggestion设置后续操作。
                # 根据不同的suggestion结果做业务上的不同处理。例如，将违规数据删除等。
                scene = sceneResult["scene"]
                suggestion = sceneResult["suggestion"]

传图片二进制内容进行检测

# coding=utf-8
# 以下代码用于调用图片同步检测接口并实时返回检测结果。
from aliyunsdkcore import client
from aliyunsdkgreen.request.v20180509 import ImageSyncScanRequest
from aliyunsdkgreenextension.request.extension import ClientUploader
from aliyunsdkgreenextension.request.extension import HttpContentHelper
import json
import uuid
import sys

# 设置编码规则，以便支持本地中文路径。
# Python2中添加以下内容，Python3无需添加。
if sys.version_info[0] == 2:
    reload(sys)
    sys.setdefaultencoding('utf-8')

# 注意：此处实例化的client尽可能重复使用，提升检测性能。避免重复建立连接。
# 常见获取环境变量方式：
# 获取RAM用户AccessKey ID：os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID']
# 获取RAM用户AccessKey Secret：os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
clt = client.AcsClient("建议从环境变量中获取RAM用户AccessKey ID", "建议从环境变量中获取RAM用户AccessKey Secret", "cn-shanghai")
# 每次请求时需要新建request，请勿复用request对象。
request = ImageSyncScanRequest.ImageSyncScanRequest()
request.set_accept_format('JSON')

# 读取本地文件作为二进制数据，模拟二进制数据检测。
# 请修改成您的本地文件路径。
f = open('d:/test/test.jpg',"rb+")
imageBytes = f.read()
f.close()

# 上传二进制文件到服务端。
uploader = ClientUploader.getImageClientUploader(clt)
url = uploader.uploadBytes(imageBytes)

# 取上传后的URL进行检测。
task = {
    "dataId": str(uuid.uuid1()),
    "url":url
}

# 设置待检测的图片，一张图片对应一个检测任务。
# 多张图片同时检测时，处理时间由最后一张处理完的图片决定。
# 通常情况下批量检测的平均响应时间比单任务检测长，一次批量提交的图片数越多，响应时间被拉长的概率越高。
# 代码中以单张图片检测作为示例，如果需要批量检测多张图片，请自行构建多个任务。
# 一次请求中可以检测多张图片，每张图片支持检测多个风险场景，计费按照单图片单场景检测叠加计算。
# 例如，检测2张图片，场景传递porn、terrorism，则计费按照2张图片鉴黄和2张图片暴恐检测计算。
request.set_content(HttpContentHelper.toValue({"tasks": [task], "scenes": ["porn"]}))
response = clt.do_action_with_exception(request)
print(response)
result = json.loads(response)
if 200 == result["code"]:
    taskResults = result["data"]
    for taskResult in taskResults:
        if (200 == taskResult["code"]):
            sceneResults = taskResult["results"]
            for sceneResult in sceneResults:
                # 根据scene和suggestion设置后续操作。
                # 根据不同的suggestion结果做业务上的不同处理。例如，将违规数据删除等。
                scene = sceneResult["scene"]
                suggestion = sceneResult["suggestion"]

提交图片异步检测任务

使用Python SDK对图片进行风险检测，通过异步请求提交检测任务，结果可以通过提交请求时设置callback回调通知获取，也可以通过接口轮询获取。

异步检测支持的输入内容与同步检测一致（本地文件、图片URL、图片二进制流），以下仅以图片URL作为输入示例。

接口	描述	支持的地域
ImageAsyncScanRequest	提交图片异步检测任务，对图片进行多个风险场景的识别，包括色情、暴恐涉政、广告、二维码、不良场景、Logo（商标台标）识别。	cn-shanghai：华东2（上海） cn-beijing：华北2（北京） cn-shenzhen：华南1（深圳） ap-southeast-1：新加坡

示例代码

# coding=utf-8
# 以下代码用于调用图片异步检测接口，您需要根据接口返回的taskId查询检测结果。
from aliyunsdkcore import client
from aliyunsdkcore.profile import region_provider
from aliyunsdkgreen.request.v20180509 import ImageAsyncScanRequest
from aliyunsdkgreenextension.request.extension import HttpContentHelper
import json
import uuid

# 注意：此处实例化的client尽可能重复使用，提升检测性能。避免重复建立连接。
# 常见获取环境变量方式：
# 获取RAM用户AccessKey ID：os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID']
# 获取RAM用户AccessKey Secret：os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
clt = client.AcsClient("建议从环境变量中获取RAM用户AccessKey ID", "建议从环境变量中获取RAM用户AccessKey Secret", "cn-shanghai")
region_provider.modify_point('Green', 'cn-shanghai', 'green.cn-shanghai.aliyuncs.com')
# 每次请求需要新建request，请勿复用request对象。
request = ImageAsyncScanRequest.ImageAsyncScanRequest()
request.set_accept_format('JSON')


task1 = {
    "dataId": str(uuid.uuid1()),
    "url":"http://example.com/xxx.jpg"
}
# 设置待检测的图片，一张图片对应一个检测任务。
# 多张图片同时检测时，处理时间由最后一张处理完的图片决定。
# 通常情况下批量检测的平均响应时间比单任务检测长，一次批量提交的图片数越多，响应时间被拉长的概率越高。
# 代码中以单张图片检测作为示例，如果需要批量检测多张图片，请自行构建多个任务。
# 一次请求中可以检测多张图片，每张图片支持检测多个风险场景，计费按照单图片单场景检测叠加计算。
# 例如，检测2张图片，场景传递porn、terrorism，则计费按照2张图片鉴黄和2张图片暴恐检测计算。
request.set_content(HttpContentHelper.toValue({"tasks": [task1], "scenes": ["porn"]}))
response = clt.do_action_with_exception(request)
print(response)
result = json.loads(response)
if 200 == result["code"]:
    taskResults = result["data"]
    for taskResult in taskResults:
        if(200 == taskResult["code"]):
           taskId = taskResult["taskId"]
           # 保存taskId。间隔一段时间后使用taskId查询检测结果。
           print(taskId)

查询异步检测结果

接口	描述	支持的地域
ImageAsyncScanResultsRequest	查询图片异步检测任务的结果。支持同时查询多个检测任务的返回结果。	cn-shanghai：华东2（上海） cn-beijing：华北2（北京） cn-shenzhen：华南1（深圳） ap-southeast-1：新加坡

示例代码

# coding=utf-8
from aliyunsdkcore import client
from aliyunsdkcore.profile import region_provider
from aliyunsdkgreen.request.v20180509 import ImageAsyncScanResultsRequest
from aliyunsdkgreenextension.request.extension import HttpContentHelper
import json 

# 注意：此处实例化的client尽可能重复使用，提升检测性能。避免重复建立连接。
# 常见获取环境变量方式：
# 获取RAM用户AccessKey ID：os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID']
# 获取RAM用户AccessKey Secret：os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
clt = client.AcsClient("建议从环境变量中获取RAM用户AccessKey ID", "建议从环境变量中获取RAM用户AccessKey Secret", "cn-shanghai")
region_provider.modify_point('Green', 'cn-shanghai', 'green.cn-shanghai.aliyuncs.com')
# 每次请求需要新建request，请勿复用request对象。
request = ImageAsyncScanResultsRequest.ImageAsyncScanResultsRequest()
request.set_accept_format('JSON')

# 通过taskId查询检测结果。
taskIds = ["img5sO$Zsssss7RYuz4Yyhhe-1q51iZ"]
request.set_content(HttpContentHelper.toValue(taskIds))
response = clt.do_action_with_exception(request)
print(response)
result = json.loads(response)
if 200 == result["code"]:
    taskResults = result["data"]
    for taskResult in taskResults:
        if (200 == taskResult["code"]):
            sceneResults = taskResult["results"]
            for sceneResult in sceneResults:
                # 根据scene和suggestion设置后续操作。
                # 根据不同的suggestion结果做业务上的不同处理。例如，将违规数据删除等。
                scene = sceneResult["scene"]
                suggestion = sceneResult["suggestion"]

图片检测结果反馈

如果您认为图片检测结果与您的预期不符，可以通过图片检测结果反馈接口，对检测结果进行纠正（系统会根据您反馈的结果，将图片添加到相似图片的黑名单库或者白名单库）。当您再次提交相似的内容进行检测时，以您反馈的label返回结果。

关于接口的说明，请参见检测结果反馈。

接口	描述	支持的Region
ImageScanFeedbackRequest	提交图片检测结果的反馈，以人工反馈的检测结果纠正算法检测结果。	cn-shanghai：华东2（上海） cn-beijing：华北2（北京） cn-shenzhen：华南1（深圳） ap-southeast-1：新加坡

示例代码

# coding=utf-8

from aliyunsdkcore import client
from aliyunsdkcore.profile import region_provider
from aliyunsdkgreen.request.v20180509 import ImageScanFeedbackRequest
import json

# 请使用您的AccessKey信息。
# 注意：此处实例化的client尽可能重复使用，提升检测性能。避免重复建立连接。
clt = client.AcsClient("您的AccessKey ID", "您的AccessKey Secret", 'cn-shanghai')
region_provider.modify_point('Green', 'cn-shanghai', 'green.cn-shanghai.aliyuncs.com')
request = ImageScanFeedbackRequest.ImageScanFeedbackRequest()
request.set_accept_format('JSON')
# scenes：检测场景，支持指定多个场景。
# suggestion：期望的检测结果，pass：正常；block：违规。
request.set_content(json.dumps({"suggestion": "block", "scenes": ["ad", "terrorism"], "url": "图片链接"}))

try:
    response = clt.do_action_with_exception(request)
    print(response)
    result = json.loads(response)
    if 200 == result["code"]:
        print("response success.")
except Exception as err:
    print(err)