Amazon Lex fulfillment function - Lambda handler (Python) + utilities

lex.py

import logging
from lex_utils import elicit_slot, delegate, close, ElicitAction, DelegateAction
from utils import validate_dialog, init_or_load_session, finalize_session, actually_book_the_hotel

logger = logging.getLogger()
logger.setLevel(logging.DEBUG)

def lambda_handler(event, context):
    logger.debug('event.bot.name=%s', event['bot']['name'])
    logger.debug('userId=%s, intentName=%s', event['userId'], event['currentIntent']['name'])

    intent_name = event['currentIntent']['name']

    if intent_name == 'BookHotel':
        return book_hotel(event)
    # elif (add more intents here)
    else:
        raise Exception('Intent with name %s not supported' % intent_name)


def book_hotel(event):

    session_attributes = init_or_load_session(event)

    slots = event['currentIntent']['slots']
    name = event['currentIntent']['name']

    try:
        if event['invocationSource'] == 'DialogCodeHook':
            validate_dialog(slots, session_attributes)
    except ElicitAction as ea:
        # request a new value
        return elicit_slot(
            session_attributes,
            name,
            slots,
            ea.invalid_slot,  # elicit this invalid slot
            ea.message,       # with this custom message
        )
    except DelegateAction as da:
        # tell Lex to move on with the next slot
        return delegate(session_attributes, slots)

    # ok, we have all the slots and we can finally book the hotel!
    actually_book_the_hotel(session_attributes)
    finalize_session(session_attributes)

    return close(
        session_attributes,
        'Thanks, I have placed your hotel reservation. What shall we do next?',
    )

lex_utils.py

""" Some serialization utilities for Amazon Lex """

class ElicitAction(Exception):
    def __init__(self, invalid_slot, message):
        super(ElicitAction, self).__init__()
        self.invalid_slot = invalid_slot
        self.message = message

class DelegateAction(Exception):
    pass


def elicit_slot(session_attributes, intent_name, slots, slot_to_elicit, message):
    return {
        'sessionAttributes': session_attributes,
        'dialogAction': {
            'type': 'ElicitSlot',
            'intentName': intent_name,
            'slots': slots,
            'slotToElicit': slot_to_elicit,
            'message': message
        }
    }



def close(session_attributes, message, fulfillment_state='Fulfilled'):
    response = {
        'sessionAttributes': session_attributes,
        'dialogAction': {
            'type': 'Close',
            'fulfillmentState': fulfillment_state,
            'message': {
                'contentType': 'PlainText',
                'content': message,
            }
        }
    }

    return response


def delegate(session_attributes, slots):
    return {
        'sessionAttributes': session_attributes,
        'dialogAction': {
            'type': 'Delegate',
            'slots': slots
        }
    }


def build_validation_result(isvalid, violated_slot, message_content):
    return {
        'isValid': isvalid,
        'violatedSlot': violated_slot,
        'message': {
            'contentType': 'PlainText',
            'content': message_content,
        }
    }

utils.py

""" Some validation utilities """
import logging
import json
import datetime
import dateutil.parser

from lex_utils import build_validation_result, ElicitAction, DelegateAction

logger = logging.getLogger()
logger.setLevel(logging.DEBUG)

def generate_hotel_price(location, nights, room_type):
    """
    Generates a number within a reasonable range that might be expected for a hotel.
    The price is fixed for a pair of location and roomType.
    """

    room_types = ['queen', 'king', 'deluxe']
    cost_of_living = 0
    for i in range(len(location)):
        cost_of_living += ord(location.lower()[i]) - 97

    return nights * (100 + cost_of_living + (100 + room_types.index(room_type.lower())))


def isvalid_city(city):
    valid_cities = ['new york', 'los angeles', 'chicago', 'houston', 'philadelphia', 'phoenix', 'san antonio',
                    'san diego', 'dallas', 'san jose', 'austin', 'jacksonville', 'san francisco', 'indianapolis',
                    'columbus', 'fort worth', 'charlotte', 'detroit', 'el paso', 'seattle', 'denver', 'washington dc',
                    'memphis', 'boston', 'nashville', 'baltimore', 'portland']
    return city.lower() in valid_cities


def isvalid_room_type(room_type):
    room_types = ['queen', 'king', 'deluxe']
    return room_type.lower() in room_types


def isvalid_date(date):
    try:
        dateutil.parser.parse(date)
        return True
    except ValueError:
        return False



def validate_slots(slots):
    location = slots.get('Location')
    checkin_date = slots.get('CheckInDate')
    nights = int(slots.get('Nights') or 0)
    room_type = slots.get('RoomType')

    if location and not isvalid_city(location):
        return build_validation_result(False, 'Location', 'We currently do not support {} as a valid destination.  Can you try a different city?'.format(location))

    if checkin_date:
        if not isvalid_date(checkin_date):
            return build_validation_result(False, 'CheckInDate', 'I did not understand your check in date.  When would you like to check in?')
        if datetime.datetime.strptime(checkin_date, '%Y-%m-%d').date() <= datetime.date.today():
            return build_validation_result(False, 'CheckInDate', 'Reservations must be scheduled at least one day in advance.  Can you try a different date?')

    if not 0 < nights < 30:
        return build_validation_result(False, 'Nights', 'You can make a reservations for from one to thirty nights.  How many nights would you like to stay for?')

    if room_type and not isvalid_room_type(room_type):
        return build_validation_result(False, 'RoomType', 'I did not recognize that room type.  Would you like to stay in a queen, king, or deluxe room?')

    return {
        'isValid': True,
    }


def init_or_load_session(event):

    current = event['currentIntent']

    location = current['slots'].get('Location')
    checkin_date = current['slots'].get('CheckInDate')
    nights = int(current['slots'].get('Nights') or 0)
    room_type = current['slots'].get('RoomType')

    # serialize new session data
    reservation = {
        'ReservationType': 'Hotel',
        'Location': location,
        'RoomType': room_type,
        'CheckInDate': checkin_date,
        'Nights': nights
    }

    # fetch or initialize session
    session_attributes = event.get('sessionAttributes') or {}

    # update session data
    session_attributes['currentReservation'] = json.dumps(reservation)

    return session_attributes

def finalize_session(session_attributes):
    # clear/update session with final attributes
    reservation = session_attributes.pop('currentReservation', None)
    session_attributes['lastConfirmedReservation'] = reservation

def actually_book_the_hotel(session_attributes):
    reservation = session_attributes['lastConfirmedReservation']
    logger.debug('bookHotel under=%s', json.dumps(reservation))

    # TODO actually book the hotel :)

def validate_dialog(slots, session_attributes):
    # Validate any slots which have been specified.
    # If any are invalid, re-elicit for their value
    validation_result = validate_slots(slots)
    if not validation_result['isValid']:
        invalid_slot = validation_result['violatedSlot']
        slots[invalid_slot] = None

        raise ElicitAction(
            invalid_slot=invalid_slot,
            message=validation_result['message'],
        )

    reservation = json.loads(session_attributes['currentReservation'])

    if reservation['Location'] and reservation['Nights'] and reservation['RoomType']:
        # ok, slots are valid, we can generate the total price

        reservation['currentReservationPrice'] = generate_hotel_price(
            location=reservation['Location'],
            nights=reservation['Nights'],
            room_type=reservation['RoomType'],
        )

        # re-serialize reservation into session (with new price)
        session_attributes['currentReservation'] = json.dumps(reservation)
        raise DelegateAction()