周五在公司内部做了一个分享，手机全程录制的音频，想着整理成文字稿，但整个分享会时间 50 分钟，内容太多，就想着剥削下电脑吧，再次感慨下，懒人真是适合当程序员。
先在网上找找有没有现成的脚本可用，逛了一圈发现没有太好的，还是要亲力亲为。百度 API 和讯飞 API 是相对成熟的，开始一直鼓捣百度语音 API 接口，但莫名的 bug 一直卡着不动，就改为使用讯飞 API 了。这个开发文档写的比较完善，Demo 脚本也详细，研究了两个小时，就把我 50min 的讲座内容存到 txt 文件中了。本脚本直接使用的是讯飞提供的完整 Python 脚本，只是这个脚本不能将转录的文字存储到文件中，故结尾加了一个文件写入函数而已。
操作步骤：
- 在讯飞开放平台注册账号，点击「语音识别」，新建一个应用，再选择「语音转写」，获取密钥；
- 查看音频采样率，我用苹果手机录制的音频采样率为 48k，需要降到 16k 才能处理，音频属性需要 16k 或 8k；
- 因需要使用 Requests 请求接口，故需要联网运行；
- 更改 RequestApi 的三个输入参数，运行脚本；
- 脚本运行完毕后，会生成一个 txt 文本。转录好的文字，就存在这个文件中。
需要注意的是，讯飞提供的免费转录时间只有几个小时，之后就要收费了，所以谨慎使用。转录出来的效果差强人意，识别的不是很精准，错误较多，不过好在不用自己一个字一个字的敲了，还是能凑合用一下的，如果收费的话，那感觉真心不值。
# -*- coding: utf-8 -*-
import base64
import hashlib
import hmac
import json
import os
import time
import requests
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
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
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
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)
# 预处理
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)
# 5 . 获取结果
return self.get_result_request(taskid=taskid)
# requests版本可能要在2.20.0以上
# 输入讯飞开放平台的appid，secret_key和待转写的文件路径
if __name__ == '__main__':
api = RequestApi(appid="f3b985fe", secret_key="d46271748fc66b0fd29c7a561fb9371d", upload_file_path="voice6.wav")
txtResult = api.all_api_request()
fileToSave = open('speech2Word.txt', 'w')
listData = eval(txtResult['data'])
for ii in range(len(listData)):
print(listData[ii]['onebest'])
fileToSave.write(listData[ii]['onebest'])
# print(dictText['data'][1]['onebest'])
fileToSave.close()
参考文档：讯飞 API_语音听写