使用Python SDK视频审核接口分别检测视频中的图像和语音是否包含风险内容。

  • 支持同步检测和异步检测两种方式:同步检测只支持传递视频的截帧图片序列;(推荐)异步检测支持对原始视频或视频的截帧图片序列进行检测。
  • 视频支持传入互联网视频URL、本地视频文件路径以及二进制视频文件流。

准备工作

在进行具体的服务调用之前,请参见以下步骤,完成准备工作:
  1. 创建阿里云AccessKeyId和AccessKeySecret。具体请参见创建AccessKey
  2. 安装Python依赖。具体请参见安装Python依赖

(推荐)提交视频异步检测任务

接口描述
接口 功能 支持的Region 描述
VideoAsyncScanRequest 视频异步风险检测。
  • cn-shanghai
  • cn-beijing
  • cn-shenzhen
  • ap-southeast-1
  • us-west-1
对视频进行多个风险场景的识别,包括色情、暴恐涉政、广告 、不良场景、Logo(商标台标)识别。
示例代码
  • 提交视频url进行检测
    #coding=utf-8
    #视频异步检测服务接口
    from aliyunsdkcore import client
    from aliyunsdkgreen.request.v20180509 import VideoAsyncScanRequest
    from aliyunsdkgreen.request.extension import HttpContentHelper
    import json
    import uuid
    
    # 请替换成您的accessKeyId、accessKeySecret
    clt = client.AcsClient("您的accessKeyId", "您的accessKeySecret", "cn-shanghai") 
    request = VideoAsyncScanRequest.VideoAsyncScanRequest()
    request.set_accept_format('JSON')
    
    task = {"dataId": str(uuid.uuid1()),
             "url": "https://xxxx/xxx.mp4"
            }
    print(task)
    # 设置待检测视频,默认一次请求只支持1个task,如果需要开放到更多,请通过工单联系我们。
    # 计费按照该处传递的场景进行
    # 一次请求中可以同时检测多个视频,每个视频可以同时检测多个风险场景,计费按照场景和视频截帧数计算
    # 例如:检测2个视频,场景传递porn,terrorism,计费会按照2个视频的截帧总数乘以鉴黄场景的价格加上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 aliyunsdkgreen.request.extension import HttpContentHelper
    from aliyunsdkgreen.request.extension import ClientUploader
    import json
    import uuid
    
    # 设置一下编码, 以支持本地中文路径
    # python2使用,python3中请去掉
    if sys.version_info[0] == 2:
        reload(sys)
        sys.setdefaultencoding('utf-8')
    
    # 请替换成您的accessKeyId、accessKeySecret
    clt = client.AcsClient("您的accessKeyId", "您的accessKeySecret", "cn-shanghai") 
    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)
    # 设置待检测视频,默认一次请求只支持1个task,如果需要开放到更多,请通过工单联系我们。
    # 计费按照该处传递的场景进行
    # 一次请求中可以同时检测多个视频,每个视频可以同时检测多个风险场景,计费按照场景和视频截帧数计算
    # 例如:检测2个视频,场景传递porn,terrorism,计费会按照2个视频的截帧总数乘以鉴黄场景的价格加上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 aliyunsdkgreen.request.extension import HttpContentHelper
    from aliyunsdkgreen.request.extension import ClientUploader
    import json
    import uuid
    
    # 设置一下编码, 以支持本地中文路径
    # python2使用,python3中请去掉
    if sys.version_info[0] == 2:
        reload(sys)
        sys.setdefaultencoding('utf-8')
    
    # 请替换成您的accessKeyId、accessKeySecret
    clt = client.AcsClient("您的accessKeyId", "您的accessKeySecret", "cn-shanghai")
    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)
    # 设置待检测视频,默认一次请求只支持1个task,如果需要开放到更多,请通过工单联系我们。
    # 计费按照该处传递的场景进行
    # 一次请求中可以同时检测多个视频,每个视频可以同时检测多个风险场景,计费按照场景和视频截帧数计算
    # 例如:检测2个视频,场景传递porn,terrorism,计费会按照2个视频的截帧总数乘以鉴黄场景的价格加上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 aliyunsdkgreen.request.extension import HttpContentHelper
    import json
    import uuid
    
    # 请替换成您的accessKeyId、accessKeySecret
    clt = client.AcsClient("您的accessKeyId", "您的accessKeySecret", "cn-shanghai")
    request = VideoAsyncScanRequest.VideoAsyncScanRequest()
    request.set_accept_format('JSON')
    
    # url请替换为您的实时直播流地址
    task = {
            "dataId": str(uuid.uuid1()),
            "url": "http://xxx/xxx.flv"
            }
    print(task)
    # 设置待检测视频,默认一次请求只支持1个task,如果需要开放到更多,请通过工单联系我们。
    # 计费按照该处传递的场景进行
    # 一次请求中可以同时检测多个视频,每个视频可以同时检测多个风险场景,计费按照场景和视频截帧数计算
    # 例如:检测2个视频,场景传递porn,terrorism,计费会按照2个视频的截帧总数乘以鉴黄场景的价格加上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 aliyunsdkgreen.request.extension import HttpContentHelper
    import json
    import uuid
    
    # 请替换成您的accessKeyId、accessKeySecret
    clt = client.AcsClient("您的accessKeyId", "您的accessKeySecret", "cn-shanghai")
    request = VideoAsyncScanRequest.VideoAsyncScanRequest()
    request.set_accept_format('JSON')
    
    # url请替换为您的实时直播流地址
    task = {
            "dataId": str(uuid.uuid1()),
            "url": "http://xxx/xxx.flv"
            }
    print(task)
    # 设置待检测视频,默认一次请求只支持1个task,如果需要开放到更多,请通过工单联系我们。
    # 计费按照该处传递的场景进行
    # 一次请求中可以同时检测多个视频,每个视频可以同时检测多个风险场景,计费按照场景和视频截帧数计算
    # 例如:检测2个视频,场景传递porn,terrorism,计费会按照2个视频的截帧总数乘以鉴黄场景的价格加上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 查询视频风险异步检测任务结果。
  • cn-shanghai
  • cn-beijing
  • cn-shenzhen
  • ap-southeast-1
  • us-west-1
查询视频检测的结果,该方法需要轮询结果,建议使用callback的方式获取结果。
示例代码
#coding=utf-8
#视频异步检测服务接口
import json
from aliyunsdkcore import client
from aliyunsdkgreen.request.v20180509 import VideoAsyncScanResultsRequest
from aliyunsdkgreen.request.extension import HttpContentHelper

# 请替换成您自己的accessKeyId、accessKeySecret。您可以修改配置文件,也可以直接明文替换
clt = client.AcsClient("您的accessKeyId", "您的accessKeySecret",'cn-shanghai')
request = VideoAsyncScanResultsRequest.VideoAsyncScanResultsRequest()
request.set_accept_format('JSON')

# taskIds 是需要查询的视频检测任务id列表
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 aliyunsdkgreen.request.extension import HttpContentHelper
import json

# 请替换成您的accessKeyId、accessKeySecret
clt = client.AcsClient("您的accessKeyId", "您的accessKeySecret", "cn-shanghai")
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)
# 设置待检测视频,默认一次请求只支持1个task,如果需要开放到更多,请通过工单联系我们。
# 计费按照该处传递的场景进行
# 一次请求中可以同时检测多个视频,每个视频可以同时检测多个风险场景,计费按照场景和视频截帧数计算
# 例如:检测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:
        for result in taskResult["results"]:
            # 根据结果进行业务处理
            print(result['suggestion'])
            print(result['scene'])