|
|
# -*- coding: utf-8 -*-
|
|
|
import threading
|
|
|
import nls
|
|
|
import os
|
|
|
from GetToken import *
|
|
|
|
|
|
URL = "wss://nls-gateway-cn-shanghai.aliyuncs.com/ws/v1"
|
|
|
TOKEN = getToken() # 参考https://help.aliyun.com/document_detail/450255.html获取token
|
|
|
APPKEY = "90RJcqjlN4ZqymGd" # 获取Appkey请前往控制台：https://nls-portal.console.aliyun.com/applist
|
|
|
|
|
|
TEXT = '大壮正想去摘取花瓣，谁知阿丽和阿强突然内讧'
|
|
|
|
|
|
|
|
|
class TestTts:
|
|
|
def __init__(self, tid, test_file):
|
|
|
self._th = threading.Thread(target=self.__test_run)
|
|
|
self._id = tid
|
|
|
self._test_file = test_file
|
|
|
self._f = None
|
|
|
|
|
|
def start(self, text):
|
|
|
self._text = text
|
|
|
# 确保目录存在
|
|
|
os.makedirs(os.path.dirname(self._test_file), exist_ok=True)
|
|
|
self._f = open(self._test_file, "wb")
|
|
|
self._th.start()
|
|
|
|
|
|
def test_on_metainfo(self, message, *args):
|
|
|
print("on_metainfo message=>{}".format(message))
|
|
|
|
|
|
def test_on_error(self, message, *args):
|
|
|
print("on_error args=>{}".format(args))
|
|
|
|
|
|
def test_on_close(self, *args):
|
|
|
print("on_close: args=>{}".format(args))
|
|
|
if self._f:
|
|
|
self._f.close()
|
|
|
|
|
|
def test_on_data(self, data, *args):
|
|
|
if self._f:
|
|
|
self._f.write(data)
|
|
|
|
|
|
def test_on_completed(self, message, *args):
|
|
|
print("on_completed:args=>{} message=>{}".format(args, message))
|
|
|
|
|
|
def __test_run(self):
|
|
|
print("thread:{} start..".format(self._id))
|
|
|
tts = nls.NlsSpeechSynthesizer(
|
|
|
url=URL,
|
|
|
token=TOKEN,
|
|
|
appkey=APPKEY,
|
|
|
#long_tts=True,
|
|
|
on_metainfo=self.test_on_metainfo,
|
|
|
on_data=self.test_on_data,
|
|
|
on_completed=self.test_on_completed,
|
|
|
on_error=self.test_on_error,
|
|
|
on_close=self.test_on_close,
|
|
|
callback_args=[self._id]
|
|
|
)
|
|
|
print("{}: session start".format(self._id))
|
|
|
r = tts.start(self._text, voice="ailun")
|
|
|
print("{}: tts done with result:{}".format(self._id, r))
|
|
|
|
|
|
|
|
|
def multiruntest(num=1):
|
|
|
threads = []
|
|
|
for i in range(0, num):
|
|
|
name = "thread" + str(i)
|
|
|
t = TestTts(name, "tests/test_tts.mp3")
|
|
|
t.start(TEXT)
|
|
|
threads.append(t)
|
|
|
|
|
|
# 等待所有线程完成
|
|
|
for t in threads:
|
|
|
t._th.join()
|
|
|
|
|
|
|
|
|
nls.enableTrace(True)
|
|
|
multiruntest(1)
|
