FastAPI・Cloud Run・Firestore・LIFFで毎朝「傘いる？」を通知するLINE Botを作った

警告

下記の問題が確認されました．

• 予報の取得に失敗
  • 原因調査中
• 雨ではないが傘が必要と判定
  • 傘の判定基準を変更予定

はじめに

出発前に天気アプリを開いて、降水確率と雨量を見比べて、「普通の傘か、折りたたみ傘
か」を考えるのが少し面倒でした。そこで、複数地域の時間別予報を判定し、指定した
曜日・時刻にLINEへ傘の種類を通知するBotを作りました。

この記事では、次の技術要素を中心に紹介します。

• FastAPIでWebhook、LIFF API、Cloud Scheduler用APIを実装
• Open-Meteoの時間別予報をドメインモデルへ正規化
• FirestoreとCloud Schedulerで定時通知
• LINE LoginのIDトークンをサーバー側で検証
• Flex Messageで全体判定と地域詳細を表示
• リトライ時の二重送信を防止
• Cloud LoggingとCloud Monitoringで障害を検知

なお、この記事とGitHubで公開するのは技術リファレンスです。実際に友だち追加できる
LINE公式アカウントや、本番のLIFF URLは公開していません。

設定画面は、LINEのリッチメニューから開く構成にしました。

全体構成

主な構成は次のとおりです。

領域	使用技術
API	Python 3.11 / FastAPI
実行環境	Cloud Run
永続化	Firestore Native mode
定時実行	Cloud Scheduler
認証	LINE Webhook署名 / LINE Login IDトークン / Google OIDC
天気	Open-Meteo Forecast API / Geocoding API
UI	LIFF / Vanilla JavaScript / CSS
テスト	pytest / pytest-asyncio / HTTPX

最初にドメインロジックを外部サービスから分離した

傘判定をFastAPIのルートやOpen-Meteoのレスポンス処理へ直接書くと、境界値テストが
難しくなります。そこで、構成をおおむね次の4層に分けました。

src/umbrella_bot/
├── domain/          # 傘判定、時間区間、メッセージ
├── application/     # ユースケースとPort
├── infrastructure/  # Open-Meteo、LINE、Firestore
└── api/             # FastAPIルートと認証依存

外部APIのレスポンスはAdapterで共通の予報モデルへ変換し、FirestoreはRepository経由で
扱います。ドメインテストはネットワークにも実トークンにも依存しません。

傘判定を3段階にした

傘レベルは「不要」「折りたたみ傘」「普通の傘」の3段階です。初期閾値は気象機関の
公式な傘基準ではなく、このBot独自の設定です。

@dataclass(frozen=True, slots=True)
class JudgementConfig:
    rain_amount_interval_threshold_mm: float = 0.1
    rain_possibility_probability_threshold_percent: float = 40.0
    standard_umbrella_total_rain_mm: float = 3.0
    standard_umbrella_max_hourly_rain_mm: float = 2.0
    standard_umbrella_long_rain_duration_minutes: int = 180
    standard_umbrella_long_rain_min_total_mm: float = 1.0

普通の傘は、合計雨量3.0 mm以上、最大1時間雨量2.0 mm以上、長時間の雨、雷雨、強い
降水のいずれかで判定します。そこまで強くなくても、雨量0.1 mm以上、降水確率40%以上、
雨・雪コードがあれば折りたたみ傘です。

複数地域を登録した場合は、地域ごとに判定してから最も強い傘レベルを全体判定にします。
同率の地域はすべて「判定基準」として残します。

地域ごとの予報
    ↓
時間帯を抽出・集計
    ↓
地域ごとの傘レベル
    ↓
最大レベルを全体判定へ採用

一部地域の予報取得に失敗し、取得できた地域がすべて「傘不要」だった場合は、傘不要と
断定せず、安全側の「折りたたみ傘」へ倒します。

定時実行では「送ったか」だけでなく実行そのものを永続化した

Cloud Schedulerは毎分、内部APIを呼びます。

POST /internal/scheduler/run-due

単に「現在時刻と同じ通知設定」を検索してPushすると、再試行や同時実行で二重送信が
起こり得ます。そのため、通知設定とは別にNotificationRunを保存します。

実行IDは通知設定IDと予定時刻から決定的に作り、processingには期限付きリースを持たせ
ます。プロセスが途中で落ちても、リース期限後に同じ実行を回収できます。

LINE Push APIには、同じ通知実行から生成した安定したX-Line-Retry-Keyを付けます。
アプリ側の実行IDとLINE側のリトライキーを対応させることで、再試行時の重複送信を
抑えました。

3種類の入口を別々に認証した

LINE Webhook

WebhookはJSONへ変換する前の生ボディに対して、Channel Secretを使ったHMAC-SHA256署名を
検証します。署名を確認する前にリクエスト内容を信用しません。

LIFF設定API

ブラウザでデコードしたユーザーIDをAPIへ送る方式にはしていません。LIFF SDKが取得した
生のIDトークンをBearerトークンとしてバックエンドへ送り、バックエンドからLINE Login
v2.1の検証APIへ問い合わせます。iss、aud、sub、expも確認します。

Cloud Scheduler

Cloud Runサービス自体はWebhookとLIFFのため公開していますが、内部Scheduler APIでは
Google署名に加え、次を検証します。

• OIDC audienceがCloud RunのサービスURLと一致
• 呼出元メールがScheduler専用サービスアカウントと一致
• 認証なしは401、別サービスアカウントは403

Firestoreで必要になった複合インデックス

定時処理では、次のような複合条件を使用します。

• 有効な通知設定をnext_run_at順で取得
• pending実行をscheduled_at順で取得
• 期限切れのprocessing実行をlease_until順で取得

そのため、リポジトリにはfirestore.indexes.jsonを含めました。インデックスが
CREATINGの間にSchedulerを有効化すると、クエリが失敗してCloud Runが503を返します。
すべてREADYになってからSchedulerを再開するのがポイントでした。

LIFF設定画面で詰まったところ

以下は実際のHTML/CSSを使い、記事用のダミー地域と通知設定だけを表示した画面です。
本番のユーザー情報、ID、URLは含めていません。

LIFF SDK初期化が終わらない

最初は「LINEに接続しています…」から進まないことがありました。原因はCSPとLIFF SDKの
実行条件でした。API全体のCSPを緩めるのではなく、/liff/だけに必要なLINE公式ドメインと
SDK実行条件を許可しました。

初期化にはタイムアウトを設け、失敗時に再試行ボタンと問い合わせ番号を表示しています。

短い市町村名を検索できない

Geocoding APIでは「蒲郡」より「蒲郡市」のほうが安定して検索できる場合がありました。
そこで、1〜2文字の日本語検索では「市」「町」「村」を補った候補を優先検索し、結果を
重複排除しています。「堺」のような1文字の市にも対応できます。

通知は全体判定と地域詳細を分けた

最初のBubbleに全地域の結論を置き、続くBubbleに地域ごとの時間帯や雨量を載せました。
画像はFlex Messageの配色と項目を使った記事用の表示再現で、実際のLINEアカウントを
撮影したものではありません。

日付と説明文の余白を端末向けに調整した

日付のwrap: trueによって、右側に余白が見えても曜日だけ次の行へ送られる端末が
ありました。日付はwrap: falseとadjustMode: shrink-to-fitを組み合わせ、必要な端末だけ
フォントを縮小しています。

また、親Boxにspacingを設定していても、子要素のmargin: noneが上書きしていたため、
傘レベルと「雨予報」が密着していました。不要なmarginを外して親の余白を効かせました。

構造化ログと監視

標準出力へ1行JSONで次のイベントを記録します。

event	用途
http_request_finished	HTTPステータスと処理時間
location_forecast_evaluated	地域ごとの判定結果
notification_run_finished	通知実行の最終結果
scheduler_run_finished	Scheduler 1回分の集計

生のLINEユーザーIDは記録せず、名前空間付きハッシュへ変換します。IDトークン、Webhook署名、
Channel Access Token、外部APIのレスポンス本文もログへ出しません。

Cloud Loggingのログベース指標から、通知失敗、Scheduler失敗、Cloud Run 5xx、Cloud
Scheduler失敗のアラートを作成し、通知チャネルまで実際にテストしました。

Cloud Runへ配置するときの注意

秘密値はコマンドへ直接書かず、Secret Managerの明示バージョンを参照します。

gcloud run services update "${SERVICE}" \
  --region="${REGION}" \
  --update-secrets="LINE_CHANNEL_SECRET=line-channel-secret:1,LINE_CHANNEL_ACCESS_TOKEN=line-channel-access-token:1,LOCATION_SELECTION_SECRET=location-selection-secret:1"

シェルコマンドをコピーするとき、イメージタグやSecret名の途中で改行すると、その改行も
値に入り、Dockerイメージ名やSecret名の解析エラーになります。変数への代入は1行に保ち、
実行前にprintf '%s\n' "${IMAGE}"で確認すると安全です。

テスト

python3.11 -m venv .venv
source .venv/bin/activate
python -m pip install -e '.[dev]'
pytest
ruff check .

境界値、時間区間結合、部分障害、安全側判定、Webhook署名、IDトークン検証、Firestoreの
競合、Schedulerのリース、LINEの二重送信防止、Flex Message上限などを自動テストして
います。外部通信はHTTPXのモックを使い、実トークンや実サービスへ依存しません。

ソースコード

YamatoDoi/umbrella-line-bot
で公開しています。

公開リポジトリには本番のプロジェクトID、Channel ID、LIFF ID、Cloud Run URL、秘密値を
含めていません。実際に動かす場合は、各自のGoogle CloudプロジェクトとLINEチャネルを
作成してください。

参考にした公式資料

• Verify webhook signature | LINE Developers
• LINE Login v2.1 API reference | LINE Developers
• Send Flex Messages | LINE Developers
• Use authentication with HTTP targets | Cloud Scheduler
• Authenticating service-to-service | Cloud Run
• Manage indexes in Cloud Firestore | Firebase
• Terms & Privacy | Open-Meteo

おわりに

最も時間がかかったのは、傘判定そのものよりも「失敗しても二重送信しないこと」と、
LINE・Google Cloud・ブラウザの3種類の認証境界を整理することでした。

ドメインロジックを先に外部サービスから分離したことで、後からFirestore、LIFF、Flex
Messageを追加しても判定部分を崩さずに済みました。小さなBotでも、定時実行、再試行、
秘密管理、監視まで含めると、実運用に近い題材になります。

Weather data by Open-Meteo.com under
CC BY 4.0。無料APIを利用する場合は、
非商用条件と最新の利用上限を確認してください。