testwit.py

from lisa_api.lisa.logger import logger
import requests
import json


class WitNLP(object):
    def __init__(self):
        self.token_wit = 'XXXXXXXXXXXXXXX'
        self.nlp_json = None
        self.lisa_api_url = 'http://127.0.0.1:8000'
        self.lisa_api_intent = None
        self.lisa_auth = 'Basic xxxxxxxxx='
        self.lisa_intent = None
        self.parsed_args = {}

    def _get_values_from_entities(self, key):
        """
        This function retrieve the value of a desired key (entity) in the wit json
        :param key:
        :return values: List of all values of an entity
        """
        entity_values = []
        for outcome in self.nlp_json['outcomes']:
            for entity in outcome['entities']:
                plugin, intent = entity.split('_', 1)
                if intent == key:
                    for values in outcome['entities'][entity]:
                        entity_values.append(values['value'])
        return entity_values

    def get_options(self, url, method):
        """
        Contact the LISA API to know the format of arguments
        :param url: The url to call for the options
        :param method: The method which will be used for the Intent
        :param entities: The json containing all parameters we will need to format
        :return: A dictionnary of arguments and their type of field
        """
        url_dict = {}
        for outcome in self.nlp_json['outcomes']:
            for entity in outcome['entities']:
                plugin, intent = entity.split('_', 1)
                url_dict[intent] = outcome['entities'][entity][0]['value']

        self.lisa_api_intent = ''.join([self.lisa_api_url, url]).format(**url_dict)

        r = requests.options(self.lisa_api_intent,
                             headers={"Authorization": self.lisa_auth,
                                      "content-type": "application/json"})
        options = r.json()
        self.parsed_args = {}

        if method.upper() in options['actions']:
            for argument in options['actions'][method.upper()]:
                value = self._get_info_field(options['actions'][method.upper()][argument])
                if value:
                    self.parsed_args[argument] = value
        return self.parsed_args

    def _get_info_field(self, argument):
        """
        This private method check the type of a field and cast the list values in the type
        :param argument: The metadata argument of a field
        :return: The value casted in the desired type
        """
        if argument.get('type') == 'list' and argument.get('child'):
            values = self._get_values_from_entities(argument['label'].lower())
            value = []
            for child_value in values:
                if argument['child']['type'] == 'string':
                    value.append(str(child_value))
                if argument['child']['type'] == 'integer':
                    value.append(int(child_value))
        elif argument.get('type') == 'string':
            value = str(self._get_values_from_entities(argument['label'].lower()))
        elif argument.get('type') == 'integer':
            value = int(self._get_values_from_entities(argument['label'].lower()))
        else:
            value = None
        return value

    def get_nlp_intent(self, sentence):
        """
        Contact the Wit API to have the json of the sentence
        :param sentence: The sentence said by the user
        :return: Return the wit json answer
        """
        self.nlp_json = None
        r_wit = requests.get('https://api.wit.ai/message?v=20141022&q={sentence}'.format(sentence=sentence),
                             headers={"Authorization": "Bearer %s" % self.token_wit})
        if r_wit.ok:
            self.nlp_json = r_wit.json()
            return self.nlp_json
        else:
            logger.debug('There was a problem contacting Wit.ai')
            logger.debug(r_wit.content)
            return False

    def get_lisa_intent(self):
        """
        Contact the LISA API to know the mapping between intent and url API
        :return: The intent metadata of lisa plugin
        """
        self.lisa_intent = None
        r_lisa_intent = requests.get('{lisa_api_url}/api/v1/core/intents/{intent_name}/'.format(
            lisa_api_url=self.lisa_api_url,
            intent_name=self.nlp_json['outcomes'][0]['intent']),
            headers={"Authorization": self.lisa_auth,
                     "content-type": "application/json"})
        if r_lisa_intent.ok:
            self.lisa_intent = r_lisa_intent.json()
            return self.lisa_intent
        else:
            logger.debug('There was a problem contacting LISA API')
            logger.debug(r_lisa_intent.content)
            return False

    def query_api(self, url, method, args):
        """
        This function call the LISA API's Plugin endpoint
        :param url: The url endpoint of the plugin
        :param method: The method used to query the endpoint
        :param args: A dictionnary of arguments to send to the endpoint
        :return: A string containing the answer of LISA API's Plugin
        """
        query = {
            'post': requests.post,
            'get': requests.get,
            'delete': requests.delete,
            'put': requests.put,
            'patch': requests.patch,
            'options': requests.options
        }

        r = query[method.lower()](url=url,
                                  data=json.dumps(args),
                                  headers={"Authorization": self.lisa_auth,
                                           "content-type": "application/json"})
        return r.content

    def process_sentence(self, sentence):
        """
        This function will do the magic with all other functions in the class to make api calls
        :param sentence: A sentence said by the user
        :return answer: A string containing the answer of LISA API's Plugin
        """
        self.get_nlp_intent(sentence=sentence)
        self.get_lisa_intent()
        self.get_options(url=self.lisa_intent['api_url'],
                         method=self.lisa_intent['method'])
        answer = self.query_api(url=self.lisa_api_intent,
                                method=self.lisa_intent['method'],
                                args=self.parsed_args)
        logger.debug(answer)
        return answer

wit = WitNLP()
wit.process_sentence('enleve les tomates de la liste de course carrefour')