技术原理1. API接口介绍 科大讯飞提供了丰富的语音处理API,其中非实时语音转写API主要包括两个主要接口: - 上传接口:用于上传待转写的音频文件。
- 结果查询接口:用于查询转写结果。
2. 签名机制为了保证API的安全性,科大讯飞采用了签名机制。每次请求都需要生成一个签名字符串,该签名字符串由应用ID、时间戳和密钥通过特定的算法计算得到。 3. 请求流程上传音频文件:
- 生成签名。
- 构建上传参数。
- 发送POST请求上传音频文件。
- 获取上传成功的响应,包括订单ID。
查询转写结果:
- 使用订单ID构建查询参数。
- 发送POST请求查询转写结果。
- 定期查询,直到任务完成。
代码实现
1. 导入必要的库
- <b>import base64</b>
- <b>import hashlib</b>
- <b>import hmac</b>
- <b>import json</b>
- <b>import logging</b>
- <b>import os</b>
- <b>import time</b>
- <b>import requests</b>
- <b>import urllib.parse</b>
复制代码
2. 配置日志
- <font face="-apple-system, blinkmacsystemfont, Helvetica Neue, helvetica, segoe ui, arial, roboto, PingFang SC, miui, Hiragino Sans GB, Microsoft Yahei, sans-serif"><font color="#2c2c36">logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')</font></font>
复制代码
3. 定义API类
- <b>class RequestApi(object):</b>
- <b> def __init__(self, appid, secret_key, upload_file_path):</b>
- <b> """</b>
- <b> 初始化RequestApi类的实例。</b>
- <b> :param appid: 应用ID</b>
- <b> :param secret_key: 应用密钥</b>
- <b> :param upload_file_path: 待上传文件的路径</b>
- <b> """</b>
- <b> if not all([appid, secret_key, upload_file_path]):</b>
- <b> raise ValueError("appid, secret_key and upload_file_path must be provided.")</b>
- <b> if not os.path.isfile(upload_file_path):</b>
- <b> raise FileNotFoundError(f"File {upload_file_path} does not exist.")</b>
- <b> self.appid = appid</b>
- <b> self.secret_key = secret_key</b>
- <b> self.upload_file_path = upload_file_path</b>
- <b> self.ts = str(int(time.time())) # 当前时间戳</b>
- <b> self.signa = self.get_signa() # 计算签名</b>
- <b> def get_signa(self):</b>
- <b> """</b>
- <b> 生成签名。</b>
- <b> :return: 签名字符串</b>
- <b> """</b>
- <b> m2 = hashlib.md5()</b>
- <b> m2.update((self.appid + self.ts).encode('utf-8'))</b>
- <b> md5 = m2.hexdigest().encode('utf-8')</b>
- <b> signa = hmac.new(self.secret_key.encode('utf-8'), md5, hashlib.sha1).digest()</b>
- <b> signa = base64.b64encode(signa).decode('utf-8')</b>
- <b> return signa</b>
- <b> def upload(self):</b>
- <b> """</b>
- <b> 上传文件到科大讯飞服务器。</b>
- <b> :return: 上传结果</b>
- <b> """</b>
- <b> logging.info("开始上传文件...")</b>
- <b> try:</b>
- <b> file_len = os.path.getsize(self.upload_file_path)</b>
- <b> file_name = os.path.basename(self.upload_file_path)</b>
- <b> param_dict = {</b>
- <b> 'appId': self.appid,</b>
- <b> 'signa': self.signa,</b>
- <b> 'ts': self.ts,</b>
- <b> 'fileSize': file_len,</b>
- <b> 'fileName': file_name,</b>
- <b> 'duration': '200', # 文件持续时间(秒)</b>
- <b> }</b>
- <b> with open(self.upload_file_path, 'rb') as f:</b>
- <b> data = f.read(file_len)</b>
- <b> response = requests.post(</b>
- <b> url=lfasr_host + api_upload + "?" + urllib.parse.urlencode(param_dict),</b>
- <b> headers={"Content-type": "application/octet-stream"},</b>
- <b> data=data</b>
- <b> )</b>
- <b> response.raise_for_status()</b>
- <b> result = response.json()</b>
- <b> logging.info("文件上传成功,响应: %s", result)</b>
- <b> return result</b>
- <b> except Exception as e:</b>
- <b> logging.error("文件上传失败: %s", str(e))</b>
- <b> raise</b>
- <b> def get_result(self, max_retries=10):</b>
- <b> """</b>
- <b> 获取语音转写的结果。</b>
- <b> :param max_retries: 最大重试次数</b>
- <b> :return: 转写结果</b>
- <b> """</b>
- <b> logging.info("开始获取结果...")</b>
- <b> try:</b>
- <b> upload_response = self.upload()</b>
- <b> order_id = upload_response['content']['orderId']</b>
- <b> param_dict = {</b>
- <b> 'appId': self.appid,</b>
- <b> 'signa': self.signa,</b>
- <b> 'ts': self.ts,</b>
- <b> 'orderId': order_id,</b>
- <b> 'resultType': 'transfer',</b>
- <b> 'callbackUrl': 'https://dev.pdfxd.com/third/vorec/asr/callback'</b>
- <b> }</b>
- <b> status = 3</b>
- <b> retries = 0</b>
- <b> while status == 3 and retries < max_retries:</b>
- <b> response = requests.post(</b>
- <b> url=lfasr_host + api_get_result + "?" + urllib.parse.urlencode(param_dict),</b>
- <b> headers={"Content-type": "application/json"}</b>
- <b> )</b>
- <b> response.raise_for_status()</b>
- <b> result = response.json()</b>
- <b> status = result['content']['orderInfo']['status']</b>
- <b> logging.info("当前状态: %s", status)</b>
- <b> if status == 4:</b>
- <b> break</b>
- <b> retries += 1</b>
- <b> time.sleep(5)</b>
- <b> if status != 4:</b>
- <b> logging.warning("达到最大重试次数,但任务状态仍为处理中。")</b>
- <b> else:</b>
- <b> logging.info("任务完成,获取结果...")</b>
- <b> order_result = json.loads(result['content']['orderResult'])</b>
- <b> lattice = order_result['lattice']</b>
- <b> with open("result.txt", "w", encoding='utf-8') as file:</b>
- <b> for item in lattice:</b>
- <b> json_1best = item['json_1best']</b>
- <b> rt = json.loads(json_1best)["st"]["rt"]</b>
- <b> for word_segment in rt:</b>
- <b> for word in word_segment['ws']:</b>
- <b> for char in word['cw']:</b>
- <b> file.write(char['w'])</b>
- <b> logging.info("结果已保存至 result.txt")</b>
- <b> return result</b>
- <b> except Exception as e:</b>
- <b> logging.error("获取结果失败: %s", str(e))</b>
- <b> raise</b>
复制代码
4. 主程序
- <b>if __name__ == '__main__':</b>
- <b> # 替换为你的appid和secret_key</b>
- <b> api = RequestApi(</b>
- <b> appid="your_app_id",</b>
- <b> secret_key="your_secret_key",</b>
- <b> upload_file_path=r"path_to_your_file.pcm"</b>
- <b> )</b>
- <b> api.get_result()</b>
复制代码
运行步骤
1.安装依赖:pip install requests
2.配置参数:将your_app_id和your_secret_key替换为你的科大讯飞应用ID和密钥。
将path_to_your_file.pcm替换为你要上传的音频文件路径。
3.运行代码: python your_script.py
结果输出运行脚本后,音频文件将被上传到科大讯飞服务器进行转写。转写完成后,结果将被保存到result.txt文件中。
|