Login With Github

ChatterBot Input Adapters

ChatterBot's input adapters are designed to allow a chat bot to have a versatile method of receiving or retrieving input from a given source.

The goal of an input adapter is to get input from some source, and then to convert it into a format that ChatterBot can understand. This format is the Statement object found in ChatterBot'sconversation module.

Creating a new input adapter

You can write your own input adapters by creating a new class that inherits from InputAdapter and overrides the necessary methods established in the base InputAdapter class.

from __future__ import unicode_literals
from chatterbot.adapters import Adapter


[docs]class InputAdapter(Adapter):
    """
    This is an abstract class that represents the
    interface that all input adapters should implement.
    """

    def process_input(self, *args, **kwargs):
        """
        Returns a statement object based on the input source.
        """
        raise self.AdapterMethodNotImplementedError()

    def process_input_statement(self, *args, **kwargs):
        """
        Return an existing statement object (if one exists).
        """
        input_statement = self.process_input(*args, **kwargs)

        self.logger.info('Received input statement: {}'.format(input_statement.text))

        existing_statement = self.chatbot.storage.find(input_statement.text)

        if existing_statement:
            self.logger.info('"{}" is a known statement'.format(input_statement.text))
            input_statement = existing_statement
        else:
            self.logger.info('"{}" is not a known statement'.format(input_statement.text))

        return input_statement

This is an abstract class that represents the interface that all input adapters should implement.

To create your own input adapter you must override the process_input method to return a Statement object.

Note that you may need to extend the __init__ method of your custom input adapter if you intend to save a kwarg parameter that was passed into the chat bot's constructor. (An API key might be an example of a parameter you would want to access here.)

from __future__ import unicode_literals
from chatterbot.adapters import Adapter


class InputAdapter(Adapter):
    """
    This is an abstract class that represents the
    interface that all input adapters should implement.
    """

    def process_input(self, *args, **kwargs):
        """
        Returns a statement object based on the input source.
        """
        raise self.AdapterMethodNotImplementedError()

    def process_input_statement(self, *args, **kwargs):
        """
        Return an existing statement object (if one exists).
        """
        input_statement = self.process_input(*args, **kwargs)

        self.logger.info('Received input statement: {}'.format(input_statement.text))

        existing_statement = self.chatbot.storage.find(input_statement.text)

        if existing_statement:
            self.logger.info('"{}" is a known statement'.format(input_statement.text))
            input_statement = existing_statement
        else:
            self.logger.info('"{}" is not a known statement'.format(input_statement.text))

        return input_statement

Variable input type adapter

from __future__ import unicode_literals
from chatterbot.input import InputAdapter
from chatterbot.conversation import Statement


[docs]class VariableInputTypeAdapter(InputAdapter):

    JSON = 'json'
    TEXT = 'text'
    OBJECT = 'object'
    VALID_FORMATS = (JSON, TEXT, OBJECT, )

    def detect_type(self, statement):
        import sys

        if sys.version_info[0] < 3:
            string_types = basestring # NOQA
        else:
            string_types = str

        if hasattr(statement, 'text'):
            return self.OBJECT
        if isinstance(statement, string_types):
            return self.TEXT
        if isinstance(statement, dict):
            return self.JSON

        input_type = type(statement)

        raise self.UnrecognizedInputFormatException(
            'The type {} is not recognized as a valid input type.'.format(
                input_type
            )
        )

    def process_input(self, statement):
        input_type = self.detect_type(statement)

        # Return the statement object without modification
        if input_type == self.OBJECT:
            return statement

        # Convert the input string into a statement object
        if input_type == self.TEXT:
            return Statement(statement)

        # Convert input dictionary into a statement object
        if input_type == self.JSON:
            input_json = dict(statement)
            text = input_json['text']
            del input_json['text']

            return Statement(text, **input_json)

    class UnrecognizedInputFormatException(Exception):
        """
        Exception raised when an input format is specified that is
        not in the VariableInputTypeAdapter.VALID_FORMATS variable.
        """

        def __init__(self, value='The input format was not recognized.'):
            self.value = value

        def __str__(self):
            return repr(self.value)

The variable input type adapter allows the chat bot to accept a number of different input types using the same adapter. This adapter accepts stringsdictionaries and Statements.

chatbot = ChatBot(
    "My ChatterBot",
    input_adapter="chatterbot.input.VariableInputTypeAdapter"
)

Terminal input adapter

from __future__ import unicode_literals
from chatterbot.input import InputAdapter
from chatterbot.conversation import Statement
from chatterbot.utils import input_function


[docs]class TerminalAdapter(InputAdapter):
    """
    A simple adapter that allows ChatterBot to
    communicate through the terminal.
    """

    def process_input(self, *args, **kwargs):
        """
        Read the user's input from the terminal.
        """
        user_input = input_function()
        return Statement(user_input)

A simple adapter that allows ChatterBot to communicate through the terminal.

The input terminal adapter allows a user to type into their terminal to communicate with the chat bot.

chatbot = ChatBot(
    "My ChatterBot",
    input_adapter="chatterbot.input.TerminalAdapter"
)

Gitter input adapter

from __future__ import unicode_literals
from time import sleep
from chatterbot.input import InputAdapter
from chatterbot.conversation import Statement


[docs]class Gitter(InputAdapter):
    """
    An input adapter that allows a ChatterBot instance to get
    input statements from a Gitter room.
    """

    def __init__(self, **kwargs):
        super(Gitter, self).__init__(**kwargs)

        self.gitter_host = kwargs.get('gitter_host', 'https://api.gitter.im/v1/')
        self.gitter_room = kwargs.get('gitter_room')
        self.gitter_api_token = kwargs.get('gitter_api_token')
        self.only_respond_to_mentions = kwargs.get('gitter_only_respond_to_mentions', True)
        self.sleep_time = kwargs.get('gitter_sleep_time', 4)

        authorization_header = 'Bearer {}'.format(self.gitter_api_token)

        self.headers = {
            'Authorization': authorization_header,
            'Content-Type': 'application/json',
            'Accept': 'application/json'
        }

        # Join the Gitter room
        room_data = self.join_room(self.gitter_room)
        self.room_id = room_data.get('id')

        user_data = self.get_user_data()
        self.user_id = user_data[0].get('id')
        self.username = user_data[0].get('username')

    def _validate_status_code(self, response):
        code = response.status_code
        if code not in [200, 201]:
            raise self.HTTPStatusException('{} status code recieved'.format(code))

    def join_room(self, room_name):
        """
        Join the specified Gitter room.
        """
        import requests

        endpoint = '{}rooms'.format(self.gitter_host)
        response = requests.post(
            endpoint,
            headers=self.headers,
            json={'uri': room_name}
        )
        self.logger.info('{} joining room {}'.format(
            response.status_code, endpoint
        ))
        self._validate_status_code(response)
        return response.json()

    def get_user_data(self):
        import requests

        endpoint = '{}user'.format(self.gitter_host)
        response = requests.get(
            endpoint,
            headers=self.headers
        )
        self.logger.info('{} retrieving user data {}'.format(
            response.status_code, endpoint
        ))
        self._validate_status_code(response)
        return response.json()

    def mark_messages_as_read(self, message_ids):
        """
        Mark the specified message ids as read.
        """
        import requests

        endpoint = '{}user/{}/rooms/{}/unreadItems'.format(
            self.gitter_host, self.user_id, self.room_id
        )
        response = requests.post(
            endpoint,
            headers=self.headers,
            json={'chat': message_ids}
        )
        self.logger.info('{} marking messages as read {}'.format(
            response.status_code, endpoint
        ))
        self._validate_status_code(response)
        return response.json()

    def get_most_recent_message(self):
        """
        Get the most recent message from the Gitter room.
        """
        import requests

        endpoint = '{}rooms/{}/chatMessages?limit=1'.format(self.gitter_host, self.room_id)
        response = requests.get(
            endpoint,
            headers=self.headers
        )
        self.logger.info('{} getting most recent message'.format(
            response.status_code
        ))
        self._validate_status_code(response)
        data = response.json()
        if data:
            return data[0]
        return None

    def _contains_mention(self, mentions):
        for mention in mentions:
            if self.username == mention.get('screenName'):
                return True
        return False

    def should_respond(self, data):
        """
        Takes the API response data from a single message.
        Returns true if the chat bot should respond.
        """
        if data:
            unread = data.get('unread', False)

            if self.only_respond_to_mentions:
                if unread and self._contains_mention(data['mentions']):
                    return True
                else:
                    return False
            elif unread:
                return True

        return False

    def remove_mentions(self, text):
        """
        Return a string that has no leading mentions.
        """
        import re
        text_without_mentions = re.sub(r'@\S+', '', text)

        # Remove consecutive spaces
        text_without_mentions = re.sub(' +', ' ', text_without_mentions.strip())

        return text_without_mentions

    def process_input(self, statement):
        new_message = False

        while not new_message:
            data = self.get_most_recent_message()
            if self.should_respond(data):
                self.mark_messages_as_read([data['id']])
                new_message = True
            sleep(self.sleep_time)

        text = self.remove_mentions(data['text'])
        statement = Statement(text)

        return statement

    class HTTPStatusException(Exception):
        """
        Exception raised when unexpected non-success HTTP
        status codes are returned in a response.
        """

        def __init__(self, value):
            self.value = value

        def __str__(self):
            return repr(self.value)

An input adapter that allows a ChatterBot instance to get input statements from a Gitter room.

chatbot = ChatBot(
    "My ChatterBot",
    input_adapter="chatterbot.input.Gitter",
    gitter_api_token="my-gitter-api-token",
    gitter_room="my-room-name",
    gitter_only_respond_to_mentions=True,
)

HipChat input adapter

from __future__ import unicode_literals
from time import sleep
from chatterbot.input import InputAdapter
from chatterbot.conversation import Statement


[docs]class HipChat(InputAdapter):
    """
    An input adapter that allows a ChatterBot instance to get
    input statements from a HipChat room.
    """

    def __init__(self, **kwargs):
        super(HipChat, self).__init__(**kwargs)

        self.hipchat_host = kwargs.get('hipchat_host')
        self.hipchat_access_token = kwargs.get('hipchat_access_token')
        self.hipchat_room = kwargs.get('hipchat_room')
        self.session_id = str(self.chatbot.default_session.uuid)

        import requests
        self.session = requests.Session()
        self.session.verify = kwargs.get('ssl_verify', True)

        authorization_header = 'Bearer {}'.format(self.hipchat_access_token)

        self.headers = {
            'Authorization': authorization_header,
            'Content-Type': 'application/json'
        }

        # This is a list of the messages that have been responded to
        self.recent_message_ids = self.get_initial_ids()

    def get_initial_ids(self):
        """
        Returns a list of the most recent message ids.
        """
        data = self.view_recent_room_history(
            self.hipchat_room,
            max_results=75
        )

        results = set()

        for item in data['items']:
            results.add(item['id'])

        return results

    def view_recent_room_history(self, room_id_or_name, max_results=1):
        """
        https://www.hipchat.com/docs/apiv2/method/view_recent_room_history
        """

        recent_histroy_url = '{}/v2/room/{}/history?max-results={}'.format(
            self.hipchat_host,
            room_id_or_name,
            max_results
        )

        response = self.session.get(
            recent_histroy_url,
            headers=self.headers
        )

        return response.json()

    def get_most_recent_message(self, room_id_or_name):
        """
        Return the most recent message from the HipChat room.
        """
        data = self.view_recent_room_history(room_id_or_name)

        items = data['items']

        if not items:
            return None
        return items[-1]

    def process_input(self, statement):
        """
        Process input from the HipChat room.
        """
        new_message = False

        response_statement = self.chatbot.storage.get_latest_response(
            self.session_id
        )

        if response_statement:
            last_message_id = response_statement.extra_data.get(
                'hipchat_message_id', None
            )
            if last_message_id:
                self.recent_message_ids.add(last_message_id)

        while not new_message:
            data = self.get_most_recent_message(self.hipchat_room)

            if data and data['id'] not in self.recent_message_ids:
                self.recent_message_ids.add(data['id'])
                new_message = True
            else:
                pass
            sleep(3.5)

        text = data['message']

        statement = Statement(text)
        statement.add_extra_data('hipchat_message_id', data['id'])

        return statement

An input adapter that allows a ChatterBot instance to get input statements from a HipChat room.

This is an input adapter that allows a ChatterBot instance to communicate through HipChat.

Be sure to also see the documentation for the HipChat output adapter.

chatbot = ChatBot(
    "My ChatterBot",
    input_adapter="chatterbot.input.HipChat",
    hipchat_host="https://mydomain.hipchat.com",
    hipchat_room="my-room-name",
    hipchat_access_token="my-hipchat-access-token",
)

Mailgun input adapter

from __future__ import unicode_literals
import datetime
from chatterbot.input import InputAdapter
from chatterbot.conversation import Statement


[docs]class Mailgun(InputAdapter):
    """
    Get input from Mailgun.
    """

    def __init__(self, **kwargs):
        super(Mailgun, self).__init__(**kwargs)

        # Use the bot's name for the name of the sender
        self.name = kwargs.get('name')
        self.from_address = kwargs.get('mailgun_from_address')
        self.api_key = kwargs.get('mailgun_api_key')
        self.endpoint = kwargs.get('mailgun_api_endpoint')

    def get_email_stored_events(self):
        import requests

        yesterday = datetime.datetime.now() - datetime.timedelta(1)
        return requests.get(
            '{}/events'.format(self.endpoint),
            auth=('api', self.api_key),
            params={
                'begin': yesterday.isoformat(),
                'ascending': 'yes',
                'limit': 1
            }
        )

    def get_stored_email_urls(self):
        response = self.get_email_stored_events()
        data = response.json()

        for item in data.get('items', []):
            if 'storage' in item:
                if 'url' in item['storage']:
                    yield item['storage']['url']

    def get_message(self, url):
        import requests

        return requests.get(
            url,
            auth=('api', self.api_key)
        )

    def process_input(self, statement):
        urls = self.get_stored_email_urls()
        url = list(urls)[0]

        response = self.get_message(url)
        message = response.json()

        text = message.get('stripped-text')

        return Statement(text)

Get input from Mailgun.

The Mailgun adapter allows a chat bot to receive emails using the Mailgun API.

# -*- coding: utf-8 -*-
from chatterbot import ChatBot
from settings import MAILGUN

'''
To use this example, create a new file called settings.py.
In settings.py define the following:

MAILGUN = {
    "CONSUMER_KEY": "my-mailgun-api-key",
    "API_ENDPOINT": "https://api.mailgun.net/v3/my-domain.com/messages"
}
'''

# Change these to match your own email configuration
FROM_EMAIL = "mailgun@salvius.org"
RECIPIENTS = ["gunthercx@gmail.com"]

bot = ChatBot(
    "Mailgun Example Bot",
    mailgun_from_address=FROM_EMAIL,
    mailgun_api_key=MAILGUN["CONSUMER_KEY"],
    mailgun_api_endpoint=MAILGUN["API_ENDPOINT"],
    mailgun_recipients=RECIPIENTS,
    input_adapter="chatterbot.input.Mailgun",
    output_adapter="chatterbot.output.Mailgun",
    storage_adapter="chatterbot.storage.SQLStorageAdapter",
    database="../database.db"
)

# Send an example email to the address provided
response = bot.get_response("How are you?")
print("Check your inbox at ", RECIPIENTS)

Microsoft Bot Framework input adapter

from __future__ import unicode_literals
from time import sleep
from chatterbot.input import InputAdapter
from chatterbot.conversation import Statement


[docs]class Microsoft(InputAdapter):
    """
    An input adapter that allows a ChatterBot instance to get
    input statements from a Microsoft Bot using *Directline client protocol*.
    https://docs.botframework.com/en-us/restapi/directline/#navtitle
    """

    def __init__(self, **kwargs):
        super(Microsoft, self).__init__(**kwargs)
        import requests
        from requests.packages.urllib3.exceptions import InsecureRequestWarning
        requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

        self.directline_host = kwargs.get('directline_host', 'https://directline.botframework.com')

        # NOTE: Direct Line client credentials are different from your bot's
        # credentials
        self.direct_line_token_or_secret = kwargs.\
            get('direct_line_token_or_secret')

        authorization_header = 'BotConnector  {}'.\
            format(self.direct_line_token_or_secret)

        self.headers = {
            'Authorization': authorization_header,
            'Content-Type': 'application/json',
            'Accept': 'application/json',
            'charset': 'utf-8'
        }

        conversation_data = self.start_conversation()
        self.conversation_id = conversation_data.get('conversationId')
        self.conversation_token = conversation_data.get('token')

    def _validate_status_code(self, response):
        code = response.status_code
        if not code == 200:
            raise self.HTTPStatusException('{} status code recieved'.
                                           format(code))

    def start_conversation(self):
        import requests

        endpoint = '{host}/api/conversations'.format(host=self.directline_host)
        response = requests.post(
            endpoint,
            headers=self.headers,
            verify=False
        )
        self.logger.info('{} starting conversation {}'.format(
            response.status_code, endpoint
        ))
        self._validate_status_code(response)
        return response.json()

    def get_most_recent_message(self):
        import requests

        endpoint = '{host}/api/conversations/{id}/messages'\
            .format(host=self.directline_host,
                    id=self.conversation_id)

        response = requests.get(
            endpoint,
            headers=self.headers,
            verify=False
        )

        self.logger.info('{} retrieving most recent messages {}'.format(
            response.status_code, endpoint
        ))

        self._validate_status_code(response)

        data = response.json()

        if data['messages']:
            last_msg = int(data['watermark'])
            return data['messages'][last_msg - 1]
        return None

    def process_input(self, statement):
        new_message = False
        data = None
        while not new_message:
            data = self.get_most_recent_message()
            if data and data['id']:
                new_message = True
            else:
                pass
            sleep(3.5)

        text = data['text']
        statement = Statement(text)
        self.logger.info('processing user statement {}'.format(statement))

        return statement

    class HTTPStatusException(Exception):
        """
        Exception raised when unexpected non-success HTTP
        status codes are returned in a response.
        """

        def __init__(self, value):
            self.value = value

        def __str__(self):
            return repr(self.value)

An input adapter that allows a ChatterBot instance to get input statements from a Microsoft Bot using Directline client protocolhttps://docs.botframework.com/en-us/restapi/directline/#navtitle

This is an input adapter that allows a ChatterBot instance to communicate through Microsoft using direct line client protocol.

Be sure to also see the documentation for the Microsoft output adapter.

chatbot = ChatBot(
    "My ChatterBot",
    input_adapter="chatterbot.input.Microsoft",
    directline_host="https://directline.botframework.com",
    directline_conversation_id="IEyJvnDULgn",
    direct_line_token_or_secret="RCurR_XV9ZA.cwA.BKA.iaJrC8xpy8qbOF5xnR2vtCX7CZj0LdjAPGfiCpg4Fv0",
)

0 Comment

temp