音频转文字分析
工作里面需要用到音频转成文字做数据分析,自己就想着测试下看看,现在Python音频转文字的方法用很多种,其中比较常用的有:
1.调用百度AI,科大讯飞,google等api
2.自己做数据挖掘,利用神经网络等算法训练
因为自己去做音频转化需要非常专业的音频方面的知识,不是研究这方面的数据分析师是很难做出很好的算法,而百度和科大讯飞本身已经做的很好的了,本身也赠送了一定的免费使用时间,我这里就使用他们的api做一下调用,分析下他们的转化效果.
PS:我这里的代码是使用百度和科大讯飞的demo去写的,感兴趣的可以直接去官方查看,链接如下:
科大讯飞:https://www.xfyun.cn/
百度ai:https://ai.baidu.com/ai-doc/
转化的文字在下面,这里直接分析的结果:
- 百度翻译的内容完全看不懂啊…
- 科大讯飞不愧是语音起家的,翻译的结果确实比百度ai的要好一点,但是涉及专业的部分及语音问题仍然不够好
- 如果使用自己公司的数据库,添加专业用词,进行算法训练,建立模型,可能会对实际的业务有更好的适用性,需要测试
- 隐私内容已用xxxxx消敏
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import requests
import json
import sys
import base64
import hashlib
import hmac
import os
import time
import datetime
import re
import ssl
#百度api
IS_PY3 = sys.version_info.major == 3
if IS_PY3:
from urllib.request import urlopen
from urllib.request import Request
from urllib.error import URLError
from urllib.parse import urlencode
timer = time.perf_counter
else:
from urllib2 import urlopen
from urllib2 import Request
from urllib2 import URLError
from urllib import urlencode
if sys.platform == "win32":
timer = time.clock
else:
# On most other platforms the best timer is time.time()
timer = time.time
import ssl
ssl._create_default_https_context = ssl._create_unverified_context
#填写百度控制台中相关开通了“音频文件转写”接口的应用的的API_KEY及SECRET_KEY
API_KEY = 'xxxxx'
SECRET_KEY = 'xxxxx'
""" 获取请求TOKEN start 通过开通音频文件转写接口的百度应用的API_KEY及SECRET_KEY获取请求token"""
class DemoError(Exception):
pass
TOKEN_URL = 'https://openapi.baidu.com/oauth/2.0/token'
# SCOPE = 'brain_bicc' # 有此scope表示有asr能力，没有请在网页里勾选 bicc
SCOPE = 'brain_asr_async' # 有此scope表示有asr能力，没有请在网页里勾选
# SCOPE = 'brain_enhanced_asr' # 有此scope表示有asr能力，没有请在网页里勾选
def fetch_token():
params = { 'grant_type': 'client_credentials',
'client_id': API_KEY,
'client_secret': SECRET_KEY}
post_data = urlencode(params)
if (IS_PY3):
post_data = post_data.encode( 'utf-8')
req = Request(TOKEN_URL, post_data)
try:
f = urlopen(req)
result_str = f.read()
except URLError as err:
print('token http response http code : ' + str(err.code))
result_str = err.read()
if (IS_PY3):
result_str = result_str.decode()
# print(result_str)
result = json.loads(result_str)
# print(result)
if ('access_token' in result.keys() and 'scope' in result.keys()):
if not SCOPE in result['scope'].split(' '):
raise DemoError('scope is not correct')
# print('SUCCESS WITH TOKEN: %s ; EXPIRES IN SECONDS: %s' % (result['access_token'], result['expires_in']))
return result['access_token']
else:
raise DemoError('MAYBE API_KEY or SECRET_KEY not correct: access_token or scope not found in token response')
""" 获取鉴权结束，TOKEN end """
""" 发送查询结果请求 """
#转写任务id列表，task_id是通过创建音频转写任务时获取到的，每个音频任务对应的值
task_id_list = [
"xxxxx"
]
for task_id in task_id_list:
url = 'https://aip.baidubce.com/rpc/2.0/aasr/v1/query' #查询音频任务转写结果请求地址
body = {
"task_ids": [task_id],
}
token = { "access_token":fetch_token()}
headers = { 'content-type': "application/json"}
response = requests.post(url,params=token,data = json.dumps(body), headers = headers)
# 科大讯飞
lfasr_host = 'http://raasr.xfyun.cn/api'
# 请求的接口名
api_prepare = '/prepare'
api_upload = '/upload'
api_merge = '/merge'
api_get_progress = '/getProgress'
api_get_result = '/getResult'
# 文件分片大小10M
file_piece_sice = 10485760
# ——————————————————转写可配置参数————————————————
# 参数可在官网界面（https://doc.xfyun.cn/rest_api/%E8%AF%AD%E9%9F%B3%E8%BD%AC%E5%86%99.html）查看，根据需求可自行在gene_params方法里添加修改
# 转写类型
lfasr_type = 0
# 是否开启分词
has_participle = 'false'
has_seperate = 'true'
# 多候选词个数
max_alternatives = 0
# 子用户标识
suid = ''
class SliceIdGenerator:
"""slice id生成器"""
def __init__(self):
self.__ch = 'aaaaaaaaa`'
def getNextSliceId(self):
ch = self.__ch
j = len(ch) - 1
while j >= 0:
cj = ch[j]
if cj != 'z':
ch = ch[:j] + chr(ord(cj) + 1) + ch[j + 1:]
break
else:
ch = ch[:j] + 'a' + ch[j + 1:]
j = j - 1
self.__ch = ch
return self.__ch
class RequestApi(object):
def __init__(self, appid, secret_key, upload_file_path):
self.appid = appid
self.secret_key = secret_key
self.upload_file_path = upload_file_path
# 根据不同的apiname生成不同的参数,本示例中未使用全部参数您可在官网(https://doc.xfyun.cn/rest_api/%E8%AF%AD%E9%9F%B3%E8%BD%AC%E5%86%99.html)查看后选择适合业务场景的进行更换
def gene_params(self, apiname, taskid=None, slice_id=None):
appid = self.appid
secret_key = self.secret_key
upload_file_path = self.upload_file_path
ts = str(int(time.time()))
m2 = hashlib.md5()
m2.update((appid + ts).encode('utf-8'))
md5 = m2.hexdigest()
md5 = bytes(md5, encoding='utf-8')
# 以secret_key为key, 上面的md5为msg， 使用hashlib.sha1加密结果为signa
signa = hmac.new(secret_key.encode('utf-8'), md5,
hashlib.sha1).digest()
signa = base64.b64encode(signa)
signa = str(signa, 'utf-8')
file_len = os.path.getsize(upload_file_path)
file_name = os.path.basename(upload_file_path)
param_dict = { }
if apiname == api_prepare:
# slice_num是指分片数量，如果您使用的音频都是较短音频也可以不分片，直接将slice_num指定为1即可
slice_num = int(file_len / file_piece_sice) + (0 if (
file_len % file_piece_sice == 0) else 1)
param_dict['app_id'] = appid
param_dict['signa'] = signa
param_dict['ts'] = ts
param_dict['file_len'] = str(file_len)
param_dict['file_name'] = file_name
param_dict['slice_num'] = str(slice_num)
elif apiname == api_upload:
param_dict['app_id'] = appid
param_dict['signa'] = signa
param_dict['ts'] = ts
param_dict['task_id'] = taskid
param_dict['slice_id'] = slice_id
elif apiname == api_merge:
param_dict['app_id'] = appid
param_dict['signa'] = signa
param_dict['ts'] = ts
param_dict['task_id'] = taskid
param_dict['file_name'] = file_name
elif apiname == api_get_progress or apiname == api_get_result:
param_dict['app_id'] = appid
param_dict['signa'] = signa
param_dict['ts'] = ts
param_dict['task_id'] = taskid
return param_dict
# 请求和结果解析，结果中各个字段的含义可参考：https://doc.xfyun.cn/rest_api/%E8%AF%AD%E9%9F%B3%E8%BD%AC%E5%86%99.html
def gene_request(self, apiname, data, files=None, headers=None):
response = requests.post(lfasr_host + apiname,
data=data,
files=files,
headers=headers)
result = json.loads(response.text)
if result["ok"] == 0:
print("{} success:".format(apiname) + str(result))
return result
else:
print("{} error:".format(apiname) + str(result))
exit(0)
return result
# 预处理
def prepare_request(self):
return self.gene_request(apiname=api_prepare,
data=self.gene_params(api_prepare))
# 上传
def upload_request(self, taskid, upload_file_path):
file_object = open(upload_file_path, 'rb')
try:
index = 1
sig = SliceIdGenerator()
while True:
content = file_object.read(file_piece_sice)
if not content or len(content) == 0:
break
files = {
"filename": self.gene_params(api_upload).get("slice_id"),
"content": content
}
response = self.gene_request(
api_upload,
data=self.gene_params(api_upload,
taskid=taskid,
slice_id=sig.getNextSliceId()),
files=files)
if response.get('ok') != 0:
# 上传分片失败
print('upload slice fail, response: ' + str(response))
return False
print('upload slice ' + str(index) + ' success')
index += 1
finally:
'file index:' + str(file_object.tell())
file_object.close()
return True
# 合并
def merge_request(self, taskid):
return self.gene_request(api_merge,
data=self.gene_params(api_merge,
taskid=taskid))
# 获取进度
def get_progress_request(self, taskid):
return self.gene_request(api_get_progress,
data=self.gene_params(api_get_progress,
taskid=taskid))
# 获取结果
def get_result_request(self, taskid):
return self.gene_request(api_get_result,
data=self.gene_params(api_get_result,
taskid=taskid))
def all_api_request(self):
# 1. 预处理
pre_result = self.prepare_request()
taskid = pre_result["data"]
# 2 . 分片上传
self.upload_request(taskid=taskid,
upload_file_path=self.upload_file_path)
# 3 . 文件合并
self.merge_request(taskid=taskid)
# 4 . 获取任务进度
while True:
# 每隔20秒获取一次任务进度
progress = self.get_progress_request(taskid)
progress_dic = progress
if progress_dic['err_no'] != 0 and progress_dic['err_no'] != 26605:
print('task error: ' + progress_dic['failed'])
return
else:
data = progress_dic['data']
task_status = json.loads(data)
if task_status['status'] == 9:
print('task ' + taskid + ' finished')
break
print('The task ' + taskid +
' is in processing, task status: ' + str(data))
# 每次获取进度间隔20S
time.sleep(20)
APP_ID = "xxxxx"
SECRET_KEY = "xxxxx"
file_path = r"xxxxx"
api = RequestApi(appid=APP_ID, secret_key=SECRET_KEY, upload_file_path=file_path)
api.all_api_request()
# 注意：如果出现requests模块报错："NoneType" object has no attribute 'read', 请尝试将requests模块更新到2.20.0或以上版本(本demo测试版本为2.20.0)
# 输入讯飞开放平台的appid，secret_key和待转写的文件路径
/prepare success:{'data': 'xxxxx ', 'err_no': 0, 'failed': None, 'ok': 0}
/upload success:{'data': None, 'err_no': 0, 'failed': None, 'ok': 0}
upload slice 1 success
/merge success:{'data': None, 'err_no': 0, 'failed': None, 'ok': 0}
/getProgress success:{'data': '{"status":2,"desc":"音频合并完成"}', 'err_no': 0, 'failed': None, 'ok': 0}
The task xxxxx is in processing, task status: {"status":2,"desc":"音频合并完成"}
/getProgress success:{'data': '{"status":3,"desc":"音频转写中"}', 'err_no': 0, 'failed': None, 'ok': 0}
The task xxxxx is in processing, task status: {"status":3,"desc":"音频转写中"}
/getProgress success:{'data': '{"status":9,"desc":"转写结果上传完成"}', 'err_no': 0, 'failed': None, 'ok': 0}
task xxxxx finished
# 科大讯飞输出结果,提取输出文字
kx_result = api.get_result_request(taskid='xxxxx')
kx_result1 = json.loads(kx_result['data'])
kx_result2 = [i['onebest'] for i in kx_result1]
kx_result2
# 百度ai输出结果,提取输出文字
baiduai_result = json.dumps(response.json(), ensure_ascii=False)
baiduai_result1 = json.loads(baiduai_result)
baiduai_result2 = baiduai_result1['tasks_info'][-1]['task_result']['result']
baiduai_result2
然后这个这个客服中心，请问是什么？八零，嗯， 你好，我这边买了那个xxxxx的那个设定了这个时间哦，跟你咨询一下。那个啊，我现在就是把那个时间已经校长了423 11分，今天是星期五，然后呢，我是按照那个定时。就是我按照是那个唉。就是按照那个定时设定的，是从周一到周五早晨八点半，然后到晚上六点钟早早晨八点半起，然后晚上六点钟停。然后我就设定了然后电源电源线我也接上去了然后手动那时候的他们不是有那个活动嘛，对吧？手动手动开手动关都是好的，就是我再把他带回到那个自动的时候，嗯，他们有没有启动吧？自动车没启动对。因为现在是12点12点半了，中午12点半对吧？嗯，我做的那个时间，今天又是周五嘛，说的时间去了，这个时间段时间，现在的我这样子，她没有启动。没有启动，对不对？我按那个手动自动那个在调那个开和关都可以。呃，十元是吧说着。怎么了？让他下哦。说那些啊！这个时间是比较长时间。之类的。我看好了。进入磁场。星期星期六，星期六，星期六。我我那个就是按照那个定时之后啊，他现在选择的是一开，然后上面显示的是，呃，周一周二周三周四周五。你先这个做好以后，然后等下等下等下要不要开学了？你吃什么？我这样说你们。弄好以后，然后。下面显示auto吗？哦，先上a写字楼。建设是吧对。那时候说话也不知道了。你这个就是我，我是我这样，我这样说你不知道你能够能够理解它不是第一就是设定的，一开始一路开第一路关闭第一路关闭第一路上的时间，从时间从周一是周一，周二，周三，周四，周五，周日这样的，那他是不是？就是说他就要检测到今天是周一到周五之间的话，嗯，在这个点都会启动和坚持是个概念嘛？就是你四个时间段当中会停止和关闭。那我是不是准备问题啊？那倒没有，这么看你是没有充电吗？充电要什么充电方面，电池一直有的呀，对吧？对呀，然后现在上面的电源也插进去了呀。而且我手动手动的话，你那个左下角的那个自动一一行让手动那个那个也是正常的呀。收到消息时，你先等一下，是xxxxx是吧？xxxxx其送到xxxxx先生。105 KO 5睇龙牛肉馆压罗罗罗罗斯鸡肉像是高高兴兴的时候报核心是吧？你把他分分心就不感到心里不？那现在这些是什么东西呢？行了行了。继续啊，对对对对对对。对对对对。就能20。要不要王者做了？嗯嗯，继续的时候是吧？我涨死了。这个问题对不对啊？我去死，谁来管呀？嗯，关键所有的在线。没有。官思索的说的是有，就是那一个。xxxxx。是不是？一起。就要6k了。走路过来。哦。嗯好不行。好不行。为。为。这是不认识，我认识。这锁定。不用。真是。是的。真是。哦。说你。是不是？吃饭。你们是说的，所以。是啊！看到这个叫做把它撕开，就是跟着一个这个实验才知道他在吃吃吃，他不用去。’
print(kx_result2)
‘您好，这里xxxxx客服中心，请问什么帮您？ ‘, ‘呃你好，我这边买的那个xxxxx的’, ‘那个设定的这个事情，我跟你咨询一下。’, ‘那个行。’, ‘呃我现在就是把那个时间已经校准了，是xxxxx，今天是xxxxx，’, ‘然后呢我是按照那个定时’, ‘就是我按的是那个’, ‘就是按照那个定时设定的是从周一到周五早晨8:30，然后到’, ‘嗯晚上xxxxx起，然后晚上6:00 6:00停，’, ‘然后我就设定了然后xxxxx，我也接上去了。’, ‘然后手动的是好的，它上面不是有那个手动，嘛’, ‘对吧？’, ‘主动’, ‘手动开手动关都是好的。’, ‘就是’, ‘我再把它再归到那个自动的时候，’, ‘呃他没有没有启动，吧’, ‘自动车没启动，’, ‘因为现在是xxxxx了，中午xxxxx了对吧？’, ‘嗯我做的那个时间今天就是周五嘛做的时间，就在这个时间段，’, ‘直接’, ‘现在他我这样去他没有启动。’, ‘没有启动，’, ‘我按那个手动自动那个再调那个开和关都可以。’, ‘嗯史云是吧？所指？’, ‘什么？啊’, ‘好吧？’, ‘这个时间是比较’, ‘我没有地方。’, ‘是的了。’, ‘那太好了。’, ‘呃’, ‘坚持啊’, ‘星期一星期二、星期三、星期四、星期五，’, ‘我我那个就是按照那个定时之后，啊’, ‘它现在显示的是一开，然后上面显示的是’, ‘呃周一周二周三周四周五’, ‘先这个测试好以后，然后’, ‘等一下说不定要不要开学了。’, ‘我这样说你们’, ‘弄好以后，然后’, ‘上面显示xxxxx吗？’, ‘呃选项a选择了a’, ‘写册是吧？’, ‘对。’, ‘凭什么去外部了？’, ‘你这个就是我我是我这样我这样说您不知道您能不能够理解他不是。’, ‘第一就是设定了一开对吧？’, ‘第一路开第一路关我第一路我上面时间从是从’, ‘周一周一周二周三周四周五都是这样的。’, ‘那他是不是’, ‘就是说它就要检测到今天是周一’, ‘到周五之间的话，’, ‘呃在这个点都会启动和停止，’, ‘是这个概念吗？’, ‘就是你是不是那个时间段都会停止和关闭？’, ‘那我是不是就没有问题？’, ‘啊’, ‘嗯好，没有这种卡你是没有充电吗？’, ‘充电要什么充电它的电池一直有的。呀’, ‘记住了是吧？’, ‘对，啊然后现在上面的电源也接进去了。呀’, ‘而且我手动手动的话，你那个左下角的那个’, ‘自动一一横杠手动，’, ‘那个那个也是正常的。’, ‘呀’, ‘稍等一下先生，您先等一下，是xxxxx是吧？’, ‘对。’, xxxxx, ‘现在稍等一下先生。’, ‘xxxxx。’, ‘估计农民用款呀农民用了，’, ‘是这一步。’, ‘最高兴就是了报’, ‘核心内容。啊’, ‘你把它这个东西你不’, ‘感到心里有数。’, ‘那那行。’, ‘这这就是人性的问题。’, ‘所以’, ‘就是’, ‘就是’, ‘噢对’, ‘噢对。’, ‘在那就好。’, ‘唉对。’, ‘嗯’, ‘嗯啊’, ‘啊’, ‘只要能转账了。’, ‘嗯’, ‘我就写着说是吧？’, ‘因为’, ‘可对。啊’, ‘我就是谁。’, ‘喂喂’, ‘嗯或者是说你赚钱，’, ‘唉就’, ‘不管是说的说的是哟就是哪一个，’, ‘啊’, ’ sorry，’, ‘是不是’, ‘啊’, ‘嗯’, ‘知道吗？’, ‘做了不少。’, ‘对，还不清。’, ‘他们去。’, ‘喂’, ‘喂’, ‘真是的。’, ‘真的是否定。’, ‘对。嗯’, ‘嗯’, ‘平时’, ‘是的。’, ‘就是’, ‘噢’, ’ Sorry’, ‘是不是？啊’, ‘共xxxxx，’, ‘现在就是跳’, ‘xxxxx，’, ‘他就不是要到这个时间才去了，’, ‘啊这些事事件还不如去。’
