はじめに
今回はWebスクレイピングとIFFFTを用いてプロ野球の試合結果をLINE通知するようにしてみる。筆者はプロ野球の中でもDeNAファンなので、DeNAの試合結果を通知することにする。Webスクレイピングに前から興味があったし、今まで試合結果はYahoo!開いて、スポーツ開いて、プロ野球開いて、、といった感じで手順が多くて面倒だったので、「じゃあLINE通知されたらいいんじゃね？」と思ったのがきっかけ。
全体構成はこんな感じ。IFTTT的に表現すると「DeNAの試合が終わったら、結果をLINEに通知する」ということになる。
①DeNAの公式サイトの試合結果を監視
②試合結果が終了になったことを検知
③IFTTTのWebhookに試合結果を通知
④Webhookの検出（Thisの部分）
⑤LINEに試合結果を通知（Thatの部分）
実際に通知されるLINEはこんな感じ。
IFFFTとは？
IFTTTとはIF This Then Thatの略称で「あることが起きたら、あることをする」ことができる外部アプリ連携サービスである。本サイトはこちらをどうぞ。
例えば、本サービスを使えば「天気予報サービスで雨予報が起きたら、LINEに通知する」といったことが可能になる。
サイトの登録方法についてはこちらをどうぞ。
また、ThisについてWebhookにより作ることが可能（Thenも可能らしい）で、今回はこれを利用する。
Webスクレイピングとは？
Webから情報をスクレイピング（抽出）することを意味する。具体的には、Webページ上のhtml構文を解析して、好きな情報を取得することができる。例えば、複数の２ちゃんねるのサイトから特定ジャンルのタイトルのスレッドのリンクのみを抽出して、自分専用のまとめページを作ったりなんかができる。
実装
Webスクレイピングにより、DeNAの試合結果が終了になることを検出
pythonのpyqueryを用いて、DeNAの公式サイトのページを取得する。ソースコードは以下の通り。
ざっくりと説明すると、以下のような感じ。
- 30分に一度公式サイトの試合結果ページを見に行く
URLの末尾2桁は01~05のいずれかが設定されるのだが法則がわからずのため、すべてを見に行って、ページがあるときに次の処理に進むようにした。また、すでに取得済みかどうかの判断は./log以下にその日付のファイル名があるかどうかで判断する。ある場合は次の日になるまでスキップする。ない場合は試合結果を取得し、./log以下に日付のファイルを作成する。 - 試合終了の場合に、結果を取得する
試合終了を表すhtmlの属性を取得して、"終"の文字がある場合は試合終了とみなし、結果（スコア、先攻後攻など）を取りに行く。 - 取得した試合結果をIFTTTのWebhookに通知する
2.で取得した試合結果をIFTTTのWebhookに必要な値（LINE通知した文字列）を詰めてPOSTする。Webhookの詳細については後ほど記事を書くことにします。ソースコード中の以下の部分が次の記事のIFFFTのPOST受け口（Webhook）になります。
//maker.ifttt.com/trigger/dena_result/with/key/【key_path】"
※また、pythonスクリプトの場所にlogフォルダを追加が必要です。（試合結果の送信に成功したら、その日は実行されないようにするため）
denascraping.py
# -*- coding:utf-8 -*-
import requests
from datetime import datetime
import urllib
import json
import time
import sys
import os
from pyquery import PyQuery as pq
from pathlib import Path
def main():
while True:
today = datetime.now()
date = today.strftime('%Y%m%d')
flag_bat_first = False
# ファイルがあるか確認あれば終了
if os.path.isfile('./log/' + date):
wait_while(date + ' file is exist...')
continue
# Webページが取得できるか確認
flag_page_found = False
for i in range(5):
print(date)
url = 'https://www.baystars.co.jp/game/result/' + date + '0' + str(i+1)
print('url: '+url )
query = pq(requests.get(url).content)
print('query: ' +query(".tag").text() )
search_result = query(".focNavy01b").text()
query_tag = query(".tag").text()
if search_result == "":
print("page found!!")
flag_page_found = True
break
# Webページが取得できていなければ終了
if flag_page_found == False:
wait_while("cannot get web page...")
continue
# 試合が終わっているか確認("終"の文字があるかで確認)
flag_game_over = False
for text in query_tag:
print('text: ' + text)
if text == "終":
flag_game_over = True
# 試合が終わっていなければ終了
if flag_game_over == False:
wait_while("the game is not stared or playing...")
continue
# 先攻後攻の判定
print("-------GAME TEAM--------")
team_names = query(".game--score-left").text()
if team_names.split('\n').index('横浜DeNA') == 0:
print("DeNA is bat first!")
flag_bat_first = True
else:
print("DeNA is not bat first!")
flag_bat_first = False
# 試合結果の取得
print("-------GAME RESULT--------")
game_result = query(".game--score-right").text()
game_score = game_result.split('\n')
print("bat_first: " + game_score[1])
print("bat_not_first: " + game_score[2])
DeNA_NAME_TEXT = "DeNA"
opponent_name = query(".enemy").text()
post_score_text = ""
print("opponent name: " + opponent_name[0])
if flag_bat_first == True:
dena_score = int(game_score[1])
opponent_score = int(game_score[2])
post_score_text = DeNA_NAME_TEXT + " " + game_score[1] + " - " + game_score[2] + " " + opponent_name + "<br>"
else:
dena_score = int(game_score[2])
opponent_score = int(game_score[1])
post_score_text = opponent_name + " " + game_score[1] + " - " + game_score[2] + " " + DeNA_NAME_TEXT + "<br>"
print(post_score_text)
# 試合結果の勝ち負けを取得
result_text = ""
win_pitcher = ""
save_pitcher = ""
lose_pitcher = ""
if dena_score > opponent_score:
print("DeNA win!")
result_text = "WIN!"
win_pitcher = "【勝】" + query(".visitor__boxgame--win").parent()(".hover__type--one").text()
save_pitcher = "【Ｓ】" + query(".visitor__boxgame--save").parent()(".hover__type--one").text()
lose_pitcher = "【負】" + query(".visitor__boxgame--lose").parent()(".text__color--main").text()
elif dena_score < opponent_score:
print("DeNA lose...")
result_text = "LOSE..."
win_pitcher = "【勝】" + query(".visitor__boxgame--win").parent()(".text__color--main").text()
save_pitcher = "【Ｓ】" + query(".visitor__boxgame--save").parent()(".text__color--main").text()
lose_pitcher = "【負】" + query(".visitor__boxgame--lose").parent()(".hover__type--one").text()
else:
print("Draw\n")
result_text = "Draw"
post_result_text = '<br>'.join([result_text, win_pitcher, save_pitcher, lose_pitcher]) + "\n"
print(post_result_text)
post_url_text = url
# IFTTTのWebHookに試合結果を通知
URL = "https://maker.ifttt.com/trigger/dena_result/with/key/【key_path】"
METHOD = "POST"
HEADERS = {"Content-Type" : "application/json"}
# PythonオブジェクトをJSONに変換する
obj = {"value1" : post_score_text, "value2" : post_result_text, "value3" : post_url_text}
json_data = json.dumps(obj).encode("utf-8")
# httpリクエストを準備してPOST
request = urllib.request.Request(URL, data=json_data, method=METHOD, headers=HEADERS)
with urllib.request.urlopen(request) as response:
response_body = response.read().decode("utf-8")
print(response_body)
# logファイルを作成する
Path('./log/' + date).touch()
print(date + ' logged!')
def wait_while(reason_text):
print(reason_text)
time.sleep(60 * 30)
if __name__=='__main__':
main()
おわりに
作成したプロセスは自宅のラズパイで常時走らせるようにしている。AWSを使ってもよいかなと思う。ソースコードなどは趣味程度で書いた可読性無視のコードなので、あくまで参考程度に。。。
続く
続きはこちらをどうぞ！