'commit'

AI/WxMini/GetToken.py

@@ -12,7 +12,7 @@ def getToken():
     client = AcsClient(
         'LTAI5t5jxkgJtRK8wew8fnbq', 'b8HXNGz7IkI3Dhv7BZx9BNBEZy1uku',
         "cn-shanghai"
-    );
+    )
 
     # 创建request，并设置参数。
     request = CommonRequest()

AI/WxMini/Start.py

@@ -1,79 +0,0 @@
-# -*- 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)

AI/WxMini/tts.py

@@ -0,0 +1,48 @@
+# -*- 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
+
+
+
+class TTS:
+    def __init__(self,  _file):
+        self._th = threading.Thread(target=self._run)
+        self._file = _file
+        self._f = None
+
+    def start(self, text):
+        self._text = text
+        # 确保目录存在
+        os.makedirs(os.path.dirname(self._file), exist_ok=True)
+        self._f = open(self._file, "wb")
+        self._th.start()
+
+    def on_close(self, *args):
+        if self._f:
+            self._f.close()
+
+    def on_data(self, data, *args):
+        if self._f:
+            self._f.write(data)
+
+    def _run(self):
+        tts = nls.NlsSpeechSynthesizer(
+            url=URL,
+            token=TOKEN,
+            appkey=APPKEY,
+            on_data=self.on_data,
+            on_close=self.on_close
+        )
+        tts.start(self._text, voice="xiaobei", aformat="mp3")
+
+
+if __name__ == '__main__':
+    TEXT = '大壮正想去摘取花瓣，谁知阿丽和阿强突然内讧'
+    t = TTS("tests/test_tts.mp3")
+    t.start(TEXT)