LambdaとEventBridgeで天気予報を自動通知する　#424

モリユウキ|YM202110

2023年10月7日 21:54

前回の記事でLINE NotifyをLambdaで動かしてみたので、その続きです。

無料の天気予報APIを使いつつ、それをEventBridgeで定期実行されるLambdaで実行し、結果を要約してLINE Notifyに通知します。

朝、LINEをチェックする際についでに天気予報が見れたらいいなと思って作ってみました。既存機能でそういうのはあるらしいですが、ここはエンジニアらしく自作してみようと。

使わせていただく天気予報APIは以下です。
無料かつ登録も認証も不要なのでとても便利です。

Lambda関数を作成する

ここはさっくりいきます。
単純にAPIを叩き、結果を整形します。urllib.requestはややクセがあって、それがまた勉強になるのでここで学んだことは別でまとめたいと思ってます。

以下のコードになりました。

import os
import json
import urllib.request


def format_value(value):
    return "-" if value is None else value

# Tenki API
tokyo = 130010
tenki_api = f'https://weather.tsukumijima.net/api/forecast/city/{tokyo}/'
tenki_request_headers = {
    # urllib.requestはデフォルトでユーザーエージェントを使用しておらず、天気予報APIに弾かれる
    'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
}

# LINE Notify
TOKEN = os.environ['LINE_NOTIFY']
api_url = 'https://notify-api.line.me/api/notify'
line_request_headers = {
    'Content-Type': 'application/x-www-form-urlencoded',
    'Authorization': 'Bearer' + ' ' + TOKEN
}


def lambda_handler(event, context):
    tenki_req = urllib.request.Request(tenki_api, headers=tenki_request_headers, method='GET')
    conn = urllib.request.urlopen(tenki_req)
    tenki_data = json.loads(conn.read().decode('utf-8'))
    
    original_title = tenki_data['title'].split(" ")
    title = original_title[1] + original_title[2]

    today = tenki_data['forecasts'][0]
    tomorrow = tenki_data['forecasts'][1]

    message = (
        f"{title}\n\n"
        f"{today['dateLabel']}\n"
        f"【{today['detail']['weather'].replace('　', '')}】\n"
        f"最高{format_value(today['temperature']['max']['celsius'])}℃, 最低{format_value(today['temperature']['min']['celsius'])}℃\n\n"
        f"{tomorrow['dateLabel']}\n"
        f"【{tomorrow['detail']['weather'].replace('　', '')}】\n"
        f"最高{format_value(tomorrow['temperature']['max']['celsius'])}℃, 最低{format_value(tomorrow['temperature']['min']['celsius'])}℃"
    )
    
    params = {'message': message}
    # data引数はbytes型を期待しているためbytes型に変換
    data = urllib.parse.urlencode(params).encode('ascii')
    
    # LINE Notifyに通知
    req = urllib.request.Request(api_url, headers=line_request_headers, data=data, method='POST')
    conn = urllib.request.urlopen(req)