lex-manager.py

#!/usr/bin/env python

##########################################################################
# Copyright 2017-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
#
# Licensed under the Amazon Software License (the "License"). You may not use this file
# except in compliance with the License. A copy of the License is located at
#
# http://aws.amazon.com/asl/
#
# or in the "license" file accompanying this file. This file is distributed on an "AS IS"
# BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, express or implied. See the
# License for the specific language governing permissions and limitations under the License.
##########################################################################
"""Lex Model Building Service helper script

Used to import/export/delete Lex bots and associated resources
(i.e. intents, slot types).

Can be run as a shell script or used as a Lambda Function for CloudFormation
Custom Resources.
"""

import logging
import json

from lexutils import LexBotImporter, LexBotExporter, LexBotDeleter

DEFAULT_LOGGING_LEVEL = logging.INFO
logging.basicConfig(
    format='[%(levelname)s] %(message)s',
    level=DEFAULT_LOGGING_LEVEL
)
logger = logging.getLogger(__name__)
logger.setLevel(DEFAULT_LOGGING_LEVEL)

BOT_DEFINITION_FILENAME = 'bot-definition.json'
BOT_EXPORT_FILENAME = 'bot-definition-export.json'

def read_bot_definition_file(file_name=BOT_DEFINITION_FILENAME):
    with open(file_name) as bot_json_file:
        bot_definition = json.load(bot_json_file)
    logger.info(
        'sucessfully read bot definition from file: {}'.format(file_name)
    )
    return bot_definition

def write_bot_definition_file(bot_definition, file_name=BOT_EXPORT_FILENAME):
    with open(file_name, 'w') as bot_json_file:
        json.dump(bot_definition, bot_json_file,
            indent=2,
            sort_keys=True,
            default=str
        )
    logger.info(
        'sucessfully wrote bot definition to file: {}'.format(file_name)
    )

def import_bot(bot_definition=None, definition_filename=BOT_DEFINITION_FILENAME):
    if (bot_definition is None):
        bot_definition = read_bot_definition_file(definition_filename)

    bot_importer = LexBotImporter(
        bot_definition=bot_definition,
        logging_level=DEFAULT_LOGGING_LEVEL,
    )
    bot_importer.import_bot()

    return bot_definition

def export_bot(bot_name=None, file_name=None):
    if (bot_name is None):
        bot_definition = read_bot_definition_file()
        bot_name = bot_definition['bot']['name']

    bot_exporter = LexBotExporter(
        bot_name=bot_name,
        logging_level=DEFAULT_LOGGING_LEVEL,
    )

    bot_definition = bot_exporter.export()
    if (file_name is None):
        print(
            json.dumps(bot_definition, indent=2, sort_keys=True, default=str)
        )
    else:
        write_bot_definition_file(bot_definition, file_name)

    return bot_definition

def delete_bot(bot_name=None):
    if (bot_name is None):
        bot_definition = read_bot_definition_file()
        bot_name = bot_definition['bot']['name']

    bot_deleter = LexBotDeleter(
        bot_name=bot_name,
        logging_level=DEFAULT_LOGGING_LEVEL,
    )
    bot_definition = bot_deleter.bot_definition
    bot_deleter.delete()

    return bot_definition

def get_parsed_args():
    """ Parse arguments passed when running as a shell script
    """
    parser = argparse.ArgumentParser(
        description='Lex bot manager. Import, export or delete a Lex bot.'
            ' Used to import/export/delete Lex bots and associated resources'
            ' (i.e. intents, slot types).'
    )
    format_group = parser.add_mutually_exclusive_group()
    format_group.add_argument('-i', '--import',
        nargs='?',
        default=argparse.SUPPRESS,
        const=BOT_DEFINITION_FILENAME,
        metavar='file',
        help='Import bot definition from file into account. Defaults to: {}'
            .format(BOT_DEFINITION_FILENAME),
    )
    format_group.add_argument('-e', '--export',
        nargs='?',
        default=argparse.SUPPRESS,
        metavar='botname',
        help='Export bot definition as JSON to stdout'
            ' Defaults to reading the botname from the definition file: {}'
            .format(BOT_DEFINITION_FILENAME),
    )
    format_group.add_argument('-d', '--delete',
        nargs=1,
        default=argparse.SUPPRESS,
        metavar='botname',
        help='Deletes the bot passed as argument and its associated resources.'
    )

    args = parser.parse_args()
    if not bool(vars(args)):
        parser.print_help()
        sys.exit(1)

    return args

def main(argv):
    """ Main function used when running as a shell script
    """
    parsed_args = get_parsed_args()

    if 'import' in parsed_args:
        try:
            # using the keyword import is problematic
            # turning to dict as workaround
            import_bot(definition_filename=vars(parsed_args)['import'])
        except Exception as e:
            error = 'failed to import bot {}'.format(e)
            logging.error(error);
            sys.exit(1)

    if 'export' in parsed_args:
        try:
            export_bot(bot_name=parsed_args.export)
        except Exception as e:
            error = 'failed to export bot {}'.format(e)
            logging.error(error);
            sys.exit(1)

    if 'delete' in parsed_args:
        try:
            delete_bot(parsed_args.delete.pop())
        except Exception as e:
            error = 'failed to delete bot {}'.format(e)
            logging.error(error);
            sys.exit(1)

def add_prefix(bot_definition, prefix='WebUi'):
    """ Adds a prefix to resource names in a bot definition

    Used to differentiate bots from the same definition file in CloudFormation
    when running multiple stacks in the same region and to make policy easier
    """
    bot = bot_definition['bot']
    bot['name'] = prefix + bot['name']
    bot['intents'] = list(map(
        lambda intent: dict(
            intentName=(prefix + intent.pop('intentName')),
            **intent
        ),
        bot_definition['bot']['intents']
    ))

    slot_types = list(map(
        lambda slot_type: dict(
            name=(prefix + slot_type.pop('name')),
            **slot_type
        ),
        bot_definition['slot_types']
    ))

    intents = map(
        lambda intent: dict(
            name=(prefix + intent.pop('name')),
            slots=list(map(
                lambda slot: dict(
                    slotType=(prefix + slot.pop('slotType')),
                    **slot
                ) if (prefix + slot['slotType']) in [s['name'] for s in slot_types]
                else slot,
                intent.pop('slots')
            )),
            **intent
        ),
        bot_definition['intents']
    )

    return dict(
        bot=bot,
        intents=list(intents),
        slot_types=slot_types
    )

def test_handler():
    import time
    handler(
        event=dict(
            RequestType='Create',
            ResourceProperties=dict(
                NamePrefix='WebUiTest',
            )
        ),
        context={},
    )
    sleep_time = 30
    logger.info('sleeping for {} seconds before deleting.'.format(sleep_time))
    time.sleep(sleep_time)
    handler(
        event=dict(
            RequestType='Delete',
            ResourceProperties=dict(
                NamePrefix='WebUiTest',
            )
        ),
        context={},
    )

def handler(event, context):
    """ CloudFormation Custom Resource Lambda Handler
    """
    import cfnresponse

    logger.info('event: {}'.format(cfnresponse.json_dump_format(event)))
    request_type = event.get('RequestType')
    resource_properties = event.get('ResourceProperties')

    response_status = cfnresponse.SUCCESS
    response = {}
    response_id = event.get('RequestId')
    reason = request_type
    error = ''

    name_prefix = resource_properties.get('NamePrefix')
    should_delete = resource_properties.get('ShouldDelete', True)

    bot_definition = read_bot_definition_file(BOT_DEFINITION_FILENAME)
    bot_definition_prefix =  add_prefix(bot_definition, name_prefix)

    if (request_type in ['Create', 'Update']):
        try:
            response_import = import_bot(bot_definition=bot_definition_prefix)
            response['BotName'] = response_import['bot']['name']
        except Exception as e:
            error = 'failed to {} bot: {}'.format(request_type, e)
            pass

    if (request_type == 'Delete' and should_delete != 'false'):
        try:
            bot_definition = read_bot_definition_file()
            bot_name = name_prefix + bot_definition['bot']['name']
            delete_bot(bot_name)
        except Exception as e:
            error = 'failed to delete bot: {}'.format(e)
            pass

    if error:
        logger.error(error)
        response_status = cfnresponse.FAILED
        reason = error

    if bool(context):
        cfnresponse.send(
            event,
            context,
            response_status,
            response,
            response_id,
            reason
        )

if __name__ == '__main__':
    #from IPython.core.debugger import Pdb ; Pdb().set_trace() # XXX
    import sys
    import argparse
    # test lambda handler from shell with -t
    if len(sys.argv) > 1 and sys.argv[1] == '-t':
        test_handler()
    # otherwise call main and parse arguments
    else:
        main(sys.argv)