[PYTHON] [AWS] Ich habe BOT mit LINE WORKS (Implementierung) in Erinnerung gerufen.

LINEWORKS Adventskalender Tag 14

Dieses Mal werden wir die Implementierung des Erinnerungs-BOT vorstellen, der in LINE WORKS Adventskalender Tag 7 eingeführt wurde.

[Repost] BOT-Bildschirm und Gesamtkonfiguration

Der Erinnerungs-BOT besteht aus drei Lambdas und ist in Python 3.7 implementiert.

① Verarbeitung von Nachrichten, die von LINE WORKS gesendet wurden, und Benachrichtigung an SQS ② In der Tabelle gespeicherte Abfrageereignisse und Benachrichtigung von SQS ③ Benachrichtigen Sie den LINE WORKS-Server über die von SQS empfangene Nachricht

Dieses Mal werde ich mich auf ① konzentrieren.

Statusübergangstabelle und Nachrichtenliste

Der BOT-Austausch mit dem Benutzer wird in der Zustandsübergangstabelle ausgedrückt. Der Erinnerungs-BOT behandelt die folgenden vier Ereignisse.

--Benutzerbeteiligung

• Tritt auf, wenn der Benutzer einen BOT hinzufügt --Text Eingabe
• Der Benutzer kann beliebigen Text für BOT eingeben --Drücken Sie die Ereigniseingabetaste --Drücken Sie auf "Ereignisregistrierung", die im Menü in BOT angezeigt wird --Drücken Sie die Ereignisausgabetaste --Drücken Sie auf "Ereignisreferenz", die im Menü in BOT angezeigt wird

Es verwaltet vier Zustände für jedes Benutzerereignis. BOT antwortet dem Benutzer mit einer Nachricht, die dem Benutzerereignis und dem Status des BOT entspricht. Der Inhalt der Nachricht wird als Nachrichtenliste definiert.

Lambda-Implementierung

Hier ist also die Lambda-Implementierung des Hauptthemas.

Die erste ist die Gesamtverarbeitung der Lambda-Funktion. Rufen Sie Ihre eigene Funktion "on_event" auf, die für die Validierung des Anforderungshauptteils und die Verarbeitung der Hauptnachricht verantwortlich ist. Die Validierung des Anforderungskörpers basiert auf dem Wert von "x-works-Signatur" im 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-Signaturvalidierung
    """

    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):
    """
Hauptfunktion
    """

    #Körpervalidierung anfordern
    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"])

    #Hauptnachrichtenverarbeitung
    reminderbot.on_event(body)

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

Als nächstes folgt die Funktion on_event. Definieren Sie 4 Zustände, 4 Benutzerereignisse und eine Nachrichtenliste, die diesmal vorbestimmt sind, mit Konstanten.

"""
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

#Definieren Sie vier Zustände basierend auf der Zustandsübergangstabelle
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"

#Definiert basierend auf der Nachrichtenliste
MESSAGE_LIST = [
    "Hallo, ich erinnere Bot. Drücken Sie die Menütaste.",
    "Bitte geben Sie den Veranstaltungsnamen ein",
    "Drücken Sie die Menütaste.",
    "Klicken Sie hier für die Details der Veranstaltung!",
    "Bitte geben Sie die Veranstaltungszeit ein.",
    "Abschluss der Registrierung!",
    "Es ist ein Fehler. Bitte geben Sie es erneut ein.",
]

#Benutzerereignis als Postback-Ereignis definieren
#Stellen Sie bei der Registrierung des BOT-Menüs den Wert des folgenden Postback-Ereignisses ein.
POSTBACK_START = "start"
POSTBACK_MESSAGE = "message"
POSTBACK_PUSH_PUT_EVENT_BUTTON = "push_put_event_button"
POSTBACK_PUSH_GET_EVENT_BUTTON = "push_get_event_button"

#Tabelle, die den Status verwaltet
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table("lineworks-sample-table")


def on_event(event):
    """
Behandlung des gesamten Ereignisses des Bots
    """

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

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

    #Überprüfen Sie den aktuellen Status des Benutzers
    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"]
    
    #Jedes Benutzerereignis(postback)Verzweigungsverarbeitung für jeden
    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]
    
    #Benachrichtigen Sie SQS über den Nachrichteninhalt
    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

Schließlich die Implementierung der Verarbeitung für jedes Ereignis. In jedem Fall wird die Verzweigungsverarbeitung für jeden Zustand basierend auf der Zustandsübergangstabelle implementiert. Die doppelte Verarbeitung wird zusammengefasst.

def on_join(account_id, status):
    """
Ereignisbehandlung beim Hinzufügen eines Bots
    """

    #Filialverarbeitung nach 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):
    """
Behandlung von Ereignissen bei der Texteingabe
    """

    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 ist ein reserviertes Wort#Ersetzen Sie durch st
            },
            ExpressionAttributeValues={
                ":s": STATUS_WATING_FOR_TIME_INPUT,
                ":t": text,
            },
        )
        return MESSAGE_LIST[4]

    elif status == STATUS_WATING_FOR_TIME_INPUT:

        # dateutil.Konvertieren Sie Daten mit 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 ->Nehmen Sie einen Unterschied von 9 Stunden für die japanische Zeitumrechnung
                # utc ->Ursprünglicher Plan+Nach 1 Stunde löschen
                "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):
    """
Ereignisverarbeitung, wenn die Schaltfläche "Ereignisregistrierung" gedrückt wird
    """

    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):
    """
Ereignisverarbeitung, wenn die Schaltfläche "Ereignis durchsuchen" gedrückt wird
    """

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

    #Ereigniserfassungsprozess
    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 Keine"
        message += "\n-----"

    for item in items:

        message += "\n-----"
        message += "\n Titel: {title}".format(title=item["Title"]) 
        message += "\n Datum und Uhrzeit: {time}".format(time=item["Range"]) 
        message += "\n-----"
    
    return message

Zusammenfassung

Welche Art von Verarbeitung sollte bei jedem Ereignis implementiert werden, indem eine Zustandsübergangstabelle erstellt wird? Da klar wurde, welche Nachricht zurückgegeben werden sollte, konnte ich sie ohne zu zögern umsetzen.

Diesmal war es eine einfache App, daher ist die Anzahl der Zustände und Ereignisse gering. Ich denke, dass die Zustandsübergangstabelle nützlicher ist, wenn Sie versuchen, BOT zu einer komplizierteren Verarbeitung zu bewegen.