【スクレイピング】Yahoo!天気予報をスクレイピングして定期ツイートする方法Part1【Python, AWS Lambda, Twitter API】

はじめに

　今回、とある企業アカウントのTwitter運用をすることになったのですが、どう考えても単純作業部分をマンパワーで回すのは「非効率的すぎるなぁ」と思ったので、自動化できる部分は自動化してやろうと思い以下のようなプログラムを書くことにしました。

　他にも色々と自動化しちゃったのですが、今回お見せするのは“毎朝のご挨拶ツイート”を自動化している部分です。Part1ではプログラムの中心となるPythonコードを、Part2ではAWS Lambdaに載せて走らせる方法を書き記していきます。

　ここら辺の知識が欲しい駆け出しPythonエンジニアさんの参考になったら嬉しいです。

ちなみに、、、

　スクレイピングという行為自体は違法ではないものの、それを禁止しているサイトに対して行なってしまうと、違法行為になってしまう恐れがあります。判例などもあるので、「ドメイン/robots.txt」を参照して、サイト運営者がスクレイピングを許可しているかどうかをよく調べてから行うようにしてください。

　幸い、Yahoo!天気予報はスクレイピングを禁止するなどの記述はないようなので、このまま進めます。

使うもの

Python, AWS Lambda, Twitter, Yahoo!天気予報のサイト

STEP1 仮想環境の用意（必要であれば）

python3 -m venv hoge（任意の名前）
. hoge/bin/activate
コマンドラインの先頭に「(hoge)」と出ていればok

STEP2　使用するパッケージのインストール

pip install pip -U
pip install bs4 requests requests_oauthlib

STEP3 Twitter APIの取得（割愛）

QiitaやWeb上にいくらでも転がってるので、そちらを参照してください。
Twitter API を利用するには
Twitter APIの申請が通らない！！！！

STEP4 AWS Lambdaのレイヤー設定（Part2にて）

こちらの記事にて解説していますので、ご覧ください。

STEP5 Twitter APIを呼び出すクラス作成

　まずは各パッケージをimportし、次にTwitter API系を呼び出しやすいよう「TwitterAPI」というクラス名でまとめています。

　ここまで丁寧にやらなくても、一個の関数内で書いちゃっても良いのですが…今回のコードをコピペして改造しやすいように、できるだけ役割ごとに関数とその所属クラスを分けてみました。異論は受け付けます。

from bs4 import BeautifulSoup
import requests
from requests_oauthlib import OAuth1Session
import time
import random

#Twitter API系はクラス化して呼び出せるようにしておく
class TwitterAPI:
    #初期化
    def __init__(self):

        self.params = {
            "CK": "TwitterのAPIキー",
            "CS": "TwitterのAPIシークレット",
            "AT": "Twitterのアクセストークン",
            "ATS": "Twitterのアクセストークンシークレット"
        }

    #Twitter認証部分
    def get_credentials(self):

        CK = self.params["CK"]
        CS = self.params["CS"]
        AT = self.params["AT"]
        ATS = self.params["ATS"]
        twitter = OAuth1Session(CK, CS, AT, ATS)
        #OAuth1Sessionそのものを返すようにする
        return twitter

    #postに対応する関数
    def post_api(self, endpoints, params, twitter):
        res = twitter.post(endpoints, params)
        #一応デバッグなどでレスポンスをJSON化して返している
        return res.json()

    #Tweetをポストできる関数
    #今回はv1.1のエンドポイントを使用
    def post_tweet(self, query):
        endpoint = "https://api.twitter.com/1.1/statuses/update.json"
        params = {"status": query}
        #先ほどの認証部分を呼び出し
        twitter = self.get_credentials()
        #post関数に食わせる
        self.post_api(endpoint, params, twitter)

　OAuth1Sessionモジュールのクラスに正しいTwitter API情報を渡すと認証が成立します。認証情報を使って、任意のエンドポイントとクエリを渡すことで狙ったAPIを動かすことができるわけですね。

　今回は、「get_credentials」という関数で認証を、「post_api」という関数でpost処理を、「post_tweet」という関数でツイートを行うようにしました。

STEP6 スクレイピングし、テキスト化し、ツイート

　次に、天気予報をスクレイピングしツイートする部分。
　まずは「GenerateTweets」というクラスを作っていて、そこに先ほどの「TwitterAPI」クラスを継承させています。これにより、Twitter APIまわりの処理関数をサクッと呼び出せるようにしています。

　次に、スクレイピングを行ったあと、情報をどのように処理するかという部分を関数化しています。ここは、ぶっちゃけ個人の自由なカスタマイズで処理を変えて良いと思います。
　そして、スクレイピングを行う関数を作ります。今回は「Beautifulsoup」「requests」を使った非常に簡易的なスクレイピング方法を採用しています。

　あとは、ここまで記述したプログラムを一気にまとめて、ツイートする関数を呼び出すだけです。

#ツイート作成系もクラス化
#先ほどのTwitter API系クラスを継承しておくことで呼び出しを簡易化
#（...というより見た目の整頓）
class GenerateTweets(TwitterAPI):

    #気温と降水確率の判定部分
    #一般的に言われている気温と降水確率、それらに対応する過ごしやすさを比較している関数
    def weather_judge(self, max_temp, prob_precip):

        if int(max_temp) >= 18 and int(max_temp) <= 22:
            temp_status = "過ごしやすい気温"
        elif int(max_temp) < 18 and int(max_temp) > 12:
            temp_status = "少し涼しい気温"
        elif int(max_temp) < 12:
            temp_status = "肌寒い気温"
        elif len(str(max_temp)) < 10:
            temp_status = "冷え込む気温"
        elif int(max_temp) > 22 and int(max_temp) <= 25:
            temp_status = "すこし暖かい気温"
        elif int(max_temp) > 25 and int(max_temp) <= 28:
            temp_status = "暖かい気温"
        elif int(max_temp) > 28:
            temp_status = "暑い気温"

        if int(prob_precip) < 20:
            prob_status = "雨の心配は少ない"
        if int(prob_precip) >= 20 and int(prob_precip) < 60:
            prob_status = "ちょっとだけ雨模様が心配"
        if int(prob_precip) >= 60:
            prob_status = "傘を持って出かけたほうがいい"

        #気温の過ごしやすさと、降水確率の不安度をテキスト化したものを返す
        return [temp_status, prob_status]

    #天気予報をスクレイピングする関数
    def weather_scraping(self):
        #URLを指定し
        url = "https://weather.yahoo.co.jp/weather/"
        #requestsでGETする
        html = requests.get(url, timeout=10.0)
        #そしてbs4のパーサーに食わせる
        soup = BeautifulSoup(html.content, "html.parser")

        #最高気温は以下のセレクタで取れる（東京の場合）
        max_temp = soup.select_one('#map > ul > li.point.pt4410 > a > dl > dd > p.temp > em.high').text
        #最低気温も同様の方法（今回は使ってない）
        #min_temp = soup.select_one('#map > ul > li.point.pt4410 > a > dl > dd > p.temp > em.low').text
        #降水確率も同様のほうほう
        prob_precip = soup.select_one('#map > ul > li.point.pt4410 > a > dl > dd > p.precip').text.replace('%', '')

        #先ほどの気温と降水確率の判定を呼び出し
        weather_status = self.weather_judge(max_temp, prob_precip)

        #辞書にまとめて返す
        total_status = {'weather_status': weather_status, 'temps': [max_temp, prob_precip]}
        return total_status

    #以上の関数をテキストにハメてrunするための関数
    def weather_tweet(self):

        #先ほどの気温と降水確率の判定を辞書化したものを呼び出し
        status = self.weather_scraping()
        #任意のテキストに落とし込む
        query = f"おはようございます！\n今日の天気予報です。\n\n今日の最高気温は{status['temps'][0]}で、{status['weather_status'][0]}です\n降水確率は{status['temps'][1]}で、{status['weather_status'][1]}となっております。\n\n今日もがんばりましょう！"
        #Twitterは同時刻に似たようなツイートをするとBOT判定を喰らう可能性があるので、20秒単位で時間をずらす
        rand_time = random.randrange(20, 120, 20)
        time.sleep(rand_time)
        #TwitterAPIクラスのpost関数でツイート
        self.post_tweet(query)

　「weather_judge」という関数では、最高気温と降水確率に対応して“どのような文を生成するか”という部分を決めています。例えば、一番上の条件分岐では『気温が18度以上22度以下の場合は、過ごしやすい気温です』というふうになっていますね。引数はmax_temp, prob_precipとなっており、それぞれ最高気温と降水確率を受け取ることで機能するようになっています。関数の最後にreturn [temp_status, prob_status]としているのは、後述する処理を容易に済ますためです。

　「weather_scraping」という関数では、いわゆるスクレイピングを行う処理をします。BeautifulSoupを使い、該当のCSSセレクターを食わせることで任意のテキストを取得しています。
BeautifulSoupの使い方はこちら
　max_tempとprob_precipで取得した最高気温と降水確率を先ほどの「weather_judge」関数に食わせ、返り値をtotal_statusという辞書へと格納しています。ちなみに、後述の処理にて扱うため、この辞書の「temps」というキーには最高気温と降水確率をリストとして渡していています。

　「weather_tweet」という関数は、これまで出てきたクラスと関数をひとまとめのテキストにして、ツイートするためのようなものです。
　status = self.weather_scraping()で“その日の過ごしやすさ”と“最高気温, 降水確率”を取得し、その返り値をqueryのテキストフォーマットへと埋め込んでいます。
　rand_time = random.randrange(20, 120, 20)とtime.sleep(rand_time)を書くことにより、Twitter運営からのBOT判定を避けています。

　あとは、self.post_tweet(query)に先ほどのテキストを食わせれば、必要なコード記述は終わりです。

全体コードはこちら

from bs4 import BeautifulSoup
import requests
from requests_oauthlib import OAuth1Session
import time
import random

#Twitter API系はクラス化して呼び出せるようにしておく
class TwitterAPI:
    #初期化
    def __init__(self):

        self.params = {
            "CK": "TwitterのAPIキー",
            "CS": "TwitterのAPIシークレット",
            "AT": "Twitterのアクセストークン",
            "ATS": "Twitterのアクセストークンシークレット"
        }

    #Twitter認証部分
    def get_credentials(self):

        CK = self.params["CK"]
        CS = self.params["CS"]
        AT = self.params["AT"]
        ATS = self.params["ATS"]
        twitter = OAuth1Session(CK, CS, AT, ATS)
        #OAuth1Sessionそのものを返すようにする
        return twitter

    #postに対応する関数
    def post_api(self, endpoints, params, twitter):
        res = twitter.post(endpoints, params)
        #一応デバッグなどでレスポンスをJSON化して返している
        return res.json()

    #Tweetをポストできる関数
    #今回はv1.1のエンドポイントを使用
    def post_tweet(self, query):
        endpoint = "https://api.twitter.com/1.1/statuses/update.json"
        params = {"status": query}
        #先ほどの認証部分を呼び出し
        twitter = self.get_credentials()
        #post関数に食わせる
        self.post_api(endpoint, params, twitter)

#ツイート作成系もクラス化
#先ほどのTwitter API系クラスを継承しておくことで呼び出しを簡易化
#（...というより見た目の整頓）
class GenerateTweets(TwitterAPI):

    #気温と降水確率の判定部分
    #一般的に言われている気温と降水確率、それらに対応する過ごしやすさを比較している関数
    def weather_judge(self, max_temp, prob_precip):

        if int(max_temp) >= 18 and int(max_temp) <= 22:
            temp_status = "過ごしやすい気温"
        elif int(max_temp) < 18 and int(max_temp) > 12:
            temp_status = "少し涼しい気温"
        elif int(max_temp) < 12:
            temp_status = "肌寒い気温"
        elif len(str(max_temp)) < 10:
            temp_status = "冷え込む気温"
        elif int(max_temp) > 22 and int(max_temp) <= 25:
            temp_status = "すこし暖かい気温"
        elif int(max_temp) > 25 and int(max_temp) <= 28:
            temp_status = "暖かい気温"
        elif int(max_temp) > 28:
            temp_status = "暑い気温"

        if int(prob_precip) < 20:
            prob_status = "雨の心配は少ない"
        if int(prob_precip) >= 20 and int(prob_precip) < 60:
            prob_status = "ちょっとだけ雨模様が心配"
        if int(prob_precip) >= 60:
            prob_status = "傘を持って出かけたほうがいい"

        #気温の過ごしやすさと、降水確率の不安度をテキスト化したものを返す
        return [temp_status, prob_status]

    #天気予報をスクレイピングする関数
    def weather_scraping(self):
        #URLを指定し
        url = "https://weather.yahoo.co.jp/weather/"
        #requestsでGETする
        html = requests.get(url, timeout=10.0)
        #そしてbs4のパーサーに食わせる
        soup = BeautifulSoup(html.content, "html.parser")

        #最高気温は以下のセレクタで取れる（東京の場合）
        max_temp = soup.select_one('#map > ul > li.point.pt4410 > a > dl > dd > p.temp > em.high').text
        #最低気温も同様の方法（今回は使ってない）
        #min_temp = soup.select_one('#map > ul > li.point.pt4410 > a > dl > dd > p.temp > em.low').text
        #降水確率も同様のほうほう
        prob_precip = soup.select_one('#map > ul > li.point.pt4410 > a > dl > dd > p.precip').text.replace('%', '')

        #先ほどの気温と降水確率の判定を呼び出し
        weather_status = self.weather_judge(max_temp, prob_precip)

        #辞書にまとめて返す
        total_status = {'weather_status': weather_status, 'temps': [max_temp, prob_precip]}
        return total_status

    #以上の関数をテキストにハメてrunするための関数
    def weather_tweet(self):

        #先ほどの気温と降水確率の判定を辞書化したものを呼び出し
        status = self.weather_scraping()
        #任意のテキストに落とし込む
        query = f"おはようございます！\n今日の天気予報です。\n\n今日の最高気温は{status['temps'][0]}で、{status['weather_status'][0]}です\n降水確率は{status['temps'][1]}で、{status['weather_status'][1]}となっております。\n\n今日もがんばりましょう！"
        #Twitterは同時刻に似たようなツイートをするとBOT判定を喰らう可能性があるので、20秒単位で時間をずらす
        rand_time = random.randrange(20, 120, 20)
        time.sleep(rand_time)
        #TwitterAPIクラスのpost関数でツイート
        self.post_tweet(query)

#ここはローカルでデバッグする場合と本番で回す場合で別れる
#ローカルの場合
if __name__ == "__main__":
    command = input('Command?(g = GenerateTweets)')
    if command == 'g':
        g = GenerateTweets()
        g.weather_tweet()
        
#本番環境の場合
def lambda_handler(event, context):
    # TODO implement
    g = GenerateTweets()
    g.weather_tweet()
    return

さいごに

　プログラムを走らせるためのコード記述、お疲れ様でした。

　次回のPart2では、AWS Lambdaで毎朝の定期実行を行うための設定、必要な処理を記事化していくので、そちらも併せてご覧ください。
Part2→【AWS Lambda定期実行】Part2! LambdaとPythonを使った定期実行をわかりやすく解説【Python, AWS Lambda, Twitter API】

　さいごに、私はTwitterも細々とやっておりますので、よかったらフォロー＆コメントいただけると幸いです！
　Twitterアカウントはこちら