経緯

これまで家計の立て替えを行った際には忘れないようにNotionで作成した支出管理用のテーブルに直接入力しておりました。

ただスマホから入力するにあたりNotionアプリの起動→対象のページに移動→1つ1つセルを選択して項目や金額や日付を埋める等手間が多く、同居人がめんどくさがって自身で入力してくれないという悩みがありました。

そこでその悩みを解決すべく考えたのが、LINEに項目と金額を入力して送信すれば勝手にNotionのテーブルに内容を反映してくれる仕組みになります。

作ったもの

今回作成した仕組みの全体アーキは以下のようにLINEとAWS LambdaとNotionを組み合わせたサーバレス構成になります。

作成したLINEのBotとのトーク画面で「項目金額」と入力することでNotionに登録され、以下のようなレスポンスが返ってきます。

Notion側にはこんな感じで反映され、支払日は今日の日付を自動取得、支払い者の名前はLINEから自動で取得されるようになっております。

前提

LINEをインストール&アカウント発行済み
Notionをインストール&アカウント発行済み
- 無料プランで問題なし
AWSアカウント発行済み

大まかな作業手順

手順1：LINE Bot用のMessaging APIチャネルを作成
手順2：Notionでテーブルを作成
手順3：Notionでインテグレーションを作成・テーブル(DB)に接続
手順4：AWS Lambda関数作成
手順5：AWS API gatewayの作成・lambdaと接続
手順6：AWS API gatewayとLINE Messaging APIチャネルを接続

詳細手順

手順1：LINE Bot用のMessaging APIチャネルを作成

LINE Deveopersにアクセスします。
LINE Developersアカウントを持っていない方はまず、アカウントを登録します。

アカウントが準備できたらコンソールにログインします。

コンソールのトップ > {自身の名前} > チャネル設定から新規チャネル作成を行います。
今回のケースではMessaging APIを選択します。
チャネル名など必要項目を埋めて作成します。（今回私はチャネル名を「支出管理くん」に設定。）

今回のケースと同様にAWS lambdaからLINEにメッセージを返す仕組みを作りたい場合は、コンソールのトップ > {自身の名前} > {作成したチャネル名} > Messaging API設定の下の方にあるチャネルアクセストークン（長期）を発行しメモしておきます。

手順2：Notionでテーブルを作成

自身のNotionアカウントでページを追加します。

「/テーブルビュー」と入力するとテーブルが作成されます。
テーブルのカラムのプロパティを設定します。
今回は以下のカラム名/種類で設定いたしました。

立て替え項目/タイトル
発生日/日付
支払い者/テキスト
立て替え金額/数値
精算ステータス/セレクト
- 未精算(デフォルト)、精算済み

手順3：Notionでインテグレーションを作成・テーブル(DB)に接続

インテグレーションとは、外部アプリケーションがNotionと通信するための機能です。

インテグレーションのページから新しいインテグレーションを選択。
関連ワークスペースは自身のワークスペース、種類は内部で設定し保存してください。

作成したインテグレーションの「内部インテグレーションシークレット」は手順4で使用するのでメモしておいてください。

先ほどの手順2でテーブルを作成したページの右上の「・・・」の下の方にあるコネクト > 接続先から作成したインテグレーションを選択します。

また、「・・・」の上の方にある「リンクをコピー」を押しメモ帳にペーストすると、"https://www.notion.so/xxxxxxxxxxx?yyyyyyyyyyy" のようなURLが現れます。

上記のURLのxxxxxxxxxxx部分がテーブル(DB)のIDになりますのでメモしておきます。
こちらも手順4で使用します。

手順4：AWS Lambda関数作成

AWSにアクセスしコンソールにサインインします。

Lambdaのページにアクセスし、関数の作成を行います。
関数名は自身で適当な名前をつけていただき、ランタイムは今回はPython 3.11を使用、アーキテクチャはarm64としました。

※今回はAPI gateway経由でLINEからLambdaにアクセスしますが、API gatewayを経由せずLambdaの関数URL経由でアクセスすることも可能なので、この場合は詳細設定で「関数URLを有効化」にチェックを入れます。

実装したい処理コードを記載していきます。今回は以下のファイル構成としました。

デフォルトでエントリーポイントはlambda_function.pyとなっております。
参考までに各スクリプトの処理内容を以下に記載します。

lambda_function.py

"""LINE Messaging APIからPOSTを受け取りAWS Lambdaでメッセージを整形し、notion DBに登録する."""

import json
import logging

from line import get_user_profile, handle_error, parse_user_message, reply_to_line
from notion import write_to_notion_database

logger = logging.getLogger()
logger.setLevel(logging.INFO)


def lambda_handler(event, context) -> dict:
    """AWS Lambda関数のエントリポイント.

    LINEメッセージを受け取り、指定のフォーマットで入力された支払い情報をNotionデータベースに登録し、結果をLINEに返信する.

    Args:
        event (dict): API Gatewayなどから渡されるイベント情報。LINEからのメッセージデータを含む
        context (object): ランタイム情報を提供するオブジェクト（使用されないが、Lambdaで必要）

    Returns:
        dict: Lambda関数の実行結果として、HTTPステータスコードとメッセージを含むレスポンス
    """
    logger.info(event)

    try:
        # LINEからのイベントをパース
        body = json.loads(event["body"])
        reply_token = body["events"][0]["replyToken"]
        user_id = body["events"][0]["source"]["userId"]
        user_message = body["events"][0]["message"]["text"]

        # 項目名と金額を取得
        name, amount = parse_user_message(user_message)
        logger.info(f"Parsed name: {name}, amount: {amount}")

        # ユーザーのプロフィールを取得
        user_profile = get_user_profile(user_id)
        user_name = user_profile.get("displayName", "名無し")

        # Notion DBに書き込み
        result = write_to_notion_database(item=name, amount=amount, payer=user_name)

        # 結果に応じてLINEにメッセージを返信
        message = (
            f"登録が成功しました\n・支払い者: {user_name}\n・支払い項目: {name}\n・支払い金額: {str(amount)}円"
            if result["statusCode"] == 200
            else f"登録が失敗しました\nエラーコード: {result['statusCode']}"
        )
        reply_to_line(reply_token, message)

        return {
            "statusCode": result["statusCode"],
            "body": json.dumps({"message": result["message"]}),
        }
    except ValueError as e:
        return handle_error(e, "入力形式が正しくありません。", reply_token)
    except Exception as e:
        logger.error(f"Unexpected error: {str(e)}")
        return handle_error(e, "内部サーバーエラー", reply_token)

line.py

"""LINEに関わる処理を提供するモジュール.

このモジュールは、ユーザープロフィールの取得、ユーザーメッセージのパース、
およびLINEにメッセージを返信する機能を提供します。
"""

import json
import logging
import os
import urllib.request

LINE_CHANNEL_ACCESS_TOKEN = os.environ["LINE_CHANNEL_ACCESS_TOKEN"]
REQUEST_URL = "https://api.line.me/v2/bot/message/reply"
PROFILE_URL = "https://api.line.me/v2/bot/profile/"
REQUEST_HEADERS = {
    "Authorization": "Bearer " + LINE_CHANNEL_ACCESS_TOKEN,
    "Content-Type": "application/json",
}


def handle_error(error: Exception, message: str, reply_token: str) -> dict:
    """エラーハンドリング用の共通関数.

    エラー発生時にログを記録し、LINEにエラーメッセージを返信する。

    Args:
        error (Exception): 発生したエラーオブジェクト
        message (str): エラーメッセージ
        reply_token (str): LINEへの返信に使用するトークン

    Returns:
        dict: エラー時のレスポンス
    """
    logging.error(f"Error: {str(error)}")
    reply_to_line(reply_token, message)
    return {
        "statusCode": 400,
        "body": json.dumps({"message": message}),
    }


def get_user_profile(user_id: str) -> dict:
    """LINEユーザーのプロフィールを取得する関数."""
    profile_request_url = PROFILE_URL + user_id
    request = urllib.request.Request(profile_request_url, headers=REQUEST_HEADERS)

    try:
        response = urllib.request.urlopen(request)
        return json.loads(response.read().decode("utf-8"))
    except Exception as e:
        logging.error(f"Failed to get user profile: {str(e)}")
        return {}


def parse_user_message(user_message: str) -> tuple[str, int]:
    """ユーザーからのメッセージをパースして項目名と金額を取得する関数."""
    user_message = user_message.replace("　", " ").replace("(", "").replace(")", "")
    parts = user_message.split()

    if len(parts) != 2 or not parts[1].isdigit():
        raise ValueError("メッセージの形式が不正です。")

    return parts[0], int(parts[1])


def reply_to_line(reply_token: str, message: str) -> None:
    """LINEにメッセージを返信する関数."""
    payload = {
        "replyToken": reply_token,
        "messages": [{"type": "text", "text": message}],
    }
    request = urllib.request.Request(
        REQUEST_URL,
        data=json.dumps(payload).encode("utf-8"),
        headers=REQUEST_HEADERS,
        method="POST",
    )

    try:
        urllib.request.urlopen(request)
        logging.info("Successfully sent reply to LINE")
    except Exception as e:
        logging.error(f"Failed to send reply to LINE: {str(e)}")

notion.py

"""Notionに関わる処理を提供するモジュール.

このモジュールは、指定されたデータ（項目名、金額、支払い者、発生日）を
Notionデータベースに書き込むための関数を提供します。

使用するAPIバージョン: 2022-06-28
"""

import json
import logging
import os
import urllib.request
from datetime import datetime
from urllib.error import HTTPError

# 環境変数からNotionのシークレットキーとデータベースIDを取得
NOTION_TOKEN = os.environ["NOTION_SECRET_KEY"]
DATABASE_ID = os.environ["NOTION_DB_ID"]
NOTION_API_URL = "https://api.notion.com/v1/pages"
NOTION_HEADERS = {
    "Accept": "application/json",
    "Authorization": f"Bearer {NOTION_TOKEN}",
    "Content-Type": "application/json",
    "Notion-Version": "2022-06-28",  # Notion APIバージョン
}


def write_to_notion_database(
    item: str, amount: int, payer: str, date: str | None = None
) -> dict[str, str | int]:
    """Notionデータベースに特定の項目（項目、金額、支払い者、発生日）を書き込む関数.

    指定された項目情報（項目名、金額、支払い者、発生日）をNotionデータベースに追加します。

    Args:
        item (str): データの名前（タイトルに相当）
        amount (int): 金額（数値フィールドに相当）
        payer (str): 支払い者の名前
        date (str|None): 発生日の日付（YYYY-MM-DD形式）。デフォルトは今日の日付。

    Returns:
        dict[str, str | int]: 結果メッセージとステータスコードを含む辞書。
    """
    # 日付が指定されていない場合は日本時間での日付を使用
    if date is None:
        date = datetime.now(ZoneInfo("Asia/Tokyo")).strftime("%Y-%m-%d")
        
    # Notion APIに送信するデータのペイロード
    payload = {
        "parent": {"database_id": DATABASE_ID},
        "properties": {
            "立て替え項目": {"title": [{"text": {"content": item}}]},
            "立て替え金額": {"number": amount},
            "支払い者": {"rich_text": [{"text": {"content": payer}}]},
            "発生日": {"date": {"start": date}},
        },
    }

    # リクエストの準備
    request = urllib.request.Request(
        NOTION_API_URL,
        data=json.dumps(payload).encode("utf-8"),
        headers=NOTION_HEADERS,
        method="POST",
    )

    try:
        # リクエスト送信とレスポンスの受信
        with urllib.request.urlopen(request) as response:
            response_body = response.read().decode("utf-8")
            logging.info(f"Response: {response_body}")
            return {"statusCode": 200, "message": "データが正常に追加されました"}
    except HTTPError as e:
        logging.error(f"HTTP Error: {e.code}, {e.reason}")
        return {
            "statusCode": e.code,
            "message": "データ追加に失敗しました",
            "details": e.reason,
        }
    except Exception as e:
        logging.error(f"Error: {str(e)}")
        return {"statusCode": 500, "message": "内部サーバーエラー", "details": str(e)}

line.pyで使用されているLINE_CHANNEL_ACCESS_TOKENや、notion.pyで使用されているNOTION_SECRET_KEYやNOTION_DB_IDなどの環境変数は、Lambda > 関数 > {関数名} > 設定 > 環境変数でキーと値を設定しています。

LINE_CHANNEL_ACCESS_TOKEN：手順1で発行したチャネルアクセストークン（長期）
NOTION_SECRET_KEY：手順3で発行した内部インテグレーションシークレット
NOTION_DB_ID：手順3で発行したテーブル(DB)のID

手順5：AWS API gatewayの作成・lambdaと接続

API gatewayのページにアクセスし、APIを作成をクリックします。

今回はREST APIを使用するのでREST APIの横の「構築」を選択します。
新しいAPIを選択しAPI名を入力、APIを作成をクリックします。
リソースを作成を選択、適当なリソース名を入力、CORS (クロスオリジンリソース共有)にチェックを入れてリソースを作成します。

メソッドを作成を選択し、メソッドタイプ：POST、統合タイプ：Lambda関数、Lambda プロキシ統合：ON、Lambda関数：先ほど作成したものを選択し、メソッドを作成をクリックする。

「APIをデプロイ」をクリック、ステージ：新しいステージ、ステージ名を入力しデプロイします。

手順6：AWS API gatewayとLINE Messaging APIチャネルを接続

ステージ名を「aaa」、リソース名を「my-resorce」とした場合は以下のようになります。

デプロイされたステージのPOSTメソッドを選択し、「URLのを呼び出す」に表示されているURL（赤丸の部分）を、LINE Developersのコンソールトップ > {自身の名前} > {作成したチャネル名} > Messaging API設定にある Webhook設定 > Webhook URL へ設定します。
Webhookの利用をONにします。

以上で設定は完了です！

実行

LINE Developersのコンソールトップ > {自身の名前} > {作成したチャネル名} > Messaging API設定にQRコードが表示されているのでそれをスマホのカメラで読み取ります。

LINEアプリが起動しBotの画面が表示されるのでトーク画面にアクセスします。

「{項目名}　{金額}」を入力し送信を押して、登録成功のメッセージが返ってくれば無事完成です！

notionの方にもきちんと記録されているはずです。

おわりに

まずはシンプルな支出転記機能をクイックに実装してみました。

今後はLambda関数のソースコードをGithubで管理し、Github Actionsを使ってpush等をトリガーに自動デプロイしてくれる仕組みを構築することで、スマートに機能改善を図っていこうと考えてます。
こちらについても追々記事にする予定です。

きっと同居人もLINE Botを使って自身で支出を入力してくれるようになると思います。