Python讯飞语音转文字保存到文件

因为有朋友需要将录音转成文字, 给我的是m4a格式, 我给转成txt发给他的.

我找了找, 发现网上很多都是收费软件,而且转换结果不尽人意., 最后决定使用讯飞服务来完成转换, 讯飞语音转写api支持大文件, 转换结果也十分准确, 感谢讯飞及制作demo的大佬yanmeng2, 嘻嘻

1先注册账号, 获取key…

在讯飞开放平台(https://passport.xfyun.cn/login)注册账号, 然后创建应用.

再去新用户送5小时转换时间, 需要去领取 (http://www.xfyun.cn/services/lfasr).

选择已创建应用程序的已收到奖励

[En]

Select the received reward to the created application

进入应用(如下页面)拿到需要的参数

Python讯飞语音转文字保存到文件

; 2 修改代码, 将值输入并确定,

Python讯飞语音转文字保存到文件

3 执行

打开cmd, 输入 python D:\workspace\python\weblfasr_python3_demo\weblfasr_python3_demo.py

Python讯飞语音转文字保存到文件

执行完毕后, 就可以在 代码设置的to_file中看到结果了

源码:


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'

file_piece_sice = 10485760

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, to_file):
        self.appid = appid
        self.secret_key = secret_key
        self.upload_file_path = upload_file_path
        self.to_file = to_file

    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')

        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 = 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)
            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):

        pre_result = self.prepare_request()
        taskid = pre_result["data"]

        self.upload_request(taskid=taskid, upload_file_path=self.upload_file_path)

        self.merge_request(taskid=taskid)

        while True:

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

            time.sleep(20)

        res = self.get_result_request(taskid=taskid)
        r = json.loads(res["data"])
        with open(self.to_file,'w', encoding='utf-8') as file_obj:
            for iterating_var in r:
                file_obj.write(iterating_var["onebest"])
                file_obj.write("\n")

if __name__ == '__main__':
    api = RequestApi(
        appid="xxx",
        secret_key="xxx",
        upload_file_path=r"C:/Users/DELL/Desktop/录音文件/20201104_163233.m4a",
        to_file=r"C:/Users/DELL/Desktop/录音文件/20201104_163233.txt"
    )
    api.all_api_request()

Original: https://blog.csdn.net/yynan555/article/details/109536995
Author: yynan555
Title: Python讯飞语音转文字保存到文件

原创文章受到原创版权保护。转载请注明出处:https://www.johngo689.com/525937/

转载文章受原作者版权保护。转载请注明原作者出处!

(0)

大家都在看

亲爱的 Coder【最近整理,可免费获取】👉 最新必读书单  | 👏 面试题下载  | 🌎 免费的AI知识星球