Alexa Skill for DHL POC - Code

alexa_skill_dhl_poc_code.py

# -*- coding: utf-8 -*-

# This sample demonstrates handling intents from an Alexa skill using the Alexa Skills Kit SDK for Python.
# Please visit https://alexa.design/cookbook for additional examples on implementing slots, dialog management,
# session persistence, api calls, and more.
# This sample is built using the handler classes approach in skill builder.
import logging
import ask_sdk_core.utils as ask_utils
import os
import json
import locale
import requests
import gettext
from ask_sdk_s3.adapter import S3Adapter
s3_adapter = S3Adapter(bucket_name=os.environ["S3_PERSISTENCE_BUCKET"])

from alexa import data

from ask_sdk_core.skill_builder import CustomSkillBuilder
from ask_sdk_core.dispatch_components import (
    AbstractRequestHandler, AbstractRequestInterceptor, AbstractExceptionHandler)
from ask_sdk_core.handler_input import HandlerInput

from ask_sdk_model import Response

from ask_sdk_core.utils import is_request_type, is_intent_name, get_account_linking_access_token

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)


OL_SUBDOMAIN = "mddemo"


def sync_info(handler_input, access_token):
    email_str = ""
    userinfo_url = "https://%s.onelogin.com/oidc/2/me" % OL_SUBDOMAIN
    headers = {"Authorization": "Bearer %s" % access_token}
    r = requests.get(userinfo_url, headers=headers)
    data_sync = None
    email_str = ""
    if r.status_code == 200:

        userdata = r.json()
        if userdata['email']:
            email_str = " %s" % userdata['email']

        if userdata and 'params' in userdata:
            
            countryCode = postalCode = None
            if 'countryCode' in userdata['params'] and userdata['params']['countryCode']:
                countryCode = userdata['params']['countryCode']
            if 'postalCode' in userdata['params'] and userdata['params']['postalCode']:
                postalCode = userdata['params']['postalCode']

            if countryCode or postalCode:
                data_sync = ""
                session_attr = handler_input.attributes_manager.session_attributes
                if countryCode:
                    session_attr['countryCode'] = countryCode
                    data_sync = " Country Code: %s" % countryCode
                if postalCode:
                    session_attr['postalCode'] = postalCode
                    data_sync = "%s Postal Code: %s" % (data_sync, postalCode)

                # save session attributes as persistent attributes
                handler_input.attributes_manager.persistent_attributes = session_attr
                handler_input.attributes_manager.save_persistent_attributes()

    return (handler_input, email_str, data_sync)


class LaunchRequestHandler(AbstractRequestHandler):
    """
    Handler for Skill Launch
    """

    def can_handle(self, handler_input):
        return is_request_type("LaunchRequest")(handler_input)

    def handle(self, handler_input):
        data = handler_input.attributes_manager.request_attributes["_"]
        
        access_token = get_account_linking_access_token(handler_input)
        
        
        if access_token is None:
            speech = data["ACC_LINKING_MSG"]
        else:
            handler_input, email_str, data_sync = sync_info(handler_input, access_token)

            speech = data["WELCOME_MSG"].format(email_str)
            if data_sync:
                speech += data["SYNC_MSG"].format(data_sync)
            else:
                speech += " \r\n%s" % data["WELCOME_REPROMPT_MSG"]

        handler_input.response_builder.speak(speech).ask(speech)
        return handler_input.response_builder.response


class HasLocationtHandler(AbstractRequestHandler):
    """Handler for launch after they have set their location info"""

    def can_handle(self, handler_input):
        # extract persistent attributes and check if they are all present
        attr = handler_input.attributes_manager.persistent_attributes
        attributes_are_present = ("countryCode" in attr and "postalCode" in attr and attr['countryCode'] and attr['postalCode'])

        return attributes_are_present and (ask_utils.is_request_type("LaunchRequest")(handler_input) or is_intent_name("ShowOfficesIntent")(handler_input))

    def handle(self, handler_input):
        attr = handler_input.attributes_manager.persistent_attributes

        countryCode = attr['countryCode']
        postalCode = attr['postalCode']

        data = handler_input.attributes_manager.request_attributes["_"]

        url = "https://api.dhl.com/location-finder/v1/find-by-address?countryCode={}&postalCode={}&radius=2500&limit=5".format(countryCode, postalCode)
        headers = {'DHL-API-Key': 'demo-key', 'accept': 'application/json'}
        r = requests.get(url, headers=headers)
        
        if r.status_code == 200:
            offices = []
            data = r.json()
            if data and data['locations']:
                for location in data['locations']:
                    offices.append(location['name'])

            if not offices:
                speak_output = "No results"
            else:
                speak_output = ", ".join(offices)
        else:
            speak_output = data["API_ERROR_MSG"].format(r.status_code)

        return (
            handler_input.response_builder
                .speak(speak_output)
                .ask(speak_output)
                .response
        )

class ShowOfficesIntentHandler(AbstractRequestHandler):
    """Handler when offices is requested but not data"""

    def can_handle(self, handler_input):
        # extract persistent attributes and check if they are all present
        attr = handler_input.attributes_manager.persistent_attributes
        attributes_are_present = ("countryCode" in attr and "postalCode" in attr and attr['countryCode'] and attr['postalCode'])

        return is_intent_name("ShowOfficesIntent")(handler_input) and not attributes_are_present

    def handle(self, handler_input):
        data = handler_input.attributes_manager.request_attributes["_"]
        speak_output = data["WELCOME_REPROMPT_MSG"]

        return (
            handler_input.response_builder
                .speak(speak_output)
                .ask(speak_output)
                .response
        )


class CaptureDHLOfficeIntentHandler(AbstractRequestHandler):
    """
    Handler for Capturing the Office Info
    """

    def can_handle(self, handler_input):
        return is_intent_name("CaptureDHLOfficeIntent")(handler_input)

    def handle(self, handler_input):
        data = handler_input.attributes_manager.request_attributes["_"]
        slots = handler_input.request_envelope.request.intent.slots
        skill_locale = handler_input.request_envelope.request.locale

        # extract slot values
        countryCode = slots["countryCode"].value

        postalCode = slots["postalCode"].value

        # save slots into session attributes
        session_attr = handler_input.attributes_manager.session_attributes
        if countryCode:
            session_attr['countryCode'] = countryCode
        else:
            countryCode = ""
        if postalCode:
            session_attr['postalCode'] = postalCode
        else:
            postalCode = ""

        # save session attributes as persistent attributes
        handler_input.attributes_manager.persistent_attributes = session_attr
        handler_input.attributes_manager.save_persistent_attributes()

        speech = data["REGISTER_LOCATION_MSG"].format(postalCode, countryCode)
        handler_input.response_builder.speak(speech)
        handler_input.response_builder.set_should_end_session(True)
        return handler_input.response_builder.response


class SyncLocationIntentnHandler(AbstractRequestHandler):
    """Handler for Sync location Intent."""
    def can_handle(self, handler_input):
        return ask_utils.is_intent_name("SyncLocationIntent")(handler_input)

    def handle(self, handler_input):
        data = handler_input.attributes_manager.request_attributes["_"]
        
        access_token = get_account_linking_access_token(handler_input)
        
        if access_token is None:
            speech = reprompt = data["ACC_LINKING_MSG"]
        else:
            handler_input, email_str, data_sync = sync_info(handler_input, access_token)

            if data_sync:
                speech = data["SYNC_MSG"].format(data_sync)
            else:
                speech = data["NO_SYNC_MSG"]

        handler_input.response_builder.speak(speech).ask(speech)
        return handler_input.response_builder.response
    

class GetLocationIntentHandler(AbstractRequestHandler):
    """Handler for Get location Intent."""
    def can_handle(self, handler_input):
        return ask_utils.is_intent_name("GetLocationIntent")(handler_input)

    def handle(self, handler_input):
        data = handler_input.attributes_manager.request_attributes["_"]

        # get slots from session attributes
        attr = handler_input.attributes_manager.persistent_attributes
        
        if "countryCode" in attr and attr['countryCode']:
            countryCode = attr['countryCode']
        else:
            countryCode = ""
        if "postalCode" in attr and attr['postalCode']:
            postalCode = attr['postalCode']
        else:
            postalCode = ""

        if postalCode or countryCode:
            speak_output = data["SHOW_LOCATION_MSG"].format(postalCode, countryCode)
        else:
            speak_output = data["SHOW_EMPTY_LOCATION_MSG"]

        return (
            handler_input.response_builder
                .speak(speak_output)
                .ask(speak_output)
                .response
        )


class ResetLocationIntentHandler(AbstractRequestHandler):
    """Handler for Reset Intent."""
    def can_handle(self, handler_input):
        return ask_utils.is_intent_name("ResetLocationIntent")(handler_input)

    def handle(self, handler_input):
        data = handler_input.attributes_manager.request_attributes["_"]

        handler_input.attributes_manager.persistent_attributes = {}
        handler_input.attributes_manager.save_persistent_attributes()

        speak_output = data["RESET_MSG"]

        return (
            handler_input.response_builder
                .speak(speak_output)
                .response
        )


class HelpIntentHandler(AbstractRequestHandler):
    """Handler for Help Intent."""
    def can_handle(self, handler_input):
        # type: (HandlerInput) -> bool
        return ask_utils.is_intent_name("AMAZON.HelpIntent")(handler_input)

    def handle(self, handler_input):
        # type: (HandlerInput) -> Response

        data = handler_input.attributes_manager.request_attributes["_"]
        speak_output = data["HELP_MSG"]

        handler_input.attributes_manager.persistent_attributes = {}
        handler_input.attributes_manager.save_persistent_attributes()

        return (
            handler_input.response_builder
                .speak(speak_output)
                .ask(speak_output)
                .response
        )

class StopIntentHandler(AbstractRequestHandler):
    """Handler for Stop Intent."""
    def can_handle(self, handler_input):
        # type: (HandlerInput) -> bool
        return ask_utils.is_intent_name("AMAZON.StopIntent")(handler_input)

    def handle(self, handler_input):
        # type: (HandlerInput) -> Response
        data = handler_input.attributes_manager.request_attributes["_"]
        speak_output = data["GOODBYE_MSG"]

        return (
            handler_input.response_builder
                .speak(speak_output)
                .response
        )


class CancelIntentHandler(AbstractRequestHandler):
    """Handler for Cancel Intent."""
    def can_handle(self, handler_input):
        # type: (HandlerInput) -> bool
        return ask_utils.is_intent_name("AMAZON.CancelIntent")(handler_input)

    def handle(self, handler_input):
        # type: (HandlerInput) -> Response
        data = handler_input.attributes_manager.request_attributes["_"]
        speak_output = data["GOODBYE_MSG"]

        return (
            handler_input.response_builder
                .speak(speak_output)
                .response
        )


class SessionEndedRequestHandler(AbstractRequestHandler):
    """Handler for Session End."""
    def can_handle(self, handler_input):
        # type: (HandlerInput) -> bool
        return ask_utils.is_request_type("SessionEndedRequest")(handler_input)

    def handle(self, handler_input):
        # type: (HandlerInput) -> Response

        # Any cleanup logic goes here.

        return handler_input.response_builder.response


class IntentReflectorHandler(AbstractRequestHandler):
    """The intent reflector is used for interaction model testing and debugging.
    It will simply repeat the intent the user said. You can create custom handlers
    for your intents by defining them above, then also adding them to the request
    handler chain below.
    """
    def can_handle(self, handler_input):
        # type: (HandlerInput) -> bool
        return ask_utils.is_request_type("IntentRequest")(handler_input)

    def handle(self, handler_input):
        # type: (HandlerInput) -> Response
        data = handler_input.attributes_manager.request_attributes["_"]
        intent_name = ask_utils.get_intent_name(handler_input)
        speak_output = data["REFLECTOR_MSG"].format(intent_name)

        return (
            handler_input.response_builder
                .speak(speak_output)
                # .ask("add a reprompt if you want to keep the session open for the user to respond")
                .response
        )


class CatchAllExceptionHandler(AbstractExceptionHandler):
    """Generic error handling to capture any syntax or routing errors. If you receive an error
    stating the request handler chain is not found, you have not implemented a handler for
    the intent being invoked or included it in the skill builder below.
    """
    def can_handle(self, handler_input, exception):
        # type: (HandlerInput, Exception) -> bool
        return True

    def handle(self, handler_input, exception):
        # type: (HandlerInput, Exception) -> Response
        logger.error(exception, exc_info=True)
        data = handler_input.attributes_manager.request_attributes["_"]
        speak_output = "Error"

        return (
            handler_input.response_builder
                .speak(speak_output)
                .ask(speak_output)
                .response
        )

# The SkillBuilder object acts as the entry point for your skill, routing all request and response
# payloads to the handlers above. Make sure any new handlers or interceptors you've
# defined are included below. The order matters - they're processed top to bottom.


class LocalizationInterceptor(AbstractRequestInterceptor):
    """
    Add function to request attributes, that can load locale specific data.
    """

    def process(self, handler_input):
        skill_locale = handler_input.request_envelope.request.locale

        # localized strings stored in language_strings.json
        with open("language_strings.json") as language_prompts:
            language_data = json.load(language_prompts)
        # set default translation data to broader translation
        data = language_data[skill_locale[:2]]
        # if a more specialized translation exists, then select it instead
        # example: "fr-CA" will pick "fr" translations first, but if "fr-CA" translation exists,
        #          then pick that instead
        if skill_locale in language_data:
            data.update(language_data[skill_locale])
        handler_input.attributes_manager.request_attributes["_"] = data

        # configure the runtime to treat time according to the skill locale
        skill_locale = skill_locale.replace('-', '_')
        locale.setlocale(locale.LC_TIME, skill_locale)
        

sb = CustomSkillBuilder(persistence_adapter=s3_adapter)

sb.add_request_handler(HasLocationtHandler())
sb.add_request_handler(LaunchRequestHandler())
sb.add_request_handler(SyncLocationIntentnHandler())
sb.add_request_handler(CaptureDHLOfficeIntentHandler())
sb.add_request_handler(GetLocationIntentHandler())
sb.add_request_handler(ShowOfficesIntentHandler())
sb.add_request_handler(ResetLocationIntentHandler())
sb.add_request_handler(HelpIntentHandler())
sb.add_request_handler(CancelIntentHandler())
sb.add_request_handler(StopIntentHandler())
sb.add_request_handler(SessionEndedRequestHandler())
sb.add_request_handler(IntentReflectorHandler()) # make sure IntentReflectorHandler is last so it doesn’t override your custom intent handlers

sb.add_exception_handler(CatchAllExceptionHandler())

sb.add_global_request_interceptor(LocalizationInterceptor())


lambda_handler = sb.lambda_handler()