【サーバレスアプリ】LINEでAWSとチャットしてみよう

import os

import boto3
from linebot import LineBotApi, WebhookHandler
from linebot.models import MessageEvent, TextMessage, TextSendMessage, ImageMessage


# LINEからメッセージを受信したAPI Gatewayが
# どのLambda関数を呼び出すかを管理するためのWebhookハンドラを作成する。
handler = WebhookHandler(os.getenv('LINE_CHANNEL_SECRET'))

# Lambda関数からLINEへ処理結果を送信するAPIを作成する。
line_bot_api = LineBotApi(os.getenv('LINE_CHANNEL_ACCESS_TOKEN'))


def image_feeling(result):
    """ 画像感情分析結果の中で最も確信度が高い感情に対応したメッセージを返す """

    message = ""

    if len(result["FaceDetails"]) <= 0:
        message = "表情が読み取れませんでした・・・"
    else:
        # ※画像の人物は1人に限定する。
        emotions = result["FaceDetails"][0]["Emotions"]

        max_conf = 0
        type = ""
        for e in emotions:
            if max_conf < e["Confidence"]:
                max_conf = e["Confidence"]
                type = e["Type"]

        if type == "CALM":
            message = "穏やかな表情ですね"
        elif type == "HAPPY":
            message = "楽しそうですね"
        elif type == "SURPRISED":
            message = "驚いた様子ですね"
        elif type == "SAD":
            message = "なんだか悲しそうですね"
        elif type == "CONFUSED":
            message = "困惑した様子ですね"
        elif type == "ANGRY":
            message = "怒っていますね"
        elif type == "DISGUSTED":
            message = "気に障ったことでもあったでしょうか"
        elif type == "FEAR":
            message = "おびえなくてもいいですよ"
        else:
            message = "良い表情ですね"

    return message


# AWS API Gatewayが呼び出すLambdaハンドラ関数
def lambda_handler(event, context):
    headers = event["headers"]
    body = event["body"]

    signature = headers['x-line-signature']
    handler.handle(body, signature)

    return {"statusCode": 200, "body": "OK"}


# WebhookハンドラへLambda関数を登録する。
# LINEからテキストメッセージを受信したときに呼ばれるLambda関数を登録する。
@handler.add(MessageEvent, message=TextMessage)
def handle_text_message(event):

    input_text = event.message.text

    # テキスト感情分析のためにComprehendを使用する。
    client = boto3.client('comprehend', 'ap-northeast-1')

    # Comprehendで感情分析を行う。
    result = client.detect_sentiment(Text=input_text, LanguageCode='ja')

    # 最もスコアの高い感情を選択する。
    max_score = 0
    max_sentiment = 'Neutral'
    for senti, sco in result['SentimentScore'].items():
        if max_score <= sco:
            max_score = sco
            max_sentiment = senti

    # 返信メッセージを決める。
    message_reply = ''
    if max_sentiment == 'Positive':
        message_reply = 'ハッピーですね。その調子でいきましょう！'
    elif max_sentiment == 'Negative':
        message_reply = '大変ですね。お気を確かに！'
    elif max_sentiment == 'Neutral':
        message_reply = 'こんにちは。いつもどおりで何よりです。'
    elif max_sentiment == 'Mixed':
        message_reply = 'うーん、なんと言ったらいいのでしょう。'
    else:
        message_reply = input_text

    # LINEへメッセージを返信する。
    line_bot_api.reply_message(
        event.reply_token,
        TextSendMessage(text=message_reply))


# WebhookハンドラへLambda関数を登録する。
# LINEから画像を受信したときに呼ばれるLambda関数を登録する。
@handler.add(MessageEvent, message=ImageMessage)
def handle_image_message(event):

    # LINEユーザから送られてきた画像コンテンツを取得する。
    message_content = line_bot_api.get_message_content(event.message.id)

    # 画像コンテンツを、AWSのTemporalディレクトリへ一時的に置く。
    # 【改善課題】S3バケットを活用すべき。
    file_path = '/tmp/sent-image.jpg'
    with open(file_path, 'wb') as fd:
        for chunk in message_content.iter_content():
            fd.write(chunk)

    # 画像感情分析のため、Rekognitionを使用する。
    client = boto3.client('rekognition', 'ap-northeast-1')

    # Rekognitionで画像の感情分析をする。
    with open(file_path, 'rb') as fd:
        sent_image_binary = fd.read()
        response = client.detect_faces(Image={"Bytes": sent_image_binary}, Attributes=["ALL"])

    # 分析した感情をもとに返信メッセージを決める
    message_reply = image_feeling(response)

    # LINEへメッセージを返信する。
    line_bot_api.reply_message(
        event.reply_token,
        TextSendMessage(text=message_reply))

    # Temporalディレクトリに置いた画像を削除しておく。
    os.remove(file_path)

{
  "AWSTemplateFormatVersion": "2010-09-09",
  "Transform": "AWS::Serverless-2016-10-31",
  "Parameters": {
    "LineChannelAccessToken": {"Type": "String", "Description": "LINE のアクセストークン"},
    "LineChannelSecret": {"Type": "String", "Description": "LINE のチャンネルシークレット"}
  },
  "Resources": {
    "EndPointFunction": {
      "Type": "AWS::Serverless::Function",
      "Properties": {
        "Runtime": "python3.8",
        "CodeUri": "src",
        "Handler": "bot-crowd.lambda_handler",
        "Environment": {"Variables": {
            "LINE_CHANNEL_ACCESS_TOKEN": {"Ref": "LineChannelAccessToken"},
            "LINE_CHANNEL_SECRET": {"Ref": "LineChannelSecret"}
        }},
        "Policies":[{"RekognitionDetectOnlyPolicy": {}},
                    {"ComprehendBasicAccessPolicy": {}}],
        "Events": {
          "API": {
            "Type": "Api",
            "Properties": {"Path": "/api_endpoint", "Method": "post"}
          }
        }
      }
    }
  },
  "Outputs": {
    "ApiEndpointURL": {
      "Description": "API Endpoint URL",
      "Value": {"Fn::Sub": "https://${ServerlessRestApi}.execute-api.${AWS::Region}.amazonaws.com/${ServerlessRestApi.Stage}/api_endpoint"}
    }
  }
}

sam build -t template-bot-crowd.json

sam deploy --guided