#!/usr/bin/env python
#coding: utf-8
import requests
import json
import argparse
import base64


# 获取access_token用于鉴权
def get_access_token(client_secret, client_id):
    grant_type = "client_credentials"
    url = "https://openapi.data-baker.com/oauth/2.0/token?grant_type={}&client_secret={}&client_id={}".format(grant_type, client_secret, client_id)
    response = requests.post(url)
    access_token = json.loads(response.text).get('access_token')

    return access_token


# 获取转换后音频
def get_audio(data):
    url = "https://openapi.data-baker.com/tts?access_token={}&domain={}&language={}&voice_name={}&text={}&audiotype={}" \
          "&enable_subtitles={}&interval={}".format(data['access_domain'], data['domain'], data['language'],
           data['voice_name'], data['text'], data['audiotype'], data['enable_subtitles'], data['interval'])
    response = requests.post(url)
    content_type = response.headers['Content-Type']
    if 'audio' not in content_type:
        raise Exception(response.text)

    length = int.from_bytes(response.content[:4], byteorder='big', signed=False)
    json_data = json.loads((response.content[4: length + 4]).decode(encoding='gbk'))
    subtitles_list = json.loads((base64.b64decode(json_data.get('data')['interval-subtitles'])).decode()).get('subtitles_list')
    for subtitles in subtitles_list:
        print(subtitles)
    return response.content[length + 4:]

# 获取命令行输入参数
def get_args():
    text = '标贝科技'
    parser = argparse.ArgumentParser(description='ASR')
    parser.add_argument('-client_secret', type=str, required=True)
    parser.add_argument('-client_id', type=str, required=True)
    parser.add_argument('-file_save_path', type=str, required=True)
    parser.add_argument('--text', type=str, default=text)
    parser.add_argument('--audiotype', type=str, default='6')
    parser.add_argument('--domain', type=str, default='1')
    parser.add_argument('--language', type=str, default='zh')
    parser.add_argument('--voice_name', type=str, default='Jiaojiao')
    args = parser.parse_args()

    return args


if __name__ == '__main__':
    try:
        args = get_args()

        # 获取access_token
        client_secret = args.client_secret
        client_id = args.client_id
        access_token = get_access_token(client_secret, client_id)

        # 读取参数
        audiotype = args.audiotype
        domain = args.domain
        language = args.language
        voice_name = args.voice_name
        text = args.text
        data = {'access_domain': access_token, 'audiotype': audiotype, 'domain': domain, 'language': language,
                'voice_name': voice_name, 'text': text, 'enable_subtitles': "1", 'interval': '1'}
        content = get_audio(data)

        #保存音频文件
        with open('test.wav', 'wb') as audio:
            audio.write(content)
        print("task finished successfully")
    except Exception as e:
        print(e)

检查http 响应头Content-Type判断合成请求是否成功，如返回Content-Type以audio开头说明合成成功，响应结构如下。

如返回Content-Type:application/json说明合成出现错误，响应体为json格式

data内部结构：

interval-subtitles内部结构：

subtitles_list内部结构：

subtitles内部结构：

{
  'data': {
'interval-info': 'b=0.072346&iao=0.238304&b=0.295952&ei=0.409770&k=0.519053&e=0.630216&j=0.775269&i=0.936000&SIL=1.236000',
'interval-info-x': 'L=1&T=1,L=1&T=2,L=1&T=1,L=1&T=2,L=1&T=1,L=1&T=2,L=1&T=1,L=1&T=2,L=1&T=5',
'interval-subtitles': 'eyJzdWJ0aXRsZXNfbGlzdCI6W3sic3VidGl0bGVzIjpbeyJiZWdpbl9pbmRleCI6MCwiYmVnaW5fdGltZSI6MCwiZW5kX2luZGV4IjoyLCJlbmRfdGltZSI6MjM4LCJwcm9zb2R5IjoiU1AwIiwidGV4dCI6IuaghyJ9LHsiYmVnaW5faW5kZXgiOjMsImJlZ2luX3RpbWUiOjIzOCwiZW5kX2luZGV4Ijo1LCJlbmRfdGltZSI6NDA5LCJwcm9zb2R5IjoiU1AwIiwidGV4dCI6Iui0nSJ9LHsiYmVnaW5faW5kZXgiOjYsImJlZ2luX3RpbWUiOjQwOSwiZW5kX2luZGV4Ijo4LCJlbmRfdGltZSI6NjMwLCJwcm9zb2R5IjoiU1AwIiwidGV4dCI6IuenkSJ9LHsiYmVnaW5faW5kZXgiOjksImJlZ2luX3RpbWUiOjYzMCwiZW5kX2luZGV4IjoxMSwiZW5kX3RpbWUiOjkzNiwicHJvc29keSI6IlNQMyIsInRleHQiOiLmioAifV0sInRleHQiOiLmoIfotJ3np5HmioAifV19Cg=='
	},
	'err_msg': 'succ',
	'err_no': 0,
	'log_id': 1233745494,
	'sn': '60bb11ab-c8ea-4305-b689-aa912248a4fc'
}

interval-subtitles解码后结构：

'interval-subtitles': {
	'subtitles_list': [{
		'subtitles': [{
			'begin_index': 0,
			'begin_time': 0,
			'end_index': 2,
			'end_time': 238,
			'prosody': 'SP0',
			'text': '标'
		}, {
			'begin_index': 3,
			'begin_time': 238,
			'end_index': 5,
			'end_time': 409,
			'prosody': 'SP0',
			'text': '贝'
		}, {
			'begin_index': 6,
			'begin_time': 409,
			'end_index': 8,
			'end_time': 630,
			'prosody': 'SP0',
			'text': '科'
		}, {
			'begin_index': 9,
			'begin_time': 630,
			'end_index': 11,
			'end_time': 936,
			'prosody': 'SP3',
			'text': '技'
		}],
		'text': '标贝科技'
	}]
}

import argparse
import json
import base64
from threading import Thread
import requests
import websocket
import wave


#websocket客户端
class Client:
    def __init__(self, data, uri):
        self.data = data
        self.uri = uri
        self.audio_data = b""

    #建立连接
    def connect(self):
        ws_app = websocket.WebSocketApp(uri,
                                        on_open=self.on_open,
                                        on_message=self.on_message,
                                        on_error=self.on_error,
                                        on_close=self.on_close)
        ws_app.run_forever()

    # 建立连接后发送消息
    def on_open(self, ws):
        print("sending..")
        def run(*args):
            ws.send(self.data)

        Thread(target=run).start()

    # 接收消息
    def on_message(self, ws, message):
        message = json.loads(message)
        code = message["code"]
        if 'subtitles_list' in message["data"]:
            print(message["data"]['subtitles_list'])
        if code != 90000:
            # 打印接口错误
            print(message)
        else:
            self.audio_data += base64.b64decode(bytes(message["data"]["audio_data"], encoding='utf-8'))
            if message.get("data")["end_flag"] == 1:
                with wave.open('test.wav', 'wb') as wavfile:
                    wavfile.setparams((1, 2, 16000, 0, 'NONE', 'NONE'))
                    wavfile.writeframes(self.audio_data)
                ws.close()
                print("task finished successfully")

    # 打印错误
    def on_error(self, ws, error):
        print("error: ", str(error))

    # 关闭连接
    def on_close(ws):
        print("client closed.")


# 准备数据
def prepare_data(args, access_token):

    # 填写Header信息
    audiotype= args.audiotype
    voice_name = args.voice_name

    text = args.text
    # 单次调用最长300个汉字
    if len(text) > 300:
        raise Exception("Text is too long. The maximum length of chinese character is 300")
    text_bytes = text.encode(encoding='UTF-8')
    text = str(base64.b64encode(text_bytes), encoding='UTF-8')
    tts_params = {"language": "ZH", "voice_name": voice_name, "audiotype": audiotype, "domain": "1", "text": text, 'interval': '1', 'enable_subtitles': '1'}

    data = {"access_token": access_token, "version": "1.0", "tts_params": tts_params}
    data = json.dumps(data)

    return data


# 获取命令行输入参数
def get_args():
    text = '标贝科技'
    parser = argparse.ArgumentParser(description='tts')
    parser.add_argument('-client_secret', type=str, required=True)
    parser.add_argument('-client_id', type=str, required=True)
    parser.add_argument('-file_save_path', type=str, required=True)
    parser.add_argument('--text', type=str, default=text)
    parser.add_argument('--audiotype', type=str, default='4')
    parser.add_argument('--voice_name', type=str, default='beixi')
    args = parser.parse_args()

    return args


# 获取access_token用于鉴权
def get_access_token(client_secret, client_id):
    grant_type = "client_credentials"
    url = "https://openapi.data-baker.com/oauth/2.0/token?grant_type={}&client_secret={}&client_id={}" \
        .format(grant_type, client_secret, client_id)

    try:
        response = requests.post(url)
        response.raise_for_status()
    except Exception as e:
        print(response.text)
        raise Exception
    else:
        access_token = json.loads(response.text).get('access_token')
        return access_token


if __name__ == '__main__':
    try:
        args = get_args()

        # 获取access_token
        client_secret = args.client_secret
        client_id = args.client_id
        access_token = get_access_token(client_secret, client_id)

        # 准备数据
        data = prepare_data(args, access_token)

        uri = "wss://openapi.data-baker.com/tts/wsapi"
        # 建立Websocket连接
        client = Client(data, uri)
        client.connect()

        print('end')
    except Exception as e:
        print(e)

data内部结构：

subtitles_list内部结构：

subtitles内部结构：

{
	'code': 90000,
	'data': {
		'audio_data': '音频数据',
		'audio_type': 'audio/pcm',
		'end_flag': 1,
		'idx': 1,
		'interval': 'b=0.072346&iao=0.238304&b=0.295952&ei=0.409770&k=0.519053&e=0.630216&j=0.775269&i=0.936000&SIL=1.236000',
		'interval_x': 'L=1&T=1,L=1&T=2,L=1&T=1,L=1&T=2,L=1&T=1,L=1&T=2,L=1&T=1,L=1&T=2,L=1&T=5',
		'subtitles_list': [{
			'subtitles': [{
				'begin_index': 0,
				'begin_time': 0,
				'end_index': 2,
				'end_time': 238,
				'prosody': 'SP0',
				'text': '标'
			}, {
				'begin_index': 3,
				'begin_time': 238,
				'end_index': 5,
				'end_time': 409,
				'prosody': 'SP0',
				'text': '贝'
			}, {
				'begin_index': 6,
				'begin_time': 409,
				'end_index': 8,
				'end_time': 630,
				'prosody': 'SP0',
				'text': '科'
			}, {
				'begin_index': 9,
				'begin_time': 630,
				'end_index': 11,
				'end_time': 936,
				'prosody': 'SP3',
				'text': '技'
			}],
			'text': '标贝科技'
		}]
	},
	'message': 'Success',
	'trace_id': '1656986976642904'
}