この記事について
本記事は、
2025 Japan AWS Jr. Champion Qiitaリレー夏
4日目の記事となります。
はじめに
「AWS上でLINEメッセージを拾って格納する仕組み」を作ってみよう、と思い立ったのでやってみました。
構成図
できるだけシンプル、かつLINE-AWSで完結するようにしました。
LINE Messaging APIについて
LINE Messaging APIは、LINEプラットフォーム上でのボットやサービスアカウントの開発を可能にするインターフェースです。
料金プランについてはこちらに載っている中から、月200メッセージの利用できる無料プランを使用しました。
利用に際しては、公式ドキュメントを参考に以下の手順で実施しました。
- LINE Developersアカウント及びLINE Developers Consoleのアカウントを作成
- APIを利用するプロバイダーを登録
- Messaging APIチャネルを作成し、アクセストークンとチャネルシークレットを取得
それぞれ適当に命名。
なお、2024年9月にコンソールから直接チャネルを作成できなくなったようなので注意です。
AWS側のリソースについて
CloudFormationで作成しました。
yaml
AWSTemplateFormatVersion: '2010-09-09'
Description: 'LINE Messaging API Integration with Lambda, DynamoDB and Secrets Manager'
Parameters:
LineChannelSecret:
Type: String
Description: LINE Channel Secret for request validation
NoEcho: true
LineChannelAccessToken:
Type: String
Description: LINE Channel Access Token for sending messages
NoEcho: true
ResourcePrefix:
Type: String
Description: Prefix for resource names to avoid conflicts
Default: line
Resources:
# Secrets Manager - LINE API Secrets
LineApiSecrets:
Type: AWS::SecretsManager::Secret
Properties:
Name: !Sub ${ResourcePrefix}-api-secrets
Description: Secrets for LINE Messaging API
SecretString: !Sub '{"channelSecret":"${LineChannelSecret}","channelAccessToken":"${LineChannelAccessToken}"}'
Tags:
- Key: Project
Value: LineMessagingIntegration
# DynamoDB Table
LineMessagesTable:
Type: AWS::DynamoDB::Table
Properties:
TableName: !Sub ${ResourcePrefix}-messages
BillingMode: PAY_PER_REQUEST
AttributeDefinitions:
- AttributeName: messageId
AttributeType: S
- AttributeName: timestamp
AttributeType: S
KeySchema:
- AttributeName: messageId
KeyType: HASH
- AttributeName: timestamp
KeyType: RANGE
Tags:
- Key: Project
Value: LineMessagingIntegration
# Lambda Function
LineWebhookFunction:
Type: AWS::Lambda::Function
Properties:
FunctionName: !Sub ${ResourcePrefix}-webhook-processor
Runtime: python3.11
Handler: index.lambda_handler
Role: !GetAtt LambdaExecutionRole.Arn
Timeout: 30
MemorySize: 128
Environment:
Variables:
TABLE_NAME: !Ref LineMessagesTable
SECRET_ARN: !Ref LineApiSecrets
Code:
ZipFile: |
import json
import boto3
import os
import time
import uuid
import base64
import hmac
import hashlib
from datetime import datetime
# クライアントの初期化
dynamodb = boto3.resource('dynamodb')
secretsmanager = boto3.client('secretsmanager')
table = dynamodb.Table(os.environ['TABLE_NAME'])
# シークレットの取得
def get_secrets():
secret_arn = os.environ['SECRET_ARN']
response = secretsmanager.get_secret_value(SecretId=secret_arn)
secret_string = response['SecretString']
return json.loads(secret_string)
# シグネチャの検証
def verify_signature(event, channel_secret):
signature = event['headers'].get('x-line-signature')
if not signature:
return False
body = event['body']
hash = hmac.new(channel_secret.encode('utf-8'), body.encode('utf-8'), hashlib.sha256).digest()
calculated_signature = base64.b64encode(hash).decode('utf-8')
return signature == calculated_signature
def lambda_handler(event, context):
try:
# シークレットの取得
secrets = get_secrets()
channel_secret = secrets['channelSecret']
# シグネチャの検証
if not verify_signature(event, channel_secret):
return {
'statusCode': 200, # LINEプラットフォームには常に200を返す
'body': json.dumps({'message': 'Invalid signature'})
}
# リクエストボディの取得
body = json.loads(event['body'])
# LINE Messaging APIからのイベントを処理
if 'events' in body:
for line_event in body['events']:
# イベントタイプの確認（メッセージ、フォロー、ブロックなど）
event_type = line_event.get('type')
# メッセージイベントの場合
if event_type == 'message':
message = line_event.get('message', {})
user_id = line_event.get('source', {}).get('userId')
# DynamoDBに保存するアイテムを作成
item = {
'messageId': message.get('id', str(uuid.uuid4())),
'timestamp': line_event.get('timestamp', str(int(time.time() * 1000))),
'userId': user_id,
'type': message.get('type'),
'replyToken': line_event.get('replyToken'),
'receivedAt': datetime.utcnow().isoformat(),
'rawEvent': json.dumps(line_event)
}
# メッセージタイプに応じて追加情報を格納
if message.get('type') == 'text':
item['text'] = message.get('text')
elif message.get('type') == 'image':
item['contentProvider'] = message.get('contentProvider')
elif message.get('type') == 'location':
item['title'] = message.get('title')
item['address'] = message.get('address')
item['latitude'] = message.get('latitude')
item['longitude'] = message.get('longitude')
# DynamoDBにアイテムを保存
table.put_item(Item=item)
# その他のイベントタイプ（フォロー、ブロックなど）
else:
# 基本情報を保存
item = {
'messageId': line_event.get('webhookEventId', str(uuid.uuid4())),
'timestamp': line_event.get('timestamp', str(int(time.time() * 1000))),
'type': event_type,
'userId': line_event.get('source', {}).get('userId'),
'receivedAt': datetime.utcnow().isoformat(),
'rawEvent': json.dumps(line_event)
}
# DynamoDBにアイテムを保存
table.put_item(Item=item)
# LINE Messaging APIへの応答（ステータスコード200を返す必要がある）
return {
'statusCode': 200,
'body': json.dumps({'message': 'Event received and processed successfully'})
}
except Exception as e:
print(f"Error processing event: {str(e)}")
# エラーが発生しても200を返す（LINEプラットフォームの要件）
return {
'statusCode': 200,
'body': json.dumps({'message': 'Event received with processing errors'})
}
# Lambda実行ロール
LambdaExecutionRole:
Type: AWS::IAM::Role
Properties:
AssumeRolePolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Principal:
Service: lambda.amazonaws.com
Action: sts:AssumeRole
ManagedPolicyArns:
- arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
Policies:
- PolicyName: DynamoDBAccess
PolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Action:
- dynamodb:PutItem
- dynamodb:GetItem
- dynamodb:UpdateItem
- dynamodb:Query
Resource: !GetAtt LineMessagesTable.Arn
- PolicyName: SecretsManagerAccess
PolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Action:
- secretsmanager:GetSecretValue
Resource: !Ref LineApiSecrets
# API Gateway
LineWebhookApi:
Type: AWS::ApiGateway::RestApi
Properties:
Name: !Sub ${ResourcePrefix}-webhook-api
Description: API for LINE Messaging webhook
EndpointConfiguration:
Types:
- REGIONAL
# APIリソース
WebhookResource:
Type: AWS::ApiGateway::Resource
Properties:
RestApiId: !Ref LineWebhookApi
ParentId: !GetAtt LineWebhookApi.RootResourceId
PathPart: webhook
# POSTメソッド
WebhookMethod:
Type: AWS::ApiGateway::Method
Properties:
RestApiId: !Ref LineWebhookApi
ResourceId: !Ref WebhookResource
HttpMethod: POST
AuthorizationType: NONE
Integration:
Type: AWS_PROXY
IntegrationHttpMethod: POST
Uri: !Sub arn:aws:apigateway:${AWS::Region}:lambda:path/2015-03-31/functions/${LineWebhookFunction.Arn}/invocations
# APIデプロイメント
ApiDeployment:
Type: AWS::ApiGateway::Deployment
DependsOn: WebhookMethod
Properties:
RestApiId: !Ref LineWebhookApi
StageName: prod
# Lambda関数の実行権限
LambdaPermission:
Type: AWS::Lambda::Permission
Properties:
Action: lambda:InvokeFunction
FunctionName: !Ref LineWebhookFunction
Principal: apigateway.amazonaws.com
SourceArn: !Sub arn:aws:execute-api:${AWS::Region}:${AWS::AccountId}:${LineWebhookApi}/*/POST/webhook
Outputs:
WebhookUrl:
Description: URL for LINE webhook
Value: !Sub https://${LineWebhookApi}.execute-api.${AWS::Region}.amazonaws.com/prod/webhook
DynamoDBTableName:
Description: DynamoDB table name
Value: !Ref LineMessagesTable
SecretArn:
Description: ARN of the Secrets Manager secret
Value: !Ref LineApiSecrets
アクセストークンとシークレットはSecret Managerに格納して参照する構成を取りました。
LINEとAWSの連携部分
両リソースが作成できたので、これを繋げます。
API Gatewayから払い出したURL(https://xxxxxxxxxx.execute-api.ap-northeast-1.amazonaws.com/prod/webhook)を、
LINE Developersコンソール>Messaging API設定>Webhook設定から登録し、「Webhookの利用」を有効にします。
動かしてみる
メッセージを送信してみる。
この時気づきましたが、帰って来るメールのカスタマイズも今後必要ですね。
DynamoDB側で、送信メッセージが格納されていることが無事確認できました！
今後の展望
連携方法は理解できたので、近々家計簿など用途を考えてみたいと思います。
構成としては格納先をAWS外で無料で使えるもの(notionなど)にしてみる、
LINE側の返答のカスタマイズ、項目やテーブルのカスタマイズなど、
やってみたいことはたくさんあるので、しばらく遊べそうです。
Views: 0
