阿里云语音合成,录音文件识别,自然语言分析,rest 调用 python实现

本文涉及的产品
NLP 自学习平台,3个模型定制额度 1个月
NLP自然语言处理_基础版,每接口每天50万次
NLP自然语言处理_高级版,每接口累计50万次
简介: 阿里云语音合成,录音文件识别,自然语言分析,rest 调用 python实现

最近研究阿里云语音合成,录音文件识别,自然语言分析。

image.png

自然语言分析官网文档:

https://help.aliyun.com/document_detail/61378.html?spm=a2c4g.11186623.6.547.9q3U1C

智能语音官网文档:

https://help.aliyun.com/product/30413.html?spm=a2c4g.11186623.3.1.yS0DIK


决定采用阿里云提供的restful api接口,主要利用python urllib库来实现调用,整合里官方给出的demo,总结出调用代码。话不多说,贴出代码:


python环境

python3.5.1


http请求代理工具类

在工程目录中建立utils目录,在目录中建立init.py文件代码如下

import hashlib
import urllib.request
import hmac
import base64
import datetime
import ssl
import uuid
from urllib.error import HTTPError
class http_proxy:
    """
    Http工具类,封装了鉴权
    """
    def __init__(self, ak_id, ak_secret):
        self.__ak_id = ak_id
        self.__ak_secret = ak_secret
    def __current_gmt_time(self):
        date = datetime.datetime.strftime(datetime.datetime.utcnow(), "%a, %d %b %Y %H:%M:%S GMT")
        return date
    def __md5_base64(self, strbody):
        hash = hashlib.md5()
        hash.update(strbody.encode('utf-8'))
        print(hash.digest())
        return base64.b64encode(hash.digest()).decode('utf-8')
    def __sha1_base64(self, str_to_sign, secret):
        hmacsha1 = hmac.new(secret.encode('utf-8'), str_to_sign.encode('utf-8'), hashlib.sha1)
        return base64.b64encode(hmacsha1.digest()).decode('utf-8')
    def send_request(self, url, body):
        gmtnow = self.__current_gmt_time()
        print(gmtnow)
        body_md5 = self.__md5_base64(body)
        print(body_md5)
        str_to_sign = "POST\napplication/json\n" + body_md5 + "\napplication/json\n" + gmtnow
        print(str_to_sign)
        signature = self.__sha1_base64(str_to_sign, self.__ak_secret)
        print(signature)
        auth_header = "Dataplus " + self.__ak_id + ":" + signature
        print(auth_header)
        ssl._create_default_https_context = ssl._create_unverified_context
        req = urllib.request.Request(url)
        req.add_header("Accept", "application/json")
        req.add_header("Content-Type", "application/json")
        req.add_header("Date", gmtnow)
        req.add_header("Authorization", auth_header)
        data = body.encode('utf-8')
        f = urllib.request.urlopen(req, data)
        return f.read().decode('utf-8')
    def send_get(self, url, task_id):
        gmtnow = self.__current_gmt_time()
        print(gmtnow)
        accept = "application/json"
        content_type = "application/json"
        str_to_sign = "GET\n" + accept + "\n" + "" + "\n" + content_type + "\n" + gmtnow
        print(str_to_sign)
        signature = self.__sha1_base64(str_to_sign, self.__ak_secret)
        print(signature)
        auth_header = "Dataplus " + self.__ak_id + ":" + signature
        print(auth_header)
        ssl._create_default_https_context = ssl._create_unverified_context
        url += "/" + task_id
        req = urllib.request.Request(url)
        req.add_header("Accept", "application/json")
        req.add_header("Content-Type", "application/json")
        req.add_header("Date", gmtnow)
        req.add_header("Authorization", auth_header)
        f = urllib.request.urlopen(req)
        return f.read().decode('utf-8')
        # try:
        #     f = urllib.request.urlopen(req)
        # except HTTPError as e:
        #     print(e)
        #     return None
        # else:
        #     return f.read().decode('utf-8')
    def send_requestForNlp(self, path, content):
        method = "POST"
        content_type = "application/json;chrset=utf-8"
        accept = "application/json"
        host = "nlp.cn-shanghai.aliyuncs.com"
        gmtnow = self.__current_gmt_time()
        print(gmtnow)
        body_md5 = self.__md5_base64(content)
        print(body_md5)
        uuidstr =  uuid.uuid4().hex
        str_to_sign = method+"\n"+accept+"\n" + body_md5 + "\n"+content_type+"\n" + gmtnow+ "\nx-acs-signature-method:HMAC-SHA1\n" + "x-acs-signature-nonce:" + uuidstr + "\n" + path;
        print(str_to_sign)
        signature = self.__sha1_base64(str_to_sign, self.__ak_secret)
        print(signature)
        auth_header = "acs " + self.__ak_id + ":" + signature
        print(auth_header)
        ssl._create_default_https_context = ssl._create_unverified_context
        req = urllib.request.Request("http://"+host+path)
        req.add_header("Accept",accept)
        req.add_header("Content-Type", content_type)
        req.add_header("Content-MD5", body_md5)
        req.add_header("Date", gmtnow)
        req.add_header("Host", host)
        req.add_header("x-acs-signature-nonce", uuidstr)
        req.add_header("x-acs-signature-method", "HMAC-SHA1")
        req.add_header("Authorization", auth_header)
        data = content.encode('utf-8')
        f = urllib.request.urlopen(req, data)
        return f.read().decode('utf-8')
    def sendTtsPost(self, textData, ttsRequestParam,fileRootPath):
        url = 'http://nlsapi.aliyun.com/speak?'#语音合成接口
        #拼接
        paramArray = []
        for key in ttsRequestParam:
            paramArray.append(key+"="+ttsRequestParam[key])
        url+=url+'&'.join(paramArray)
        method = "POST"
        content_type = "text/plain"
        accept = "audio/" + ttsRequestParam['encode_type'] + ",application/json"
        gmtnow = self.__current_gmt_time()
        body_md5 = self.__md5_base64(textData)
        print(body_md5)
        str_to_sign = method + "\n" + accept + "\n" + body_md5 + "\n" + content_type + "\n" + gmtnow
        print(str_to_sign)
        signature = self.__sha1_base64(str_to_sign, self.__ak_secret)
        print(signature)
        auth_header = "Dataplus " + self.__ak_id + ":" + signature
        print(auth_header)
        ssl._create_default_https_context = ssl._create_unverified_context
        req = urllib.request.Request(url)
        req.add_header("accept", accept)
        req.add_header("content-Type", content_type)
        req.add_header("date", gmtnow)
        req.add_header("Authorization", auth_header)
        req.add_header("Content-Length", len(textData))
        data = textData.encode('utf-8')
        f = urllib.request.urlopen(req, data)
        if f.status ==200:
            file = 'g:audio/' + uuid.uuid4().hex + ".wav"
            content = f.read()
            with open(file, 'wb') as f:
                f.write(content)
            print("success"+file)
        else:
            print('失败!')

调用demo

在目录中建立demo.py ,注意和上面的utils目录同级

import sys
import utils
import json
# app_key   语音数据格式  领域
# nls-service-realtime-8k   8kHz采样率 智能客服服务领域,比如电话客服等
# nls-service-multi-domain  16kHz采样率    汉语通用识别
# nls-service   16kHz采样率    输入法,社交聊天
# nls-service-tv    16kHz采样率    家庭娱乐
# nls-service-shopping  16kHz采样率    电商购物
# nls-realtime-fangyan  16kHz采样率    支持东北、河南、四川等方言
# nls-service-yue-streaming 16kHz采样率    粤语
# nls-service-en    16kHz采样率    英语
ak_id = "";  ##数加管控台获得的accessId
ak_secret = ""  ## 数加管控台获得的accessSecret
url = "https://nlsapi.aliyun.com/transcriptions"
# 录音文件提交
def request():
    body = {
        'app_key': 'nls-service-multi-domain',
        'oss_link': 'http://网址/audio/zl4.mp3',
    }
    bodyStr = json.dumps(body)
    httpProxy = utils.http_proxy(ak_id, ak_secret)
    result = httpProxy.send_request(url, bodyStr)
    return result
# 录音文件识别结果查询
def query(id):
    httpProxy = utils.http_proxy(ak_id, ak_secret)
    result = httpProxy.send_get(url, id)
    return result
# 自然语音分析 分词
def nlpTest():
    path = '/nlp/api/wordpos/general'
    postBody = {
        'text': '为什么世界是这个样子,人们都不诚实,我要努力,获得成功,让别人尊敬',
        'lang': 'ZH',
    }
    bodyStr = json.dumps(postBody)
    httpProxy = utils.http_proxy(ak_id, ak_secret)
    result = httpProxy.send_requestForNlp(path,bodyStr)
    return result
#语音合成
def ttsTest():
    text = '在打招呼的时候直视对方的人在交往中往往具有攻击性,'
    fileRootPath='g:audio/'
    ttsRequestParam ={
        'encode_type':'wav',#合成语音的编码格式,支持pcm/wav/mp3/alaw
        'voice_name':'xiaogang',#xiaogang - 男,xiaoyun - 女
        'volume':'50',#0~100
        'sample_rate':'16000',#抽样频率率 8000/16000
        'speech_rate':'0',#语速 -500~500
        'pitch_rate':'0',#语调 -500~500
        'tts_nus':'1',#0 - 通过参数合成语音,1 - 拼接原始录音
        'background_music_id':'1',#播放语音时可选背景音乐,0,1
        'background_music_offset':'0',#背景音乐播放偏移时长,毫秒。当启用背景音乐时生效
        'background_music_volume':'100'#背景音乐音量,当启用背景音乐时生效,0~100
    }
    httpProxy = utils.http_proxy(ak_id, ak_secret)
    filepath = httpProxy.sendTtsPost(text,ttsRequestParam,fileRootPath)
    print(filepath)
if __name__ == '__main__':
    # print(request())
    # print(query('2324ec1ed63549318b9477f1bf3eaf8a'))
    print( nlpTest())
    # print(ttsTest())
相关文章
|
18天前
|
并行计算 安全 Java
Python GIL(全局解释器锁)机制对多线程性能影响的深度分析
在Python开发中,GIL(全局解释器锁)一直备受关注。本文基于CPython解释器,探讨GIL的技术本质及其对程序性能的影响。GIL确保同一时刻只有一个线程执行代码,以保护内存管理的安全性,但也限制了多线程并行计算的效率。文章分析了GIL的必要性、局限性,并介绍了多进程、异步编程等替代方案。尽管Python 3.13计划移除GIL,但该特性至少要到2028年才会默认禁用,因此理解GIL仍至关重要。
97 16
Python GIL(全局解释器锁)机制对多线程性能影响的深度分析
|
27天前
|
缓存 Rust 算法
从混沌到秩序:Python的依赖管理工具分析
Python 的依赖管理工具一直没有标准化,主要原因包括历史发展的随意性、社区的分散性、多样化的使用场景、向后兼容性的挑战、缺乏统一治理以及生态系统的快速变化。依赖管理工具用于处理项目中的依赖关系,确保不同环境下的依赖项一致性,避免软件故障和兼容性问题。常用的 Python 依赖管理工具如 pip、venv、pip-tools、Pipenv、Poetry 等各有优缺点,选择时需根据项目需求权衡。新工具如 uv 和 Pixi 在性能和功能上有所改进,值得考虑。
84 35
|
28天前
|
数据采集 数据可视化 数据挖掘
金融波动率的多模型建模研究:GARCH族与HAR模型的Python实现与对比分析
本文探讨了金融资产波动率建模中的三种主流方法:GARCH、GJR-GARCH和HAR模型,基于SPY的实际交易数据进行实证分析。GARCH模型捕捉波动率聚类特征,GJR-GARCH引入杠杆效应,HAR整合多时间尺度波动率信息。通过Python实现模型估计与性能比较,展示了各模型在风险管理、衍生品定价等领域的应用优势。
251 66
金融波动率的多模型建模研究:GARCH族与HAR模型的Python实现与对比分析
|
1月前
|
人工智能 自然语言处理 测试技术
阿里云通义实验室自然语言处理方向负责人黄非:通义灵码2.0,迈入 Agentic AI
在通义灵码 2.0 发布会上,阿里云通义实验室自然语言处理方向负责人黄非分享了代码大模型的演进。过去一年来,随着大模型技术的发展,特别是智能体技术的深入应用,通义灵码也在智能体的基础上研发了针对于整个软件研发流程的不同任务的智能体,这里既包括单智能体,也包括多智能体合并框架,在这样的基础上我们研发了通义灵码2.0。
179 21
|
1月前
|
数据采集 缓存 API
python爬取Boss直聘,分析北京招聘市场
本文介绍了如何使用Python爬虫技术从Boss直聘平台上获取深圳地区的招聘数据,并进行数据分析,以帮助求职者更好地了解市场动态和职位需求。
|
1月前
|
机器学习/深度学习 运维 数据可视化
Python时间序列分析:使用TSFresh进行自动化特征提取
TSFresh 是一个专门用于时间序列数据特征自动提取的框架,支持分类、回归和异常检测等机器学习任务。它通过自动化特征工程流程,处理数百个统计特征(如均值、方差、自相关性等),并通过假设检验筛选显著特征,提升分析效率。TSFresh 支持单变量和多变量时间序列数据,能够与 scikit-learn 等库无缝集成,适用于大规模时间序列数据的特征提取与模型训练。其工作流程包括数据格式转换、特征提取和选择,并提供可视化工具帮助理解特征分布及与目标变量的关系。
75 16
Python时间序列分析:使用TSFresh进行自动化特征提取
|
1月前
|
机器学习/深度学习 数据可视化 数据挖掘
使用Python实现基于矩阵分解的长期事件(MFLEs)时间序列分析
在现代数据分析中,高维时间序列数据的处理和预测极具挑战性。基于矩阵分解的长期事件(MFLEs)分析技术应运而生,通过降维和时间序列特性结合,有效应对大规模数据。MFLE利用矩阵分解提取潜在特征,降低计算复杂度,过滤噪声,并发现主要模式。相比传统方法如ARIMA和深度学习模型如LSTM,MFLE在多变量处理、计算效率和可解释性上更具优势。通过合理应用MFLE,可在物联网、金融等领域获得良好分析效果。
64 0
使用Python实现基于矩阵分解的长期事件(MFLEs)时间序列分析
|
1月前
|
机器学习/深度学习 人工智能 分布式计算
我的阿里云社区年度总结报告:Python、人工智能与大数据领域的探索之旅
我的阿里云社区年度总结报告:Python、人工智能与大数据领域的探索之旅
117 35
|
1月前
|
数据可视化 算法 数据挖掘
Python时间序列分析工具Aeon使用指南
**Aeon** 是一个遵循 scikit-learn API 风格的开源 Python 库,专注于时间序列处理。它提供了分类、回归、聚类、预测建模和数据预处理等功能模块,支持多种算法和自定义距离度量。Aeon 活跃开发并持续更新至2024年,与 pandas 1.4.0 版本兼容,内置可视化工具,适合数据探索和基础分析任务。尽管在高级功能和性能优化方面有提升空间,但其简洁的 API 和完整的基础功能使其成为时间序列分析的有效工具。
80 37
Python时间序列分析工具Aeon使用指南
|
2月前
|
分布式计算 大数据 数据处理
技术评测:MaxCompute MaxFrame——阿里云自研分布式计算框架的Python编程接口
随着大数据和人工智能技术的发展,数据处理的需求日益增长。阿里云推出的MaxCompute MaxFrame(简称“MaxFrame”)是一个专为Python开发者设计的分布式计算框架,它不仅支持Python编程接口,还能直接利用MaxCompute的云原生大数据计算资源和服务。本文将通过一系列最佳实践测评,探讨MaxFrame在分布式Pandas处理以及大语言模型数据处理场景中的表现,并分析其在实际工作中的应用潜力。
116 2

热门文章

最新文章

推荐镜像

更多