Audio anti-spam detection-AI Guardrails(AI Guardrails)-阿里云帮助中心

Feature description

Audio stream and audio file moderation are both asynchronous. You can retrieve the moderation results by polling or using callbacks. For more information about the scenes request parameter, the label classification parameter, and the suggestion parameter in the response, see Asynchronous audio moderation.

Audio moderation is billed per minute based on the duration of the moderated audio files or streams. The total duration is calculated daily. If the total duration for a day is less than one minute, it is billed as one minute.

Note

This SDK supports only public audio URLs. It does not support local files or binary data.
Supported URL types:
- Public HTTP/HTTPS URLs that do not exceed 2048 characters in length.
- Alibaba Cloud OSS URL: oss://<bucket-name>.<endpoint>/<object-name>. You must authorize AI Guardrails to access the destination OSS bucket. The bucket and the AI Guardrails service must be in the same region. For more information, see Authorize AI Guardrails to access an OSS bucket.

Prerequisites

Python dependencies are installed. For more information, see Installation.

Note
You must use the required Python version described in the Installation topic to install the dependencies. Otherwise, subsequent operation calls fail.
The Extension.Uploader utility class is downloaded and imported into your project.

Submit asynchronous audio detection tasks

Interface

Description

Supported region

VoiceAsyncScanRequest

Asynchronously detects violations in audio streams or audio files.

Supported audio stream protocols: HTTP, RTMP, and RTSP
Supported audio stream formats: M3U8 and FLV

cn-shanghai: China (Shanghai)

Sample code

Submit an audio URL for detection

# coding=utf-8
# The following code shows how to call the asynchronous audio detection API operation.
from aliyunsdkcore import client
from aliyunsdkcore.profile import region_provider
from aliyunsdkgreen.request.v20180509 import VoiceAsyncScanRequest
from aliyunsdkgreenextension.request.extension import HttpContentHelper
import json

# Note: To improve detection performance, reuse the instantiated client and avoid creating repeated connections.
# Common methods to obtain environment variables:
# Obtain the AccessKey ID of the RAM user: os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID']
# Obtain the AccessKey secret of the RAM user: os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
clt = client.AcsClient("yourAccessKeyId", "yourAccessKeySecret", "cn-shanghai")
request = VoiceAsyncScanRequest.VoiceAsyncScanRequest()
request.set_accept_format('JSON')

task = {
    "url": "https://example.com/example.flv",
    }

# To detect an audio stream, set the live parameter to true.
request.set_content(HttpContentHelper.toValue({"tasks": [task], "scenes": ["antispam"], "live":false}))
response = clt.do_action_with_exception(request)
print(response)
result = json.loads(response)
if 200 == result["code"]:
    taskResults = result["data"]
    for taskResult in taskResults:
        code = taskResult["code"]
        if 200 == code:
            # Save the taskId. You can use it to query the detection result.
            print(taskResult["taskId"])

Submit a local audio file for detection

# coding=utf-8
# The following code shows how to call the asynchronous audio detection API operation.
from aliyunsdkcore import client
from aliyunsdkcore.profile import region_provider
from aliyunsdkgreen.request.v20180509 import VoiceAsyncScanRequest
from aliyunsdkgreenextension.request.extension import HttpContentHelper
from aliyunsdkgreenextension.request.extension import ClientUploader
import json

# Note: To improve detection performance, reuse the instantiated client and avoid creating repeated connections.
# Common methods to obtain environment variables:
# Obtain the AccessKey ID of the RAM user: os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID']
# Obtain the AccessKey secret of the RAM user: os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
clt = client.AcsClient("yourAccessKeyId", "yourAccessKeySecret", "cn-shanghai")
request = VoiceAsyncScanRequest.VoiceAsyncScanRequest()
request.set_accept_format('JSON')

# Upload the local file to the server. Change the path to your local file path.
uploader = ClientUploader.getVoiceClientUploader(clt)
url = uploader.uploadFile('d:/local_audio_file.mp3')
# Set the type parameter to file to indicate that you are detecting an audio file.
task = {
    "url": url,
    "type": "file"
    }

request.set_content(HttpContentHelper.toValue({"tasks": [task], "scenes": ["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:
        code = taskResult["code"]
        if 200 == code:
            # Save the taskId. You can use it to query the detection result.
            print(taskResult["taskId"])

Query the results of asynchronous audio detection tasks

API	Description	Supported regions
VoiceAsyncScanResultsRequest	Queries asynchronous audio moderation results.	`cn-shanghai`: China (Shanghai)

Sample code

# coding=utf-8
# The following code shows how to query the results of an asynchronous audio detection task.
from aliyunsdkcore import client
from aliyunsdkcore.profile import region_provider
from aliyunsdkgreen.request.v20180509 import VoiceAsyncScanResultsRequest
import json
import time

# Note: To improve detection performance, reuse the instantiated client and avoid creating repeated connections.
# Common methods to obtain environment variables:
# Obtain the AccessKey ID of the RAM user: os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID']
# Obtain the AccessKey secret of the RAM user: os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
clt = client.AcsClient("yourAccessKeyId", "yourAccessKeySecret", "cn-shanghai")

def get_task(task_id):
    request = VoiceAsyncScanResultsRequest.VoiceAsyncScanResultsRequest()
    request.set_accept_format('JSON')
    task = {
        "taskId": task_id,
    }
    request.set_content(bytearray(json.dumps([task]), "utf-8"))
    response = clt.do_action_with_exception(request)
    result = json.loads(response)
    if 200 == result["code"]:
        taskResults = result["data"]
        for taskResult in taskResults:
            code = taskResult["code"]
            if 280 == code:
                print(taskResult["msg"])
            else:
                print(response)
            return code
    result["code"]

while True:
    code = get_task("input taskid")
    if code == 280:
        print("Scanning:")
        time.sleep(10)
    elif code == 200:
        print("=====SUCCESS=====")
        break
    else:
        print("=====ERROR=====")
        break

Synchronous detection of short audio clips

Sample code

# coding=utf-8
# The following code shows how to call the synchronous audio detection API operation.

import json

from aliyunsdkcore import client
from aliyunsdkcore.profile import region_provider
from aliyunsdkgreen.request.v20180509 import VoiceSyncScanRequest

# Note: To improve detection performance, reuse the instantiated client and avoid creating repeated connections.
# Common methods to obtain environment variables:
# Obtain the AccessKey ID of the RAM user: os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID']
# Obtain the AccessKey secret of the RAM user: os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
clt = client.AcsClient("yourAccessKeyId", "yourAccessKeySecret", "cn-shanghai")

request = VoiceSyncScanRequest.VoiceSyncScanRequest()
request.set_accept_format('JSON')
# Synchronous audio detection is time-consuming. Set the call timeout period to 15 seconds.
request.set_read_timeout(15000)

task = {
    "url": "http://example.com/example.mp3",
}

request.set_content(json.dumps({"tasks": [task], "scenes": ["antispam"]}))
response = clt.do_action_with_exception(request)
result = json.loads(response)
print(json.dumps(result, ensure_ascii=False, indent=2))