GCP Cloud Functionsで気象庁天気予報情報を定期的にスクレイピングして、BigQueryへinsertする方法

#はじめに
スクレイピングのための、BeautifulSoup4ライブラリを扱う練習がてら、気象庁の天気予報情報をスクレイピングしました。
気象庁の過去天気はcsvで取得できますが、天気「予報」情報は、csvなどで取得できない（はず）ため、スクレイピングして取得する必要があります。

スクレイピングしただけは面白くないので、定期的にスクレイピングをGCP Cloud Functionsを実行しつつ、取得したデータをBigQueryに格納するコードを説明します。

注意

注意として、今回のコードを定期実行させるタイミングは、縛りがあります。昼11時頃に、翌日以降7日間の天気予報を公開してくれる（ような）ので、その時間以降にて、htmlをスクレイピングを試みます。今回はとりあえず毎日12:30に実行させています。

それより前の時間帯に動作させると、想定通りには動きません。
また、気象庁のリンクやhtmlの書き方が変化したら使えなくなりますので、ご注意ください。

こんなユーザが対象です。

• 何らかのサイトをスクレイピングしてみたい(requests)
• CloudFunctionsを定期実行して、BigQueryへデータを保存したい
• Python3を使いたい！

##環境

• CloudFunctions(Python 3.7(ベータ版))
• Functionsの定期実行に、pubsub Trigger
• WEBアクセス機能に、requestsライブラリ
• スクレイピング機能に、BeautifulSoup4ライブラリ
• 定期実行のために、GoogleAppEngine StandardEnvironment (Python3.7(ベータ版))
• データ保存に、BigQuery

##簡易構成図と簡単な流れは、以下です。

1. 事前準備(pubsubトピック作成、BQ dataset/table作成)
2. GAEをcronで定期実行
3. GAEからFunctionsをトリガーするためのpubsubトピックを呼び出し
4. pubsub経由でFunctionsを実行
5. Functionsからrequestsで、気象庁の天気予報htmlを取得(東京地域)
6. Functionsで、BeautifulSoup4を利用してスクレイピング実施
7. FunctionsからBigQueryへデータをinsert

1. 事前準備(pubsubトピック作成、BQ dataset/table作成)

GCPのCloudFunctionsは、定期実行する機能がないため、回避策としては、GAEからpubsubを呼び出し、pubsubトリガーでFunctionsを実行させます。
https://developers-jp.googleblog.com/2017/04/how-to-schedule-cron-jobs-with-cloud.html
なので、事前にpubsubを作成しておきます。

トピック名(例)
weather_forecast

また、今回は、データをBQ(BigQuery)へ保存するため、事前にtableを作っておきます。

• date:予報対象の日付(例えば2018-10-15)
• previous_forecast:上記の日付を何日前に予報したのかの情報（例えば-1であれば前日に予報されたデータであることを指します。最大-7)
• max_temp,min_temp:最大気温、最低気温

フィールド名	タイプ	モード
date	DATE	NULLABLE
previous_forecast	INTEGER	NULLABLE
max_temp	INTEGER	NULLABLE
min_temp	INTEGER	NULLABLE

2. GAEをcronで定期実行

GAEをcronで実行させます。main.pyは後述します。
定期実行間隔は、毎日12:30とする場合は、以下のように指定をします。

cron.yaml

cron:
- description: Push to pubsub every day
  url: /event/trigger
  schedule: every day of month 12:30
  timezone: Asia/Tokyo

3. GAEからFunctionsをトリガーするためのpubsubトピックを呼び出し

GAEの実行する機能は以下のようなものです。
PUBSUBトピックとしては、1.で作成したものを指定します。

main.py

PROJECT_ID = '<replace to your ProjectID>'
PUBSUB_TOPIC = '<pubsub Topic for Functions>'

@app.route("/event/trigger", methods=['GET'])
def push_pubsub():
        #pubsubへpush
        publisher = pubsub.PublisherClient()
        topic_path = publisher.topic_path(
            PROJECT_ID,
            PUBSUB_TOPIC
            )
        message = 'test message'
        # message must be bytestream
        message_byte = message.encode('utf-8')
        publisher.publish(topic_path, data=message_byte)
        return 'success'

4. pubsub経由でFunctionsを実行

pubsubトリガーで動作するFunctionを作成します。

• [トリガー]:Cloud Pub/Sub

5. Functionsからrequestsで、気象庁の天気予報htmlを取得(東京地域)

ここからCloudFunctionsのコード説明となります。以降で、利用するライブラリは以下の通りです。requirements.txtでも読み込みライブラリを定義します。lxmlも定義しないとエラーが起こりますので注意ください。

main.py

import base64
from datetime import datetime, timedelta, timezone
import requests
from google.cloud import bigquery
import re
from bs4 import BeautifulSoup

requirements.txt

requests==2.19.1
bs4==0.0.1
lxml==4.2.5
google-cloud-bigquery==1.6.0

アクセスするURLは以下で、requestsライブラリを利用して、東京都の天気予報情報を取得できます。

main.py

url = 'https://www.jma.go.jp/jp/week/319.html'
response = requests.get(url)

参考までに、当該サイトは、以下のようなものです。
今回はこのサイトから、向こう1週間の最高気温(℃)と、最低気温(℃)を取得したいと思います。

サイトから取得したデータを、BeautifulSoupで読み込みます。この際、HTMLパーサーはlxmlを利用しました。

main.py

soup = BeautifulSoup(response.text, 'lxml')

##6. Functionsで、BeautifulSoup4を利用してスクレイピング実施

スクレイピングの流れは、以下です。

1. htmlのソースコードから、読み込みたいhtmlテーブルを、条件指定(id/class)で取得
2. 多数あるtrタグのなかで、5番目のタグと6番目のタグに、最高気温と最低気温があることが分かるため、当該タグのテキスト部を取得
3. スペースや改行が含まれているため、replaceで削除
4. 最高気温や最低気温は2日後以降の予報では幅の情報(xx～xx)が含まれているため、あとでBQで扱いやすいように幅情報も削除

main.py

#1.の工程（抜粋）
table = soup.find_all("table",attrs={"id":"infotablefont","class":"forecast-top"})

#2.の工程（抜粋）
count = 0
for targetrow in row:
    if count == 5:
        if targetrow.find('th').string == '東京' and re.search('最高', targetrow.find('td').string):

#3.の工程（抜粋）
b = a.text.replace("\t","").replace("\n","")

#4.の工程（抜粋）
kekka = re.search("\(",b)
if kekka != None:
    c = b[:kekka.start()]

##7. FunctionsからBigQueryへデータをinsert

予報情報をBQに格納するために、Date情報をListで作っておきます。予報は7日分存在するので、7日分作ります。

main.py

days = []
days.append((datetime.now(JST)+timedelta(days=1)).strftime("%Y-%m-%d"))
days.append((datetime.now(JST)+timedelta(days=2)).strftime("%Y-%m-%d"))
days.append((datetime.now(JST)+timedelta(days=3)).strftime("%Y-%m-%d"))
days.append((datetime.now(JST)+timedelta(days=4)).strftime("%Y-%m-%d"))
days.append((datetime.now(JST)+timedelta(days=5)).strftime("%Y-%m-%d"))
days.append((datetime.now(JST)+timedelta(days=6)).strftime("%Y-%m-%d"))
days.append((datetime.now(JST)+timedelta(days=7)).strftime("%Y-%m-%d"))

BigQueryにデータをインサートするためには、事前にBigQueryテーブルを作っておき、以下のように、データを入れます。

• days[dayscount]は、先ほど作った日付情報で、daycountは繰り返しカウントです
• previousは、-1から-7まで繰り返し、BigQueryのprevious_forecast=何日前に予報したのかの情報です
• maxtemp_list[dayscount]や、mintemp_list[dayscount]は、対象日付の最高気温と最低気温情報です。

main.py

#インサートする情報を定義(抜粋)
rows_to_insert = [(days[dayscount], previous, maxtemp_list[dayscount], mintemp_list[dayscount])]

#BigQueryへinsert(抜粋)
response = client.insert_rows(table, rows_to_insert)

参考

BigQuery Client Library
PythonとBeautiful Soupでスクレイピング

おわりに

BeautifulSoup4ライブラリはやはり便利で、スクレイピングは結構柔軟に対応できそうです。また、今回の手法が最適とは思えませんが、とりあえず形になったので良しとします。
定期的に様子を見て、気象庁の予報情報と確定情報を比較してみても面白そうです。
コードはgithubに公開しました。