👋

【個人開発】お弁当屋さんの日替わりメニューを通知する Line Bot を作った

2022/12/11に公開

Google Cloud

はじめに

先日お気に入りのお弁当屋さんの日替わり弁当を通知する Line Bot を作りました。
難しいことはしておらず、お弁当屋さんが毎朝ツイートする日替わりメニューの内容を LINE へ転送するシンプルな Bot です。
もし似たような仕組みを実現したいという人がいれば参考にしてください。

背景

私の住まいの近くにお気に入りのお弁当屋さんがあります。
そちらでは毎日違う種類の日替わり弁当を出しているのですが、その種類が非常に豊富です。
週一くらいの頻度で訪ねてますが、毎回違うメニューが提供されているのです。
訪れるたびにどんなメニューか楽しみな反面、実は美味しい日替わりを見逃しているのでは?という疑問が浮かびました。
そこで、日替わり弁当のメニューを毎日通知してくれるサービスを作り、日替わりの内容を常に把握できるような仕組みを作ってみます。

成果物

最終的に毎日お昼の 11 時に日替わりメニューを通知する LINE Bot を作成しました。
メッセージ内容は、お弁当屋さんが毎朝ツイートしている日替わりメニュー内容をそのまま転送しています。

Screenshot of line bot
非公式のためお店の名前は伏せます

実装内容はこちらです。

要件定義

日替わり弁当のメニューを通知するためには、メニューの内容を取得する必要があります。
幸いに、そのお弁当屋さんは Twitter を利用しており、毎朝日替わりメニューをツイートしています。
それを見れば十分ですね、今回は開発の必要ありません、、、とはなりませんでした。

わがままなユーザー(私)は今どき Twitter をやっていません。
このために Twitter を見たくないという、とてもめんどくさいユーザーでした。

以上の条件から、今回開発する日替わり通知システムでは以下二つの機能が必要があります。

システムが日替わりメニューツイートを取得する
Twitter をやっていないユーザーへメニューツイートを配信する

お弁当屋さんは基本 1 日 1 ツイートです。
たまに 2 種類あったり、特別仕様のときは 1 日 2,3 ツイート行います。
また、日替わりメニューの写真もツイートに含まれています。

これらの仕様を踏まえた上で実装を行います。

技術選定

要件定義で決まった二つの機能をどう実装するか考えます。

日替わりメニューの取得は Twitter API を利用すればすぐに可能でしょう。
その通知方法ですが、多くの日本人が利用している LINE API を利用することにします。

その上で開発言語は Python を使用することにしました。
理由は、サードパティ製の Twitter API Wrapper tweepy と公式の LINE bot SDK どちらも過去に使ったことがあるからです。

次にデプロイ先ですが、今回はシンプルな機能なのでサーバーレス関数として実装できそうです。
また毎日日替わり弁当の通知をするため、定期的な自動実行が必要です。
これだけならメジャーなサービスのどれでも実現できそうです。
今回は個人的な好みで GCP の Cloud Functions と Cloud Scheduler の組み合わせを採用しました。

ということで、最終的に以下のような形で実装しました。

Overview of this architecture

日替わりメニューツイートの取得

Twitter API の利用準備

Twitter API でお弁当屋さんのツイートを取得します。

API 利用のために、まず Twitter Developer Portal でアプリを作成します。
公開ツイートの取得だけできれば良いので、Standalone Apps を作成しました。
良いドキュメントがなかったので、簡単に説明します。

Twitter Developer Portal へアクセス
サイドメニューから Projects & Apps メニューの Overview を開く
ページ最下部の「+ Create App」をクリック

Create App button is shown at the bottom of page

アプリ名を入力すると API Key などが取得できます。
Access Token も必要なのですが、なぜか作成直後は取得できません。
Projects & Apps の Overview へもう一度もどり、作成されたアプリを確認します。
名前の横に鍵マークがあるのでそれをクリックすると Access Token 取得画面に移動します。

Secret key link is shown as key icon

以下 4 つをメモしたら実装を始めます。

API Key
API Secret
Access Token
Access Token Secret

Python でツイートを取得

Python で Twitter API を利用する際は tweepy を利用すると簡単です。
まずは仮想環境へインストールします。

poetry add tweepy

あるユーザーの最新ツイートを取得するコードは以下のとおりです。

title=get_user_timelines.py

import tweepy
from tweepy.models import Status

# API認証、API Key などは各々の物を使うこと
auth = tweepy.OAuthHandler("TWITTER_API_KEY", "TWITTER_API_SECRET")
auth.set_access_token("TWITTER_ACCESS_TOKEN", "TWITTER_ACCESS_TOKEN_SECRET")

api = tweepy.API(auth)

# 指定した User ID の最新ツイートを count 数分取得
# Status モデルはツイートとほぼ同意
result: list[Status] = api.user_timeline(user_id="user_id", count=5)

user_timeline メソッドはデフォルトで 20 件取得します。
お弁当屋さんはツイートは多くても 3, 4 件なので、5 件だけ取得しています。

次に、取得したツイートを当日分だけにフィルターします。
きちんと作るなら、ツイート内容を解析して日替わりメニューかそうでないか判別したいところですが、そこまで複雑なものは面倒なので作りません。
幸いにもお弁当屋さんは日替わりメニューに関連する内容しかツイートしないため、当日のツイートをすべて配信する方針で良さそうです、

本日のツイートだけをフィルターするコードは以下のとおりです。
ツイート取得部分は前回と同じなので省略しています。

from datetime import datetime, timedelta, timezone


def is_today(tweet: Status) -> bool:
    if isinstance(tweet.created_at, datetime):
        now = datetime.now(JST)
        tweeted_at = tweet.created_at.astimezone(JST)
        delta = now - tweeted_at

        if delta.days > 1:
            return False
        else:
            # timedelta が 24 時間以内で、日付が同じなら本日のデータとする
            return tweeted_at.day == now.day
    else:
        # 日付がわからないときは判別不可能なので、何もしない
        return False

result: list[Status] = get_user_timeline(user_id="user_id", count=5)

# ついでにツイート時刻の昇順にする
tweets_of_today = reversed(list(filter(is_today, result)))

本日のツイート判定は、ツイート時刻が現在時刻から24時間以内かつ日付の数値が同じ、という条件でやっています。
もっときれいにアルゴリズムで実装できそうですが、なにかアイデアがあれば教えて下さい。

取得した Tweet を LINE Bot から配信

LINE Messaging API の使用準備

LINE Bot の作成には LINE Messaging API を利用します。
API 利用のためにまずプロバイダーを作成します。
プロバイダーとは、ビジネスやサービスの提供者という意味で使われてます。
具体的な作り方はドキュメントに任せます。

作成したプロバイダーの Messaging API Channel Access Token を取得します。
コンソールのプロバイダー設定 → Messaging API 設定ページから確認できます。
Access Token を取得したら準備完了です、実装に移りましょう。

Python LINE　ｂｏｔ SDK でメッセージ送信

公式の Python LINE bot SDK をインストールします。

poetry add line-bot-sdk

LINE bot からメッセージ送信する方法はいくつかあります。
今回は、bot と友達登録しているユーザー全員に送信する機能 broadcast を利用します。

ツイートを取得したあとのメッセージ送信機能は以下のようになります。
ツイート取得部分は省略していますのであしからず。

from linebot import LineBotApi
from linebot.models import TextSendMessage

tweets_of_today = get_tweets_of_today(user_id="user_id")

line = LineBotApi("LINE_BOT_ACCESS_TOKEN")

for tweet in tweets_of_today:
    line.broadcast(TextSendMessage(text=message))

難しい部分はありませんので、読めばなんとなくわかると思います。
メッセージを LINE SDK が提供する専用のクラス TextSendMessage に変換することだけ忘れずに。

これだけで必要最低限の機能は完成です。
しかし日替わりメニューツイートには、美味しそうな弁当の写真も添付されています。
今回はそちらも LINE へ転送するようにします。

ツイートから画像　URL を取得して　LINE で配信

ここの実装に一番時間がかかりました。
というのも、 tweepy で得られるツイート情報には同じ画像を表示する URL が 4 種類ほど取得できます。
しかし、この内の 1 種類でしか LINE で正しく送信できませんでした。
最初に間違った URL を使ってしまったため、無駄にハマってしまいました。

実際に画像を送るためのコードは以下のとおりです。

from linebot import LineBotApi
from linebot.models import ImageSendMessage, TextSendMessage

tweets_of_today = get_tweets_of_today(user_id="user_id")

line = LineBotApi("LINE_BOT_ACCESS_TOKEN")

for tweet in tweets:
    send_message(tweet.text)

    # `tweepy` で取得したデータに画像が含まれていることをチェック
    if hasattr(tweet, "extended_entities"):
        for media in tweet.extended_entities["media"]:
            url = media["media_url_https"]
            line.broadcast(
                ImageSendMessage(
                    original_content_url=url,
                    preview_image_url=url,
                )
            )

tweepy で取得したデータに画像が含まれていると extended_entities という dict 型のメンバが付与されます。
このメンバの media.media_url_https で取得できる URL だけが LINE で正しく送信できます。
それ以外の URL は直接開けば画像を確認できるのですが、LINE bot で送ることができないので注意してください。

Cloud Functions で定期実行

GCP で定期実行用のイベントを定義

ここまでで実装は完了です。
最後に Cloud Functions にデプロイして実際に動かすところまで解説します。

GCP はすでに利用できる前提で話しますが、初めて使う場合は以下のドキュメントを参考にしてください。

まず GCP 上で定期的に Cloud Functions を実行するためのイベントを定義します。
これには Cloud Scheduler と Pub/Sub という機能を使います。
実装方法はドキュメントが詳しいです。

簡単にやっていることを説明します。

Pub/Sub で任意の名前のイベントを定義
Cloud Scheduler で定義したイベントを指定のタイミングでトリガー

イベントはデプロイ時に指定する必要があるので、わかりやすいものにしましょう。
Scheduler は cron 形式で任意のタイミングを指定できるます。
以下は毎日 11 時にイベントをトリガーする例です。

0 11 * * *

コードを Cloud Functions にデプロイ

Cloud Functions にデプロイする前に、コードに少し変更を加えます。

まず functions-framework をインストールします。
こちらは Python 用の Cloud Function フレームワークです。

poetry add functions-framework

次に今までの実装を Cloud Functions で呼び出せるように関数化します。

import functions_framework


@functions_framework.cloud_event
def main(cloud_event):
    # 今までの実装内容をここに書く

定義した関数は GCP 上でのイベントをトリガーに実行されることを意味するデコレータ @functions_framework.cloud_event をつけています。
これは定期実行するために必要な部分です。

ここまでできたらデプロイしましょう。
デプロイコマンドは以下を参考に自分の環境に合わせてカスタマイズしてください。

gcloud functions deploy function-name --gen2 --runtime=python310 --region=asia-northeast1 --source=. --entry-point=main --trigger-topic=event-name --allow-unauthenticated --env-vars-file .env.yaml

Python をイベントトリガーで実行するには第2世代 Cloud Functions である必要があるため、 --gen2 は必須です。
また --trigger-topic=event-name には、Pub/Sub で定義したイベント名を指定します。

デプロイ後は Cloud Scheduler の管理画面上から "ジョブを強制実行する" というコマンドで動作確認ができます。

終わりに

今回はお弁当屋さんの日替わり弁当ツイートを LINE で通知する Bot を作ってみました。
機能が少なく各サービスのドキュメントも充実していたため、数時間で完成しました。
巨人は偉大ですね。

これで美味しそうな日替わりを見逃す心配もなくなりました。
今度は逆に毎日弁当を食べたくなる欲求と戦うはめになりましたが、それはまた別のお話。

もし参考になったと思ったらいいねよろしくお願いします。

それでは。

GitHubで編集を提案