找回密码
立即注册
搜索
热搜: 活动 交友 discuz
发新帖

15

积分

0

好友

1

主题
发表于 2024-10-8 11:02:59 | 查看: 162| 回复: 0
技术原理1. API接口介绍
科大讯飞提供了丰富的语音处理API,其中非实时语音转写API主要包括两个主要接口:
  • 上传接口:用于上传待转写的音频文件。
  • 结果查询接口:用于查询转写结果。
2. 签名机制
为了保证API的安全性,科大讯飞采用了签名机制。每次请求都需要生成一个签名字符串,该签名字符串由应用ID、时间戳和密钥通过特定的算法计算得到。
3. 请求流程
  • 上传音频文件:

    • 生成签名。
    • 构建上传参数。
    • 发送POST请求上传音频文件。
    • 获取上传成功的响应,包括订单ID。
  • 查询转写结果:

    • 使用订单ID构建查询参数。
    • 发送POST请求查询转写结果。
    • 定期查询,直到任务完成。

代码实现
1. 导入必要的库
  1. <b>import base64</b>
  2. <b>import hashlib</b>
  3. <b>import hmac</b>
  4. <b>import json</b>
  5. <b>import logging</b>
  6. <b>import os</b>
  7. <b>import time</b>
  8. <b>import requests</b>
  9. <b>import urllib.parse</b>
复制代码


2. 配置日志
  1. <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类
  1. <b>class RequestApi(object):</b>
  2. <b>    def __init__(self, appid, secret_key, upload_file_path):</b>
  3. <b>        """</b>
  4. <b>        初始化RequestApi类的实例。</b>

  5. <b>        :param appid: 应用ID</b>
  6. <b>        :param secret_key: 应用密钥</b>
  7. <b>        :param upload_file_path: 待上传文件的路径</b>
  8. <b>        """</b>
  9. <b>        if not all([appid, secret_key, upload_file_path]):</b>
  10. <b>            raise ValueError("appid, secret_key and upload_file_path must be provided.")</b>
  11. <b>        if not os.path.isfile(upload_file_path):</b>
  12. <b>            raise FileNotFoundError(f"File {upload_file_path} does not exist.")</b>

  13. <b>        self.appid = appid</b>
  14. <b>        self.secret_key = secret_key</b>
  15. <b>        self.upload_file_path = upload_file_path</b>
  16. <b>        self.ts = str(int(time.time()))  # 当前时间戳</b>
  17. <b>        self.signa = self.get_signa()  # 计算签名</b>

  18. <b>    def get_signa(self):</b>
  19. <b>        """</b>
  20. <b>        生成签名。</b>

  21. <b>        :return: 签名字符串</b>
  22. <b>        """</b>
  23. <b>        m2 = hashlib.md5()</b>
  24. <b>        m2.update((self.appid + self.ts).encode('utf-8'))</b>
  25. <b>        md5 = m2.hexdigest().encode('utf-8')</b>
  26. <b>        signa = hmac.new(self.secret_key.encode('utf-8'), md5, hashlib.sha1).digest()</b>
  27. <b>        signa = base64.b64encode(signa).decode('utf-8')</b>
  28. <b>        return signa</b>

  29. <b>    def upload(self):</b>
  30. <b>        """</b>
  31. <b>        上传文件到科大讯飞服务器。</b>

  32. <b>        :return: 上传结果</b>
  33. <b>        """</b>
  34. <b>        logging.info("开始上传文件...")</b>
  35. <b>        try:</b>
  36. <b>            file_len = os.path.getsize(self.upload_file_path)</b>
  37. <b>            file_name = os.path.basename(self.upload_file_path)</b>
  38. <b>            param_dict = {</b>
  39. <b>                'appId': self.appid,</b>
  40. <b>                'signa': self.signa,</b>
  41. <b>                'ts': self.ts,</b>
  42. <b>                'fileSize': file_len,</b>
  43. <b>                'fileName': file_name,</b>
  44. <b>                'duration': '200',  # 文件持续时间(秒)</b>
  45. <b>            }</b>

  46. <b>            with open(self.upload_file_path, 'rb') as f:</b>
  47. <b>                data = f.read(file_len)</b>

  48. <b>            response = requests.post(</b>
  49. <b>                url=lfasr_host + api_upload + "?" + urllib.parse.urlencode(param_dict),</b>
  50. <b>                headers={"Content-type": "application/octet-stream"},</b>
  51. <b>                data=data</b>
  52. <b>            )</b>
  53. <b>            response.raise_for_status()</b>
  54. <b>            result = response.json()</b>
  55. <b>            logging.info("文件上传成功,响应: %s", result)</b>
  56. <b>            return result</b>
  57. <b>        except Exception as e:</b>
  58. <b>            logging.error("文件上传失败: %s", str(e))</b>
  59. <b>            raise</b>

  60. <b>    def get_result(self, max_retries=10):</b>
  61. <b>        """</b>
  62. <b>        获取语音转写的结果。</b>

  63. <b>        :param max_retries: 最大重试次数</b>
  64. <b>        :return: 转写结果</b>
  65. <b>        """</b>
  66. <b>        logging.info("开始获取结果...")</b>
  67. <b>        try:</b>
  68. <b>            upload_response = self.upload()</b>
  69. <b>            order_id = upload_response['content']['orderId']</b>
  70. <b>            param_dict = {</b>
  71. <b>                'appId': self.appid,</b>
  72. <b>                'signa': self.signa,</b>
  73. <b>                'ts': self.ts,</b>
  74. <b>                'orderId': order_id,</b>
  75. <b>                'resultType': 'transfer',</b>
  76. <b>                'callbackUrl': 'https://dev.pdfxd.com/third/vorec/asr/callback'</b>
  77. <b>            }</b>
  78. <b>            status = 3</b>
  79. <b>            retries = 0</b>
  80. <b>            while status == 3 and retries < max_retries:</b>
  81. <b>                response = requests.post(</b>
  82. <b>                    url=lfasr_host + api_get_result + "?" + urllib.parse.urlencode(param_dict),</b>
  83. <b>                    headers={"Content-type": "application/json"}</b>
  84. <b>                )</b>
  85. <b>                response.raise_for_status()</b>
  86. <b>                result = response.json()</b>
  87. <b>                status = result['content']['orderInfo']['status']</b>
  88. <b>                logging.info("当前状态: %s", status)</b>
  89. <b>                if status == 4:</b>
  90. <b>                    break</b>
  91. <b>                retries += 1</b>
  92. <b>                time.sleep(5)</b>
  93. <b>            if status != 4:</b>
  94. <b>                logging.warning("达到最大重试次数,但任务状态仍为处理中。")</b>
  95. <b>            else:</b>
  96. <b>                logging.info("任务完成,获取结果...")</b>
  97. <b>                order_result = json.loads(result['content']['orderResult'])</b>
  98. <b>                lattice = order_result['lattice']</b>
  99. <b>                with open("result.txt", "w", encoding='utf-8') as file:</b>
  100. <b>                    for item in lattice:</b>
  101. <b>                        json_1best = item['json_1best']</b>
  102. <b>                        rt = json.loads(json_1best)["st"]["rt"]</b>
  103. <b>                        for word_segment in rt:</b>
  104. <b>                            for word in word_segment['ws']:</b>
  105. <b>                                for char in word['cw']:</b>
  106. <b>                                    file.write(char['w'])</b>
  107. <b>                logging.info("结果已保存至 result.txt")</b>
  108. <b>            return result</b>
  109. <b>        except Exception as e:</b>
  110. <b>            logging.error("获取结果失败: %s", str(e))</b>
  111. <b>            raise</b>
复制代码


4. 主程序
  1. <b>if __name__ == '__main__':</b>
  2. <b>    # 替换为你的appid和secret_key</b>
  3. <b>    api = RequestApi(</b>
  4. <b>        appid="your_app_id",</b>
  5. <b>        secret_key="your_secret_key",</b>
  6. <b>        upload_file_path=r"path_to_your_file.pcm"</b>
  7. <b>    )</b>
  8. <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文件中。


您需要登录后才可以回帖 登录 | 立即注册

Archiver|手机版|小黑屋|流星火雨Aioo

GMT+8, 2024-11-23 17:43 , Processed in 0.028178 second(s), 18 queries .

Powered by Discuz! X3.5

© 2001-2024 Discuz! Team.

快速回复 返回顶部 返回列表