文档

视频审核

更新时间:

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

功能描述

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

  • 同步检测只支持传递视频的截帧图片序列。关于参数的详细介绍,请参见同步检测

  • (推荐)异步检测支持对原始视频或视频的截帧图片序列进行检测。关于参数的详细介绍,请参见异步检测

前提条件

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

接口

描述

支持的地域

VideoAsyncScanRequest

提交视频异步检测任务,对视频进行多个风险场景的识别,包括色情、暴恐涉政、广告 、不良场景、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 VideoAsyncScanRequest
    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")
    # 每次请求时需要新建request,请勿复用request对象。
    request = VideoAsyncScanRequest.VideoAsyncScanRequest()
    request.set_accept_format('JSON')
    
    task = {"dataId": str(uuid.uuid1()),
             "url": "https://www.aliyundoc.com/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
    # 以下代码将调用视频异步检测接口。
    import sys
    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')
    
    # 注意:此处实例化的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 = 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
    # 以下代码将调用视频异步检测接口。
    import sys
    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')
    
    # 注意:此处实例化的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 = 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
    
    # 注意:此处实例化的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 = VideoAsyncScanRequest.VideoAsyncScanRequest()
    request.set_accept_format('JSON')
    
    # 请将url替换为您的实时直播流地址。
    task = {
            "dataId": str(uuid.uuid1()),
            "url": "http://www.aliyundoc.com/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
    
    # 注意:此处实例化的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 = VideoAsyncScanRequest.VideoAsyncScanRequest()
    request.set_accept_format('JSON')
    
    # 请将url替换为您的实时直播流地址。
    task = {
            "dataId": str(uuid.uuid1()),
            "url": "http://www.aliyundoc.com/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"])

查询视频异步检测结果

接口

描述

支持的地域

VideoAsyncScanResultsRequest

查询视频异步检测任务的结果。

说明

该方法需要轮询结果,建议使用callback的方式获取结果。

  • cn-shanghai:华东2(上海)

  • cn-beijing:华北2(北京)

  • cn-shenzhen:华南1(深圳)

  • ap-southeast-1:新加坡

示例代码

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

# 注意:此处实例化的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 = 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'])

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

接口

描述

支持的地域

VideoSyncScanRequest

提交视频同步检测任务,同步检测视频中的风险内容。

说明

同步检测只支持传递视频帧序列,不支持检测视频文件,推荐使用异步检测接口。

  • cn-shanghai:华东2(上海)

  • cn-beijing:华北2(北京)

  • cn-shenzhen:华南1(深圳)

  • ap-southeast-1:新加坡

示例代码

#coding=utf-8
# 以下代码将调用视频同步检测接口,仅支持传入视频截帧序列进行检测。
from aliyunsdkcore import client
from aliyunsdkgreen.request.v20180509 import VideoSyncScanRequest
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")
# 每次请求时需要新建request,请勿复用request对象。
request = VideoSyncScanRequest.VideoSyncScanRequest()
request.set_accept_format('JSON')

task = {
            "frames":[
                {"offset" : 0, "url" : "https://example.com/test1.jpg"},
                {"offset" : 2, "url" : "https://example.com/test2.jpg"},
                {"offset" : 3, "url" : "https://example.com/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'])

视频检测结果反馈

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

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

接口

描述

支持的Region

VideoFeedbackRequest

提交视频检测结果的反馈,以人工反馈的检测结果纠正算法检测结果。

  • 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 VideoFeedbackRequest
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 = VideoFeedbackRequest.VideoFeedbackRequest()
request.set_accept_format('JSON')
# scenes:检测场景,支持指定多个场景。
# suggestion:期望的检测结果,pass:正常;block:违规。
request.set_content(
    json.dumps({"dataId": "检测数据ID", "taskId": "视频审核任务ID", "url": "视频链接",
                "suggestion": "block", "frames": [{"url": "视频截帧链接_1", "offset": "视频截帧offset_1"}],
                "scenes": ["ad", "terrorism"], "note": "备注信息"}))

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)
  • 本页导读 (0)