使用Python SDK语音反垃圾接口检测实时语音流或语音文件中的垃圾内容。

说明
  • 语音流检测和语音文件检测均为异步检测,检测结果需要您以轮询或者回调的方式获取。关于调用请求中的检测场景参数scene,返回结果中的分类参数label,以及操作建议参数suggestion的说明,请参见语音异步检测
  • 语音检测按照检测的语音文件/语音流的时间长度进行计费,计费粒度为分钟,每天累计检测总时长进行计量统计,每天检测总时长不足一分钟的按照1分钟进行计费。

准备工作

在进行具体的服务调用之前,请参见以下步骤,完成准备工作:
  1. 创建阿里云AccessKeyId和AccessKeySecret。具体请参见创建AccessKey
  2. 安装Python依赖。具体请参见安装Python依赖

提交语音异步检测任务

接口描述
接口 功能 支持的Region 描述
VoiceAsyncScanRequest 提交异步语音检测。 cn-shanghai
异步检测语音流或语音文件中是否包含违规内容。语音流格式支持:
  • rtmp
  • http
  • flv
提交语音URL示例代码
# coding=utf-8
# 异步语音检测
from aliyunsdkcore import client
from aliyunsdkcore.profile import region_provider
from aliyunsdkgreen.request.v20180509 import VoiceAsyncScanRequest
from aliyunsdkgreen.request.extension import HttpContentHelper
import json

# 请替换成您自己的accessKeyId、accessKeySecret。
clt = client.AcsClient("您的accessKeyId", "您的accessKeySecret", 'cn-shanghai')
region_provider.modify_point('Green', 'cn-shanghai', 'green.cn-shanghai.aliyuncs.com')

request = VoiceAsyncScanRequest.VoiceAsyncScanRequest()
request.set_accept_format('JSON')

task = {
    "url": "https://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 aliyunsdkgreen.request.extension import HttpContentHelper
from aliyunsdkgreen.request.extension import ClientUploader
import json

# 请替换成您自己的accessKeyId、accessKeySecret。
clt = client.AcsClient("您的accessKeyId", "您的accessKeySecret", 'cn-shanghai')
region_provider.modify_point('Green', 'cn-shanghai', 'green.cn-shanghai.aliyuncs.com')

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

查询语音异步检测结果

接口描述
接口 功能 支持的Region 描述
VoiceAsyncScanResultsRequest 查询异步语音检测结果。 cn-shanghai 查询语音检测的结果。
示例代码
# coding=utf-8
# 查询异步语音检测结果
from aliyunsdkcore import client
from aliyunsdkcore.profile import region_provider
from aliyunsdkgreen.request.v20180509 import VoiceAsyncScanResultsRequest
import json
import time

# 请替换成您自己的accessKeyId、accessKeySecret。
clt = client.AcsClient("您的accessKeyId", "您的accessKeySecret",'cn-shanghai')
region_provider.modify_point('Green', 'cn-shanghai', 'green.cn-shanghai.aliyuncs.com')

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("这里填写提交异步语音检测任务返回的taskid")
    if code == 280:
        print("Scanning:")
        time.sleep(10)
    elif code == 200:
        print("=====SUCCESS=====")
        break
    else:
        print("=====ERROR=====")
        break