はじめに
皆さんはLINEをメモ帳代わりに使ったことはありませんか？
私は自分しかいないチャネルを一つだけ作って、それをメモ帳代わりにしています。
そのメモ帳のチャネルに、検索機能をつけてみようと思いました。
今回はAWSとLINEのAPIを使って、その機能を実装してみようと思います
構成
構成としてはAWSのDynamoDBにデータを蓄積して、メモデータを蓄積します。その蓄積したデータを取り出す際に、
AWSのAPIゲートウェイに対して、リクエストを送信して、検索キーワードに該当するMessage をすべて抜き出して表示する
という機能を実装します。
完成系の動作
何もしなければ、LINEに投降したコメントはメッセージとしてDBに保存されて蓄積されます
検索 と打鍵してから キーワード （下記の例では「テスト」）と打鍵するとDynamoDBから「テスト」というキーワードを含む
データを取り出してきてその日付とタイムスタンプとともに表示します。
全体の流れ
１ LINE APIの設定準備
２ DynamoDBの準備
３ lambdaのプログラムを書く
４ API gatewayの設定を完了
５ テスト
上記の順番で行います
LINEチャネルの準備
https://developers.line.biz/ から登録します。右上の「コンソールにログイン」からログインします。ここではLINE Business IDの登録の行い方は
割愛させていただきます
「LINEアカウントでログイン」でログイン
- LINEチャネルの設定:
- LINE Developersコンソールでチャネルを作成し、チャネルアクセストークンを取得します。ここでは作成の方法は省略します。
今回は 「AWSLine」というチャネルを作成します。チャネルの基本設定から チャネルIDを取得できます
「Messaging API設定」から「チャネルのアクセストークン」をコピペして保存しておきます
チャネルの基本設定からシークレットを取得できますので、コピペして保存しておいてください。後ほどLambdaの環境変数として利用します
AWSの設定
Lambdaの設定
Lambdaコンソールで新しいLambda関数を作成します。
「一から作成」をクリックして、ランタイムはPython3.9 もしくは3.11
を選択ロール名は作成しておいたものを利用します
環境変数に以下を設定します これは先ほどLINE のチャンネルで作成してコピペしたものを使います
- IS_TEST_ENV false に設定
- LINE_CHANNEL_SECRET: LINEのチャネルシークレット
以下のPythonコードをLambdaに設定します。ここでは説明は省略いたしますが、コメントを見れば大体わかると思います
import json
import boto3
from datetime import datetime
import os
import traceback
import re
from boto3.dynamodb.conditions import Key, Attr
from linebot import LineBotApi
from linebot.models import TextSendMessage
from linebot.exceptions import LineBotApiError
# LINEチャネルアクセストークンを環境変数から取得し、LINE Bot APIを初期化します
LINE_CHANNEL_ACCESS_TOKEN = os.environ.get('LINE_CHANNEL_ACCESS_TOKEN')
LINE_BOT_API = LineBotApi(LINE_CHANNEL_ACCESS_TOKEN)
# テスト環境の判定フラグを環境変数から取得（デフォルトはFalse）
IS_TEST_ENV = os.environ.get('IS_TEST_ENV', 'false').lower() == 'true'
# DynamoDBリソースを初期化し、テーブルインスタンスを取得します
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('LINE_BOT2')
def lambda_handler(event, context):
"""
Lambda関数のエントリーポイント。LINEからのイベントを受け取り、
必要な処理を行います。
"""
print(f"Received event: {json.dumps(event, indent=2)}")
try:
# イベントのbodyキーをチェック
if 'body' not in event:
raise KeyError("'body' key is missing in the event")
# JSON形式のボディをパース
try:
body = json.loads(event['body'])
except json.JSONDecodeError as e:
print(f"JSON decode error: {str(e)}")
raise ValueError(f"Invalid JSON in event body: {str(e)}")
print(f"Parsed body: {json.dumps(body, indent=2)}")
# イベントに「events」キーが存在するかを確認
if 'events' not in body:
raise KeyError("'events' key is missing in the request body")
# 各LINEイベントを処理
for line_event in body['events']:
handle_event(line_event)
return {
'statusCode': 200,
'body': json.dumps('OK')
}
except KeyError as e:
print(f"KeyError: {str(e)}")
print(traceback.format_exc())
return error_response(400, f"Bad Request: {str(e)}")
except ValueError as e:
print(f"ValueError: {str(e)}")
print(traceback.format_exc())
return error_response(400, f"Bad Request: {str(e)}")
except Exception as e:
print(f"Unexpected error: {str(e)}")
print(traceback.format_exc())
return error_response(500, "Internal Server Error")
def handle_event(event):
"""
個々のLINEイベントを処理します。メッセージイベントの場合、特定のキーワードに応じて
DynamoDBに保存したり、キーワード検索を行います。
"""
print(f"Processing event: {json.dumps(event, indent=2)}")
try:
# イベントタイプがメッセージかつメッセージタイプがテキストであることを確認
if event['type'] == 'message' and event['message']['type'] == 'text':
user_id = event['source']['userId']
message_text = event['message']['text']
# 「検索 keyword」形式のメッセージをチェックし、検索処理を行います
search_match = re.match(r'検索\s+(.+)', message_text)
if search_match:
# キーワード部分を抽出して検索関数に渡す
keyword = search_match.group(1)
search_messages(event.get('replyToken'), user_id, keyword)
else:
# キーワード検索以外の場合はメッセージをDynamoDBに保存
save_to_dynamodb(user_id, message_text)
send_response(event.get('replyToken'), "メッセージを保存しました。")
except KeyError as e:
print(f"KeyError in handle_event: {str(e)}")
print(traceback.format_exc())
except Exception as e:
print(f"Unexpected error in handle_event: {str(e)}")
print(traceback.format_exc())
def save_to_dynamodb(user_id, message_text):
"""
DynamoDBにメッセージを保存します。user_idやメッセージ、日付などを記録します。
Parameters:
user_id (str): メッセージを送信したユーザーのID
message_text (str): ユーザーが送信したメッセージの内容
"""
try:
# 現在の時刻を取得し、タイムスタンプとフォーマットされた日付・時間を生成
current_time = datetime.now()
timestamp = int(current_time.timestamp() * 1000) # UNIXタイムスタンプをミリ秒に変換
# 保存するアイテムのデータを定義
item = {
'userId': user_id,
'timestamp': timestamp,
'message': message_text,
'date': current_time.strftime('%Y-%m-%d'),
'time': current_time.strftime('%H:%M:%S')
}
# DynamoDBにデータを追加
response = table.put_item(Item=item)
print(f"Message saved: {message_text}")
print(f"DynamoDB response: {json.dumps(response, indent=2)}")
except Exception as e:
print(f"Error saving to DynamoDB: {str(e)}")
print(traceback.format_exc())
def search_messages(reply_token, user_id, keyword):
"""
DynamoDBから指定されたキーワードを含むメッセージを検索し、結果を返信します。
結果の形式は「日付 時間: メッセージ内容」となります。
Parameters:
reply_token (str): LINEでユーザーに返信するためのトークン
user_id (str): メッセージを送信したユーザーのID
keyword (str): 検索するキーワード
"""
try:
# キーワードを含むメッセージを検索
response = table.scan(
FilterExpression=Attr('userId').eq(user_id) & Attr('message').contains(keyword)
)
# 検索結果をまとめて返信します
messages = response['Items']
if messages:
# 日付、時間、メッセージをフォーマット
result_text = f"「{keyword}」を含むメッセージ:\n" + "\n".join(
[f"{m['date']} {m['time']}: {m['message']}" for m in messages]
)
else:
result_text = f"「{keyword}」を含むメッセージはありません。"
send_response(reply_token, result_text)
except Exception as e:
print(f"Error searching messages: {str(e)}")
print(traceback.format_exc())
def send_response(reply_token, message):
"""
LINEメッセージをユーザーに返信します。テスト環境ではコンソールに出力し、本番環境では
実際にLINEに送信します。
Parameters:
reply_token (str): LINEで返信を行うためのトークン
message (str): 返信メッセージの内容
"""
if IS_TEST_ENV:
# テスト環境の場合はコンソールに出力
print(f"テスト環境: 以下のメッセージを送信します\n{message}")
else:
if reply_token:
try:
# LINE Bot APIを使用してメッセージを返信
LINE_BOT_API.reply_message(
reply_token,
TextSendMessage(text=message)
)
except LineBotApiError as e:
print(f"Error sending message to LINE: {str(e)}")
else:
print(f"実行環境: replyTokenがないため、メッセージを送信できません\n{message}")
def error_response(status_code, message):
"""
エラーが発生した場合のレスポンスを生成します。
Parameters:
status_code (int): HTTPステータスコード
message (str): エラーメッセージの内容
"""
return {
'statusCode': status_code,
'body': json.dumps({'error': message})
}レイヤーの作成
あとrequestモジュールがないのでレイヤーを作成します。この設定をしておかないと、importしたモジュールを利用することができません。
レイヤーは
https://api.klayers.cloud//api/v2/p3.8/layers/latest/ap-northeast-1/html
に
arn:aws:lambda:ap-northeast-1:770693421928:layer:Klayers-p38-requests:18
が用意されているのでこれを使わせてもらうことにします
レイヤーを追加しましょう。場所がわかりにくいのですが、画面左側から、「レイヤー」を選択
レイヤーを作成をクリックします
ランタイムでハンドラー名も下記のようになっているか確認
コードをデプロイします。Deployボタンを押してください。
DynamoDBの設定
DynamoDBコンソールに移動し、新しいテーブルを作成します。テーブル名はLINE_BOT2とします
「テーブルの作成」をクリックします
テーブル名、パーティションキー、ソートキーを下記のように入力します
上記の手順で一応テーブルは作成できるので、これに対して「項目を作成」で他のカラムも作成します
下記のようにすでにUserIdとtimestamp は作成されているので、
・date 日付
・message LINEから投稿したメッセージ
・time 時刻
「新しい項目の追加」をクリックして message 、date, time をすべて文字列として追加します
この状態で 「項目を作成」ボタンをクリックすれば
カラムの作成は完了します。これでDBの作成は完了です
API Gatewayの設定
API Gatewayコンソールで新しいREST APIを作成します。REST APIを選択します
新しいAPIを選択して、API名を指定します。APIのエンドポイントタイプはリージョンにします
「リソース」を追加し、「メソッドの作成」ボタンを押して、
「メソッドタイプ」からPOSTを選択します。当然Lambdaを使うので「統合タイプ」には 「Lambda関数」を選択します
「Lambdaプロキシ統合」を必ずチェック
CORS(クロスオリジンリソース共有）チェックしてください
リソース名はここでは Messges として、「リソースを作成」を押します
下記のように、リソースは作成されます。この時点で「APIをデプロイ」ボタンを押していったんデプロイしてみます
「デプロイ」を行い、エンドポイントURLを取得してLINEのWebhook URLに設定します。
デプロイ出来たら、「URLを呼び出す」の欄で、LINEから呼び出すためのURLが表示されますので、これを
コピペしてメモしておきます。・
POSTリクエストに CORSを有効にして、下記の レスポンスヘッダーを有効にしました
| Access-Control-Allow-Headers |
| Access-Control-Allow-Methods |
| Access-Control-Allow-Origin |
テスト
LINEでAPIの呼び出しテストします
これが失敗する場合は場合は lambda prxoy がONになっていない可能性あああるのでご確認ください
下記の lambda プロキシ統合 が false になっていたら失敗します
LINE側のテスト
LINE側から、チャネルに対して、テストを行います。例えば下記のようにテストを行うと
メッセージをDynamoDBに対して、保存します。
上記のエントリーはDynamoDBでも確認できます。DynamoDBから該当するテーブルを選択して
「テーブルアイテムの検索」をクリック
条件に文字列を入力すれば検索できます。これでLINEからのメッセージが無事DynamoDBまで届いて
保存されたことが確認できました。
次にLambdaからのテストを行います。このテストに関しては、下記のようなJSONを書く必要があります
lambdaから今回書いたプログラムに進んで「テスト」のタブをクリックします
下記のようにテストイベントを作成してテストを行います
問題なければ、テストが成功して、キーワードを含むエントリーが列挙されるはずです
テスト用のJSONを記載しておきますので参考にしてください 下記の「テスト」を適当な文字に書き換えれば、その文字を検索します
{
"body": "{\"events\":[{\"type\":\"message\",\"message\":{\"type\":\"text\",\"text\":\"検索 テスト\"},\"source\":{\"userId\":\"test_user_id\"},\"replyToken\":\"test_reply_token\"}]}",
"resource": "/{proxy+}",
"path": "/path/to/resource",
"httpMethod": "POST",
"isBase64Encoded": false,
"queryStringParameters": null,
"pathParameters": {
"proxy": "/path/to/resource"
},
"stageVariables": null,
"headers": {
"Content-Type": "application/json",
"X-Forwarded-Proto": "https",
"Host": "your-api-id.execute-api.us-east-1.amazonaws.com",
"User-Agent": "Custom User Agent String"
},
"requestContext": {
"accountId": "123456789012",
"resourceId": "linebot_search",
"stage": "prod",
"requestId": "test-request-id",
"identity": {
"sourceIp": "127.0.0.1",
"userAgent": "Custom User Agent"
},
"path": "/prod/path/to/resource",
"resourcePath": "/{proxy+}",
"httpMethod": "POST",
"apiId": "1234567890",
"protocol": "HTTP/1.1"
}
}LINE Developer からのAPI設定とテスト
LINE Developpers にログインして、作成したチャネルの「Messaging API設定」
から WebhookのURLを設定します。先ほど、AWSのAPIゲートウェイで設定した
URLを入力して、「Webhookの利用」をONにします
黒い「検証」ボタンを押せば 下記のように成功と出力されます
LINEから,検索してみて、Dynamo DBに格納されている値を拾えるかチェックします。
下記のように、文字列を返すようになればテスト合格です
