From e4524ad6507ac5b4f724db2021e5e641dce5a617 Mon Sep 17 00:00:00 2001 From: Jipok Date: Thu, 23 Nov 2023 20:48:16 +0500 Subject: [PATCH] Add auto_tts plugin --- bot/openai_helper.py | 2 +- bot/plugin_manager.py | 6 ++-- bot/plugins/auto_tts.py | 44 ++++++++++++++++++++++++++ bot/plugins/crypto.py | 2 +- bot/plugins/ddg_image_search.py | 2 +- bot/plugins/ddg_translate.py | 2 +- bot/plugins/ddg_web_search.py | 2 +- bot/plugins/deepl.py | 2 +- bot/plugins/gtts_text_to_speech.py | 2 +- bot/plugins/plugin.py | 2 +- bot/plugins/spotify.py | 2 +- bot/plugins/weather.py | 2 +- bot/plugins/webshot.py | 2 +- bot/plugins/whois_.py | 2 +- bot/plugins/wolfram_alpha.py | 2 +- bot/plugins/worldtimeapi.py | 2 +- bot/plugins/youtube_audio_extractor.py | 2 +- 17 files changed, 63 insertions(+), 17 deletions(-) create mode 100644 bot/plugins/auto_tts.py diff --git a/bot/openai_helper.py b/bot/openai_helper.py index 6346eff9..951d745b 100644 --- a/bot/openai_helper.py +++ b/bot/openai_helper.py @@ -301,7 +301,7 @@ async def __handle_function_call(self, chat_id, response, stream=False, times=0, return response, plugins_used logging.info(f'Calling function {function_name} with arguments {arguments}') - function_response = await self.plugin_manager.call_function(function_name, arguments) + function_response = await self.plugin_manager.call_function(function_name, self, arguments) if function_name not in plugins_used: plugins_used += (function_name,) diff --git a/bot/plugin_manager.py b/bot/plugin_manager.py index f0adeccf..370b1c3e 100644 --- a/bot/plugin_manager.py +++ b/bot/plugin_manager.py @@ -1,6 +1,7 @@ import json from plugins.gtts_text_to_speech import GTTSTextToSpeech +from plugins.auto_tts import AutoTextToSpeech from plugins.dice import DicePlugin from plugins.youtube_audio_extractor import YouTubeAudioExtractorPlugin from plugins.ddg_image_search import DDGImageSearchPlugin @@ -36,6 +37,7 @@ def __init__(self, config): 'dice': DicePlugin, 'deepl_translate': DeeplTranslatePlugin, 'gtts_text_to_speech': GTTSTextToSpeech, + 'auto_tts': AutoTextToSpeech, 'whois': WhoisPlugin, 'webshot': WebshotPlugin, } @@ -47,14 +49,14 @@ def get_functions_specs(self): """ return [spec for specs in map(lambda plugin: plugin.get_spec(), self.plugins) for spec in specs] - async def call_function(self, function_name, arguments): + async def call_function(self, function_name, helper, arguments): """ Call a function based on the name and parameters provided """ plugin = self.__get_plugin_by_function_name(function_name) if not plugin: return json.dumps({'error': f'Function {function_name} not found'}) - return json.dumps(await plugin.execute(function_name, **json.loads(arguments)), default=str) + return json.dumps(await plugin.execute(function_name, helper, **json.loads(arguments)), default=str) def get_plugin_source_name(self, function_name) -> str: """ diff --git a/bot/plugins/auto_tts.py b/bot/plugins/auto_tts.py new file mode 100644 index 00000000..7118b963 --- /dev/null +++ b/bot/plugins/auto_tts.py @@ -0,0 +1,44 @@ +import datetime +import tempfile +from typing import Dict + +from .plugin import Plugin + + +class AutoTextToSpeech(Plugin): + """ + A plugin to convert text to speech using Openai Speech API + """ + + def get_source_name(self) -> str: + return "TTS" + + def get_spec(self) -> [Dict]: + return [{ + "name": "translate_text_to_speech", + "description": "Translate text to speech using OpenAI API", + "parameters": { + "type": "object", + "properties": { + "text": {"type": "string", "description": "The text to translate to speech"}, + }, + "required": ["text"], + }, + }] + + async def execute(self, function_name, helper, **kwargs) -> Dict: + try: + bytes, text_length = await helper.generate_speech(text=kwargs['text']) + with tempfile.NamedTemporaryFile(delete=False, suffix='.opus') as temp_file: + temp_file.write(bytes.getvalue()) + temp_file_path = temp_file.name + except Exception as e: + logging.exception(e) + return {"Result": "Exception: " + str(e)} + return { + 'direct_result': { + 'kind': 'file', + 'format': 'path', + 'value': temp_file_path + } + } diff --git a/bot/plugins/crypto.py b/bot/plugins/crypto.py index 42e35e32..5559a22c 100644 --- a/bot/plugins/crypto.py +++ b/bot/plugins/crypto.py @@ -26,5 +26,5 @@ def get_spec(self) -> [Dict]: }, }] - async def execute(self, function_name, **kwargs) -> Dict: + async def execute(self, function_name, helper, **kwargs) -> Dict: return requests.get(f"https://api.coincap.io/v2/rates/{kwargs['asset']}").json() diff --git a/bot/plugins/ddg_image_search.py b/bot/plugins/ddg_image_search.py index 61918243..a1db0353 100644 --- a/bot/plugins/ddg_image_search.py +++ b/bot/plugins/ddg_image_search.py @@ -49,7 +49,7 @@ def get_spec(self) -> [Dict]: }, }] - async def execute(self, function_name, **kwargs) -> Dict: + async def execute(self, function_name, helper, **kwargs) -> Dict: with DDGS() as ddgs: image_type = kwargs.get('type', 'photo') ddgs_images_gen = ddgs.images( diff --git a/bot/plugins/ddg_translate.py b/bot/plugins/ddg_translate.py index c5b1211d..294ff05e 100644 --- a/bot/plugins/ddg_translate.py +++ b/bot/plugins/ddg_translate.py @@ -26,6 +26,6 @@ def get_spec(self) -> [Dict]: }, }] - async def execute(self, function_name, **kwargs) -> Dict: + async def execute(self, function_name, helper, **kwargs) -> Dict: with DDGS() as ddgs: return ddgs.translate(kwargs['text'], to=kwargs['to_language']) diff --git a/bot/plugins/ddg_web_search.py b/bot/plugins/ddg_web_search.py index fbd3d78b..07060a90 100644 --- a/bot/plugins/ddg_web_search.py +++ b/bot/plugins/ddg_web_search.py @@ -46,7 +46,7 @@ def get_spec(self) -> [Dict]: }, }] - async def execute(self, function_name, **kwargs) -> Dict: + async def execute(self, function_name, helper, **kwargs) -> Dict: with DDGS() as ddgs: ddgs_gen = ddgs.text( kwargs['query'], diff --git a/bot/plugins/deepl.py b/bot/plugins/deepl.py index a43a6f8c..d2236c20 100644 --- a/bot/plugins/deepl.py +++ b/bot/plugins/deepl.py @@ -33,7 +33,7 @@ def get_spec(self) -> [Dict]: }, }] - async def execute(self, function_name, **kwargs) -> Dict: + async def execute(self, function_name, helper, **kwargs) -> Dict: if self.api_key.endswith(':fx'): url = "https://api-free.deepl.com/v2/translate" else: diff --git a/bot/plugins/gtts_text_to_speech.py b/bot/plugins/gtts_text_to_speech.py index 50f01207..544de9f6 100644 --- a/bot/plugins/gtts_text_to_speech.py +++ b/bot/plugins/gtts_text_to_speech.py @@ -31,7 +31,7 @@ def get_spec(self) -> [Dict]: }, }] - async def execute(self, function_name, **kwargs) -> Dict: + async def execute(self, function_name, helper, **kwargs) -> Dict: tts = gTTS(kwargs['text'], lang=kwargs.get('lang', 'en')) output = f'gtts_{datetime.datetime.now().timestamp()}.mp3' tts.save(output) diff --git a/bot/plugins/plugin.py b/bot/plugins/plugin.py index b028b270..c9c734dd 100644 --- a/bot/plugins/plugin.py +++ b/bot/plugins/plugin.py @@ -23,7 +23,7 @@ def get_spec(self) -> [Dict]: pass @abstractmethod - async def execute(self, function_name, **kwargs) -> Dict: + async def execute(self, function_name, helper, **kwargs) -> Dict: """ Execute the plugin and return a JSON serializable response """ diff --git a/bot/plugins/spotify.py b/bot/plugins/spotify.py index 3c0db65c..a578ed2c 100644 --- a/bot/plugins/spotify.py +++ b/bot/plugins/spotify.py @@ -111,7 +111,7 @@ def get_spec(self) -> [Dict]: } ] - async def execute(self, function_name, **kwargs) -> Dict: + async def execute(self, function_name, helper, **kwargs) -> Dict: time_range = kwargs.get('time_range', 'short_term') limit = kwargs.get('limit', 5) diff --git a/bot/plugins/weather.py b/bot/plugins/weather.py index abfce458..7b2b1f29 100644 --- a/bot/plugins/weather.py +++ b/bot/plugins/weather.py @@ -57,7 +57,7 @@ def get_spec(self) -> [Dict]: } ] - async def execute(self, function_name, **kwargs) -> Dict: + async def execute(self, function_name, helper, **kwargs) -> Dict: url = f'https://api.open-meteo.com/v1/forecast' \ f'?latitude={kwargs["latitude"]}' \ f'&longitude={kwargs["longitude"]}' \ diff --git a/bot/plugins/webshot.py b/bot/plugins/webshot.py index 4eb0e03b..fa925629 100644 --- a/bot/plugins/webshot.py +++ b/bot/plugins/webshot.py @@ -26,7 +26,7 @@ def generate_random_string(self, length): characters = string.ascii_letters + string.digits return ''.join(random.choice(characters) for _ in range(length)) - async def execute(self, function_name, **kwargs) -> Dict: + async def execute(self, function_name, helper, **kwargs) -> Dict: try: image_url = f'https://image.thum.io/get/maxAge/12/width/720/{kwargs["url"]}' diff --git a/bot/plugins/whois_.py b/bot/plugins/whois_.py index 7814558a..91b81b5f 100644 --- a/bot/plugins/whois_.py +++ b/bot/plugins/whois_.py @@ -24,7 +24,7 @@ def get_spec(self) -> [Dict]: }, }] - async def execute(self, function_name, **kwargs) -> Dict: + async def execute(self, function_name, helper, **kwargs) -> Dict: try: whois_result = whois.query(kwargs['domain']) if whois_result is None: diff --git a/bot/plugins/wolfram_alpha.py b/bot/plugins/wolfram_alpha.py index fc97e95c..b151ae03 100644 --- a/bot/plugins/wolfram_alpha.py +++ b/bot/plugins/wolfram_alpha.py @@ -32,7 +32,7 @@ def get_spec(self) -> [Dict]: } }] - async def execute(self, function_name, **kwargs) -> Dict: + async def execute(self, function_name, helper, **kwargs) -> Dict: client = wolframalpha.Client(self.app_id) res = client.query(kwargs['query']) try: diff --git a/bot/plugins/worldtimeapi.py b/bot/plugins/worldtimeapi.py index 9ec15d7e..2e09e366 100644 --- a/bot/plugins/worldtimeapi.py +++ b/bot/plugins/worldtimeapi.py @@ -35,7 +35,7 @@ def get_spec(self) -> [Dict]: }, }] - async def execute(self, function_name, **kwargs) -> Dict: + async def execute(self, function_name, helper, **kwargs) -> Dict: timezone = kwargs.get('timezone', self.default_timezone) url = f'https://worldtimeapi.org/api/timezone/{timezone}' diff --git a/bot/plugins/youtube_audio_extractor.py b/bot/plugins/youtube_audio_extractor.py index 5fc4d801..804663a8 100644 --- a/bot/plugins/youtube_audio_extractor.py +++ b/bot/plugins/youtube_audio_extractor.py @@ -28,7 +28,7 @@ def get_spec(self) -> [Dict]: }, }] - async def execute(self, function_name, **kwargs) -> Dict: + async def execute(self, function_name, helper, **kwargs) -> Dict: link = kwargs['youtube_link'] try: video = YouTube(link)