视频审核_Python SDK_SDK参考_内容安全

背景信息

视频审核支持同步检测和异步检测两种方式。

同步检测只支持传入视频的截帧图片序列进行检测。
异步检测支持传入原始视频或视频的截帧图片序列进行检测。推荐您使用异步检测。

支持传入检测的视频内容媒介包括互联网视频URL、本地视频文件路径和二进制视频文件流。

准备工作

在进行具体的服务调用之前，请完成以下准备工作：

创建阿里云AccessKey。具体请参见创建AccessKey。
安装Python依赖。具体请参见安装Python依赖。

（推荐）提交视频异步检测任务


接口	描述	支持的Region
VideoAsyncScanRequest	提交视频异步检测任务，对视频进行多个风险场景的识别，包括色情、暴恐涉政、广告、不良场景、Logo（商标台标）识别。	cn-shanghai cn-beijing cn-shenzhen ap-southeast-1 us-west-1

示例代码

提交视频URL进行检测

#coding=utf-8
# 以下代码将调用视频异步检测接口。
from aliyunsdkcore import client
from aliyunsdkgreen.request.v20180509 import VideoAsyncScanRequest
from aliyunsdkgreenextension.request.extension import HttpContentHelper
import json
import uuid

# 请使用您自己的AccessKey信息。
clt = client.AcsClient("yourAccessKeyId", "yourAccessKeySecret","cn-shanghai")
# 每次请求时需要新建request，请勿复用request对象。
request = VideoAsyncScanRequest.VideoAsyncScanRequest()
request.set_accept_format('JSON')

task = {"dataId": str(uuid.uuid1()),
         "url": "https://xxxx/xxx.mp4"
        }
print(task)
# 设置待检测的视频。默认一次请求只支持检测一个视频。如果需要开放到多个，请通过工单联系我们。
# 一次请求中可以同时检测多个视频，每个视频可以同时检测多个风险场景，计费按照单视频截帧单场景检测叠加计算。
# 例如：检测2个视频，场景传入porn和terrorism，则计费按照2个视频的截帧总数*（鉴黄检测的单价+暴恐检测的单价）计算。
request.set_content(HttpContentHelper.toValue({"tasks": [task], "scenes": ["terrorism"]}))
response = clt.do_action_with_exception(request)
print(response)
result = json.loads(response)
if 200 == result["code"]:
    taskResults = result["data"]
    for taskResult in taskResults:
        # 保存返回的taskId，用于查询视频检测的结果。
        print(taskResult["taskId"])

提交本地视频文件进行检测

#coding=utf-8
# 以下代码将调用视频异步检测接口。
from aliyunsdkcore import client
from aliyunsdkgreen.request.v20180509 import VideoAsyncScanRequest
from aliyunsdkgreenextension.request.extension import HttpContentHelper
from aliyunsdkgreenextension.request.extension import ClientUploader
import json
import uuid

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

# 请使用您自己的AccessKey信息。
clt = client.AcsClient("yourAccessKeyId", "yourAccessKeySecret","cn-shanghai")
# 每次请求时需要新建request，请勿复用request对象。
request = VideoAsyncScanRequest.VideoAsyncScanRequest()
request.set_accept_format('JSON')

# 上传本地文件到服务端。请修改成您自己的本地文件路径。
uploader = ClientUploader.getVideoClientUploader(clt)
url = uploader.uploadFile('d:/暴恐涉政1.mp4')

task = {"dataId": str(uuid.uuid1()),
         "url": url
        }
print(task)
# 设置待检测的视频。默认一次请求只支持检测一个视频。如果需要开放到多个，请通过工单联系我们。
# 一次请求中可以同时检测多个视频，每个视频可以同时检测多个风险场景，计费按照单视频截帧单场景检测叠加计算。
# 例如：检测2个视频，场景传入porn,terrorism，则计费按照2个视频的截帧总数*（鉴黄检测的单价+暴恐检测的单价）计算。
request.set_content(HttpContentHelper.toValue({"tasks": [task], "scenes": ["terrorism"]}))
response = clt.do_action_with_exception(request)
print(response)
result = json.loads(response)
if 200 == result["code"]:
    taskResults = result["data"]
    for taskResult in taskResults:
        # 保存返回的taskId，用于查询视频检测的结果。
        print(taskResult["taskId"])

提交视频文件二进制文件进行检测

#coding=utf-8
# 以下代码将调用视频异步检测接口。
from aliyunsdkcore import client
from aliyunsdkgreen.request.v20180509 import VideoAsyncScanRequest
from aliyunsdkgreenextension.request.extension import HttpContentHelper
from aliyunsdkgreenextension.request.extension import ClientUploader
import json
import uuid

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

# 请使用您自己的AccessKey信息。
clt = client.AcsClient("yourAccessKeyId", "yourAccessKeySecret","cn-shanghai")
# 每次请求时需要新建request，请勿复用request对象。
request = VideoAsyncScanRequest.VideoAsyncScanRequest()
request.set_accept_format('JSON')
# 读取本地文件作为二进制数据，模拟二进制数据检测。
# 请修改成您自己的本地文件路径。
f = open('d:/暴恐涉政1.mp4', "rb+")
videoBytes = f.read()
f.close()

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

task = {"dataId": str(uuid.uuid1()),
         "url": url
        }
print(task)
# 设置待检测的视频。默认一次请求只支持检测一个视频。如果需要开放到多个，请通过工单联系我们。
# 一次请求中可以同时检测多个视频，每个视频可以同时检测多个风险场景，计费按照单视频截帧单场景检测叠加计算。
# 例如：检测2个视频，场景传入porn,terrorism，则计费按照2个视频的截帧总数*（鉴黄检测的单价+暴恐检测的单价）计算。
request.set_content(HttpContentHelper.toValue({"tasks": [task], "scenes": ["terrorism"]}))
response = clt.do_action_with_exception(request)
print(response)
result = json.loads(response)
if 200 == result["code"]:
    taskResults = result["data"]
    for taskResult in taskResults:
        # 保存返回的taskId，用于查询视频检测的结果。
        print(taskResult["taskId"])

提交视频直播流进行检测

#coding=utf-8
# 以下代码将调用视频异步检测接口。
from aliyunsdkcore import client
from aliyunsdkgreen.request.v20180509 import VideoAsyncScanRequest
from aliyunsdkgreenextension.request.extension import HttpContentHelper
import json
import uuid

# 请使用您自己的AccessKey信息。
clt = client.AcsClient("yourAccessKeyId", "yourAccessKeySecret","cn-shanghai")
# 每次请求时需要新建request，请勿复用request对象。
request = VideoAsyncScanRequest.VideoAsyncScanRequest()
request.set_accept_format('JSON')

# 请将url替换为您的实时直播流地址。
task = {
        "dataId": str(uuid.uuid1()),
        "url": "http://xxx/xxx.flv"
        }
print(task)
# 设置待检测的视频。默认一次请求只支持检测一个视频。如果需要开放到多个，请通过工单联系我们。
# 一次请求中可以同时检测多个视频，每个视频可以同时检测多个风险场景，计费按照单视频截帧单场景检测叠加计算。
# 例如：检测2个视频，场景传入porn,terrorism，则计费按照2个视频的截帧总数*（鉴黄检测的单价+暴恐检测的单价）计算。
request.set_content(HttpContentHelper.toValue({"tasks": [task], "scenes": ["terrorism"], "live": True}))
response = clt.do_action_with_exception(request)
print(response)
result = json.loads(response)
if 200 == result["code"]:
    taskResults = result["data"]
    for taskResult in taskResults:
        # 保存返回的taskId，用于查询视频检测的结果。
        print(taskResult["taskId"])

提交视频语音进行综合检测

#coding=utf-8
# 以下代码将调用视频异步检测接口。
from aliyunsdkcore import client
from aliyunsdkgreen.request.v20180509 import VideoAsyncScanRequest
from aliyunsdkgreenextension.request.extension import HttpContentHelper
import json
import uuid

# 请使用您自己的AccessKey信息。
clt = client.AcsClient("yourAccessKeyId", "yourAccessKeySecret","cn-shanghai")
# 每次请求时需要新建request，请勿复用request对象。
request = VideoAsyncScanRequest.VideoAsyncScanRequest()
request.set_accept_format('JSON')

# 请将url替换为您的实时直播流地址。
task = {
        "dataId": str(uuid.uuid1()),
        "url": "http://xxx/xxx.flv"
        }
print(task)
# 设置待检测的视频。默认一次请求只支持检测一个视频。如果需要开放到多个，请通过工单联系我们。
# 一次请求中可以同时检测多个视频，每个视频可以同时检测多个风险场景，计费按照单视频截帧单场景检测叠加计算。
# 例如：检测2个视频，场景传入porn,terrorism，则计费按照2个视频的截帧总数*（鉴黄检测的单价+暴恐检测的单价）计算。
# 如果同时检测视频画面和视频中的语音，视频中的画面按照上述示例计费，语音部分按照视频语音的总时长进行计费。
request.set_content(HttpContentHelper.toValue({"tasks": [task], "scenes": ["terrorism"], "live": True, "audioScenes": ["antispam"]}))
response = clt.do_action_with_exception(request)
print(response)
result = json.loads(response)
if 200 == result["code"]:
    taskResults = result["data"]
    for taskResult in taskResults:
        # 保存返回的taskId，用于查询视频检测的结果。
        print(taskResult["taskId"])

查询视频异步检测结果


接口	描述	支持的region
VideoAsyncScanResultsRequest	查询视频异步检测任务的结果。说明该方法需要轮询结果，建议使用callback的方式获取结果。	cn-shanghai cn-beijing cn-shenzhen ap-southeast-1 us-west-1

示例代码

#coding=utf-8
# 以下代码将调用视频异步检测结果查询接口。
import json
from aliyunsdkcore import client
from aliyunsdkgreen.request.v20180509 import VideoAsyncScanResultsRequest
from aliyunsdkgreenextension.request.extension import HttpContentHelper

# 请使用您自己的AccessKey信息。
clt = client.AcsClient("yourAccessKeyId", "yourAccessKeySecret","cn-shanghai")
# 每次请求时需要新建request，请勿复用request对象。
request = VideoAsyncScanResultsRequest.VideoAsyncScanResultsRequest()
request.set_accept_format('JSON')

# 传入需要查询的视频检测任务的taskId列表。
taskIds = ['vi3pX@vXC94hPnWsss39WOQ9-1q52ZG']
request.set_content(HttpContentHelper.toValue(taskIds))
response = clt.do_action_with_exception(request)
result = json.loads(response)
if 200 == result["code"]:
    taskResults = result["data"]
    for taskResult in taskResults:
        # 每个task的results中包含该视频的截帧检测结果。
        print(taskResult['results'])

提交视频截帧同步检测任务


接口	描述	支持的Region
VideoSyncScanRequest	提交视频同步检测任务，同步检测视频中的风险内容。说明同步检测只支持传递视频帧序列，不支持检测视频文件，推荐使用异步检测接口。	cn-shanghai cn-beijing cn-shenzhen ap-southeast-1 us-west-1

示例代码

#coding=utf-8
# 以下代码将调用视频同步检测接口，仅支持传入视频截帧序列进行检测。
from aliyunsdkcore import client
from aliyunsdkgreen.request.v20180509 import VideoSyncScanRequest
from aliyunsdkgreenextension.request.extension import HttpContentHelper
import json

# 请使用您自己的AccessKey信息。
clt = client.AcsClient("yourAccessKeyId", "yourAccessKeySecret","cn-shanghai")
# 每次请求时需要新建request，请勿复用request对象。
request = VideoSyncScanRequest.VideoSyncScanRequest()
request.set_accept_format('JSON')

task = {
            "frames":[
                {"offset" : 0, "url" : "https://test1.jpg"},
                {"offset" : 2, "url" : "https://test2.jpg"},
                {"offset" : 3, "url" : "https://test3.jpg"}
            ]
        }
print(task)
# 设置待检测的视频。默认一次请求只支持检测一个视频。如果需要开放到多个，请通过工单联系我们。
# 一次请求中可以同时检测多个视频，每个视频可以同时检测多个风险场景，计费按照单视频截帧单场景检测叠加计算。
# 例如：检测2个视频，场景传入porn,terrorism，则计费按照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:
        for result in taskResult["results"]:
            # 根据结果设置后续操作。
            print(result['suggestion'])
            print(result['scene'])