Messaging APIを用いたPython Line bot

Maki@Sunwood.ai.labs

2023年11月3日 21:32

はじめに

Messaging APIの使い方はこちら

設定ファイルローダー

下記のコードは設定ファイルローダーのクラスです。このクラスは、JSON形式の設定ファイルを読み込み、その内容をオブジェクトとして保持します。


# config_loader.py
import json  # jsonモジュールをインポートします。これによりJSON形式のファイルを扱うことができます。

# ConfigLoaderという名前のクラスを定義します。
class ConfigLoader:
    # クラスの初期化メソッドです。インスタンスが作成されるときに自動的に呼び出されます。
    def __init__(self, file_name):
        self.file_name = file_name  # file_nameという引数をインスタンス変数に代入します。
        self.settings = self.load_settings()  # load_settingsメソッドを呼び出して、設定を読み込みます。

    # 設定をファイルから読み込むメソッドです。
    def load_settings(self):
        # ファイルを読み込むコンテキストを開始します。'r'は読み込みモードを意味します。
        with open(self.file_name, 'r') as file:
            # json.loadを使って、開いたファイルの内容をJSONとして読み込みます。
            # 読み込んだデータはPythonの辞書形式に変換されます。
            return json.load(file)

公開ツール

以下のコードは、NgrokManagerという名前のクラスです。ngrokはローカルで動いているサーバーをインターネットに一時的に公開するツールです。このクラスはngrokを管理し、特定のローカルポートで実行しているサービスを外部からアクセス可能にするためのものです。


# ngrok_manager.py
import os  # osモジュールをインポートして、オペレーティングシステムの機能にアクセスできるようにします。
from pyngrok import ngrok  # pyngrokモジュールからngrokをインポートします。これによりPythonからngrokを操作できます。
from pyngrok.conf import PyngrokConfig  # pyngrokの設定を管理するためのクラスをインポートします。

# NgrokManagerクラスを定義します。
class NgrokManager:
    # クラスの初期化メソッドです。インスタンスが作成されるときに自動的に呼び出されます。
    def __init__(self, authtoken):
        self.authtoken = authtoken  # ngrokの認証トークンをインスタンス変数に代入します。
        self.webhook_url = self.start_ngrok()  # start_ngrokメソッドを呼び出して、ngrokを開始します。

    # ngrokを開始するメソッドです。
    def start_ngrok(self):
        # 既に実行中のngrokプロセスがあれば、それを終了させます。
        # 'pgrep ngrok'でngrokのプロセスIDを取得し、'kill -9'で強制終了します。
        os.system('kill -9 $(pgrep ngrok)')
        
        # ngrokを新しいセッションで開始します。ローカルアドレス '127.0.0.1:5000' にngrokのトンネルを作成します。
        # PyngrokConfigを使って新しいセッションをスタートするように設定しています。
        return ngrok.connect(addr='127.0.0.1:5000', pyngrok_config=PyngrokConfig(start_new_session=True))

LineBot

このコードは、LINE Messaging APIを使用してLINEプラットフォーム上で動作するボットを作成するためのPythonスクリプトです。LINEボットは、LINEの友達やグループにメッセージを送受信することができます。ここでは、FlaskというWebフレームワークを使用しています。Flaskは、簡単なWebサーバーを作成し、LINEからのWebhookを受け取るために使用されています。

モジュールのインポート


from flask import Flask, request, abort
from linebot import LineBotApi, WebhookHandler
from linebot.exceptions import InvalidSignatureError
from linebot.models import MessageEvent, TextMessage, TextSendMessage

ここでは、必要なモジュールをインポートしています。FlaskモジュールはWebサーバー機能を提供し、`linebot`モジュールはLINE Messaging APIとやりとりするために使用します。

`LineBot` クラスの定義


class LineBot:
    ...

`LineBot`という名前のクラスを定義しています。このクラスは、LINE Botの基本機能を備えています。

コンストラクタ


def __init__(self, access_token, channel_secret):
        ...

コンストラクタは、`LineBot`クラスのインスタンスが生成されるときに呼ばれる特別なメソッドです。
`access_token`と`channel_secret`を引数に取り、これらはLINEプラットフォームと安全に通信するために必要な認証情報です。

`LineBotApi`と`WebhookHandler`の初期化


self.line_bot_api = LineBotApi(access_token)
        self.handler = WebhookHandler(channel_secret)

`LineBotApi`オブジェクトを初期化してアクセストークンを設定し、LINEのメッセージ送信などの機能を利用できるようにしています。
`WebhookHandler`は、LINEからのWebhookを処理するために使用します。チャネルシークレットを使ってリクエストの署名を検証します。

Flaskアプリケーションの作成


def create_app(self):
        ...
        return app

`create_app`メソッドは、Flaskアプリケーションのインスタンスを作成し、Webhook用のエンドポイントを定義しています。

テスト用のエンドポイント定義


@app.route("/test")
        def test():
            return "TEST OK"

`/test`というURLにアクセスがあった場合に、`"TEST OK"`と返すシンプルなルーティングをしています。

LINE Webhookのエンドポイント定義


@app.route("/", methods=['POST'])
        def callback():
            ...

`/`というURLにPOSTリクエストがあった場合に実行される関数`callback`を定義しています。これがLINEからのWebhookリクエストを処理するためのメインのエンドポイントです。

署名検証とイベントハンドラの実行


try:
                self.handler.handle(body, signature)
            except InvalidSignatureError:
                ...

受け取ったリクエストの署名が有効か検証し、無効な場合は400エラーを返します。

メッセージイベントのハンドラ定義


@self.handler.add(MessageEvent, message=TextMessage)
        def handle_message(event):
            ...

LINEからのテキストメッセージイベントが発生した場合に実行される`handle_message`関数を定義しています。
受け取ったテキストメッセージをそのままユーザーに返信します。

このコードは、LINEからのリクエストを処理し、ユーザーからのメッセージに応答するための基本的なLINEボットを実装するためのものです。ボットを動作させるためには、このスクリプトを実行し、LINEプラットフォームにWebhook URLとして設定しておく必要があります。それによって、ユーザーがボットにメッセージを送るたびに、このスクリプトが呼び出されて適切な応答を行うことができます。

コード全体


# line_bot.py
from flask import Flask, request, abort
from linebot import LineBotApi, WebhookHandler
from linebot.exceptions import InvalidSignatureError
from linebot.models import MessageEvent, TextMessage, TextSendMessage

# LineBot クラスを定義します。
# このクラスは、LINE Bot APIとのやり取りを管理するためのメソッドを含んでいます。
class LineBot:
    # コンストラクタでは、LINE Bot APIとWebhookHandlerの初期化を行います。
    def __init__(self, access_token, channel_secret):
        # LineBotApiオブジェクトを作成し、LINEのアクセストークンを設定します。
        self.line_bot_api = LineBotApi(access_token)
        # WebhookHandlerオブジェクトを作成し、LINEのチャネルシークレットを設定します。
        self.handler = WebhookHandler(channel_secret)

    # Flaskアプリケーションを作成するメソッドです。
    def create_app(self):
        # Flaskのインスタンスを作成します。
        app = Flask(__name__)

        # '/test'のパスにアクセスがあった場合の処理を定義します。
        @app.route("/test")
        def test():
            # テスト用のエンドポイントなので、"TEST OK"というレスポンスを返します。
            return "TEST OK"

        # '/'のパスにPOSTリクエストがあった場合の処理を定義します。
        # これはLINE PlatformからのWebhookを処理するためのエンドポイントです。
        @app.route("/", methods=['POST'])
        def callback():
            # LINE Platformからのリクエストに含まれる署名を取得します。
            signature = request.headers['X-Line-Signature']
            # リクエストの本体（body）をテキストとして取得します。
            body = request.get_data(as_text=True)
            # リクエストの内容をログに記録します。
            app.logger.info("Request body: " + body)

            # 署名を検証し、イベントハンドラを呼び出します。
            try:
                self.handler.handle(body, signature)
            except InvalidSignatureError:
                # 署名が無効な場合は、エラーメッセージを出力し、400エラーを返します。
                print("Invalid signature. Please check your channel access token/channel secret.")
                abort(400)

            # すべて正常に処理された場合は、'OK'のレスポンスを返します。
            return 'OK'

        # MessageEventとTextMessageを処理するイベントハンドラを定義します。
        @self.handler.add(MessageEvent, message=TextMessage)
        def handle_message(event):
            # 受信したメッセージの内容をログに出力します。
            print("event.message.text:{}".format(event.message.text))
            # ユーザーに送られたテキストメッセージと同じ内容で返信します。
            self.line_bot_api.reply_message(
                event.reply_token,
                TextSendMessage(text=event.message.text),
            )

        # Flaskアプリケーションのインスタンスを返します。
        return app

Webサーバー

このコードは、LINE Botと連携するWebサーバーを立ち上げるためのPythonスクリプトです。各部分を細かく説明していきます。

スクリプトの全体的な流れ：

設定ファイルを読み込む。
`ngrok` を使用して、ローカルで動作しているWebサーバーをインターネットに公開する。
LINE Botを初期化する。
Flaskアプリケーションを作成し、起動する。

スクリプトの解説

他のファイルからクラスをインポートする


from config_loader import ConfigLoader
from ngrok_manager import NgrokManager
from line_bot import LineBot
from youri_bot import YouriBot

`config_loader.py` から `ConfigLoader` クラスをインポートします。
`ngrok_manager.py` から `NgrokManager` クラスをインポートします。
`line_bot.py` から `LineBot` クラスをインポートします。
おそらくカスタムの `youri_bot.py` から `YouriBot` クラスをインポートします。

スクリプトが直接実行されたときの処理を定義する


if __name__ == '__main__':

この行は、スクリプトが直接実行された場合（`python main.py` のように）にのみ、以下のコードが実行されることを意味します。

設定ファイルの読み込み


config_loader = ConfigLoader('settings.json')
settings = config_loader.settings

`settings.json` ファイルから設定を読み込むために `ConfigLoader` インスタンスを作成します。
読み込まれた設定は `settings` 変数に格納されます。

ngrokを使用してWebhook URLを生成


ngrok_manager = NgrokManager(settings['NGROK_AUTHTOKEN'])
print(f"Webhook URL: {ngrok_manager.webhook_url}")

`settings` 辞書から `NGROK_AUTHTOKEN` キーに対応する値を取得し、`NgrokManager` のインスタンスを作成します。
このインスタンスを使用してngrokサービスを開始し、生成されたWebhook URLをコンソールに出力します。

LINE Botの初期化


bot = LineBot(settings['LINE_ACCESS_TOKEN'], settings['LINE_CHANNEL_SECRET'])

`settings` 辞書から `LINE_ACCESS_TOKEN` と `LINE_CHANNEL_SECRET` を取得し、`LineBot` （カスタムのLINE Botクラス）のインスタンスを作成します。

Flaskアプリケーションの作成と起動


app = bot.create_app()
app.run()

`bot.create_app()` を呼び出してFlaskアプリケーションを作成します。
`app.run()` により、作成したFlaskアプリケーションを起動します。これによりWebサーバーが
実行を開始し、外部からのリクエストを待機する状態になります。

以上の手順を踏むことで、LINE Botのウェブフックを介してメッセージを受信し、対応することが可能になるWebサーバーが立ち上がります。これにより、LINEプラットフォームとの間でメッセージのやり取りが可能になります。

全体コード


# main.py
# 必要なクラスを他のファイルからインポートします。
from config_loader import ConfigLoader
from ngrok_manager import NgrokManager
from line_bot import LineBot
from youri_bot import YouriBot

# このスクリプトが直接実行された時だけ、以下のコードが実行されます。
if __name__ == '__main__':
    # 設定ファイル 'settings.json' を読み込むために ConfigLoader クラスを使用します。
    config_loader = ConfigLoader('settings.json')

    # 設定ファイルから読み込んだ設定を取得します。
    settings = config_loader.settings

    # Ngrokを管理するためのクラスをインスタンス化し、ngrokのAuthtokenを渡します。
    ngrok_manager = NgrokManager(settings['NGROK_AUTHTOKEN'])

    # ngrokによって生成されたWebhook URLを出力します。
    print(f"Webhook URL: {ngrok_manager.webhook_url}")

    # LINE Botのインスタンスを作成します。これには、設定から取得したアクセストークンと
    # チャネルシークレットを渡します。
    # bot = LineBot(settings['LINE_ACCESS_TOKEN'], settings['LINE_CHANNEL_SECRET'])
    bot = YouriBot(settings['LINE_ACCESS_TOKEN'], settings['LINE_CHANNEL_SECRET'])

    # Flaskアプリケーションを作成します。
    app = bot.create_app()

    # アプリケーションを起動します。これにより、Webサーバーが起動し、
    # LINEからのリクエストを待機する状態になります。
    app.run()

この記事が気に入ったらサポートをしてみませんか？