AI Guardrails:Video moderation

Choose a moderation mode

Video moderation provides an asynchronous mode and a synchronous mode. The following table describes the input that each mode accepts and how each mode returns moderation results.

Usage notes

Before you call the video moderation operations of the Content Moderation Python SDK, review the following items:

• Videos per request — A single request can carry multiple videos, and each video can be checked against multiple risk scenes. By default, each request is limited to one video.

• Billing — Video moderation is billed by frame: the total number of frames from all videos in the request, multiplied by the sum of the unit prices of the scenes that you specify. For example, moderating two videos for both the porn scene and the terrorism scene costs (total frames from both videos) × (unit price for porn detection + unit price for terrorism detection). When the audio track is moderated together with the video frames, the audio track is billed separately based on its total duration.

• Credentials — Each sample reads the AccessKey pair of a RAM user from the ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variables.

• Client and request objects — Reuse the client instance across requests to improve performance and avoid repeated connection overhead. Create a new request object for each call, and do not reuse request objects.

• Region and endpoint — The samples in this topic create the client for the cn-shanghai region. The video moderation feedback sample also calls region_provider.modify_point to set the endpoint of the Green service in that region. For the regions that the video moderation feedback operation supports, see the table in the "Video moderation feedback" section.

• Python 2 compatibility — The samples that read a local video file set the default encoding to utf-8 when they run on Python 2, so that local paths containing non-ASCII characters are supported. This step is not required on Python 3.

• Failure handling — Each sample processes the response only when code in the response body is 200. Add a branch for other response codes, and wrap the call in a try...except block as shown in the video moderation feedback sample.

(Recommended) Submit an asynchronous video moderation task

Asynchronous video moderation submits one or more videos for detection and returns a taskId for each task. Moderation results are not returned in the response of the submission: save each taskId, and then retrieve the results as described in Query asynchronous video moderation results.

The following samples all call the asynchronous video detection operation. The first three samples differ only in how the video reaches the operation: as a URL, as a local file that the ClientUploader utility uploads, or as binary data that the utility uploads. The last two samples add request parameters that change what is moderated:

• live — Set live to True when the url in the task points to a live stream.

• audioScenes — Specify audioScenes to moderate the audio track in addition to the video frames. The combined moderation sample submits a live stream URL, so it also sets live to True.

  Sample code

• Submit a video URL for moderation

#coding=utf-8
# The following code calls the asynchronous video detection operation.
from aliyunsdkcore import client
from aliyunsdkgreen.request.v20180509 import VideoAsyncScanRequest
from aliyunsdkgreenextension.request.extension import HttpContentHelper
import json
import uuid
import os

clt = client.AcsClient(os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'], os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET'], "cn-shanghai")
request = VideoAsyncScanRequest.VideoAsyncScanRequest()
request.set_accept_format('JSON')

task = {"dataId": str(uuid.uuid1()),
         "url": "https://www.aliyundoc.com/xxx.mp4" # Replace with your video URL.
        }
print(task)
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:
        # Save the returned taskId. You can use it to query the video detection results.
        print(taskResult["taskId"])

• Submit a local video file for moderation

#coding=utf-8
# The following code calls the asynchronous video detection operation.
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
import os

# Python 2 only: supports local paths that contain non-ASCII characters.
if sys.version_info[0] == 2:
    reload(sys)
    sys.setdefaultencoding('utf-8')

clt = client.AcsClient(os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'], os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET'], "cn-shanghai")
request = VideoAsyncScanRequest.VideoAsyncScanRequest()
request.set_accept_format('JSON')

# Upload the local file to the server. Replace the path with the path to your own video file.
uploader = ClientUploader.getVideoClientUploader(clt)
url = uploader.uploadFile('/path/to/your-video.mp4')

task = {"dataId": str(uuid.uuid1()),
         "url": url
        }
print(task)
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:
        # Save the returned taskId. You can use it to query the video detection results.
        print(taskResult["taskId"])

• Submit a video file as binary data for moderation

#coding=utf-8
# The following code calls the asynchronous video detection operation.
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
import os

# Python 2 only: supports local paths that contain non-ASCII characters.
if sys.version_info[0] == 2:
    reload(sys)
    sys.setdefaultencoding('utf-8')

clt = client.AcsClient(os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'], os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET'], "cn-shanghai")
request = VideoAsyncScanRequest.VideoAsyncScanRequest()
request.set_accept_format('JSON')
# Read a local file as binary data to simulate binary data detection.
# Replace the path with the path to your own video file.
f = open('/path/to/your-video.mp4', "rb")
videoBytes = f.read()
f.close()

# Upload the binary data to the server.
uploader = ClientUploader.getVideoClientUploader(clt)
url = uploader.uploadBytes(videoBytes)

task = {"dataId": str(uuid.uuid1()),
         "url": url
        }
print(task)
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:
        # Save the returned taskId. You can use it to query the video detection results.
        print(taskResult["taskId"])

• Submit a live stream for moderation

#coding=utf-8
# The following code calls the asynchronous video detection operation.
from aliyunsdkcore import client
from aliyunsdkgreen.request.v20180509 import VideoAsyncScanRequest
from aliyunsdkgreenextension.request.extension import HttpContentHelper
import json
import uuid
import os

clt = client.AcsClient(os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'], os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET'], "cn-shanghai")
request = VideoAsyncScanRequest.VideoAsyncScanRequest()
request.set_accept_format('JSON')

# Replace the URL with your live stream URL.
task = {
        "dataId": str(uuid.uuid1()),
        "url": "http://www.aliyundoc.com/xxx.flv"
        }
print(task)
# Set live to True because the URL points to a live stream.
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:
        # Save the returned taskId. You can use it to query the video detection results.
        print(taskResult["taskId"])

• Submit a video for combined image and audio moderation

#coding=utf-8
# The following code calls the asynchronous video detection operation.
from aliyunsdkcore import client
from aliyunsdkgreen.request.v20180509 import VideoAsyncScanRequest
from aliyunsdkgreenextension.request.extension import HttpContentHelper
import json
import uuid
import os

clt = client.AcsClient(os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'], os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET'], "cn-shanghai")
request = VideoAsyncScanRequest.VideoAsyncScanRequest()
request.set_accept_format('JSON')

# This sample moderates both the video frames and the audio track of a live stream.
# Replace the URL with your live stream URL.
task = {
        "dataId": str(uuid.uuid1()),
        "url": "http://www.aliyundoc.com/xxx.flv"
        }
print(task)
# scenes applies to the video frames, and audioScenes applies to the audio track.
# live is set to True because the URL points to a live stream.
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:
        # Save the returned taskId. You can use it to query the video detection results.
        print(taskResult["taskId"])

Query asynchronous video moderation results

Query the moderation results of asynchronous video moderation tasks with the taskId values that the submission returned. For each task, the results field contains the moderation results for the frames extracted from the video. To submit the tasks, see (Recommended) Submit an asynchronous video moderation task.

Sample code

#coding=utf-8
# The following code calls the operation to query asynchronous video detection results.
import json
from aliyunsdkcore import client
from aliyunsdkgreen.request.v20180509 import VideoAsyncScanResultsRequest
from aliyunsdkgreenextension.request.extension import HttpContentHelper
import os

clt = client.AcsClient(os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'], os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET'], "cn-shanghai")
request = VideoAsyncScanResultsRequest.VideoAsyncScanResultsRequest()
request.set_accept_format('JSON')

# Pass the list of taskIds for the video detection tasks you want to query.
# Replace the value with the taskIds returned when you submitted the tasks.
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:
        # The 'results' field for each task contains the detection results for the frames extracted from the video.
        print(taskResult['results'])

Submit a synchronous video frame moderation task

Synchronous video frame moderation returns the moderation results in the response of the same call. This operation supports only frame sequences: submit the frames in the frames array, in which each frame has a url and an offset. To submit a video URL, a video file, or a live stream, use the asynchronous video moderation operation instead.

Sample code

#coding=utf-8
# The following code calls the synchronous video detection operation, which supports only frame sequences.
from aliyunsdkcore import client
from aliyunsdkgreen.request.v20180509 import VideoSyncScanRequest
from aliyunsdkgreenextension.request.extension import HttpContentHelper
import json
import os

clt = client.AcsClient(os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'], os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET'], "cn-shanghai")
request = VideoSyncScanRequest.VideoSyncScanRequest()
request.set_accept_format('JSON')

# Replace the URLs with the URLs of your own video frames.
task = {
            "frames":[
                {"offset" : 0, "url" : "https://www.aliyundoc.com/test1.jpg"},
                {"offset" : 2, "url" : "https://www.aliyundoc.com/test2.jpg"},
                {"offset" : 3, "url" : "https://www.aliyundoc.com/test3.jpg"}
            ]
        }
print(task)
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 frameResult in taskResult["results"]:
            # Take subsequent actions based on the results.
            print(frameResult['suggestion'])
            print(frameResult['scene'])

Video moderation feedback

If a video moderation result does not match your expectation, call the video moderation feedback operation to correct the result. Based on your feedback, the system adds the video frames to the similar image blacklist or whitelist. When you submit similar content for moderation again, the result is returned with the label that you provided.

For more information about the operation, see Moderation result feedback.

The following table describes the fields in the request body of the video moderation feedback sample.

Sample code

# coding=utf-8

from aliyunsdkcore import client
from aliyunsdkcore.profile import region_provider
from aliyunsdkgreen.request.v20180509 import VideoFeedbackRequest
import json
import os

clt = client.AcsClient(os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'], os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET'], "cn-shanghai")
region_provider.modify_point('Green', 'cn-shanghai', 'green.cn-shanghai.aliyuncs.com')
request = VideoFeedbackRequest.VideoFeedbackRequest()
request.set_accept_format('JSON')
# Replace the placeholder values with the data of the video moderation task that you want to correct.
request.set_content(
    json.dumps({"dataId": "your-data-id", "taskId": "your-task-id", "url": "https://www.aliyundoc.com/xxx.mp4",
                "suggestion": "block", "frames": [{"url": "https://www.aliyundoc.com/test1.jpg", "offset": 0}],
                "scenes": ["ad", "terrorism"], "note": "your remarks"}))

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)