Video moderation-AI Guardrails(AI Guardrails)-阿里云帮助中心

Use the AI Guardrails SDK for Python to submit video moderation tasks and retrieve results.

Choose a detection mode

AI Guardrails supports two detection modes. Choose based on your input type:

Mode	Operation	Supported input	When to use
Asynchronous (recommended)	`VideoAsyncScanRequest`	Original videos, frame sequences, live streams	Most use cases — supports all video types
Synchronous	`VideoSyncScanRequest`	Frame sequences only	When you need results inline and can pre-extract frames

Note: The SDK accepts only public URLs (HTTP or HTTPS, up to 2,048 characters). It does not accept local file paths or raw binary data directly — use the ClientUploader utility to upload local files or binary streams first.

Prerequisites

Before you begin, make sure you have:

Python dependencies installed per the Installation guide — use the exact Python version specified there, otherwise operation calls fail
The Extension.Uploader utility class downloaded and imported into your project

Submit asynchronous video moderation tasks

VideoAsyncScanRequest sends asynchronous requests to moderate videos across multiple scenarios: pornography, terrorist content, ad, undesirable scene, and logo detection.

Supported regions: China (Shanghai) cn-shanghai, China (Beijing) cn-beijing, China (Shenzhen) cn-shenzhen, Singapore ap-southeast-1

All examples below use VideoAsyncScanRequest. Save the taskId from the response — you need it to query results.

Supported scenes

Scene value	Content detected
`porn`	Pornographic content
`terrorism`	Terrorist content
`ad`	Advertisements
`undesirable`	Undesirable scenes
`logo`	Logo detection

For audio moderation within live streams, use audioScenes: ["antispam"].

Billing: You are charged based on the number of frames extracted multiplied by the number of scenes. For example, moderating two videos for both porn and terrorism costs: frames extracted from both videos × (porn unit price + terrorism unit price). Audio moderation is billed separately by audio duration.

Submit a video URL

# coding=utf-8
import json
import os
import uuid
from aliyunsdkcore import client
from aliyunsdkgreen.request.v20180509 import VideoAsyncScanRequest
from aliyunsdkgreenextension.request.extension import HttpContentHelper

# Reuse the client instance across requests to improve performance
# and avoid repeated connection overhead.
clt = client.AcsClient(
    os.environ["ALIBABA_CLOUD_ACCESS_KEY_ID"],
    os.environ["ALIBABA_CLOUD_ACCESS_KEY_SECRET"],
    "cn-shanghai"
)

# Create a new request object for each request — request objects cannot be reused.
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
}

# Specify one or more scenes. Billing is based on frames extracted x number of scenes.
request.set_content(HttpContentHelper.toValue({
    "tasks": [task],
    "scenes": ["terrorism"]
}))

response = clt.do_action_with_exception(request)
result = json.loads(response)

if result["code"] == 200:
    for task_result in result["data"]:
        print(task_result["taskId"])  # Save this ID to query results later

Submit a local video file

Use ClientUploader to upload the local file to the server, then pass the returned URL as the task URL.

# coding=utf-8
import json
import os
import sys
import uuid
from aliyunsdkcore import client
from aliyunsdkgreen.request.v20180509 import VideoAsyncScanRequest
from aliyunsdkgreenextension.request.extension import ClientUploader
from aliyunsdkgreenextension.request.extension import HttpContentHelper

# Python 2 only: set default encoding to UTF-8 to support paths with 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"
)

# Upload the local video and get a public URL
uploader = ClientUploader.getVideoClientUploader(clt)
url = uploader.uploadFile("d:/example-video.mp4")  # Replace with your local file path

request = VideoAsyncScanRequest.VideoAsyncScanRequest()
request.set_accept_format("JSON")

task = {
    "dataId": str(uuid.uuid1()),
    "url": url
}

request.set_content(HttpContentHelper.toValue({
    "tasks": [task],
    "scenes": ["terrorism"]
}))

response = clt.do_action_with_exception(request)
result = json.loads(response)

if result["code"] == 200:
    for task_result in result["data"]:
        print(task_result["taskId"])

Submit a binary video stream

Read the video as binary data, upload it with ClientUploader, then submit the returned URL.

# coding=utf-8
import json
import os
import sys
import uuid
from aliyunsdkcore import client
from aliyunsdkgreen.request.v20180509 import VideoAsyncScanRequest
from aliyunsdkgreenextension.request.extension import ClientUploader
from aliyunsdkgreenextension.request.extension import HttpContentHelper

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"
)

# Read the local video as binary data and upload it
with open("d:/example-video.mp4", "rb") as f:
    video_bytes = f.read()

uploader = ClientUploader.getVideoClientUploader(clt)
url = uploader.uploadBytes(video_bytes)

request = VideoAsyncScanRequest.VideoAsyncScanRequest()
request.set_accept_format("JSON")

task = {
    "dataId": str(uuid.uuid1()),
    "url": url
}

request.set_content(HttpContentHelper.toValue({
    "tasks": [task],
    "scenes": ["terrorism"]
}))

response = clt.do_action_with_exception(request)
result = json.loads(response)

if result["code"] == 200:
    for task_result in result["data"]:
        print(task_result["taskId"])

Submit a live stream (video only)

Set the live parameter to True to enable live stream moderation.

# coding=utf-8
import json
import os
import uuid
from aliyunsdkcore import client
from aliyunsdkgreen.request.v20180509 import VideoAsyncScanRequest
from aliyunsdkgreenextension.request.extension import HttpContentHelper

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": "http://www.aliyundoc.com/xxx.flv"  # Replace with your live stream URL
}

request.set_content(HttpContentHelper.toValue({
    "tasks": [task],
    "scenes": ["terrorism"],
    "live": True
}))

response = clt.do_action_with_exception(request)
result = json.loads(response)

if result["code"] == 200:
    for task_result in result["data"]:
        print(task_result["taskId"])

Submit a live stream (video and audio)

Add audioScenes to moderate both video frames and audio content in a single request. Video frame moderation is billed by frames × scenes; audio moderation is billed separately by audio duration.

# coding=utf-8
import json
import os
import uuid
from aliyunsdkcore import client
from aliyunsdkgreen.request.v20180509 import VideoAsyncScanRequest
from aliyunsdkgreenextension.request.extension import HttpContentHelper

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": "http://www.aliyundoc.com/xxx.flv"  # Replace with your live stream URL
}

request.set_content(HttpContentHelper.toValue({
    "tasks": [task],
    "scenes": ["terrorism"],       # Video frame moderation scenes
    "live": True,
    "audioScenes": ["antispam"]    # Audio moderation scenes
}))

response = clt.do_action_with_exception(request)
result = json.loads(response)

if result["code"] == 200:
    for task_result in result["data"]:
        print(task_result["taskId"])

Query asynchronous moderation results

VideoAsyncScanResultsRequest retrieves the results of previously submitted asynchronous tasks.

Supported regions: China (Shanghai) cn-shanghai, China (Beijing) cn-beijing, China (Shenzhen) cn-shenzhen, Singapore ap-southeast-1

Note: Instead of polling with this operation, set the callback parameter when submitting tasks to receive results via callback. This reduces latency and API call overhead.

# coding=utf-8
import json
import os
from aliyunsdkcore import client
from aliyunsdkgreen.request.v20180509 import VideoAsyncScanResultsRequest
from aliyunsdkgreenextension.request.extension import HttpContentHelper

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 taskId values returned when you submitted the moderation tasks
task_ids = ["vi3pX@vXC94hPnWsss39WOQ9-1q52ZG"]
request.set_content(HttpContentHelper.toValue(task_ids))

response = clt.do_action_with_exception(request)
result = json.loads(response)

if result["code"] == 200:
    for task_result in result["data"]:
        # results contains moderation outcomes for each frame extracted from the video
        print(task_result["results"])

Submit synchronous video moderation tasks

VideoSyncScanRequest moderates a submitted sequence of consecutive frames. It supports frame sequences only — for other input types, use VideoAsyncScanRequest.

Supported regions: China (Shanghai) cn-shanghai, China (Beijing) cn-beijing, China (Shenzhen) cn-shenzhen, Singapore ap-southeast-1

# coding=utf-8
import json
import os
from aliyunsdkcore import client
from aliyunsdkgreen.request.v20180509 import VideoSyncScanRequest
from aliyunsdkgreenextension.request.extension import HttpContentHelper

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")

# Submit an ordered sequence of frames with their time offsets (in seconds)
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"}
    ]
}

# Billing: frames x number of scenes, same as asynchronous mode
request.set_content(HttpContentHelper.toValue({
    "tasks": [task],
    "scenes": ["porn"]
}))

response = clt.do_action_with_exception(request)
result = json.loads(response)

if result["code"] == 200:
    for task_result in result["data"]:
        for frame_result in task_result["results"]:
            print(frame_result["suggestion"])  # pass, block, or review
            print(frame_result["scene"])

Provide feedback on moderation results

If moderation results don't match your expectations, use VideoFeedbackRequest to correct them. AI Guardrails adds the moderated video frames to the similar image blacklist or whitelist based on your feedback. When a similar frame is submitted later, the service returns results based on your feedback label.

For parameter details, see /green/video/feedback.

Supported regions: China (Shanghai) cn-shanghai, China (Beijing) cn-beijing, China (Shenzhen) cn-shenzhen, Singapore ap-southeast-1

# coding=utf-8
import json
import os
from aliyunsdkcore import client
from aliyunsdkcore.profile import region_provider
from aliyunsdkgreen.request.v20180509 import VideoFeedbackRequest

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")

request.set_content(json.dumps({
    "dataId": "<video-data-id>",       # dataId of the moderated video
    "taskId": "<video-task-id>",       # taskId returned when you submitted the task
    "url": "<video-url>",              # URL of the moderated video
    "suggestion": "block",             # pass: normal content | block: violating content
    "frames": [
        {
            "url": "<frame-url>",      # URL of the captured frame
            "offset": "<offset>"       # Time offset of the frame (seconds)
        }
    ],
    "scenes": ["ad", "terrorism"],     # Moderation scenarios the feedback applies to
    "note": "<remarks>"               # Optional notes
}))

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

What's next

Asynchronous detection parameters — full parameter reference for VideoAsyncScanRequest
Synchronous detection parameters — full parameter reference for VideoSyncScanRequest
Installation — SDK installation and Python version requirements