基于讯飞API增加TTS能力，总配置文件样例

.gitignore

@@ -3,7 +3,7 @@ test.py
 *.log
 
 ### config file
-*.yaml
+config.yaml
 *.db
 *.bin
 
@@ -12,6 +12,7 @@ custom_plugins/microsoft_todo/token_cache.bin
 
 ### plugin file
 custom_plugins/*
+audio_tmp/*
 
 ### venv template
 # Virtualenv

README.md

@@ -1 +1,13 @@
 # myAssistant
+
+使用Python编写的个人助理程序，主程序为一个插件管理平台，可以自行编写不同的插件完成自己需要的任务
+该程序为自用，因此会持续更新
+
+to-do：
+- [x] 插件管理程序
+- [x] 定时运行(间隔、指定时间)
+- [x] 电报(telegram)机器人
+- [x] TTS
+- [ ] 前端页面
+
+鸣谢：TTS代码参考自https://github.com/CorttChan/xfyun-tts/tree/master

config/config_sample.yaml

@@ -0,0 +1,11 @@
+LOG_PATH: log # log文件夹
+PROXY:
+  http: 127.0.0.1:8089 # 代理地址
+  https: 127.0.0.1:8089
+
+
+TTS:
+  APPID: # 讯飞APPID
+  APIKey: # 讯飞APIKey
+  APISecret: # 讯飞APISecret
+  AUDIO_PATH: audio_tmp # 音频文件地址

main.py

@@ -7,7 +7,7 @@ if __name__ == "__main__":
     ## 初始化
     ## flask初始化
     app = Flask(__name__)
-    ## 配比初始化
+    ## 配置初始化
     config = Config()
     ## 插件管理器
     manager = PluginManager(config)

requirements.txt

@@ -0,0 +1,54 @@
+aiofiles==23.2.1
+aiohttp==3.9.5
+aiosignal==1.3.1
+anyio==4.3.0
+APScheduler==3.10.4
+async-timeout==4.0.3
+attrs==23.2.0
+blinker==1.7.0
+certifi==2024.2.2
+cffi==1.16.0
+charset-normalizer==3.3.2
+click==8.1.7
+cryptography==42.0.5
+exceptiongroup==1.2.1
+Flask==3.0.3
+Flask-APScheduler==1.13.1
+Flask-Script==2.0.6
+frozenlist==1.4.1
+greenlet==3.0.3
+greenletio==0.11.0
+h11==0.14.0
+h2==4.1.0
+hpack==4.0.0
+httpcore==1.0.5
+httpx==0.27.0
+Hypercorn==0.16.0
+hyperframe==6.0.1
+idna==3.6
+itsdangerous==2.1.2
+Jinja2==3.1.3
+MarkupSafe==2.1.5
+msal==1.27.0
+multidict==6.0.5
+priority==2.0.0
+pycparser==2.21
+PyJWT==2.8.0
+python-dateutil==2.9.0.post0
+python-telegram-bot==21.1.1
+pytz==2024.1
+PyYAML==6.0.1
+requests==2.31.0
+schedule==1.2.1
+six==1.16.0
+sniffio==1.3.1
+SQLAlchemy==2.0.28
+taskgroup==0.0.0a4
+tomli==2.0.1
+typing_extensions==4.10.0
+tzlocal==5.2
+urllib3==2.2.1
+uvicorn==0.29.0
+Werkzeug==3.0.2
+wsproto==1.2.0
+yarl==1.9.4

util/PluginManager.py

@@ -1,12 +1,12 @@
 import asyncio
 import importlib.util
-import os
 import threading
-from datetime import datetime
-
+from datetime import timedelta, datetime
 from util.base_plugin import BasePlugin
-
 import telegram
+from util.tts import TTSWebSocket, TTS
+from playsound import playsound
+import os
 
 class PluginManager:
     PLUGIN_DIR = "custom_plugins"
@@ -17,7 +17,6 @@ class PluginManager:
         self.running_tasks = []
         # 插件所在目录
         self.plugins_path = {}
-        # self.loop = asyncio.new_event_loop()
         self.config = config
 
     def load_plugins(self):
@@ -37,21 +36,19 @@ class PluginManager:
             member = getattr(module, member_name)
 
             if callable(member) and hasattr(member, "__bases__") and BasePlugin in member.__bases__:
-                print(member)
-                self.plugins[plugin_name] = member()
+                self.plugins[plugin_name] = member(root)
                 self.plugins_path[plugin_name] = root
                 print(f"Plugin '{plugin_name}' loaded successfully.")
 
     def execute_plugins(self, *args, **kwargs):
         for plugin_name, plugin_instance in self.plugins.items():
             print(f"Executing plugin '{plugin_name}':")
-            plugin_instance.run_plugin(self.callback, *args, **kwargs)
+            plugin_instance.run_plugin(*args, **kwargs)
 
-    def execute_plugin(self, plugin_name, *args, **kwargs):
+    async def execute_plugin(self, plugin_name, *args, **kwargs):
         if plugin_name in self.plugins:
             print(f"Executing plugin '{plugin_name}':")
-            return self.plugins[plugin_name].run_plugin(self.callback, *args, **kwargs)
-
+            return await self.plugins[plugin_name].run_plugin(*args, **kwargs)
         else:
             print(f"Plugin '{plugin_name}' not found.")
             return f"Plugin '{plugin_name}' not found."
@@ -60,18 +57,10 @@ class PluginManager:
 
         return [plugin[0] for plugin in self.plugins.items()]
 
-    def callback(self, result):
-
-        print(f"Plugin '{result['plugin_name']}' returned:")
-        print(f"Success: {result['success']}")
-        if result['success']:
-            print(f"Result: {result['result']}")
-        else:
-            print(f"Error: {result['error']}")
-
     def start_scheduler(self):
         self.loop = asyncio.new_event_loop()
         asyncio.set_event_loop(self.loop)
+
         def scheduler():
             asyncio.set_event_loop(self.loop)
             for name in self.get_plugin_name_list():
@@ -84,33 +73,94 @@ class PluginManager:
         thread.daemon = True
         thread.start()
 
-
     async def execute_plugin_async(self, plugin_name, schedule, *args, **kwargs):
+        telegram_need = False
+        tts_need = False
+        result = {'result':""}
+        APPID = ""
+        APIKey = ""
+        APISecret = ""
+        try:
+            token = self.config.get_plugin_config('TELEGRAM', self.plugins_path[plugin_name])['BOT_TOKEN']
+            chat_id = self.config.get_plugin_config('TELEGRAM', self.plugins_path[plugin_name])['CHAT_ID']
+            telegram_need = True
+        except:
+            telegram_need = False
+        try:
+            if self.config.get_plugin_config('TTS', self.plugins_path[plugin_name]) == "NEED":
+                tts_need = True
+            else:
+                tts_need = False
+        except Exception as e:
+            print(e)
+            tts_need = False
+        try:
+            APPID = self.config.get_config("TTS")["APPID"]
+            APIKey = self.config.get_config("TTS")["APIKey"]
+            APISecret = self.config.get_config("TTS")["APISecret"]
+        except Exception as e:
+            print(e)
+        sleep_time = 30
 
         while True:
             if plugin_name in self.plugins:
-                print(f"Executing plugin '{plugin_name}':")
+                # print(f"Executing plugin '{plugin_name}':")
                 if schedule['mode'] == "INTERVAL" or schedule['mode'] == "CONTIENUOUS":
-                    result = self.execute_plugin(plugin_name, *args, **kwargs)
-                    token = self.config.get_config('TELEGRAM')['BOT_TOKEN']
-                    chat_id = self.config.get_config('TELEGRAM')['CHAT_ID']
-                    await self.send_message_async(token, chat_id, result)
+                    result = await self.execute_plugin(plugin_name, *args, **kwargs)
+                    sleep_time = schedule['INTERVAL_TIME']
+                    if telegram_need:
+                        if result['result'] != "":
+                            await self.send_message_async(token, chat_id, result['result'])
+                    if tts_need:
+                        try:
+                            if result['result'] != "":
+                                tts = TTS(API_KEY=APIKey,
+                                          API_SECRET=APISecret, APP_ID=APPID)
+                                sever = TTSWebSocket(tts_obj=tts, msg=(0, result['result']))
+                                au_result = sever.run()
+                                # speakpath = os.path.join(self.config.BASE_PATH, au_result)
+                                print(au_result)
+                                playsound(au_result)
+                        except Exception as e:
+                            print(e)
                 elif schedule['mode'] == "FIX":
                     now = datetime.now()
-                    if now.hour == schedule['HOUR'] and now.minute == schedule['MINUTE']:
-                        result = self.execute_plugin(plugin_name, *args, **kwargs)
-                        token = self.config.get_config('TELEGRAM')['BOT_TOKEN']
-                        chat_id = self.config.get_config('TELEGRAM')['CHAT_ID']
-                        await self.send_message_async(token, chat_id, result)
+                    today = now.date()
+                    if now.hour <= schedule['HOUR'] and now.minute <= schedule['MINUTE']:
+                        next_day = today
+                    else:
+                        next_day = today + timedelta(days=1)
+                    next_time = datetime(year=next_day.year, month=next_day.month, day=next_day.day,
+                                         hour=schedule['HOUR'], minute=schedule['MINUTE'])
+                    sleep_time = (next_time - now).seconds
+                    if now.hour == schedule['HOUR'] and now.minute == schedule["MINUTE"]:
+                        result = await self.execute_plugin(plugin_name, *args, **kwargs)
+                        if telegram_need:
+                            if result['result'] != "":
+                                await self.send_message_async(token, chat_id, result['result'])
+                        if tts_need:
+                            try:
+                                if result['result'] != "":
+                                    tts = TTS(API_KEY=APIKey,
+                                              API_SECRET=APISecret, APP_ID=APPID)
+                                    sever = TTSWebSocket(tts_obj=tts, msg=(0, result['result']))
+                                    au_result = sever.run()
+                                    # speakpath = os.path.join(self.config.BASE_PATH, au_result)
+                                    print(au_result)
+                                    playsound(au_result)
+                            except Exception as e:
+                                print(e)
+
+
             else:
                 print(f"Plugin '{plugin_name}' not found.")
 
-            print(schedule)
-            await asyncio.sleep(schedule['INTERVAL_TIME'])
 
+            await asyncio.sleep(sleep_time)
 
     async def send_message_async(self, bot_token, chat_id, text):
-        proxy = telegram.request.HTTPXRequest(proxy_url='http://127.0.0.1:8889')
+        proxy_url = "http://" + self.config.get_proxy()['http']
+        proxy = telegram.request.HTTPXRequest(proxy_url=proxy_url)
         bot = telegram.Bot(token=bot_token, request=proxy)
         await bot.send_message(chat_id=chat_id, text=text)
 
@@ -119,13 +169,12 @@ class PluginManager:
         try:
             schedule_config = self.config.get_plugin_config('SCHEDULE', self.plugins_path[plugin_name])
             mode = schedule_config['MODE'].upper()
-            print(mode)
             schedule['mode'] = mode
             if mode == 'INTERVAL':
                 if 'INTERVAL_TIME' in schedule_config.keys():
                     interval_time = (schedule_config['INTERVAL_TIME']['HOUR'] * 3600 +
                                      schedule_config['INTERVAL_TIME']['MINUTE'] * 60 +
-                                        schedule_config['INTERVAL_TIME']['SECOND'])
+                                     schedule_config['INTERVAL_TIME']['SECOND'])
                     if interval_time < 30:
                         schedule['INTERVAL_TIME'] = 30
                     else:
@@ -133,8 +182,8 @@ class PluginManager:
                 else:
                     schedule['INTERVAL_TIME'] = 30
             elif mode == 'FIX' and "RUN_TIME" in schedule_config.keys():
-                schedule['HOUR'] = schedule_config["RUN_TIME"]['HOUR']
-                schedule['MINUTE'] = schedule_config["RUN_TIME"]['MINUTE']
+                schedule['HOUR'] = int(schedule_config["RUN_TIME"]['HOUR'])
+                schedule['MINUTE'] = int(schedule_config["RUN_TIME"]['MINUTE'])
                 schedule['INTERVAL_TIME'] = 30
             elif mode == 'CONTIENUOUS':
                 schedule['INTERVAL_TIME'] = 30

util/base_plugin.py

@@ -3,5 +3,5 @@ class BasePlugin:
         self.plugin_name = plugin_name
         self.title = title
 
-    def run_plugin(self, callback):
+    def run_plugin(self):
         raise NotImplementedError("Subclasses must implement run_plugin method")

util/config.py

@@ -1,13 +1,14 @@
 import yaml
 import os
+from util.singleton import singleton
 
-
+@singleton
 class Config:
 
     __BASE_PATH = os.getcwd()
     __CONFIG_PATH = os.path.join(__BASE_PATH, 'config')
-    __CONFIG_NAME = 'test.yaml'
-    __SCHEDULER_DB_FILE_NAME = 'schedule_db.db'
+    __CONFIG_NAME = 'config.yaml'
+    # __SCHEDULER_DB_FILE_NAME = 'schedule_db.db'
     __LOG_FILE_NAME = 'myAssistant.log'
     __CONFIG_DICT = {}
 
@@ -29,17 +30,17 @@ class Config:
     def BASE_PATH(self):
         return self.__BASE_PATH
 
-    def get_scheduler_db_file_path(self):
-        try:
-            return os.path.join(self.__BASE_PATH, self.__CONFIG_DICT['SCHEDULER_DB_PATH'],
-                                self.__SCHEDULER_DB_FILE_NAME)
-        except:
-            return os.path.join(self.__CONFIG_PATH, 'schedule_db', self.__SCHEDULER_DB_FILE_NAME)
+    # def get_scheduler_db_file_path(self):
+    #     try:
+    #         return os.path.join(self.__BASE_PATH, self.__CONFIG_DICT['SCHEDULER_DB_PATH'],
+    #                             self.__SCHEDULER_DB_FILE_NAME)
+    #     except:
+    #         return os.path.join(self.__CONFIG_PATH, 'schedule_db', self.__SCHEDULER_DB_FILE_NAME)
 
-    def set_scheduler_db_path(self, new_path):
-        self.__CONFIG_DICT['SCHEDULER_DB_PATH'] = new_path
-        with open(os.path.join(self.__CONFIG_PATH, self.__CONFIG_NAME), 'w') as f:
-            f.write(yaml.safe_dump(self.__CONFIG_DICT, sort_keys=False))
+    # def set_scheduler_db_path(self, new_path):
+    #     self.__CONFIG_DICT['SCHEDULER_DB_PATH'] = new_path
+    #     with open(os.path.join(self.__CONFIG_PATH, self.__CONFIG_NAME), 'w') as f:
+    #         f.write(yaml.safe_dump(self.__CONFIG_DICT, sort_keys=False))
 
     def get_log_file_path(self):
         try:
@@ -75,8 +76,7 @@ class Config:
             f.write(yaml.safe_dump(self.__CONFIG_DICT, sort_keys=False))
 
     def get_plugin_config(self, config_name, config_path):
-        print(config_path)
         if config_path != '':
             with open(os.path.join(config_path, self.__CONFIG_NAME)) as f:
                 config_d = yaml.safe_load(f)
-                return  config_d[config_name]
+                return config_d[config_name]

util/global_config.py

@@ -1,3 +1,6 @@
+from util.singleton import singleton
+
+@singleton
 class GlobalConfig:
     def __init__(self):
         """ 初始化 """
@@ -23,3 +26,6 @@ class GlobalConfig:
             return global_dict[name]
         except:
             return defaultvalue
+
+    def get_globalconfig_key(self):
+        return global_dict.keys()

util/singleton.py

@@ -0,0 +1,9 @@
+def singleton(cls):
+    instances = {}
+
+    def get_instance(*args, **kwargs):
+        if cls not in instances:
+            instances[cls] = cls(*args, **kwargs)
+        return instances[cls]
+
+    return get_instance

util/tts.py

@@ -0,0 +1,161 @@
+import base64
+import hashlib
+import hmac
+import json
+import ssl
+import threading
+import time
+from datetime import datetime
+from time import mktime
+from urllib.parse import urlencode
+from wsgiref.handlers import format_date_time
+import websocket
+import os
+from util.config import Config
+
+
+STATUS_FIRST_FRAME = 0  # 第一帧的标识
+STATUS_CONTINUE_FRAME = 1  # 中间帧标识
+STATUS_LAST_FRAME = 2  # 最后一帧的标识
+
+
+class TTS(object):
+
+    def __init__(self, APP_ID, API_KEY, API_SECRET):
+
+        self.APP_ID = APP_ID
+        self.API_KEY = API_KEY
+        self.API_SECRET = API_SECRET
+        self.config = Config()
+
+        # 公共参数(common)
+        self.common_args = {"app_id": self.APP_ID}
+
+        # 业务参数(business)
+        self.business_args = {
+            "aue": "lame", "sfl":1, "auf": "audio/L16;rate=16000", "vcn": "xiaoyan", "tte": "utf8"
+        }
+
+    # 生成业务数据流参数(data)
+    @staticmethod
+    def gen_data(text):
+        data = {
+            "status": 2,  # 数据状态，固定为2 注：由于流式合成的文本只能一次性传输，不支持多次分段传输，此处status必须为2。
+            "text": str(base64.b64encode(text.encode('utf-8')), "UTF8")
+        }
+        return data
+
+    # 生成url
+    def create_url(self):
+        url = 'wss://tts-api.xfyun.cn/v2/tts'
+        # 生成RFC1123格式的时间戳
+        now = datetime.now()
+        date = format_date_time(mktime(now.timetuple()))
+
+        # 拼接字符串
+        signature_origin = "host: " + "ws-api.xfyun.cn" + "\n"
+        signature_origin += "date: " + date + "\n"
+        signature_origin += "GET " + "/v2/tts " + "HTTP/1.1"
+        # 进行hmac-sha256进行加密
+        signature_sha = hmac.new(self.API_SECRET.encode('utf-8'), signature_origin.encode('utf-8'),
+                                 digestmod=hashlib.sha256).digest()
+        signature_sha = base64.b64encode(signature_sha).decode(encoding='utf-8')
+
+        authorization_origin = "api_key=\"%s\", algorithm=\"%s\", headers=\"%s\", signature=\"%s\"" % (
+            self.API_KEY, "hmac-sha256", "host date request-line", signature_sha)
+        authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8')
+        # 将请求的鉴权参数组合为字典
+        v = {
+            "authorization": authorization,
+            "date": date,
+            "host": "ws-api.xfyun.cn"
+        }
+        # 拼接鉴权参数，生成url
+        url = url + '?' + urlencode(v)
+        # print("date: ",date)
+        # print("v: ",v)
+        # 此处打印出建立连接时候的url,参考本demo的时候可取消上方打印的注释，比对相同参数时生成的url与自己代码生成的url是否一致
+        # print('websocket url :', url)
+        return url
+
+class TTSWebSocket(object):
+
+    def __init__(self, msg, tts_obj):
+        self.msg = msg
+        self.tts = tts_obj
+        self.url = tts_obj.create_url()
+        self.data = []
+        self.flag = False
+        self.audio_dir = "audio/"
+        self.ws_listener = None
+        self.config = Config()
+        websocket.enableTrace(False)
+        self.ws = websocket.WebSocketApp(self.url, on_message=self.on_message,
+                                         on_error=self.on_error, on_close=self.on_close)
+
+    def on_message(self, ws, msg):
+        try:
+            message = json.loads(msg)
+            print(message)
+            code = message["code"]
+            sid = message["sid"]
+            audio = message["data"]["audio"]
+            status = message["data"]["status"]
+
+            if code == 0:
+                self.data.append(audio)
+            else:
+                err_msg = message["message"]
+                print("sid:%s call error:%s code is:%s" % (sid, err_msg, code))
+            if status == 2:
+                print("------>数据接受完毕")
+                self.flag = True
+                self.ws.close()
+        except Exception as e:
+            print("receive msg,but parse exception:", e)
+            # print(sys.exc_info()[0])
+
+    def on_error(self, ws, error):
+        print("### error:", error)
+
+    def on_close(self, ws,  *args):
+        print("### closed ###")
+
+    def on_open(self, ws):
+        d = {"common": self.tts.common_args,
+             "business": self.tts.business_args,
+             "data": self.tts.gen_data(self.msg[1]),
+             }
+        d = json.dumps(d)
+        print("------>开始发送文本数据: {}".format(self.msg))
+        self.ws.send(d)
+
+    def get_result(self):
+        self.flag = False
+
+        if self.data:
+            audio_path = os.path.join(self.config.BASE_PATH, self.config.get_config("TTS")["AUDIO_PATH"])
+            print(self.config.get_config("TTS"))
+            audio_file = os.path.join(audio_path, "result.mp3")
+            print(audio_path)
+            with open(audio_file, 'wb') as f:
+                for _r in self.data:
+                    f.write(base64.b64decode(_r))
+            return audio_file
+        return "error：未收到任何信息"
+
+    def run(self):
+        self.ws.on_open = self.on_open
+        self.ws_listener = threading.Thread(target=self.ws.run_forever, kwargs={"sslopt": {"cert_reqs": ssl.CERT_NONE}})
+        self.ws_listener.daemon = True
+        self.ws_listener.start()
+
+        timeout = 15
+        end_time = time.time() + timeout
+        while True:
+            if time.time() > end_time:
+                raise websocket.WebSocketTimeoutException
+            if self.flag:
+                result = self.get_result()
+                return result
+