Merge pull request #468 from Jipok/tts

Add auto_tts plugin

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,)

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:
  """

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
+ }
+ }

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()

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(

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'])

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'],

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:

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)

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
  """

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)
 

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"]}' \

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"]}'
 

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:

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:

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}'
 

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)