[PYTHON] [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
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.
- Participation des utilisateurs
- Se produit lorsque l'utilisateur ajoute un BOT --Saisie de texte
- L'utilisateur peut entrer du texte arbitraire pour BOT --Appuyez sur le bouton d'entrée d'événement
- Appuyez sur "Enregistrement d'événement" affiché dans le menu du BOT --Appuyez sur le bouton de sortie d'événement --Appuyez sur "Se référer à l'événement" affiché dans le menu du BOT
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