[AWS] I made a reminder BOT with LINE WORKS (implementation)

LINEWORKS Advent Calendar Day 14

This time, I will introduce the implementation of the reminder BOT introduced in LINE WORKS Advent Calendar Day 7.

[Repost] BOT screen and overall configuration

review.png

The reminder BOT consists of three Lambdas and is implemented in Python3.7.

①. Processing of messages sent from LINE WORKS and notification to SQS ②. Polling events stored in the table and notifying SQS ③. Notify the LINE WORKS server of the message received from SQS

This time, I will focus on ①.

State transition table and message list

The BOT exchange with the user is expressed in the state transition table. The reminder BOT handles the following four events.

--User participation --Occurs when a user adds a BOT --Text input --User can enter arbitrary text for BOT --Press the event input button --Press "Event Registration" displayed in the menu in BOT --Press the event output button --Press "Event Reference" displayed in the menu in BOT

table.png

It manages four states for each user event. The BOT responds to the user with a message that corresponds to the user event and the state of the BOT. The content of the message is defined as a message list.

Lambda implementation

Now, let's implement the main subject of Lambda.

The first is the overall processing of the Lambda function. Call your own ʻon_event function, which is responsible for validating the request body and processing the main message. Request body validation is based on the value of x-works-signature` in the header.

"""
index.py
"""

import os
import json
from base64 import b64encode, b64decode
import hashlib
import hmac

import reminderbot

API_ID = os.environ.get("API_ID")


def validate(payload, signature):
    """
    x-works-Signature validation
    """

    key = API_ID.encode("utf-8")
    payload = payload.encode("utf-8")

    encoded_body = hmac.new(key, payload, hashlib.sha256).digest()
    encoded_base64_body = b64encode(encoded_body).decode()

    return encoded_base64_body == signature


def handler(event, context):
    """
main function
    """

    #Request body validation
    if not validate(event["body"], event["headers"].get("x-works-signature")):
        return {
            "statusCode": 400,
            "body": "Bad Request",
            "headers": {
                "Content-Type": "application/json"
            }
        }

    body = json.loads(event["body"])

    #Main message processing
    reminderbot.on_event(body)

    return {
        "statusCode": 200,
        "body": "OK",
        "headers": {"Content-Type": "application/json"}
    }

Next is the on_event function. Define the four states, four user events, and message list defined in advance with constants.

"""
reminderbot.py
"""

import os

import json
import datetime
import dateutil.parser
from dateutil.relativedelta import relativedelta

import boto3
from boto3.dynamodb.conditions import Key, Attr

#Define four states based on the state transition table
STATUS_NO_USER = "no_user"
STATUS_WATING_FOR_BUTTON_PUSH = "status_waiting_for_button_push"
STATUS_WATING_FOR_NAME_INPUT = "status_waiting_for_name_input"
STATUS_WATING_FOR_TIME_INPUT = "status_waiting_for_time_input"

#Defined based on message list
MESSAGE_LIST = [
    "Hello, I remind bot. Press the menu button.",
    "Please enter the event name",
    "Press the menu button.",
    "Click here for the details of the event!",
    "Please enter the event time.",
    "Completion of registration!",
    "It's an error. Please enter it again.",
]

#Define user event as postback event
#When registering the BOT menu, make it the same as the value of the following postback event.
POSTBACK_START = "start"
POSTBACK_MESSAGE = "message"
POSTBACK_PUSH_PUT_EVENT_BUTTON = "push_put_event_button"
POSTBACK_PUSH_GET_EVENT_BUTTON = "push_get_event_button"

#Table that manages status
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table("lineworks-sample-table")


def on_event(event):
    """
Handling the entire bot event
    """

    account_id = event["source"]["accountId"]
    content = event["content"]

    postback =  content.get("postback") or "message"

    #Check the current status of the user
    response = table.get_item(
        Key={
            "Hash": "status_" + account_id,
            "Range": "-"
        }
    )

    status = STATUS_NO_USER
    message = None
    
    if response.get("Item") is not None:
        status = response.get("Item")["Status"]
    
    #Each user event(postback)Branch processing for each
    try:
    
        if postback == POSTBACK_START:
            message = on_join(account_id, status)

        elif postback == POSTBACK_MESSAGE:
            text = content["text"]
            message = on_message(account_id, status, text)

        elif postback == POSTBACK_PUSH_PUT_EVENT_BUTTON:
            message = on_pushed_put_event_button(account_id, status)

        elif postback == POSTBACK_PUSH_GET_EVENT_BUTTON:
            message = on_pushed_get_event_button(account_id, status)

    except Exception as e:
        print(e)
        message = MESSAGE_LIST[6]
    
    #Notify SQS of message content
    sqs = boto3.resource("sqs")
    queue = sqs.get_queue_by_name(QueueName="lineworks-message-queue")
    
    queue.send_message(
        MessageBody=json.dumps(
            {
                "content": {
                    "type": "text",
                    "text": message,
                },
                "account_id": account_id,
            }
        ),
    )

    return True

Finally, the implementation of processing for each event. In each event, branch processing for each state is implemented based on the state transition table. Duplicate processing is summarized.

def on_join(account_id, status):
    """
Event handling when adding a bot
    """

    #Branch processing according to status
    if status == STATUS_NO_USER:

        table.put_item(
            Item={
                "Hash": "status_" + account_id,
                "Range": "-",
                "Status": STATUS_WATING_FOR_BUTTON_PUSH,
            }
        )
        return MESSAGE_LIST[0]
    
    else:

        table.delete_item(
            Key={
                "Hash": "status_" + account_id,
                "Range": "-"
            }
        )
        
        table.put_item(
            Item={
                "Hash": "status_" + account_id,
                "Range": "-",
                "Status": STATUS_WATING_FOR_BUTTON_PUSH,
            }
        )
        
        return MESSAGE_LIST[0]

def on_message(account_id, status, text):
    """
Handling of events when entering text
    """

    if status == STATUS_WATING_FOR_BUTTON_PUSH:

        table.put_item(
            Item={
                "Hash": "status_" + account_id,
                "Range": "-",
                "Status": STATUS_WATING_FOR_BUTTON_PUSH,
            }
        )
        return MESSAGE_LIST[2]

    elif status == STATUS_WATING_FOR_NAME_INPUT:

        table.update_item(
            Key={
                "Hash": "status_" + account_id,
                "Range": "-",
            },
            UpdateExpression="set #st = :s, Title = :t",
           	ExpressionAttributeNames = {
                "#st": "Status" #Status is a reserved word#Replace with st
            },
            ExpressionAttributeValues={
                ":s": STATUS_WATING_FOR_TIME_INPUT,
                ":t": text,
            },
        )
        return MESSAGE_LIST[4]

    elif status == STATUS_WATING_FOR_TIME_INPUT:

        # dateutil.Convert dates with parser
        time_dt = dateutil.parser.parse(text)
        time = time_dt.strftime("%Y/%m/%d %H:%M:%S")

        response = table.get_item(
            Key={
                "Hash": "status_" + account_id,
                "Range": "-",
            }
        )

        table.put_item(
            Item={
                "Hash": "event_" + account_id,
                "Range": time,
                "Title": response["Item"]["Title"],
                # utc ->Take a 9-hour difference for Japan time conversion
                # utc ->Original plan+Set to delete after 1h
                "ExpireTime": int((time_dt - relativedelta(hours=9) + relativedelta(hours=1)).timestamp()),
                "SentFlag": False
            }
        ),

        table.put_item(
            Item={
                "Hash": "status_" + account_id,
                "Range": "-",
                "Status": STATUS_WATING_FOR_BUTTON_PUSH,
            }
        )

        return MESSAGE_LIST[5]

def on_pushed_put_event_button(account_id, status):
    """
Event processing when the "Event registration" button is pressed
    """

    if status == STATUS_WATING_FOR_BUTTON_PUSH:
    
        table.put_item(
            Item={
                "Hash": "status_" + account_id,
                "Range": "-",
                "Status": STATUS_WATING_FOR_NAME_INPUT,
            }
        )
        return MESSAGE_LIST[1]
    
    elif status == STATUS_WATING_FOR_NAME_INPUT:

        return MESSAGE_LIST[1]
    
    elif status == STATUS_WATING_FOR_TIME_INPUT:

        table.put_item(
            Item={
                "Hash": "status_" + account_id,
                "Range": "-",
                "Status": STATUS_WATING_FOR_NAME_INPUT,
            }
        )
        return MESSAGE_LIST[1]

def on_pushed_get_event_button(account_id, status):
    """
Event processing when the "Event reference" button is pressed
    """

    current_jst_time = (datetime.datetime.utcnow() + relativedelta(hours=9)).strftime("%Y/%m/%d %H:%M:%S")

    #event acquisition process
    response = table.query(
        KeyConditionExpression=Key("Hash").eq("event_" + account_id) & Key("Range").gt(current_jst_time)
    )

    items = response["Items"] or []
    
    message = MESSAGE_LIST[3]

    if len(items) == 0:
        message += "\n-----"
        message += "\n None"
        message += "\n-----"

    for item in items:

        message += "\n-----"
        message += "\n title: {title}".format(title=item["Title"]) 
        message += "\n date and time: {time}".format(time=item["Range"]) 
        message += "\n-----"
    
    return message

Summary

What kind of processing should be implemented at each event by creating a state transition table? I was able to implement it without hesitation because it became clear which message should be returned.

This time, it was a simple app, so the number of states and events is small, but I think that the state transition table will be more useful if you try to make BOT perform more complicated processing.

Recommended Posts

[AWS] I made a reminder BOT with LINE WORKS (implementation)
[AWS] I made a reminder BOT with LINE WORKS
I made a stamp substitute bot with line
I made a LINE Bot with Serverless Framework!
I made a LINE BOT with Python and Heroku
I made a LINE BOT that returns parrots with Go
Make a LINE WORKS bot with Amazon Lex
I made a Mattermost bot with Python (+ Flask)
I made a discord bot
I made a Twitter BOT with GAE (python) (with a reference)
I made a wikipedia gacha bot
I made a fortune with Python.
Until I return something with a line bot in Django!
I made a rigid Pomodoro timer that works with CUI
I made a daemon with Python
I made a Twitter Bot with Go x Qiita API x Lambda
I made a character counter with Python
I made a Hex map with Python
I made a life game with Numpy
I made a stamp generator with GAN
I made a roguelike game with Python
I made a configuration file with Python
I made a WEB application with Django
I tried to make "Sakurai-san" a LINE BOT with API Gateway + Lambda
I wrote a Slack bot that notifies delay information with AWS Lambda
I made a competitive programming glossary with Python
I made a weather forecast bot-like with Python.
I made a GUI application with Python + PyQt5
I made my dog "Monaka Bot" with LineBot
Create a LINE BOT with Minette for Python
I made a bot to post on twitter by web scraping a dynamic site with AWS Lambda (continued)
I made a Twitter fujoshi blocker with Python ①
[Python] I made a Youtube Downloader with Tkinter.
LINE BOT with Python + AWS Lambda + API Gateway
I made a simple Bitcoin wallet with pycoin
Serverless LINE bot made with IBM Cloud Functions
I made a random number graph with Numpy
I made a bin picking game with Python
I made a LINE BOT that returns a terrorist image using the Flickr API
I made a Line Bot that uses Python to retrieve unread Gmail emails!
I made a LINE Bot that sends recommended images every day on time
[Python] I made a LINE Bot that detects faces and performs mosaic processing.
[For beginners] I made a motion sensor with Raspberry Pi and notified LINE!
In Python, I made a LINE Bot that sends pollen information from location information.
I made a ready-to-use syslog server with Play with Docker
I made a Christmas tree lighting game with Python
I made a vim learning game "PacVim" with Go
I made a window for Log output with Tkinter
I made a net news notification app with Python
Make a parrot return LINE Bot on AWS Cloud9
I made a Python3 environment on Ubuntu with direnv.
[Valentine special project] I made a LINE compatibility diagnosis!
[Super easy] Let's make a LINE BOT with Python.
I made a falling block game with Sense HAT
A story that stumbled when I made a chatbot with Transformer
I made a simple typing game with tkinter in Python
Make a LINE bot with GoogleAppEngine / py. Simple naked version
I made a package to filter time series with python
I made LINE-bot with Python + Flask + ngrok + LINE Messaging API
I made a simple book application with python + Flask ~ Introduction ~
I made a resource monitor for Raspberry Pi with a spreadsheet