ごあいさつ
大切な人を笑顔にしたい皆さん、こんにちは。
自身の披露宴でLINEボットを活用した余興アプリを作り披露したので振り返りも兼ねて記事にしたいと思います。
プロジェクト概要
- ユーザーがLINEで画像を送信すると、ボットが自動で画像を受け取る
- AWS Rekognitionを使って画像内の人物の笑顔と感情を分析
- 分析結果に基づいてスマイルスコアを計算し、LINEでフィードバックを返す
- ユーザー情報とスコアをDynamoDBに保存
- API Gatewayのマッピングテンプレートを活かしたウェブ表示機能
使用した技術
- LINE Messaging API: ユーザーとのメッセージのやり取り
- AWS Lambda: サーバーレスでボットのロジックを実行
- Amazon S3: 送信された画像の保存
- Amazon Rekognition: 顔検出と感情分析
- Amazon DynamoDB: ユーザーデータとスコアの保存
- Python: 実装言語（boto3, LINE SDK）
システム構成
- ユーザーがLINEで画像を送信
- LINE PlatformからWebhookでAWS Lambdaが起動
- 画像をS3バケットに保存
- Rekognitionで顔と感情を分析
- 結果をDynamoDBに保存
- ユーザーにスコアをLINEで返信
実装のポイント
1. LINE Messaging APIの設定
LINE Developers Consoleでチャネルを作成し、Webhook URLにLambda関数のエンドポイントを設定しました。チャネルシークレットとアクセストークンは環境変数として管理
LINE_CHANNEL_SECRET = os.environ['LINE_CHANNEL_SECRET']
LINE_CHANNEL_ACCESS_TOKEN = os.environ['LINE_CHANNEL_ACCESS_TOKEN']
line_bot_api = LineBotApi(LINE_CHANNEL_ACCESS_TOKEN)
handler = WebhookHandler(LINE_CHANNEL_SECRET)
2. 画像の受信と保存
ユーザーから送られた画像は、一意のファイル名を生成してS3バケットに保存。
@handler.add(MessageEvent, message=ImageMessage)
def handle_image_message(event):
message_content = line_bot_api.get_message_content(event.message.id)
file_name = datetime.now().strftime("%Y%m%d%H%M%S%f") + ".jpg"
s3.put_object(
Bucket=BUCKET_NAME,
Key=file_name,
Body=message_content.content,
ContentType='image/jpeg'
)
3. Amazon Rekognitionによる顔分析
S3に保存した画像をRekognitionに送信し、顔の検出と感情分析を実行。
response = rekognition.detect_faces(
Image={
'S3Object': {
'Bucket': BUCKET_NAME,
'Name': file_name
}
},
Attributes=['ALL']
)
4. 笑顔スコアの計算ロジック
検出された顔ごとに、笑顔の信頼度と感情分析の結果を組み合わせてスコアを計算。
for face in faces:
if 'Smile' in face and face['Smile']['Value']:
score = face['Smile']['Confidence']
for emotion in face['Emotions'] :
match emotion['Type']:
case 'ANGRY' | 'DISGUSTED' | 'CONFUSED' | 'SAD' | 'FEAR':
score -= emotion['Confidence'] * 1000
case 'HAPPY' :
score += emotion['Confidence'] / 100
smile_scores.append(score)
このロジックでは：
- 笑顔の基本スコアとしてRekognitionの「Smile Confidence」を使用
- ネガティブな感情（怒り、嫌悪、混乱、悲しみ、恐れ）があると大幅に減点
- 「幸せ」な感情は少し加点
複数の顔が検出された場合は平均値を取りますが、計算がガバガバで普通にマイナススコアになるので一応気を遣って最低スコアは70点に設定しました。
5. DynamoDBへのデータ保存
ユーザー情報とスコアをDynamoDBに保存して、履歴を管理します。
table.put_item(
Item={
'id': file_name,
'user_id': event.source.user_id,
'user_name': user_name,
'score': Decimal(str(ave_smile))
}
)
6. フィードバックメッセージ
↓ 奥さんの実家で飼っている鳥
ボットのキャラクター性を出すため、結果に応じて異なるメッセージを返しています：
- 顔が検出されない場合：「ビィ!!!!!(画像に人物が含まれていなくて怒っている)」
- 笑顔が検出されない場合：「ぴぃ…(笑顔の人物が見つからず悲しんでいる)」
- 笑顔が検出された場合：「ピィー！（今回のスマイルポイントは、{スコア}点だそうです）」
LINESDKはローカルでline-bot-sdkをpipしzip化してAWSコンソールからLayerとしてアップしました。
$ mkdir python
$ pip3 install line-bot-sdk -t python
$ zip -r python.zip python
(pythonディレクトリ配下にインストールしてZIP化しないと動かないので注意)
7. ランキング機能の実装
結果をプロジェクターに投影するためスコアの高い上位画像を取得する別のLambda関数を実装しました。
# DynamoDBからスコアでソートして上位3件の画像URLを取得
data = table.scan()
items = data['Items']
sorted_items = sorted(items, key=lambda x: x['score'], reverse=True)[:3]
8. Web表示機能の実装
ランキングをWeb表示するため、API Gatewayのマッピングテンプレートを活用したシンプルな方法で実装しました。Lambda関数からのレスポンスを直接HTMLに変換する方法です。
API GatewayのVelocityテンプレートマッピング
Lambda関数のJSON出力を、API Gatewayのレスポンスマッピングテンプレートで直接HTMLに変換しています。これにより、静的ファイルのホスティングやフロントエンド開発なしで、シンプルなWeb表示を実現しました。
#set($inputRoot = $util.parseJson($input.path('$').body))
<html>
<head>
<title>Image Gallery</title>
<meta charset="UTF-8">
</head>
<body>
<h1>笑顔ランキング</h1>
#foreach($item in $inputRoot)
<div>
<h2>撮影者: $item.user_name</h2>
<img src="$item.presigned_url" alt="$item.user_name" style="height:100%"/>
<h2>Score: $item.score</h2>
</div>
#end
</body>
</html>
実装のポイント
- Velocityテンプレート言語: API GatewayのVelocityテンプレート言語を使用して、JSONからHTMLへ変換
- サーバーレスなフロントエンド: S3やCloudFrontなどのホスティングサービスを使わずにWebページを実現
- 最小限のコード: HTMLを直接生成することで、JavaScriptによるDOM操作やAPI呼び出しのコードが不要に
- メンテナンスの容易さ: フロントエンドとバックエンドが一体化しているため、この規模であれば変更管理が簡単
この構成の利点
追加のAWSリソースなしでWebページが実現できました。正直、API GatewayとLambda関数だけで完結してしまったので簡単すぎて自分でも困惑しました。
もちろん、より洗練されたUIやグラフィカルな動きが必要な場合は、安定のS3 + CloudFrontでJavaScriptを使うアプローチも検討できますが、今回のユースケースでは(式直前で画面にこだわってる時間がなかったので)このシンプルな方法で十分でした。
学んだこと
1. サーバーレスアーキテクチャの威力
サーバー構築、管理という作業から解放され、ロジックに集中できたのが大きかったです。
特に今回のような一時的なイベント用アプリケーションには最適でした。
インフラ構築の手間が省けた分、機能の実装やロジックの調整に時間を使えたのが良かったです。
2. アプリケーション構築の選択肢
今回、LINE Messaging APIとAWSサービスを組み合わせることで、スマホアプリを開発せずとも多くの人が使えるサービスを構築できました。
参加者にわざわざアプリをインストールしてもらう必要がなく、普段使っているLINEだけで完結できたのはユーザー、開発者、双方良しでした
改善点
- UI: あまりに簡素な表示になっているので、ランキングらしく王冠やメダルなどの表示を追加したい
- リアルタイムランキング表示: 定期更新などリアルタイムのランキング更新機能があっても面白そう
- 自動化デプロイ: IaCを活かした自動デプロイの設定も今後取り入れたい
さいごに
LINE Messaging APIとAWSの各種サービスを組み合わせることで、比較的簡単にチャットボットを実装できました。
特にサーバーレスアーキテクチャとフルマネージドのAIサービスの組み合わせの開発効率は凄まじく、公開1週間前から実装を開始したにもかかわらず、それなりのものが作れてしまいました。
なにより、技術習得だけでなく式に集まった大切な人たちを嫌でも笑顔にできたことが一番の成果です。
会社のイベントの余興にも使えるので選択肢のひとつにいれてみてください。
以下ソースコード全貌
改めて、これだけの記述で作れちゃうのすごい。
Lambda：画像受取からLINEのフィードバックまで
import boto3
import json
import os
from decimal import Decimal
from linebot import LineBotApi, WebhookHandler
from linebot.exceptions import InvalidSignatureError
from linebot.models import MessageEvent, ImageMessage, TextSendMessage
LINE_CHANNEL_SECRET = os.environ['LINE_CHANNEL_SECRET']
LINE_CHANNEL_ACCESS_TOKEN = os.environ['LINE_CHANNEL_ACCESS_TOKEN']
line_bot_api = LineBotApi(LINE_CHANNEL_ACCESS_TOKEN)
handler = WebhookHandler(LINE_CHANNEL_SECRET)
s3 = boto3.client('s3')
dynamodb = boto3.resource('dynamodb')
rekognition = boto3.client('rekognition')
BUCKET_NAME = os.environ['S3_BUCKET_NAME']
TABLE_NAME = os.environ['DYNAMODB_TABLE_NAME']
@handler.add(MessageEvent, message=ImageMessage)
def handle_image_message(event):
message_content = line_bot_api.get_message_content(event.message.id)
file_name = datetime.now().strftime("%Y%m%d%H%M%S%f") + ".jpg"
s3.put_object(
Bucket=BUCKET_NAME,
Key=file_name,
Body=message_content.content,
ContentType='image/jpeg'
)
# 一時返答
line_bot_api.reply_message(
event.reply_token,
TextSendMessage(text="ピッ！(画像を受け取ったようだ)")
)
user_id = event.source.user_id
profile = line_bot_api.get_profile(user_id)
user_name = profile.display_name
table = dynamodb.Table(TABLE_NAME)
response = rekognition.detect_faces(
Image={
'S3Object': {
'Bucket': BUCKET_NAME,
'Name': file_name
}
},
Attributes=['ALL']
)
# 写っている人単位
faces = response['FaceDetails']
smile_scores = []
reply_message = ""
# 顔が検出されなかった場合の処理
if not faces:
reply_message = "ビィ!!!!!(画像に人物が含まれていなくて怒っている)"
else:
# 顔ごとの情報を処理
for face in faces:
if 'Smile' in face and face['Smile']['Value']:
score = face['Smile']['Confidence']
for emotion in face['Emotions'] :
match emotion['Type']:
case 'ANGRY' | 'DISGUSTED' | 'CONFUSED' | 'SAD' | 'FEAR':
score -= emotion['Confidence'] * 1000
case 'HAPPY' :
score += emotion['Confidence'] / 100
smile_scores.append(score)
if smile_scores:
ave_smile = sum(smile_scores) / len(smile_scores)
res_smile = ave_smile if ave_smile > 70 else 70
reply_message = f"ピィー！（今回のスマイルポイントは、{res_smile:.2f}点だそうです）"
# Dynamodbに情報を保存
table.put_item(
Item={
'id': file_name,
'user_id': event.source.user_id,
'user_name': user_name,
'score': Decimal(str(ave_smile))
}
)
else:
reply_message = "ぴぃ…(笑顔の人物が見つからず悲しんでいる)"
line_bot_api.push_message(user_id, TextSendMessage(text=reply_message))
def lambda_handler(event, context):
signature = event["headers"]["x-line-signature"]
body = event["body"]
try:
handler.handle(body, signature)
except InvalidSignatureError:
return {"statusCode": 400, "body": "Invalid signature"}
return {"statusCode": 200, "body": "OK"}
Lambda：画面に出力するデータを抽出〜レスポンス
import boto3
import json
from decimal import Decimal
import os
s3 = boto3.client('s3')
dynamodb = boto3.resource('dynamodb')
BUCKET_NAME = os.environ['S3_BUCKET_NAME']
TABLE_NAME = os.environ['DYNAMODB_TABLE_NAME']
def lambda_handler(event, context):
table = dynamodb.Table(TABLE_NAME)
# DynamoDBからスコアでソートして上位3件を取得
data = table.scan()
items = data['Items']
sorted_items = sorted(items, key=lambda x: x['score'], reverse=True)[:3]
for item in sorted_items:
image_key = item['id']
item['presigned_url'] = s3.generate_presigned_url('get_object',
Params={'Bucket': BUCKET_NAME, 'Key': image_key},
ExpiresIn=3600)
def convert_decimals(obj):
if isinstance(obj, list):
return [convert_decimals(x) for x in obj]
elif isinstance(obj, dict):
return {k: convert_decimals(v) for k, v in obj.items()}
elif isinstance(obj, Decimal):
if obj % 1 == 0:
return int(obj)
else:
return float(obj)
else:
return obj
converted_items = convert_decimals(sorted_items)
json_response = json.dumps(converted_items)
return {
'statusCode': 200,
'headers': {'Content-Type': 'application/json'},
'body': json_response
}