Qiita Teams that are logged in
You are not logged in to any team

Log in to Qiita Team
Community
OrganizationAdvent CalendarQiitadon (β)
Service
Qiita JobsQiita ZineQiita Blog
4
Help us understand the problem. What is going on with this article?
@kozakura16

【IFTTT】Webスクレイピング×IFTTTでプロ野球の試合結果をLINE通知する(1/2)

More than 1 year has passed since last update.

はじめに

今回はWebスクレイピングとIFFFTを用いてプロ野球の試合結果をLINE通知するようにしてみる。筆者はプロ野球の中でもDeNAファンなので、DeNAの試合結果を通知することにする。Webスクレイピングに前から興味があったし、今まで試合結果はYahoo!開いて、スポーツ開いて、プロ野球開いて、、といった感じで手順が多くて面倒だったので、「じゃあLINE通知されたらいいんじゃね?」と思ったのがきっかけ。
全体構成はこんな感じ。IFTTT的に表現すると「DeNAの試合が終わったら、結果をLINEに通知する」ということになる。
①DeNAの公式サイトの試合結果を監視
②試合結果が終了になったことを検知
③IFTTTのWebhookに試合結果を通知
④Webhookの検出(Thisの部分)
⑤LINEに試合結果を通知(Thatの部分)
図1.jpg

実際に通知されるLINEはこんな感じ。
試合結果.png

IFFFTとは?

IFTTTとはIF This Then Thatの略称で「あることが起きたら、あることをする」ことができる外部アプリ連携サービスである。本サイトはこちらをどうぞ。
例えば、本サービスを使えば「天気予報サービスで雨予報が起きたら、LINEに通知する」といったことが可能になる。
サイトの登録方法についてはこちらをどうぞ。
また、ThisについてWebhookにより作ることが可能(Thenも可能らしい)で、今回はこれを利用する。

Webスクレイピングとは?

Webから情報をスクレイピング(抽出)することを意味する。具体的には、Webページ上のhtml構文を解析して、好きな情報を取得することができる。例えば、複数の2ちゃんねるのサイトから特定ジャンルのタイトルのスレッドのリンクのみを抽出して、自分専用のまとめページを作ったりなんかができる。

実装

Webスクレイピングにより、DeNAの試合結果が終了になることを検出

pythonのpyqueryを用いて、DeNAの公式サイトのページを取得する。ソースコードは以下の通り。
ざっくりと説明すると、以下のような感じ。
1. 30分に一度公式サイトの試合結果ページを見に行く
 URLの末尾2桁は01~05のいずれかが設定されるのだが法則がわからずのため、すべてを見に行って、ページがあるときに次の処理に進むようにした。また、すでに取得済みかどうかの判断は./log以下にその日付のファイル名があるかどうかで判断する。ある場合は次の日になるまでスキップする。ない場合は試合結果を取得し、./log以下に日付のファイルを作成する。
2. 試合終了の場合に、結果を取得する
 試合終了を表すhtmlの属性を取得して、"終"の文字がある場合は試合終了とみなし、結果(スコア、先攻後攻など)を取りに行く。
3. 取得した試合結果をIFTTTのWebhookに通知する
 2.で取得した試合結果をIFTTTのWebhookに必要な値(LINE通知した文字列)を詰めてPOSTする。Webhookの詳細については後ほど記事を書くことにします。ソースコード中の以下の部分が次の記事のIFFFTのPOST受け口(Webhook)になります。
URL = "https://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 = "【S】" + 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 = "【S】" + 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を使ってもよいかなと思う。ソースコードなどは趣味程度で書いた可読性無視のコードなので、あくまで参考程度に。。。

undefined

続く

続きはこちらをどうぞ!

4
Help us understand the problem. What is going on with this article?
Why not register and get more from Qiita?
  1. We will deliver articles that match you
    By following users and tags, you can catch up information on technical fields that you are interested in as a whole
  2. you can read useful information later efficiently
    By "stocking" the articles you like, you can search right away
kozakura16
個人開発や勉強内容をメインに投稿しているエンジニアです。

Comments

No comments
Sign up for free and join this conversation.
Sign Up
If you already have a Qiita account Login
4
Help us understand the problem. What is going on with this article?