[AWS] J'ai créé un BOT de rappel avec LINE WORKS (implémentation)

LINEWORKS Calendrier de l'Avent Jour 14

Cette fois, nous présenterons l'implémentation du rappel BOT introduit dans LINE WORKS Advent Calendar Day 7.

[Republier] Ecran BOT et configuration générale

review.png

Le rappel BOT se compose de trois Lambdas et est implémenté en Python 3.7.

①. Traitement des messages envoyés depuis LINE WORKS et notification à SQS ②. Interroger les événements stockés dans la table et notifier SQS ③. Informez le serveur LINE WORKS du message reçu de SQS

Cette fois, je vais me concentrer sur ①.

Table de transition d'état et liste de messages

L'échange BOT avec l'utilisateur est exprimé dans la table de transition d'état. Le BOT de rappel gère les quatre événements suivants.

table.png

Il gère quatre états pour chaque événement utilisateur. Le BOT répond à l'utilisateur par un message correspondant à l'événement utilisateur et à l'état du BOT. Le contenu du message est défini comme une liste de messages.

Implémentation Lambda

Voici donc l'implémentation Lambda du sujet principal.

Le premier est le traitement global de la fonction Lambda. Appelez votre propre «fonction on_event», qui est responsable de la validation du corps de la requête et du traitement du message principal. La validation du corps de la demande est basée sur la valeur de x-works-signature dans l'en-tête.

"""
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-validation de signature
    """

    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):
    """
fonction principale
    """

    #Demander la validation du corps
    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"])

    #Traitement principal des messages
    reminderbot.on_event(body)

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

Vient ensuite la fonction on_event. Définissez 4 états, 4 événements utilisateur et une liste de messages, qui sont prédéterminés cette fois, avec des constantes.

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

#Définissez quatre états en fonction de la table de transition d'état
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"

#Défini en fonction de la liste des messages
MESSAGE_LIST = [
    "Bonjour, je rappelle bot. Appuyez sur le bouton de menu.",
    "Veuillez saisir le nom de l'événement",
    "Appuyez sur le bouton de menu.",
    "Cliquez ici pour les détails de l'événement!",
    "Veuillez saisir l'heure de l'événement.",
    "Fin de l'inscription!",
    "C'est une erreur. Veuillez le saisir à nouveau.",
]

#Définir l'événement utilisateur comme événement de publication
#Lors de l'enregistrement du menu BOT, définissez-le comme la valeur de l'événement de publication ci-dessous.
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 qui gère l'état
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table("lineworks-sample-table")


def on_event(event):
    """
Gérer tout l'événement du bot
    """

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

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

    #Vérifier l'état actuel de l'utilisateur
    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"]
    
    #Chaque événement utilisateur(postback)Traitement de branche pour chacun
    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]
    
    #Informer SQS du contenu du message
    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

Enfin, la mise en œuvre du traitement pour chaque événement. Dans chaque événement, le traitement de branchement pour chaque état est mis en œuvre sur la base de la table de transition d'état. Le traitement en double est résumé.

def on_join(account_id, status):
    """
Gestion des événements lors de l'ajout d'un bot
    """

    #Traitement des succursales selon le statut
    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):
    """
Gestion des événements lors de la saisie de texte
    """

    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" #Le statut est un mot réservé#Remplacer par st
            },
            ExpressionAttributeValues={
                ":s": STATUS_WATING_FOR_TIME_INPUT,
                ":t": text,
            },
        )
        return MESSAGE_LIST[4]

    elif status == STATUS_WATING_FOR_TIME_INPUT:

        # dateutil.Convertir les dates avec un analyseur
        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 ->Prendre une différence de 9 heures pour la conversion de l'heure en japonais
                # utc ->Plan original+Définir pour supprimer après 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):
    """
Traitement des événements lorsque le bouton "Enregistrement des événements" est enfoncé
    """

    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):
    """
Traitement des événements lorsque le bouton "Parcourir l'événement" est enfoncé
    """

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

    #processus d'acquisition d'événements
    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 Aucun"
        message += "\n-----"

    for item in items:

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

Résumé

Quel type de traitement doit être implémenté à chaque événement en créant une table de transition d'état Comme il est devenu clair quel message doit être renvoyé, j'ai pu le mettre en œuvre sans hésitation.

Cette fois, c'était une application simple, donc le nombre d'états et d'événements est faible, Je pense que la table de transition d'état sera plus utile si vous essayez de faire en sorte que BOT effectue un traitement plus compliqué.

Recommended Posts

[AWS] J'ai créé un BOT de rappel avec LINE WORKS (implémentation)
[AWS] J'ai créé un BOT de rappel avec LINE WORKS
J'ai fait un robot de remplacement de tampon avec une ligne
J'ai créé un Bot LINE avec Serverless Framework!
J'ai essayé de faire LINE BOT avec Python et Heroku
Créer un bot LINE WORKS avec Amazon Lex
Made Mattermost Bot avec Python (+ Flask)
J'ai fait un robot discord
J'ai fait un Twitter BOT avec GAE (python) (avec une référence)
J'ai fait un wikipedia gacha bot
J'ai fait une loterie avec Python.
Jusqu'à ce que Django retourne quelque chose avec un robot de ligne!
J'ai fait une minuterie pomodoro dure qui fonctionne avec CUI
J'ai créé un démon avec Python
J'ai créé un bot Twitter avec Go x Qiita API x Lambda
J'ai fait un compteur de caractères avec Python
J'ai fait une carte hexadécimale avec Python
J'ai fait un jeu de vie avec Numpy
J'ai fait un générateur Hanko avec GAN
J'ai fait un jeu rogue-like avec Python
J'ai créé un fichier de configuration avec Python
J'ai fait une application WEB avec Django
J'ai essayé de créer un LINE BOT "Sakurai-san" avec API Gateway + Lambda
J'ai écrit un bot Slack qui notifie les informations de retard avec AWS Lambda
J'ai fait une prévision météo de type bot avec Python.
J'ai créé une application graphique avec Python + PyQt5
J'ai fait mon chien "Monaka Bot" avec Line Bot
Créer un LINE BOT avec Minette pour Python
J'ai créé un robot pour publier sur Twitter en grattant sur le Web un site dynamique avec AWS Lambda (suite)
J'ai essayé de créer un bloqueur de filles pourries sur Twitter avec Python ①
[Python] J'ai créé un téléchargeur Youtube avec Tkinter.
LINE BOT avec Python + AWS Lambda + API Gateway
J'ai fait un simple portefeuille de Bitcoin avec pycoin
Bot LINE sans serveur conçu avec IBM Cloud Functions
J'ai fait un graphique de nombres aléatoires avec Numpy
J'ai fait un jeu de cueillette avec Python
J'ai créé un LINE BOT qui renvoie une image de riz terroriste en utilisant l'API Flickr
J'ai créé un Line Bot qui utilise Python pour récupérer les e-mails non lus de Gmail!
J'ai créé un robot LINE qui envoie des images recommandées tous les jours à l'heure
[Python] J'ai créé un LINE Bot qui détecte les visages et effectue le traitement de la mosaïque.
[Pour les débutants] J'ai fait un capteur humain avec Raspberry Pi et notifié LINE!
En Python, j'ai créé un LINE Bot qui envoie des informations sur le pollen à partir des informations de localisation.
J'ai créé un serveur syslog prêt à l'emploi avec Play with Docker
J'ai fait un jeu d'éclairage de sapin de Noël avec Python
J'ai créé une fenêtre pour la sortie du journal avec Tkinter
J'ai créé une application de notification de nouvelles en ligne avec Python
Créer un robot LINE de retour de perroquet avec AWS Cloud9
J'ai créé un environnement Python3 sur Ubuntu avec direnv.
[Projet spécial Valentine] J'ai fait un diagnostic de compatibilité LINE!
[Super facile] Faisons un LINE BOT avec Python.
J'ai fait un jeu mono tombé avec Sense HAT
Une histoire qui a trébuché lorsque j'ai créé un bot de chat avec Transformer
J'ai fait un jeu de frappe simple avec tkinter de Python
Créez un bot LINE avec GoogleAppEngine / py. Version nue simple
J'ai créé un package pour filtrer les séries chronologiques avec python
J'ai essayé de créer LINE-bot avec Python + Flask + ngrok + LINE Messaging API
J'ai créé une application de livre simple avec python + Flask ~ Introduction ~
J'ai créé un moniteur de ressources pour Raspberry Pi avec une feuille de calcul