# -*- coding: utf-8 -*- """ 科大讯飞的语音识别API. 外网ip查询:https://ip.51240.com/ voice_name: https://www.xfyun.cn/services/online_tts """ from __future__ import division from ..__load__ import * from .base import Base import requests import hashlib class Xunfei(Base): url = 'http://api.xfyun.cn/v1/service/v1/' def __init__(self): super(self.__class__, self).__init__() def setting(self, appid, api_key, voice_name='vecan', **args): self.appid = appid self.api_key = api_key self.voice_name = voice_name def getHeader(self, aue, engineType): curTime = str(int(time.time())) if aue == 'raw': param = "{\"aue\":\"" + aue + "\"" + ",\"engine_type\":\"" + engineType + "\"}" else: param = "{\"aue\":\""+aue+"\",\"auf\":\"audio/L16;rate=16000\",\"voice_name\":\"" + self.voice_name + "\",\"engine_type\":\"intp65\"}" Demeter.logger.debug("param:{}".format(param)) paramBase64 = str(base64.b64encode(param.encode('utf-8')), 'utf-8') Demeter.logger.debug("x_param:{}".format(paramBase64)) m2 = hashlib.md5() m2.update((self.api_key + curTime + paramBase64).encode('utf-8')) checkSum = m2.hexdigest() header = { 'X-CurTime': curTime, 'X-Param': paramBase64, 'X-Appid': self.appid, 'X-CheckSum': checkSum, 'Content-Type': 'application/x-www-form-urlencoded; charset=utf-8', } return header def asr(self, fp): url = self.url + "iat" binfile = open(fp, 'rb') data = {'audio': base64.b64encode(binfile.read())} r = requests.post(url, headers=self.getHeader('raw', 'sms16k'), data=data) res = json.loads(r.content.decode('utf-8')) Demeter.logger.debug(res) if 'code' in res and res['code'] == '0': return self.asrLog(1, res['data']) else: return self.asrLog(2, '') def tts(self, phrase): url = self.url + "tts" r = requests.post(url, headers=self.getHeader('lame', ''), data={'text':phrase}) contentType = r.headers['Content-Type'] if contentType == "audio/mpeg": tmpfile = writeTempFile(r.content, '.mp3') return self.ttsLog(1, tmpfile) else : return self.ttsLog(2, r.text)