90 lines
4.1 KiB
Python
90 lines
4.1 KiB
Python
from opsdroid.skill import Skill
|
|
from opsdroid.matchers import match_rasanlu, match_parse
|
|
import random
|
|
import importlib
|
|
|
|
class Respond(Skill):
|
|
# Constructor gets configs for the skill and applies the decorator
|
|
# to match using rasa.
|
|
def __init__(self, opsdroid, config):
|
|
super().__init__(opsdroid, config)
|
|
self._get_configs()
|
|
self.respond = match_rasanlu(self.intent)(self.respond)
|
|
|
|
# This utility function will grab the relevent config variables
|
|
# for this skill to function.
|
|
def _get_configs(self):
|
|
# The intent to match.
|
|
self.intent = self.config.get("intent")
|
|
# Possible responses given a match.
|
|
self.response_options = self.config.get("response_options")
|
|
# A bool to check if we want to use entities in our response.
|
|
self.use_entities = self.config.get("use_entities", False)
|
|
# Fallback responses if we do not get the required entities
|
|
# from Rasa.
|
|
self.fallback_response_options = self.config.get("fallback_response_options",
|
|
["Sorry, Rasa did not extract sufficient entities to fulfil your request"])
|
|
# An optional callback used for more advanced responses. Such
|
|
# functions should simply return the final response string and
|
|
# will be given the message and extracted entities (if enabled).
|
|
self.callback_file = self.config.get("callback_file", None)
|
|
self.callback_name = self.config.get("callback_name", None)
|
|
|
|
|
|
# Extract the entities from the rasanlu response
|
|
def _get_entities(self, message):
|
|
# This includes data such as the extractor used, which we
|
|
# don't need.
|
|
raw_entities = message.rasanlu["entities"]
|
|
entities = {}
|
|
for entity in raw_entities:
|
|
entity_type = entity["entity"]
|
|
unique_entity_type = self._get_unique_key(entity_type, entities)
|
|
entity_value = entity["value"]
|
|
entities[unique_entity_type] = entity_value
|
|
return entities
|
|
|
|
|
|
# Appends the smallest integer to the name of the key in order to
|
|
# create a new key that is not already in the dictionary.
|
|
def _get_unique_key(self, base_key, dictionary):
|
|
counter = 0
|
|
while True:
|
|
unique_key_attempt = base_key + str(counter)
|
|
if unique_key_attempt in dictionary:
|
|
counter += 1
|
|
continue
|
|
return unique_key_attempt
|
|
|
|
|
|
# Takes the response and attempts to replace all entities in it
|
|
# with the extracted entities. Failing that, we will fall back to a
|
|
# fallback response.
|
|
def _substitute_entities(self, response, entities):
|
|
try:
|
|
response = response.format(**entities)
|
|
except KeyError as e:
|
|
chosen_fallback = random.choice(self.fallback_response_options)
|
|
return chosen_fallback
|
|
return response
|
|
|
|
|
|
# This line is to add properties to respond on declaration
|
|
# as doing this in __init__ would cause an exception to be
|
|
# raised. If you don't believe me, remove it!
|
|
@match_parse("hgftgyhjknbvhgftyuihjkvhgftyuijkbjhgtyyuijhkjghfdtrytyihujkbvncgfxdtrytuyiuhkbjvbhcfdytryuhjkbvhcfdyrtuyiuhkbhj")
|
|
async def respond(self, message):
|
|
chosen_response = random.choice(self.response_options)
|
|
entities = None # In case we want callbacks without entities
|
|
if self.use_entities:
|
|
entities = self._get_entities(message)
|
|
chosen_response = self._substitute_entities(chosen_response, entities)
|
|
if self.callback_file != None and self.callback_name != None:
|
|
# The following three lines allow us to import the module using an absolute path dynamically
|
|
spec = importlib.util.spec_from_file_location("callback", self.callback_file)
|
|
callback_module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(callback_module)
|
|
# Now we can get the response from the callback.
|
|
chosen_response = eval("callback_module." + self.callback_name + "(message, entities)")
|
|
await message.respond(chosen_response)
|