本示例的過程是：
1. 音頻轉文本
2. 利用文本獲取情感傾向分析結果
3. 利用文本獲取關鍵詞提取
首先是訊飛的語音識別模塊。在這里可以找到非實時語音轉寫的相關文檔以及 Python 示例。我略作了改動，讓它可以對不同人說話作區分，並且作了一些封裝。
語音識別功能
weblfasr_python3_demo.py 文件：
1 #!/usr/bin/env python 2 # -*- coding: utf-8 -*- 3 """ 4 訊飛非實時轉寫調用demo（語音識別） 5 """ 6 import base64 7 import hashlib 8 import hmac 9 import json 10 import os 11 import time 12 13 import requests 14 15 lfasr_host = 'http://raasr.xfyun.cn/api' 16 17 # 請求的接口名 18 api_prepare = '/prepare' 19 api_upload = '/upload' 20 api_merge = '/merge' 21 api_get_progress = '/getProgress' 22 api_get_result = '/getResult' 23 # 文件分片大下52k 24 file_piece_sice = 10485760 25 26 # ——————————————————轉寫可配置參數———————————————— 27 # 參數可在官網界面（https://doc.xfyun.cn/rest_api/%E8%AF%AD%E9%9F%B3%E8%BD%AC%E5%86%99.html）查看，根據需求可自行在gene_params方法里添加修改 28 # 轉寫類型 29 lfasr_type = 0 30 # 是否開啟分詞 31 has_participle = 'false' 32 has_seperate = 'true' 33 # 多候選詞個數 34 max_alternatives = 0 35 # 子用戶標識 36 suid = '' 37 38 39 class SliceIdGenerator: 40 """slice id生成器""" 41 42 def __init__(self): 43 self.__ch = 'aaaaaaaaa`' 44 45 def getNextSliceId(self): 46 ch = self.__ch 47 j = len(ch) - 1 48 while j >= 0: 49 cj = ch[j] 50 if cj != 'z': 51 ch = ch[:j] + chr(ord(cj) + 1) + ch[j + 1:] 52 break 53 else: 54 ch = ch[:j] + 'a' + ch[j + 1:] 55 j = j - 1 56 self.__ch = ch 57 return self.__ch 58 59 60 class RequestApi(object): 61 def __init__(self, appid, secret_key, upload_file_path): 62 self.appid = appid 63 self.secret_key = secret_key 64 self.upload_file_path = upload_file_path 65 66 # 根據不同的apiname生成不同的參數,本示例中未使用全部參數您可在官網(https://doc.xfyun.cn/rest_api/%E8%AF%AD%E9%9F%B3%E8%BD%AC%E5%86%99.html)查看后選擇適合業務場景的進行更換 67 def gene_params(self, apiname, taskid=None, slice_id=None): 68 appid = self.appid 69 secret_key = self.secret_key 70 upload_file_path = self.upload_file_path 71 ts = str(int(time.time())) 72 m2 = hashlib.md5() 73 m2.update((appid + ts).encode('utf-8')) 74 md5 = m2.hexdigest() 75 md5 = bytes(md5, encoding='utf-8') 76 # 以secret_key為key, 上面的md5為msg， 使用hashlib.sha1加密結果為signa 77 signa = hmac.new(secret_key.encode('utf-8'), md5, hashlib.sha1).digest() 78 signa = base64.b64encode(signa) 79 signa = str(signa, 'utf-8') 80 file_len = os.path.getsize(upload_file_path) 81 file_name = os.path.basename(upload_file_path) 82 param_dict = {} 83 84 if apiname == api_prepare: 85 # slice_num是指分片數量，如果您使用的音頻都是較短音頻也可以不分片，直接將slice_num指定為1即可 86 slice_num = int(file_len / file_piece_sice) + (0 if (file_len % file_piece_sice == 0) else 1) 87 param_dict['app_id'] = appid 88 param_dict['signa'] = signa 89 param_dict['ts'] = ts 90 param_dict['file_len'] = str(file_len) 91 param_dict['file_name'] = file_name 92 param_dict['slice_num'] = str(slice_num) 93 elif apiname == api_upload: 94 param_dict['app_id'] = appid 95 param_dict['signa'] = signa 96 param_dict['ts'] = ts 97 param_dict['task_id'] = taskid 98 param_dict['slice_id'] = slice_id 99 elif apiname == api_merge: 100 param_dict['app_id'] = appid 101 param_dict['signa'] = signa 102 param_dict['ts'] = ts 103 param_dict['task_id'] = taskid 104 param_dict['file_name'] = file_name 105 elif apiname == api_get_progress or apiname == api_get_result: 106 param_dict['app_id'] = appid 107 param_dict['signa'] = signa 108 param_dict['ts'] = ts 109 param_dict['task_id'] = taskid 110 param_dict['has_seperate'] = has_seperate 111 return param_dict 112 113 # 請求和結果解析，結果中各個字段的含義可參考：https://doc.xfyun.cn/rest_api/%E8%AF%AD%E9%9F%B3%E8%BD%AC%E5%86%99.html 114 def gene_request(self, apiname, data, files=None, headers=None): 115 response = requests.post(lfasr_host + apiname, data=data, files=files, headers=headers) 116 result = json.loads(response.text) 117 if result["ok"] == 0: 118 # print("{} success:".format(apiname) + str(result)) 119 print('treating...') 120 return result 121 else: 122 # print("{} error:".format(apiname) + str(result)) 123 exit(0) 124 return result 125 126 # 預處理 127 def prepare_request(self): 128 return self.gene_request(apiname=api_prepare, 129 data=self.gene_params(api_prepare)) 130 131 # 上傳 132 def upload_request(self, taskid, upload_file_path): 133 file_object = open(upload_file_path, 'rb') 134 try: 135 index = 1 136 sig = SliceIdGenerator() 137 while True: 138 content = file_object.read(file_piece_sice) 139 if not content or len(content) == 0: 140 break 141 files = { 142 "filename": self.gene_params(api_upload).get("slice_id"), 143 "content": content 144 } 145 response = self.gene_request(api_upload, 146 data=self.gene_params(api_upload, taskid=taskid, 147 slice_id=sig.getNextSliceId()), 148 files=files) 149 if response.get('ok') != 0: 150 # 上傳分片失敗 151 print('upload slice fail, response: ' + str(response)) 152 return False 153 # print('upload slice ' + str(index) + ' success') 154 print('treating...') 155 index += 1 156 finally: 157 'file index:' + str(file_object.tell()) 158 file_object.close() 159 return True 160 161 # 合並 162 def merge_request(self, taskid): 163 return self.gene_request(api_merge, data=self.gene_params(api_merge, taskid=taskid)) 164 165 # 獲取進度 166 def get_progress_request(self, taskid): 167 return self.gene_request(api_get_progress, data=self.gene_params(api_get_progress, taskid=taskid)) 168 169 # 獲取結果 170 def get_result_request(self, taskid): 171 return self.gene_request(api_get_result, data=self.gene_params(api_get_result, taskid=taskid)) 172 173 def all_api_request(self): 174 # 1. 預處理 175 pre_result = self.prepare_request() 176 taskid = pre_result["data"] 177 # 2 . 分片上傳 178 self.upload_request(taskid=taskid, upload_file_path=self.upload_file_path) 179 # 3 . 文件合並 180 self.merge_request(taskid=taskid) 181 # 4 . 獲取任務進度 182 while True: 183 # 每隔20秒獲取一次任務進度 184 progress = self.get_progress_request(taskid) 185 progress_dic = progress 186 if progress_dic['err_no'] != 0 and progress_dic['err_no'] != 26605: 187 # print('task error: ' + progress_dic['failed']) 188 return 189 else: 190 data = progress_dic['data'] 191 task_status = json.loads(data) 192 if task_status['status'] == 9: 193 # print('task ' + taskid + ' finished') 194 break 195 print('The task ' + taskid + ' is in processing, task status: ' + str(data)) 196 print('processing...') 197 # 每次獲取進度間隔20S 198 time.sleep(20) 199 # 5 . 獲取結果 200 return self.get_result_request(taskid=taskid) 201 202 203 def get_text_result(upload_file_path): 204 """ 205 封裝該接口，獲取接口返回的內容 206 :param upload_file_path: 207 :return: 識別出來的文本數據 208 """ 209 api = RequestApi(appid="xxx", secret_key="xxx", upload_file_path=upload_file_path) 210 return api.all_api_request() 211 212 213 # 注意：如果出現requests模塊報錯："NoneType" object has no attribute 'read', 請嘗試將requests模塊更新到2.20.0或以上版本(本demo測試版本為2.20.0) 214 # 輸入訊飛開放平台的appid，secret_key和待轉寫的文件路徑 215 if __name__ == '__main__': 216 result = get_text_result('input/xxx.m4a') 217 print(result) 218 print(type(result))
appid 和 secret_key 需要你自己申請之后，配置上去。
配置好之后填寫需要輸入的音頻，就可以運行該腳本作測試。
python weblfasr_python3_demo.py treating... treating... treating... treating... treating... The task e3e3284aee4a4e3b86a4fd506960e0f2 is in processing, task status: {"status":2,"desc":"音頻並完成"} processing... treating... The task e3e3284aee4a4e3b86a4fd506960e0f2 is in processing, task status: {"status":3,"desc":"音頻寫中"} processing... treating... treating... {'data': '[{"bg":"480","ed":"1810","onebest":"我好高興！","speaker":"2"},{"bg":"1820","ed":"4440ebest":"啊明天就放假了！","speaker":"1"}]', 'err_no': 0, 'failed': None, 'ok': 0} <class 'dict'>
情感傾向分析功能
這里是百度情感傾向分析的文檔，可以選擇 Python SDK 或者 API 接口，我選擇的是 API 接口。並且我對它進行了一定程度的封裝。
baidu_sentiment.py 文件有如下代碼：
1 #!/usr/bin/env python 2 # -*- coding: utf-8 -*- 3 """ 4 百度情感傾向分析： 5 get_sentiment_result 用於 demo 進行調用 6 # 參數 說明 描述 7 # log_id uint64 請求唯一標識碼 8 # sentiment int 表示情感極性分類結果，0:負向，1:中性，2:正向 9 # confidence float 表示分類的置信度，取值范圍[0,1] 10 # positive_prob float 表示屬於積極類別的概率 ，取值范圍[0,1] 11 # negative_prob float 表示屬於消極類別的概率，取值范圍[0,1] 12 """ 13 import json 14 import requests 15 16 17 def get_sentiment_result(text): 18 """ 19 利用情感傾向分析API來獲取返回數據 20 :param text: 輸入文本 21 :return response: 返回的響應 22 """ 23 if text == '': 24 return '' 25 # 請求接口 26 url = 'https://aip.baidubce.com/oauth/2.0/token' 27 # 需要先獲取一個 token 28 client_id = 'xxx' 29 client_secret = 'xxx' 30 params = { 31 'grant_type': 'client_credentials', 32 'client_id': client_id, 33 'client_secret': client_secret 34 } 35 headers = {'Content-Type': 'application/json; charset=UTF-8'} 36 response = requests.post(url=url, params=params, headers=headers).json() 37 access_token = response['access_token'] 38 39 # 通用版情緒識別接口 40 url = 'https://aip.baidubce.com/rpc/2.0/nlp/v1/sentiment_classify' 41 # 定制版情緒識別接口 42 # url = 'https://aip.baidubce.com/rpc/2.0/nlp/v1/sentiment_classify_custom' 43 # 使用 token 調用情感傾向分析接口 44 params = { 45 'access_token': access_token 46 } 47 payload = json.dumps({ 48 'text': text 49 }) 50 headers = {'Content-Type': 'application/json; charset=UTF-8'} 51 response = requests.post(url=url, params=params, data=payload, headers=headers).json() 52 return response 53 54 55 if __name__ == '__main__': 56 print(get_sentiment_result('白日放歌須縱酒，青春作伴好還鄉。')) 57 print(get_sentiment_result('思悠悠，恨悠悠，恨到歸時方始休。'))
同樣，你需要在百度創建應用，配置好你的 client_id 和 client_secret。你也可以運行該腳本進行測試。
python baidu_sentiment.py {'log_id': 2676765769120607830, 'text': '白日放歌須縱酒，青春作伴好還鄉。', 'items': [{'positive_prob': 0.537741, 'confidence': 0.245186, 'negative_prob': 0.462259, 'sentiment': 1}]} {'log_id': 4078175744151108694, 'text': '思悠悠，恨悠悠，恨到歸時方始休。', 'items': [{'positive_prob': 0.345277, 'confidence': 0.232717, 'negative_prob': 0.654723, 'sentiment': 0}]}
關鍵詞提取功能
在這里可以找到訊飛的關鍵詞提取的接口文檔和示例代碼。同樣我也略作了改動，進行了封裝。
WebLtp_python3_demo.py 文件代碼：
1 #!/usr/bin/python 2 # -*- coding: UTF-8 -*- 3 """ 4 訊飛關鍵詞提取接口 5 """ 6 import time 7 import urllib.request 8 import urllib.parse 9 import json 10 import hashlib 11 import base64 12 13 # 接口地址 14 url = "http://ltpapi.xfyun.cn/v1/ke" 15 # 開放平台應用ID 16 x_appid = "xxx" 17 # 開放平台應用接口秘鑰 18 api_key = "xxx" 19 # 語言文本 20 TEXT = "漢皇重色思傾國，御宇多年求不得。楊家有女初長成，養在深閨人未識。天生麗質難自棄，一朝選在君王側。" 21 22 23 def get_keyword_result(text): 24 """ 25 這是訊飛官方文檔給出的示例 26 :param text: 輸入文本 27 :return response: 返回對象 28 """ 29 if text == '': 30 return '' 31 body = urllib.parse.urlencode({'text': text}).encode('utf-8') 32 param = {"type": "dependent"} 33 x_param = base64.b64encode(json.dumps(param).replace(' ', '').encode('utf-8')) 34 x_time = str(int(time.time())) 35 x_checksum = hashlib.md5(api_key.encode('utf-8') + 36 str(x_time).encode('utf-8') + 37 x_param).hexdigest() 38 x_header = {'X-Appid': x_appid, 39 'X-CurTime': x_time, 40 'X-Param': x_param, 41 'X-CheckSum': x_checksum} 42 req = urllib.request.Request(url, body, x_header) 43 result = urllib.request.urlopen(req) 44 result = result.read() 45 return result.decode('utf-8') 46 47 48 if __name__ == '__main__': 49 keyword_result = get_keyword_result(TEXT) 50 print(keyword_result) 51 print(type(keyword_result))
配置好你的 x_appid 和 api_key。
注意：關鍵詞提取還需要你在訊飛應用的后台設置白名單。
點擊管理，配置好自己的公網 IP。試着運行一下腳本，會有如下輸出：
python WebLtp_python3_demo.py {"code":"0","data":{"ke":[{"score":"0.646","word":"漢皇"},{"score":"0.634","word":"御宇"},{"score":"0.633","word":"重色"},{"score":"0.632","word":"王側"},{"score":"0.628","word":"思傾國"},{"score":"0.601","word":"自棄"},{"score":"0.600","word":"楊家"},{"score":"0.588","word":"深閨人未識"},{"score":"0.588","word":"求不得"},{"score":"0.586","word":"天生麗質"}]},"desc":"success","sid":"ltp000aed03@dx589210907749000100"} <class 'str'>
把所有功能組合起來
用一個 Demo 把所有功能組合起來，並把結果存儲到文件中。
demo.py 如下：
1 #!/usr/bin/env python 2 # -*- coding: utf-8 -*- 3 """ 4 這是主要的demo 5 流程是： 6 音頻->訊飛語音識別API->文本 7 文本再作兩種處理： 8 文本->百度情緒識別API->情緒識別的響應 9 文本->訊飛關鍵詞提取API->關鍵詞提取的響應 10 """ 11 import sys 12 import json 13 from weblfasr_python3_demo import get_text_result 14 from baidu_sentiment import get_sentiment_result 15 from WebLtp_python3_demo import get_keyword_result 16 17 # 硬編碼選定需要離線分析的音頻 18 # 以下是一些測試-------------------------- 19 # SOURCE_PATH = 'input/test.mp3' 20 # SOURCE_PATH = 'input/test.pcm' 21 # SOURCE_PATH = 'input/test.m4a' 22 # SOURCE_PATH = 'input/test.wav' 23 # 以上是一些測試-------------------------- 24 # 或者，通過命令行參數選定需要離線分析的音頻 25 # 如：python demo.py test.wav 26 SOURCE_PATH = 'input/' + sys.argv[1] 27 # STEP 1: 調用訊飛語音識別 API 28 # 獲取訊飛識別出來的響應 29 TEXT_RESULT = get_text_result(SOURCE_PATH) 30 31 32 def save_file(data, destin): 33 """ 34 數據持久化函數 35 :param data: 數據 36 :param destin: 目標路徑 37 :return: None 38 """ 39 data = str(data) 40 if data: 41 with open(destin, "w", encoding='utf-8') as f: 42 f.write(data) 43 44 45 def whole_method(): 46 """ 47 將音頻文本不作區分地提取（兩個人的對話不做區分） 48 :return: None 49 """ 50 # 解析語音識別出來的數據 51 data_list = json.loads(TEXT_RESULT['data']) 52 # text 用於拼接 53 text_result = '' 54 for data in data_list: 55 text_result += data['onebest'] 56 print('text_result:', text_result) 57 print('text_result completed') 58 # 把文本寫入到文件中 59 save_file(text_result, 'output/text_result.txt') 60 # STEP 2: 情感傾向分析 61 # 輸入文本，使用情緒識別函數獲取響應 62 sentiment_result = get_sentiment_result(text_result) 63 # 保存數據 64 save_file(sentiment_result, 'output/sentiment_result.txt') 65 print('sentiment_result completed') 66 # STEP 3: 關鍵詞提取 67 # 輸入文本，調用訊飛提取關鍵詞的接口，對文本做關鍵詞提取 68 keyword_result = get_keyword_result(text_result) 69 # 保存數據 70 save_file(keyword_result, 'output/keyword_result.txt') 71 print('keyword_result completed') 72 73 74 def seperate_method(): 75 """ 76 將音頻文本作區分地提取（區分兩個人的對話） 77 :return: None 78 """ 79 data_list = json.loads(TEXT_RESULT['data']) 80 text_result1 = '' 81 text_result2 = '' 82 # 假設有兩個人，把文本分別做整合 83 for data in data_list: 84 # print(data) 85 if data['speaker'] == '1': 86 text_result1 += data['onebest'] 87 else: 88 text_result2 += data['onebest'] 89 print('text_result1', text_result1) 90 print('text_result2', text_result2) 91 print('text_result1 text_result2 completed') 92 save_file(text_result1, 'output/text_result1.txt') 93 save_file(text_result2, 'output/text_result2.txt') 94 # STEP 2: 情感傾向分析 95 # 輸入文本，使用情緒識別函數獲取響應 96 # A 的對話 97 sentiment_result1 = get_sentiment_result(text_result1) 98 save_file(sentiment_result1, 'output/sentiment_result1.txt') 99 print('result_get_result1 completed') 100 # B 的對話 101 sentiment_result2 = get_sentiment_result(text_result2) 102 save_file(sentiment_result2, 'output/sentiment_result2.txt') 103 print('result_get_result2 completed') 104 # STEP 3: 關鍵詞提取 105 # 調用訊飛接口做文本的關鍵字提取 106 # A 的對話 107 keyword_result1 = get_keyword_result(text_result1) 108 save_file(keyword_result1, 'output/keyword_result1.txt') 109 print('keyword_result1 completed') 110 # B 的對話 111 keyword_result2 = get_keyword_result(text_result2) 112 save_file(keyword_result2, 'output/keyword_result2.txt') 113 print('keyword_result2 completed') 114 115 116 if __name__ == '__main__': 117 if TEXT_RESULT: 118 whole_method() 119 seperate_method()
輸出大致如下：
python demo.py test.mp3 treating... treating... treating... treating... treating... The task 8552d13470ed4839b11e0f3693f296f9 is in processing, task status: {"status":2,"desc":"音頻合並完成"} processing... treating... ... The task 8552d13470ed4839b11e0f3693f296f9 is in processing, task status: {"status":3,"desc":"音頻轉寫中"} processing... treating... treating... text_result: 喂喂你好，是xxx的機主是吧？誰？呀我是xxx的工作人員，您在今天中午12點多在我們xxx提交了xxx是吧？那怎么？...那沒有關系，我說您是否辦理xxx？什么有什么有關系，啊有什么有關系啊。 text_result completed sentiment_result completed keyword_result completed text_result1 喂喂你好，是xxx的機主是吧？呀我是xxx的工作人員，您在今天中午12點多在我們xxx提交了xxx是吧？...那沒有關系，我說您是否辦理xxx？ text_result2 誰？那怎么？...什么有什么有關系，啊有什么有關系啊。 text_result1 text_result2 completed result_get_result1 completed result_get_result2 completed keyword_result1 completed keyword_result2 completed
原文作者：雨先生
原文鏈接：https://www.cnblogs.com/noluye/p/11225024.html
許可協議：知識共享署名-非商業性使用 4.0 國際許可協議