LINE Messaging APIとAWS Lambdaで定期通知

今回は、LINE Messaging APIとAWSのサーバーレス構成を用いて、定期的なメッセージ通知を行ってみました。

本記事では、構成・各種設定方法、メッセージ送信テストまで紹介します。

本システムの概要

はじめに、本システムで出来ること、構成図をご紹介します。

できること

今回のシステムで、下記機能が実装可能です。

・LINEビジネスアカウントで、ユーザー個別に定期メッセージを送信できる
・隔週、週1、複数曜日を指定して送信できる

LINEビジネスアカウントでは各ユーザーに対して定期的にメッセージを送信することはできません。

定期メッセージを行うには、LINE Messaging APIを利用する必要があります。

EC2などの仮想サーバーを構築してAPIを利用すると高額になるため、今回はサーバーレス構成で構築を行いました。

構成

今回作成する構成は以下のようになります。

各サービスは以下のような役割を持っています。

Lambda ： LINE Messaging APIへAPIを送信。
EventBridge ： Lambdaの定期実行。毎日決まった時刻に起動。
DynamoDB ：ユーザー情報、送信内容の格納。Lambda起動時に参照される。
LINE公式アカウント： LINE Messaging API機能。ユーザーから見ると、本アカウントからメッセージが送信される。
User ： LINE公式アカウントにお友達登録をしているユーザー。送信対象の顧客。

料金

本システムで発生する料金は、ユーザーの人数にもよりますが、~40人程度であればほぼ無料で運用可能です。

AWS側のLambdaやDynamoDBは掛かっても数円程度で収まります。

しかし、LINEビジネスアカウントは、無料のメッセージ送信枠が200通しかできないため、月に200通以上送信する必要がある場合は有料プランに加入する必要があります。

引用元：LINE公式アカウント、料金プラン改定のお知らせ

構築

それでは、実際に構築していきましょう。

LINE側の設定

新規作成時に、「Messaging API」を選びアカウントを作成しましょう。

既に登録している場合は、設定＞Messaging APIを選択し、「Messaging APIを利用する」から始めましょう。

その後、Channel情報の欄に、「Channel ID」「Channel secret」が払い出されるので、控えておきましょう。

次に、設定>応答設定から、「Webhook」「チャット」「あいさつメッセージ」をONにしておきます。

（認証済みアカウントの場合は、チャット・あいさつメッセージは不要です。後述します。）

続いて、LINE Developersのコンソール画面へ移動します。

https://developers.line.biz

プロバイダー>Messaging API設定の最下部に「チャネルアクセストークン」があるため、長期で発行します。

このトークンも後のLambda設定で利用するためコピーして保管しておきましょう。

Lambda作成後の作業

後述するLambda構築が完了したら、設定>Messaging APIから「Webhook URL」を記載しましょう。

LambdaのHTTPSエンドポイント（関数URL）を入力します。

AWS側

LINE側の準備が完了したため、AWS側の構築を行います。

Lambda

Lambdaの環境、利用するライブラリ及び機能は以下です。

requestsライブラリを利用するため、別途ライブラリをzip化してアップロードしましょう。

ランタイム	Python 3.10
ライブラリ	requests
利用機能	関数 URL/環境変数

Lambda設定

▼▼▼関連記事▼▼▼

: 【Python】dulwichをパッケージ化してダウンロードする
はじめに本記事では、Pythonのモジュールであるdulwichをダウンロードしてローカルに持ってくる方法をご紹介します。 Python dulwichとは PythonのdulwichとはGitリ ...

Lambdaのコードは下記です。

今回は、対象の曜日前日にメッセージを送付するようなコードを作成しています。

import json
import requests
import boto3
import os
from datetime import datetime, timedelta

def lambda_handler(event, context):
    # Webhookイベント全体を解析
    if 'body' in event:
        body = json.loads(event['body'])  # Webhookのボディ部分を取得
    else:
        body = event  # テスト実行の場合はそのままeventを使用

    # テスト実行かWebhookイベントかを判断
    # Webhookイベントがある場合は、その内容に基づいて処理
    if 'events' in body and body['events']:
        event_type = body['events'][0]['type']
        
        # メッセージイベントを無視する前にuserIDをCloudWatchログに記録
        if event_type == 'message':
            user_id = body['events'][0]['source']['userId']
            print(f"User ID: {user_id} sent a message. Logging to CloudWatch.")
            
            # メッセージイベントを無視
            print(f"Ignoring message event from user: {user_id}")
            return {
                'statusCode': 200,
                'body': json.dumps('Message event ignored')
            }

    # Webhookでない、またはテスト実行時には前日通知を実行
    return send_notifications()

# 前日通知を行う関数
def send_notifications():
    # LINEのアクセストークンを環境変数から取得
    line_access_token = os.getenv('LINE_ACCESS_TOKEN')
    
    # DynamoDBのリソースを取得
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table('xxxxx')  # テーブル名を指定
    
    # DynamoDBからデータをスキャン（全取得）
    response = table.scan()
    
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {line_access_token}'
    }
    
    # 今日の曜日に1日追加して、翌日の曜日を取得
    tomorrow = (datetime.now() + timedelta(days=1)).strftime('%a')  # 例: Sun（短縮形）
    
    print(f"Tomorrow is: {tomorrow}")
    
    # 各ユーザーに対して処理
    for item in response['Items']:
        user_id = item['userID']  # ユーザーID
        day = item['Day']  # 曜日
        
        # 前日にメッセージを送信
        if day == tomorrow:
            payload = {
                'to': user_id,
                'messages': [{'type': 'text', 'text': 'message内容'}]
            }
            r = requests.post("https://api.line.me/v2/bot/message/push", headers=headers, data=json.dumps(payload))
            print(f"Sent message to {user_id}, response: {r.text}")
    
    return {
        'statusCode': 200,
        'body': json.dumps('Notifications sent successfully')
    }

EventBridge

EventBridge スケジュールを利用します。

cron式入力し、Lambdaを動かしたい時間を決めましょう。

DynamoDB

呼出し回数は少ないため、オンデマンドで作成します。

次に、ユーザーIDや通知する曜日を格納するテーブルを作成します。

Key - Value形式で下記のように設定していきます。

本文中に名前やその他項目を入れたい場合は、こちらに設定してLambdaに追記しましょう。

User-IDは次のCloudWatchのログから特定します。

CloudWatch

Lambda内にUser-IDをCloudWatchに記録するコードを入れています。

そのため、ユーザーからメッセージやスタンプ等を受けた場合、CloudWatch logの中にUser-IDの記載があるはずです。

ユーザーからのメッセージの時間とCloudWatchの時間を照合し、User-IDを特定しましょう。

※認証済みアカウントの場合は、User-IDがLINE上で特定できるため、本作業は不要です。

まとめ

Lambdaのテスト、もしくはEventBridgeの定期実行により、メッセージが送られていることを確認できます。

Tips1

LINE側からは、下記のようなJSONが飛んできています。

{
"destination": "xxxxxx",
"events": [
{
"type": "message",
"message": {
"type": "text",
"id": "xxxxxx",
"quoteToken": "xxxxxx",
"text": "メッセージが入ります。"
},
"webhookEventId": "xxxxxx",
"source": {
"type": "user",
"userId": "xxxxxx"
},
"replyToken": "xxxxxx",
"mode": "active"
}
]
}

Tips2

現状、下記バグがあります。

・画像やスタンプが送られた際に、webhookが起動する

ユーザーからテキスト以外のメッセージが来たときはwebhookが反応してしまうはずです。

現状textのみ無視するように書いているので、imageなどが来た際も無視するようなコードを書く必要があります。