Python调⽤百度接⼝（情感倾向分析）和讯飞接⼝（语⾳识别、关键词提取）处
理⾳频⽂件
本⽰例的过程是：
1. ⾳频转⽂本
2. 利⽤⽂本获取情感倾向分析结果
3. 利⽤⽂本获取关键词提取
⾸先是讯飞的语⾳识别模块。在可以到⾮实时语⾳转写的相关⽂档以及 Python ⽰例。我略作了改动，让它可以对不同⼈说话作区分，并且作了⼀些封装。
语⾳识别功能
weblfasr_python3_demo.py ⽂件：
1#!/usr/bin/env python
2# -*- coding: utf-8 -*-
3"""
4讯飞⾮实时转写调⽤demo（语⾳识别）
5"""
6import base64
7import hashlib
8import hmac
9import json
10import os
11import time
12
13import requests
14
15 lfasr_host = 'raasr.xfyun/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# 参数可在官⽹界⾯（doc.xfyun/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
39class SliceIdGenerator:
40"""slice id⽣成器"""
41
42def__init__(self):
43 self.__ch = 'aaaaaaaaa`'
44
45def getNextSliceId(self):
46 ch = self.__ch
47 j = len(ch) - 1
48while j >= 0:
49 cj = ch[j]
50if cj != 'z':
51 ch = ch[:j] + chr(ord(cj) + 1) + ch[j + 1:]
52break
53else:
54 ch = ch[:j] + 'a' + ch[j + 1:]
55 j = j - 1
56 self.__ch = ch
57return self.__ch
58
59
60class RequestApi(object):
61def__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⽣成不同的参数，本⽰例中未使⽤全部参数您可在官⽹(doc.xfyun/rest_api/%E8%AF%AD%E9%9F%B3%E8%BD%AC%E5%86%99.html)查看后选择适合业务场景的进⾏更换 67def 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 = w(de('utf-8'), md5, hashlib.sha1).digest()
78 signa = base64.b64encode(signa)
79 signa = str(signa, 'utf-8')
80 file_len = size(upload_file_path)
81 file_name = os.path.basename(upload_file_path)
82 param_dict = {}
83
84if 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)
93elif 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
99elif 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
105elif 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
111return param_dict
112
113# 请求和结果解析，结果中各个字段的含义可参考：doc.xfyun/rest_api/%E8%AF%AD%E9%9F%B3%E8%BD%AC%E5%86%99.html
114def 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.)
117if result["ok"] == 0:
118# print("{} success:".format(apiname) + str(result))
119print('')
120return result
121else:
122# print("{} error:".format(apiname) + str(result))
123 exit(0)
124return result
125
126# 预处理
127def prepare_request(self):
_request(apiname=api_prepare,
129 _params(api_prepare))
130
131# 上传
132def upload_request(self, taskid, upload_file_path):
133 file_object = open(upload_file_path, 'rb')
134try:
135 index = 1
136 sig = SliceIdGenerator()
137while True:
138 content = ad(file_piece_sice)
139if not content or len(content) == 0:
140break
141 files = {
142"filename": _params(api_upload).get("slice_id"),
143"content": content
144 }
145 response = _request(api_upload,
146 _params(api_upload, taskid=taskid,
147 slice_NextSliceId()),
148 files=files)
('ok') != 0:
150# 上传分⽚失败
151print('upload slice fail, response: ' + str(response))
152return False
153# print('upload slice ' + str(index) + ' success')
154print('')
155 index += 1
156finally:
157'file index:' + str(ll())
158 file_object.close()
159return True
160
161# 合并
162def merge_request(self, taskid):
_request(api_merge, _params(api_merge, taskid=taskid))
164
165# 获取进度
166def get_progress_request(self, taskid):
_request(api_get_progress, _params(api_get_progress, taskid=taskid))
168
169# 获取结果
170def get_result_request(self, taskid):
_request(api_get_result, _params(api_get_result, taskid=taskid))
172
173def 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 _request(taskid=taskid)
181# 4 . 获取任务进度
182while True:
183# 每隔20秒获取⼀次任务进度
184 progress = _progress_request(taskid)
185 progress_dic = progress
186if progress_dic['err_no'] != 0 and progress_dic['err_no'] != 26605:
187# print('task error: ' + progress_dic['failed'])
188return
189else:
190 data = progress_dic['data']
191 task_status = json.loads(data)
192if task_status['status'] == 9:
193# print('task ' + taskid + ' finished')
194break
195print('The task ' + taskid + ' is in processing, task status: ' + str(data))
196print('')
197# 每次获取进度间隔20S
198 time.sleep(20)
199# 5 . 获取结果
_result_request(taskid=taskid)
201
202
203def 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)
210return 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和待转写的⽂件路径
215if__name__ == '__main__':
216 result = get_text_result('input/xxx.m4a')
217print(result)
218print(type(result))
appid 和 secret_key 需要你⾃⼰申请之后，配置上去。
配置好之后填写需要输⼊的⾳频，就可以运⾏该脚本作测试。
python weblfasr_python3_demo.py
<
<
<
<
<
The task e3e3284aee4a4e3b86a4fd506960e0f2 is in processing, task status: {"status":2,"desc":"⾳频并完成"}
<
<
The task e3e3284aee4a4e3b86a4fd506960e0f2 is in processing, task status: {"status":3,"desc":"⾳频写中"}
<
<
<
{'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百度情感倾向分析：
5get_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"""
13import json
14import requests
15
16
17def get_sentiment_result(text):
18"""
19利⽤情感倾向分析API来获取返回数据
20 :param text: 输⼊⽂本
21 :return response: 返回的响应
22"""
23if text == '':
24return''
25# 请求接⼝
26 url = 'aip.baidubce/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 = 'aip.baidubce/rpc/2.0/nlp/v1/sentiment_classify'
41# 定制版情绪识别接⼝
42# url = 'aip.baidubce/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()
52return response
53
54
55if__name__ == '__main__':
56print(get_sentiment_result('⽩⽇放歌须纵酒，青春作伴好还乡。'))
57print(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"""
6import time
quest
8import urllib.parse
9import json
10import hashlib
11import base64
12
13# 接⼝地址
14 url = "ltpapi.xfyun/v1/ke"
15# 开放平台应⽤ID
16 x_appid = "xxx"
17# 开放平台应⽤接⼝秘钥
18 api_key = "xxx"
19# 语⾔⽂本
20 TEXT = "汉皇重⾊思倾国，御宇多年求不得。杨家有⼥初长成，养在深闺⼈未识。天⽣丽质难⾃弃，⼀朝选在君王侧。"
21
22
23def get_keyword_result(text):
24"""
25这是讯飞官⽅⽂档给出的⽰例
27 :return response: 返回对象
28"""
29if text == '':
30return''
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(de('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 = quest.Request(url, body, x_header)
43 result = quest.urlopen(req)
44 result = ad()
45return result.decode('utf-8')
46
47
48if__name__ == '__main__':
49 keyword_result = get_keyword_result(TEXT)
50print(keyword_result)
51print(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", <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"""
11import sys
12import json
13from weblfasr_python3_demo import get_text_result
14from baidu_sentiment import get_sentiment_result
15from 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
32def save_file(data, destin):
33"""
34数据持久化函数
35 :param data: 数据
36 :param destin: ⽬标路径
37 :return: None
38"""
39 data = str(data)
40if data:
41 with open(destin, "w", encoding='utf-8') as f:
42 f.write(data)
43
44
45def whole_method():
46"""
47将⾳频⽂本不作区分地提取（两个⼈的对话不做区分）
48 :return: None
49"""
50# 解析语⾳识别出来的数据
51 data_list = json.loads(TEXT_RESULT['data'])
53 text_result = ''
54for data in data_list:
55 text_result += data['onebest']
56print('text_result:', text_result)
57print('text_result completed')
58# 把⽂本写⼊到⽂件中
59 save_file(text_result, 'output/')
60# STEP 2: 情感倾向分析
61# 输⼊⽂本，使⽤情绪识别函数获取响应
62 sentiment_result = get_sentiment_result(text_result)
63# 保存数据
64 save_file(sentiment_result, 'output/')
65print('sentiment_result completed')
66# STEP 3: 关键词提取
67# 输⼊⽂本，调⽤讯飞提取关键词的接⼝，对⽂本做关键词提取
68 keyword_result = get_keyword_result(text_result)
69# 保存数据
70 save_file(keyword_result, 'output/')
71print('keyword_result completed')
72
73
74def seperate_method():
75"""
76将⾳频⽂本作区分地提取（区分两个⼈的对话）
77 :return: None
78"""
79 data_list = json.loads(TEXT_RESULT['data'])
80 text_result1 = ''
81 text_result2 = ''
82# 假设有两个⼈，把⽂本分别做整合
百度api接口83for data in data_list:
84# print(data)
85if data['speaker'] == '1':
86 text_result1 += data['onebest']
87else:
88 text_result2 += data['onebest']
89print('text_result1', text_result1)
90print('text_result2', text_result2)
91print('text_result1 text_result2 completed')
92 save_file(text_result1, 'output/')
93 save_file(text_result2, 'output/')
94# STEP 2: 情感倾向分析
95# 输⼊⽂本，使⽤情绪识别函数获取响应
96# A 的对话
97 sentiment_result1 = get_sentiment_result(text_result1)
98 save_file(sentiment_result1, 'output/')
99print('result_get_result1 completed')
100# B 的对话
101 sentiment_result2 = get_sentiment_result(text_result2)
102 save_file(sentiment_result2, 'output/')
103print('result_get_result2 completed')
104# STEP 3: 关键词提取
105# 调⽤讯飞接⼝做⽂本的关键字提取
106# A 的对话
107 keyword_result1 = get_keyword_result(text_result1)
108 save_file(keyword_result1, 'output/')
109print('keyword_result1 completed')
110# B 的对话
111 keyword_result2 = get_keyword_result(text_result2)
112 save_file(keyword_result2, 'output/')
113print('keyword_result2 completed')
114
115
116if__name__ == '__main__':
117if TEXT_RESULT:
118 whole_method()
119 seperate_method()
输出⼤致如下：
python demo.py test.mp3
<
<
<
<
<
The task 8552d13470ed4839b11e0f3693f296f9 is in processing, task status: {"status":2,"desc":"⾳频合并完成"}
<
<
...
The task 8552d13470ed4839b11e0f3693f296f9 is in processing, task status: {"status":3,"desc":"⾳频转写中"}
<
<
<
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
原⽂作者：
原⽂链接：
许可协议：
版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。
发表评论