[PYTHON] [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
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
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