如何使用PythonSDK语音反垃圾接口_内容安全(Content Moderation)-阿里云帮助中心

本文介绍了如何使用Python SDK语音反垃圾接口，检测实时语音流或语音文件中的垃圾内容。

功能描述

语音流检测和语音文件检测均为异步检测，检测结果需要您以轮询或者回调的方式获取。关于调用请求中的检测场景参数scenes，返回结果中的分类参数label，以及操作建议参数suggestion的说明，请参见语音异步检测。

语音检测按照检测的语音文件、语音流的时间长度进行计费，计费粒度为分钟，每天累计检测总时长进行计量统计，每天检测总时长不足一分钟的按照一分钟进行计费。

前提条件

安装Python依赖。关于安装Python依赖的具体操作，请参见安装Python依赖。
说明
请一定按照安装Python依赖页面中的版本安装，否则会导致调用失败。
下载并在项目工程中引入Extension.Uploader工具类。

提交语音异步检测任务

接口

描述

支持的地域

VoiceAsyncScanRequest

异步检测语音流或语音文件中是否包含违规内容。

支持的语音流协议：HTTP、RTMP、RTSP
支持的语音流格式：M3U8、FLV

cn-shanghai：华东2（上海）

示例代码

提交语音URL进行检测

# coding=utf-8
# 以下代码用于调用语音异步检测接口。
from aliyunsdkcore import client
from aliyunsdkcore.profile import region_provider
from aliyunsdkgreen.request.v20180509 import VoiceAsyncScanRequest
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 = VoiceAsyncScanRequest.VoiceAsyncScanRequest()
request.set_accept_format('JSON')

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

# 如果需要检测语音流，请将live设置为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:
            # 请保存taskId，用于查询语音检测结果。
            print(taskResult["taskId"])

提交本地语音文件进行检测

# coding=utf-8
# 以下代码用于调用语音异步检测接口。
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

# 注意：此处实例化的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 = VoiceAsyncScanRequest.VoiceAsyncScanRequest()
request.set_accept_format('JSON')

# 上传本地文件到服务端。请修改成您自己的本地文件路径。
uploader = ClientUploader.getVoiceClientUploader(clt)
url = uploader.uploadFile('d:/暴恐涉政1.mp3')
# 将type设置为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:
            # 请保存taskId，用于查询语音检测结果。
            print(taskResult["taskId"])

查询语音异步检测结果

接口	描述	支持的地域
VoiceAsyncScanResultsRequest	查询异步语音检测结果。	`cn-shanghai`：华东2（上海）

示例代码

# coding=utf-8
# 以下代码用于调用异步语音检测结果查询接口。
from aliyunsdkcore import client
from aliyunsdkcore.profile import region_provider
from aliyunsdkgreen.request.v20180509 import VoiceAsyncScanResultsRequest
import json
import time

# 注意：此处实例化的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")

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

短语音同步检测

示例代码

# coding=utf-8
# 以下代码用于调用语音同步检测接口。

import json

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

# 注意：此处实例化的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 = VoiceSyncScanRequest.VoiceSyncScanRequest()
request.set_accept_format('JSON')
# 同步语音检测耗时比较长，建议将调用超时设置为15秒。
request.set_read_timeout(15000)

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

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