LoginSignup
3
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

100万記事ありがとう！2025年もたくさんアウトプットしよう
@curlneko(ユウ)in株式会社BeeX

Lambdaを活用した試験管理Lineボットを作ってみました

Posted at

#やりたいこと
LineボットをAWSサービスと連携させ、直近の試験日程を登録し、試験の一覧を表示する機能を作りたいと考えています。

image.png

イメージとしては以下↓
image.png

①Lineボット作成＋Messaging APIの有効化

Line関連の前準備は少し複雑なので、別途ブログとしてまとめました～
以下ご覧ください～

②Lambdaの作成：ただのオウム返しボット

最初は、LambdaとLineボットのやり取りの流れを学ぶため、複雑な返答分岐を作らずにただのオウム返しを試してみたいと思います。

Lambdaを開き、「関数の作成」
image.png

「一から作成」を選び、関数名入力し、Python3.13で「作成」
image.png

作成できたら、以下のソースをデプロイ
image.png

import os
from linebot import LineBotApi, WebhookParser
from linebot.models import MessageEvent, TextMessage, TextSendMessage

# 環境変数から設定を取得
LINE_CHANNEL_SECRET = os.getenv('LINE_CHANNEL_SECRET')
LINE_CHANNEL_ACCESS_TOKEN = os.getenv('LINE_CHANNEL_ACCESS_TOKEN')

line_bot_api = LineBotApi(LINE_CHANNEL_ACCESS_TOKEN)
parser = WebhookParser(LINE_CHANNEL_SECRET)

def lambda_handler(event, context):
    try:
        print(event)
        print(context)

        # LINEのイベント情報を直接取得
        line_events = event.get("events", [])

        for e in line_events:
            # メッセージイベントの場合のみ処理
            if e["type"] == "message" and e["message"]["type"] == "text":
                reply_token = e["replyToken"]
                user_message = e["message"]["text"]

                # 返信メッセージ作成
                reply_message = TextSendMessage(text=f"あなたのメッセージ: {user_message}")
                line_bot_api.reply_message(reply_token, reply_message)

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

    except Exception as e:
        print(f"Error: {str(e)}")
        return {"statusCode": 500, "body": str(e)}[title](url)

③Line Developersでチャンネルアクセストークンを発行

Line Developersで作成した公式アカウントを選択し、Messaging API設定タブをクリックし、画面一番下の「チャネルアクセストークン」の「発行」を押します

image.png

④Lambdaの環境変数設定

「Lambda関数＞設定＞環境変数」で、以下の環境変数を設定します。
image.png

LINE_CHANNEL_ACCESS_TOKENは、先ほど発行したチャンネルアクセストークンで設定し、
image.png

LINE_CHANNEL_SECRETは、Line Developers＞作成した公式アカウント＞チャネル基本設定で、画面真ん中にある「チャネルシークレット」で設定します。
（公式アカウント作成時に自動で発行されるはずですが、発行されていない場合は「発行」を押して手動発行してください。）

image.png

⑤ApiGatewayの構築

ApiGatewayを開き、「APIを作成」
image.png

「RestAPI」で構築、
image.png

「新しいAPI」をクリックし、API名を入力し、APIエンドポイントを「リージョン」にし、「APIを作成」
image.png

⑥ApiGatewayのリソース+メソッドの作成

出来上がったAPIGatewayはこんな感じです。「リソースを作成」をクリック
image.png

リソース名を「line」で入力し、「リソースの作成」
（lineじゃなくても大丈夫です。）
image.png

リソースが作成されました。次に、リソースの下に「メソッドを作成」をクリック
image.png

メソッドタイプを「POST」で選択し、統合タイプを「Lambda関数」で選択し、先ほど作ったLambda関数を選択し、作成
image.png

すると、メソッドができました！
image.png

「APIをデプロイ」をクリックし、新しいステージとして、ステージ名を入れてデプロイ
image.png

⑦Line DevelopersでMessaging APIのWebhook設定

デプロイできたAPIGatewayのステージを開き、メソッドの「URLを呼び出す」のところをURLをコピー
image.png

Line Developersで作成した公式アカウントの「Messaging API」タブをクリックし、Webhook設定の「編集」をクリックし、
image.png

コピーしたAPIGatewayのURLで設定します。
image.png

URL設定が終わったら、「Webhookの利用」にチェックを入れます。
image.png

⑧Lineボットのテスト：オウム返し

携帯でLineボットを開き、適当なテストメッセージを入れてみると、
オウム返しの「あなたのメッセージ：テスト」が返されたが、その前に一つ変なメッセージも一緒に返されました。。。

image.png

⑨Lineボットのデフォルト応答メッセージの無効化

それは、Lineボットのデフォルト応答メッセージです。。。
作成時に自動で作られて、手動で無効化しない限り、すべてのユーザーメッセージに対して自動返信する仕組みとなっています。。。

Line Official Account Managerを開き、作成した公式アカウントのホーム画面＞ 左メニューの応答メッセージでクリックし、「メッセージありがとうございます！」のやつをクリック
image.png

「一律応答」で作成されていますね。。。いったん「利用停止」をクリックします。
image.png

すると、もう一度携帯のほうでテストメッセージを送ってみたら、今回は変な自動応答なく、想定したオウム返しの返信文のみが返されました！

image.png

⑩DynamoDB作成

DynamoDBを開き、「テーブルの作成」
image.png

テーブル名を入力し、パーティションキーをidにソートキーをnameに設定し、その他はデフォルトのままで、「テーブルを作成」
image.png

作成に成功しました！
image.png

⑪Lambdaの作成：試験管理アプリ

ただのオウム返しボットはpythonだけで実装できますが、より複雑な処理分岐をする場合は、「line-bot-sdk」ライブラリが必要です。

ローカル側の作業フォルダに入って、
image.png

以下のコマンドを実行してください。

python -m pip install line-bot-sdk -t .

すると、フォルダには大量なライブラリがダウンロードされます。
image.png

ソースコードを以下のように修正します。

import os
import re
import boto3
from linebot import LineBotApi, WebhookParser
from linebot.models import MessageEvent, TextMessage, TextSendMessage, TemplateSendMessage, ButtonsTemplate, PostbackAction
from datetime import datetime
import uuid

# DynamoDBクライアント
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(os.getenv('DYNAMODB_TABLE_NAME') 

# 環境変数からLINEの設定を取得
LINE_CHANNEL_SECRET = os.getenv('LINE_CHANNEL_SECRET')
LINE_CHANNEL_ACCESS_TOKEN = os.getenv('LINE_CHANNEL_ACCESS_TOKEN')

line_bot_api = LineBotApi(LINE_CHANNEL_ACCESS_TOKEN)
parser = WebhookParser(LINE_CHANNEL_SECRET)

# 正しい試験名のフォーマット（yyyy-mm-dd）
exam_name_pattern = re.compile(r"^[\w\sぁ-んァ-ン一-龥々ー]+:\d{4}-\d{2}-\d{2}$")

def lambda_handler(event, context):
    print(event)
    print(context)

    try:
        # LINEのイベント情報を取得
        line_events = event.get("events", [])

        for e in line_events:
            # メッセージイベント
            if e["type"] == "message" and e["message"]["type"] == "text":
                user_message = e["message"]["text"]
                reply_token = e["replyToken"]

                # 「試験登録」の場合
                if user_message == "試験登録":
                    print("試験登録")
                    reply_message = TextSendMessage(text="試験名:yyyy-mm-ddで登録してください。")
                    line_bot_api.reply_message(reply_token, reply_message)

                # 「試験名：yyyy-mm-dd」の形式の場合
                elif exam_name_pattern.match(user_message):
                    print("試験名：yyyy-mm-dd")
                    exam_name, exam_date = user_message.split(":")
                    try:
                        # DynamoDBに試験を登録
                        table.put_item(
                            Item={
                                'id': str(uuid.uuid4()),  # UUIDをパーティションキーとして設定
                                'name': exam_name.strip(),
                                'date': exam_date.strip(),
                                'created_at': datetime.now().isoformat()
                            }
                        )
                        reply_message = TextSendMessage(text="登録しました")
                    except Exception as e:
                        reply_message = TextSendMessage(text="登録に失敗しました。もう一度試してください。")
                    # reply_message = TextSendMessage(text="登録しました")
                    line_bot_api.reply_message(reply_token, reply_message)

                # 「試験一覧」の場合
                elif user_message == "試験一覧":
                    print("試験一覧")
                    # DynamoDBから試験一覧を取得
                    response = table.scan()
                    exams = response.get('Items', [])
                    if exams:
                        exam_list = "\n".join([f"{exam['name']} : {exam['date']}" for exam in exams])
                        reply_message = TextSendMessage(text=f"登録された試験一覧:\n{exam_list}")
                    else:
                        reply_message = TextSendMessage(text="まだ試験は登録されていません。")
                    # reply_message = TextSendMessage(text="まだ試験は登録されていません。")
                    line_bot_api.reply_message(reply_token, reply_message)

                # 上記以外のメッセージの場合
                else:
                    print("上記以外のメッセージの場合")
                    buttons_template = ButtonsTemplate(
                        title="選択してください",
                        text="試験登録または試験一覧を選んでください",
                        actions=[
                            PostbackAction(label="試験登録", data="action=register"),
                            PostbackAction(label="試験一覧", data="action=list")
                        ]
                    )
                    reply_message = TemplateSendMessage(alt_text="試験登録 or 試験一覧", template=buttons_template)
                    line_bot_api.reply_message(reply_token, reply_message)

        # Postbackイベントの処理
        for e in line_events:
            if e["type"] == "postback":
                postback_data = e["postback"]["data"]
                reply_token = e["replyToken"]

                if postback_data == "action=register":
                    reply_message = TextSendMessage(text="試験名:yyyy-mm-ddで登録してください。")
                    line_bot_api.reply_message(reply_token, reply_message)
                elif postback_data == "action=list":
                    # DynamoDBから試験一覧を取得
                    response = table.scan()
                    exams = response.get('Items', [])
                    if exams:
                        exam_list = "\n".join([f"{exam['name']} : {exam['date']}" for exam in exams])
                        reply_message = TextSendMessage(text=f"登録された試験一覧:\n{exam_list}")
                    else:
                        reply_message = TextSendMessage(text="まだ試験は登録されていません。")
                    # reply_message = TextSendMessage(text="まだ試験は登録されていません。")
                    line_bot_api.reply_message(reply_token, reply_message)

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

    except Exception as e:
        print(f"Error: {str(e)}")
        return {"statusCode": 500, "body": str(e)}

ライブラリを含めて、すべてのソースをZip化します。
image.png

Lambda関数を開き、「アップロード元」をクリックし、「.zipファイル」を選択
image.png

作ったZipファイルをアップロードし、「保存」
image.png

ソースの解説

このように、「試験登録」と「試験一覧」のボタンを押して、操作できるようにします。
（Lineの公式ドキュメントから抜粋）
image.png

また、直接「試験登録」と「試験一覧」のテキストを送信するパターンにも備えて、直接操作も対応できるようにしています。

⑫LambdaのIAMを修正

Lambda関数＞設定＞アクセス権限で、Lambda関数のデフォルトIAMロールをクリック
image.png

デフォルトのポリシーをクリック
image.png

「編集」をクリック
image.png

ビジュアルタブをクリックし、「許可をさらに追加」
image.png

DynamoDBに対して、すべての権限を許可
（今回は検証をスムーズに進めるために広い権限を付与しましたが、実務では最小権限の原則に従うべきだと考えます。）
image.png

「次へ」＞「変更を保存」
image.png

⑬Dynamoテーブル名をLambdaの環境に追加

「Lambda関数＞設定＞環境変数」で、Dynamoテーブル名を追加します。

image.png

⑭Lineボットのテスト：試験管理アプリ

もう一度携帯で公式アカウントを開き、操作してみましょう！

すると、今回は、こんな感じで操作できるようになりました！

image.png

参考サイト

3
1
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
3
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?