AWS Lambda + Python + LINE Botで傘が必要か教えてもらう

4/17 追記

Line Messaging API + PythonとLambdaを使って、オウム返しBotをつくるのまとめ
こちらの記事内にて、AWS Lambdaの関数の作成のlambda_funciton.pyの記述に誤りがあったことを指摘していただいたため、編集しております。
本記事は初めに作ってからできる限り手順は再現しながら書いたつもりですが、もし手違いでエラー等になる場合はお手数ですがコメントにてお知らせください。

はじめに

Python最近書いてないなぁと思ったのと、LINEはよく使うけどLINE APIは使ったことないなぁと思ったのでそれらを使ったLINE Bot を作ってみようと思います。
今回はLINE Messaging APIも一から触ってみたかったので、line-bot-sdk-pythonは使いません。
筆者はPythonの経験が少ないので記事中のコードに違和感を覚える方もいるかもしれませんが、
その際はコメント欄にてご指摘いただければ幸いです。

作るもの

傘がいらないパターンしか取れなかったんですが、地名を入力すると(おおよそ)1時間後に傘が必要かどうかを教えてくれるBotを作ります。いざ出かける前に出かける場所の地名を入力して傘を持っていくかどうかが知れると思います。

使用技術

• Python3.8
• AWS Lambda
• LINE Messaging API
• Yahoo 気象情報API
• Geocoding.jp API

開発手順

いきなり目的のものを作るのではなく、段階を経て作っていきます。

1. LINE Messaging API を使っておうむ返しするbotを作る
2. コンソール上で地点を入力したら傘の必要不要を返してくれる処理をPythonで書く
3. 1と2を組み合わせて目的のものを作る

実装

1. LINE Messaging API を使っておうむ返しするbotを作る

まず画像のようにおうむ返ししてくれるbotを作っていきます。

LINE Developersへのチャネル登録

LINE Developersへログイン(アカウントが作成できていない方はまずアカウント作成を行ってください)し、真ん中あたりの「Messaging API」を押下し、「今すぐはじめよう」を押下します。
チャネルの種類はそのまま「Messaging API」を、プロバイダーはもし以前作成したことがある場合はそちらのプロバイダーを、なければ「新規プロバイダー作成」を選択し、プロバイダ名は任意のプロバイダ名を設定してください。
チャンネルアイコンは作成するBotのアイコンに、チャネル名とチャネル説明はそれぞれ作成するBotの名前と説明になるので、用途に応じて適当に設定してください。
大業種は今回は「個人」を、小業種は「個人（IT・コンピュータ）」を選択します。
最後にメールアドレスを設定し、各利用規約の内容を確認したのち、一番下の「作成」ボタンを押下してください。
チャネル基本設定などのメニューが出ると思うので、「Messaging API設定」の一番下、「チャンネルアクセストークン」の発行ボタンを押下し、発行されたチャンネルアクセストークンを控えておいてください。

AWS Lambdaの関数の作成

AWSコンソールにログインして検索バーに「Lambda」と入力し、Lambdaの画面に遷移します。
右上の「関数を作成」ボタンを押下し、関数名、ランタイムを設定します。
ここでは関数名は「LINEBot」、ランタイムは「Python3.8」を設定します。

ここまでできたら「関数の作成」ボタンを押下してLambda関数を作成します。
関数の作成が正常に完了するとこのような画面が表示されるはずです。
ここから「lambda_function.py」を編集して実際にコードを書いていきます。
まず、LINEにてテキストメッセージが送信された場合、どのような形式で送られるのか確認します。
LINE Messaging API ドキュメントをみてみます。
公式ドキュメントによると、テキストメッセージはjson形式で、textという名前で送られていることがわかります。
次に、応答メッセージをどうやってユーザに返すかを確認します。LINE Messaging API メッセージより、
「 https://api.line.me/v2/bot/message/reply 」に対してPOST形式で、リクエストヘッダとリクエストボディをそれぞれ下記の形式で設定することでメッセージ送信できることがわかります。

リクエストヘッダ

Content-Type
application/json

Authorization
Bearer {channel access token}

リクエストボディ

replyToken
String 必須
Webhookで受信する応答トークン

messages
メッセージオブジェクトの配列 必須
送信するメッセージ
最大件数：5

notificationDisabled
Boolean 任意
true：メッセージ送信時に、ユーザーに通知されない。
false：メッセージ送信時に、ユーザーに通知される。ただし、LINEで通知をオフにしている場合は通知されません。
デフォルト値はfalseです。

リクエストボディのmessagesに記されている「メッセージオブジェクトの配列」は、テキストメッセージの場合はこちらに書いてあります。
これらの情報を参考に、「ユーザからのメッセージを受け取り、そのメッセージをそのままユーザに返す」処理をPythonで書いてみます。

lambda_function.py

import json
import urllib.request

def lambda_handler(event, context):
    for message_event in json.loads(event['body'])['events']:
        url = 'https://api.line.me/v2/bot/message/reply'
        headers = {
            'Content-Type': 'application/json',
            'Authorization': 'Bearer ' + {あなたのチャンネルアクセストークン}
        }
        body = {
            'replyToken': message_event['replyToken'],
            'messages': [
                {
                    "type": "text",
                    "text": message_event['message']['text'],
                }
            ]
        }
        
        req = urllib.request.Request(url, data=json.dumps(body).encode('utf-8'), method='POST', headers=headers)
        with urllib.request.urlopen(req) as res:
            logger.info(res.read().decode("utf-8"))
        
        
    return {
        'statusCode': 200,
        'body': json.dumps('Hello from Lambda!')
    }

{あなたのチャンネルアクセストークン}の部分はLINE Developersへのチャネル登録で発行したチャンネルアクセストークンを設定してください。
ユーザからきたjson形式で来たメッセージの中のtext部分を取得し、POSTリクエストを作成して返す処理が書けました。当然ですが、このままではこの関数は呼ばれないのでトリガーを指定していきます。

Lambda関数のトリガーの設定

Lambda　コンソール画面の上部、「関数の概要」より、「トリガーを追加」ボタンを押下します。
「トリガーを選択」では「API GateWay」、「APIを作成する」をそれぞれ選択し、APIタイプは「HTTP」を選択します。
セキュリティは今回は「オープン」を選択します。
ここまでできたら「追加」ボタンを押下します。

追加後に、設定タブのトリガーの中にあるAPI GateWayのAPIエンドポイントをコピーしておいてください。
LINE Developersへのチャネル登録で作成したチャネル画面へ戻り、
Messageing API設定の「Webhook URL」に先ほどコピーしたAPI GateWayのAPIエンドポイントを追記してください。
実際に友だち登録して、botに対してメッセージを送信すればそっくりそのまま同じ内容が返ってくるはずです。
これでおうむ返しするbotの作成ができました。

2. コンソール上で地点を入力したら傘の必要不要を返してくれる処理をPythonで書く

とりあえずおうむ返しするbotの作成ができたので、次は元々の目的である「地点を入力したら傘の必要不要を返してくれる処理」をまずはコンソール上で動くように書いていきます。

処理の流れ

ざっくり実装する処理の流れを説明していきます。

1. コンソール上から地点を入力
2. 1で入力された地点とGeocoding.jp APIを使って該当地点の緯度経度を取得
3. 2で取得した緯度経度と Yahoo 気象情報APIを使って降水強度を取得
4. 3で取得した降水強度から傘の必要不要を判定し、出力として返す

以上のような流れで目的の処理を記載していきます。

Yahoo 気象情報APIを使用するためにアプリケーションを登録

Yahoo 気象情報APIを使うためにこちらのご利用ガイドを参考にアプリケーションを登録し、Client IDを取得してください。
アプリケーションの種類は「サーバーサイド（Yahoo! ID連携 v2）」、
アプリケーション利用者情報は「個人」、
アプリケーション名は任意のアプリケーション名を、
サイトURLはそのまま「 http://example.com/ 」で、
利用するスコープはそのままチェックをつけないで大丈夫です。

必要なライブラリをインストール

緯度経度取得APIの結果を取得用に必要なライブラリをインストールします。
今回はBeautiful Soup4を使用するので、下記コマンドでインストールしておいてください。

$ pip install beautifulsoup4

コード

実際のコードがこちらです。

WeatherReport.py

import time
import urllib.request, urllib.parse
import json
import datetime
from bs4 import BeautifulSoup as bs4

# 場所から緯度経度を取得
def getLatitude(place):

    params = {
        'q': place
    }

    # 入力された地点よりクエリパラメータを生成
    params = urllib.parse.urlencode(params)

    # 緯度経度取得APIに地点をセット
    url = 'http://www.geocoding.jp/api/?' + params

    # 緯度経度取得API取得サイトにアクセス
    url = urllib.request.urlopen(url).read()

    # APIの実行結果を取得
    soup = bs4(url, 'html.parser')

    # 実行結果にエラーが含まれていた場合は空文字を返す
    if soup.find('error'):
        return '', ''

    # 緯度経度を文字列型に変換して返す
    lat = soup.find('lat').string
    lon = soup.find('lng').string

    return lat, lon

# 経度緯度から降水強度を取得
def getWeatherReport(lat, lon):
    # YahooWebAPI Client ID
    appId = {あなたのClient ID}

    # Yahoo気象情報APIのリクエストURLの大元
    baseUrl =  'https://map.yahooapis.jp/weather/V1/place?coordinates='

    # 大元のURLに緯度経度とClient IDを設定して気象情報取得用のURLを生成
    url = baseUrl + lon + ',' + lat + '&output=json&appid=' + appId

    # 気象情報取得用のURLにアクセス
    weather_report = urllib.request.urlopen(url).read()

    # json形式で取得できるので辞書型に変換
    json_tree = json.loads(weather_report)

    # 取得結果の中から必要な情報のみ抽出(現在時刻から10分感覚で取得できるので[6]:1時間後を取得)
    weather_list = json_tree['Feature'][0]['Property']['WeatherList']['Weather'][6]

    # 時刻表示用にフォーマットを整形
    dt = datetime.datetime.strptime(weather_list['Date'], '%Y%m%d%H%M')
    dt = dt.strftime("%Y/%m/%d %H:%M")

    # 降水強度に応じて傘の必要不要を判定
    if weather_list['Rainfall'] == 0.0:
        print(dt + ' : 傘は必要ありません')
    elif 0.0 < weather_list['Rainfall'] < 1.0:
        print(dt + ' : 長時間出かける場合は傘を持っていきましょう')
    elif weather_list['Rainfall'] >= 1.0:
        print(dt + ' : 絶対に傘を持って出かけてください')

def main():
    print('どこの天気を知りたい？')

    place = input('>> ')

    # 入力された地点の緯度、経度を取得
    lat, lon = getLatitude(place)

    # 緯度、経度を用いて降水強度を取得
    getWeatherReport(lat, lon)

if __name__ == '__main__':
    main()

下記コマンドで実行し、任意の地点を入力して結果が取得できればOKです。

$ python WeatherReport.py
どこの天気を知りたい？
>> 表参道
2021/04/05 23:15 : 傘は必要ありません

1と2を組み合わせて目的のものを作る

ここまでで「おうむ返しをするLINE Botの作成」と、「コンソール上で任意の地点の傘の必要不要を教えてくれる処理の実装」ができたので、いよいよこの2つを組み合わせて目的の処理を作っていきます。

環境変数の登録

環境変数の設定を参考に、Lambda コンソールで環境変数を設定します。
今回は以下2つの環境変数を設定します。
APP_ID : Yahoo 気象情報APIを使用するためにアプリケーションを登録で登録したClient ID
ChannelAccessToken : LINE Developersへのチャネル登録で登録したチャンネルアクセストークン
設定した環境変数は環境変数の取得にて取得できます。

必要なライブラリの登録

ローカルでpipコマンドを使ってインストールしたBeautiful Soup4はそのままだとLambda関数上で使えないので、Lambda レイヤーを使って登録します。
レイヤーの作成には

1. .zip ファイルをアップロード
2. Amazon S3 からファイルをアップロードする

の2つやり方があるのですが、今回はBeautiful Soup4のみアップロードしたいので、zipファイルをアップロードする形式でやっていきます。
※10 MB より大きいファイルの場合は、Amazon S3 を使用したアップロードを検討してください。
とのことなので、10MBより大きいファイルはできる限りS3からアップロードする形式をとるようにしましょう。
今回はzipファイルをアップロードする形式を取るのでまずカレントディレクトリにpythonというディレクトリを作成し、
そこにBeautiful Soup4を使うのに必要なファイルをダウンロードし、それらをzipファイルにしてアップロードします。
まずzipファイルを作成します。

$ mkdir python
$ pip install -t . beautifulsoup4
$ cd ../
$ zip python.zip python/*

これでBeautiful Soup4を使うのに必要なファイルが入ったzipファイルが作られたのでアップロードしていきます。
AWS Lambda 画面の左メニューから「レイヤー」を押下し、「レイヤーの作成」を押下します。名前と説明は任意で指定し、「zipファイルをアップロード」を選択して先ほど作成したpython.zipを指定します。
最後に「作成」ボタンを押下すればレイヤーの作成は完了です。これで必要なライブラリの登録ができました。

コードの書き換え

ここまでくれば、あとはLambda関数のコードの書き換えを行うだけです。
基本的には1. LINE Messaging API を使っておうむ返しするbotを作るでやったようにユーザの入力値を取得し、
その入力値を使って2. コンソール上で地点を入力したら傘の必要不要を返してくれる処理をPythonで書くと同じように判定するだけです。
下記にコードをまとめます。

lambda_function.py

import time
import urllib.request, urllib.parse
import json
import datetime
import os
from bs4 import BeautifulSoup as bs4

# 場所から緯度経度を取得
def getLatitude(place):

    params = {
        'q': place
    }

    # 入力された地点よりクエリパラメータを生成
    params = urllib.parse.urlencode(params)

    # 緯度経度取得APIに地点をセット
    url = 'http://www.geocoding.jp/api/?' + params

    # 緯度経度取得API取得サイトにアクセス
    url = urllib.request.urlopen(url).read()

    # APIの実行結果を取得
    soup = bs4(url, 'html.parser')

    # 実行結果にエラーが含まれていた場合は空文字を返す
    if soup.find('error'):
        return '', ''

    # 緯度経度を文字列型に変換して返す
    lat = soup.find('lat').string
    lon = soup.find('lng').string

    return lat, lon

# 経度緯度から降水強度を取得
def getWeatherReport(lat, lon):
    # YahooWebAPIアプリケーションID
    appId = os.environ['APP_ID']

    # Yahoo気象情報APIのリクエストURLの大元
    baseUrl =  'https://map.yahooapis.jp/weather/V1/place?coordinates='

    # 大元のURLに緯度経度とアプリケーションIDを設定して気象情報取得用のURLを生成
    url = baseUrl + lon + ',' + lat + '&output=json&appid=' + appId

    # 気象情報取得用のURLにアクセス
    weather_report = urllib.request.urlopen(url).read()

    # json形式で取得できるので辞書型に変換
    json_tree = json.loads(weather_report)

    # 取得結果の中から必要な情報のみ抽出(現在時刻から10分感覚で取得できるので[6]:1時間後を取得)
    weather_list = json_tree['Feature'][0]['Property']['WeatherList']['Weather'][6]

    # 時刻表示用にフォーマットを整形
    dt = datetime.datetime.strptime(weather_list['Date'], '%Y%m%d%H%M')
    dt = dt.strftime("%Y/%m/%d %H:%M")

    # 降水強度に応じて傘の必要不要を判定
    if weather_list['Rainfall'] == 0.0:
        message = dt + ' : 傘は必要ありません'
    elif 0.0 < weather_list['Rainfall'] < 1.0:
        message = dt + ' : 長時間出かける場合は傘を持っていきましょう'
    elif weather_list['Rainfall'] >= 1.0:
        message = dt + ' : 絶対に傘を持って出かけてください'
        
    return message
        
def lambda_handler(event, context):
    # 送られたメッセージから地点と返信用トークンを取得
    for message_event in json.loads(event['body'])['events']:
        place = message_event['message']['text']
        reply_token = message_event['replyToken']
    
    # 入力された地点の緯度、経度を取得
    lat, lon = getLatitude(place)

    # 緯度、経度を用いて傘の必要不要を取得
    message = getWeatherReport(lat, lon)
    
    # チャンネルアクセストークンと返信用トークンを用いてユーザにメッセージを返す
    url = 'https://api.line.me/v2/bot/message/reply'
    headers = {
        'Content-Type': 'application/json',
        'Authorization': 'Bearer ' + os.environ['ChannelAccessToken']
    }
    body = {
        'replyToken': reply_token,
        'messages': [
            {
                "type": "text",
                "text": message,
            }
        ]
    }
        
    req = urllib.request.Request(url, data=json.dumps(body).encode('utf-8'), method='POST', headers=headers)
    with urllib.request.urlopen(req) as res:
        logger.info(res.read().decode("utf-8"))
        
        
    return {
        'statusCode': 200,
        'body': json.dumps('Success!')
    }

結果

下記の画像のように、傘の必要不要がわかるようになりました！

まとめ

今回はAWS Lambda + Python + LINE Botを使って傘が必要か教えてもらうBotを作成しました。
初めに作ってからできる限り手順は再現しながら書いたつもりですが、もし手違いでエラー等になる場合はお手数ですがコメントにてお知らせください。
また、最初にも申し上げた通り筆者はPythonの経験が少ないので記事中のコードに違和感を覚える方もいるかもしれませんが、その際はコメント欄にてご指摘いただければ幸いです。
最後まで読んでいただきありがとうございました。

参考

• Messaging APIの概要
• 【LINE Botの作り方】お金をかけずにLINEのMessaging APIの投稿(push)と返信(replay)を試してみる。
• AWS LambdaのPython3.6でLINT Botを動かす
• LINE BOT + AWS Lambda + Python 作成時のメモ