文档

语音反垃圾检测

更新时间:

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

功能描述

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

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

前提条件

提交语音异步检测任务

接口

描述

支持的地域

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))
  • 本页导读 (0)
文档反馈