タイトルの通り、TwitterAPIを利用して自動検索をpythonで実現し、そこから得られた情報をLINENotifyで通知するコードを書きました。

###動機

新型コロナウイルスの流行に関連して、毎日その日の感染者数が発表されています。その数値を気にする必要性の有無はともかくとして、東京都の感染者数の情報が出てくる昼頃から夕方にかけて、いつ出るか分からない速報を確認するためにTwitterを開いては検索をかける習慣がついてしまいました。（感染者数の速報はTwitterで確認するのが最も早いと個人的には思っています）
➡︎ 無駄にTwitterで検索する作業をなくしたい！！というのが今回の動機です。

###実現したこと

Twitterの自動検索
東京都の感染者数に関するツイートから数値を取得
LINENotifyで通知する

今回の実装は東京都の感染者数に特化したものですが、少し変えるだけで様々な応用もできそうです。

###必要なもの

※以下はMac環境を想定しています。

pythonの実行環境：ここではJupyter Notebookを使用
Twitter APIの登録 ：『Twitter API 登録 (アカウント申請方法) から承認されるまでの手順まとめ　※2019年8月時点の情報』を読んで登録しました
LINENotifyの登録 ：『PythonでLINEにメッセージを送る』の方法及びコードをそのまま使わせていただきました
特別に環境構築の必要があるpythonモジュール
- tweepy：pip install tweepyで大丈夫だと思います
- mecab-python3：pythonモジュールだけでなく、mecab本体と辞書をインストールする必要があります。詳しくは『Macにmecab-python3を導入しようとして、苦戦した話』などを参考にすれば環境構築できると思います。また、Google Colabolatoryでは!pip install mecab-python3==0.7を実行するだけで必要なものが全て自動でインストールされたのでそちらの方が楽かもしれません。

これらを用意するのが、実は最も手間のかかる部分かもしれません。
『{やりたいこと}　python』でググって色んなツールを見つけていくのは楽しい作業でもありますが笑

#実装

以下、実際のコードを見ながら実装内容を説明します。

##0. 実装の概要

1. 諸々のインポート
2. Twitterアクセス用のオブジェクトを作成
3. LINE通知用のオブジェクトを作成
4. 自動検索の関数を作成

##1. 諸々のインポート

.py

import requests
import datetime
import time
import pandas as pd

from IPython.display import clear_output
#printの出力を消す関数

import tweepy
import MeCab

tagger = MeCab.Tagger("-Owakati")

最後の行の"tagger"というのは、MeCabで文章の分割を行う時に使うオブジェクトです。

具体的には、tagger.parseを用いて

.py

tagger.parse("今日はいい天気ですね")
# => '今日 は いい 天気 です ね \n'
tagger.parse("今日はいい天気ですね").split()
# => ['今日', 'は', 'いい', '天気', 'です', 'ね']

という感じです。後で出てきますが、今回の実装ではsplit()でlist型にしてツイートを扱います。

##2. Twitterアクセス用のオブジェクトを作成

TwitterAPIの登録で取得したキーやトークンを組み込んだオブジェクトを作っておきます。
参考：『Tweepyの使い方 ~その１~ 【Tweetの取得】』

.py

consumer_key = "ここに取得したキー/トークンを入力"
consumer_secret = "同上"
access_token = "同上"
access_token_secret = "同上"

auth = tweepy.OAuthHandler(consumer_key,consumer_secret)
auth.set_access_token(access_token,access_token_secret)

api = tweepy.API(auth)  # これを後で使います

##3. LINE通知用のオブジェクトを作成

先ほども引用した記事、『PythonでLINEにメッセージを送る』のコードをそのまま使用させていただきました。

.py

class LINENotifyBot:
    API_URL = 'https://notify-api.line.me/api/notify'
    def __init__(self, access_token):
        self.__headers = {'Authorization': 'Bearer ' + access_token}

    def send(
            self, message,
            image=None, sticker_package_id=None, sticker_id=None,
            ):
        payload = {
            'message': message,
            'stickerPackageId': sticker_package_id,
            'stickerId': sticker_id,
            }
        files = {}
        if image != None:
            files = {'imageFile': open(image, 'rb')}
        r = requests.post(
            LINENotifyBot.API_URL,
            headers=self.__headers,
            data=payload,
            files=files,
            )
       
access_toke_Notify = "ここにトークンを入力"
 
bot_Notify = LINENotifyBot(access_token=access_token_Notify)

これで、bot_Notify.send(message="xxxxx")とすれば、トークンの指定先にLINEが届きます。

##4. 自動検索の関数を作成

基本的なアイディアとしては、

「東京都、感染、{日付}日」のキーワードを含む最新のツイートを一定数取り出す
「n人」という表現があればその n を取得
取得したツイートが含む n のうち、最も多く登場するものを感染者数候補とする
その感染者数候補を含むツイートが一定の割合を超えた時に通知

のプロセスを一定時間おきに繰り返す、という感じです。

というわけで、最終的に実行する関数 auto_search がこちら。

.py

def auto_search(item=100,wait_time=60,rate=0.5):
    
    """
    item: 取り出すツイート数
    wait_time: 自動検索をかける時間間隔　単位 s
    rate: 感染者数と推定される数がツイートに含まれている割合
    """
   
    d = datetime.datetime.now().day
    m = datetime.datetime.now().month

    print("searching on Twitter...")
    pre_mode = 0  # 前にrateを超えた数値を記録するための変数

    while True:
        df = find_infected_num(d,item)  # "n人"のnをDataFrame型で返す関数
        num_mode = df.mode().values[0,0]  # dfの最頻値=感染者数候補 を取得
        count = df.groupby("num").size()  # nごとのツイート数の集計データ

        #num_modeの出現頻度がrateを超えている ＆ num_modeは新たに出現したもの
        if count.max() > item*rate and num_mode!=pre_mode:
            #結果の出力
            print("\n--RESULT--")
            print(count)

            #結果のLINE通知
            text = "{}月{}日\n東京都の感染者は[{}]人\n※ツイート割合は{:.2f}%".format(m,d,num_mode,count.max()/item*100)
            bot_Notify.send(message=text)  # LINEに送信

            #結果が不適切だった場合は継続できるようにする為の条件分岐
            if input("\ncontinue? y/n   ")=="n":
                break  # 終了

        waiting(2,wait_time,count)  # 待ち時間の間の表示用

        #pre_modeの更新
        if count.max() > item*rate:
            pre_mode = num_mode

find_infected_num は"n人"の n をDataFrameで返す為の関数です。
ここで、１で用意した tagger、及び２で用意した api を使っています。

.py

def find_infected_num(d,item):
    num_list = []  # nを格納するリスト
    for tweet in tweepy.Cursor(api.search, q=['感染',"東京都","{}日".format(d)]).items(item):
        split_tweet = tagger.parse(tweet.text).split()
        if "人" in split_tweet:
            index = split_tweet.index("人") - 1
            n = cut_number(split_tweet,index)  # "人"の直前の数字を返す関数
            num_list.append(n)
    return pd.DataFrame(num_list,columns=["num"])

ここに含まれる cut_number は "人" 直前の数字を取得する関数です。

.py

def cut_number(split_tweet,index):
    start_i = index  # ツイート内で、数字が始まった位置を表す変数
    
    # "人"の直前が数字のstr型でなかった場合は0を返す（10000は適当です）
    if not split_tweet[index] in list(map(str,range(0,10000))):
        return 0

    ans = split_tweet[start_i]  # "人"の直前の数字を取得
    while True:
        #順次数字が続く限りansの左側に付け加えていく
        if split_tweet[start_i-1] in list(map(str,range(0,9))):
            start_i -= 1
            ans = split_tweet[start_i] + ans
        #数字が終わったらansを返す
        else:
            return ans

なぜこんな関数が必要になるかを少し説明しておきます。
例えば、「今日の感染者は123人」という文があったとして、mecabによる分割をしてみると、

.py

tagger.parse("今日の感染者は123人").split()
# => ['今日', 'の', '感染', '者', 'は', '1', '2', '3', '人']

このように、1,2,3が区切られてしまいました。このままでは、感染者数の下一桁しか得られないので、正しく数値を取得するために cut_number を作成しました。

auto_search に出てくる関数としては他に waiting がありますが、これは次の自動検索までの残り時間を可視化してくれる関数です。
（本体の機能にはあまり関係がないのでオマケのようなものです。）

.py

def waiting(div,wait_time,count):
    clear_output()
    for i in range(1,wait_time//div+1):
        print("waiting: |"+"*"*i+" "*(wait_time//div-i)+"|")
        print("\n--RESULT--")
        print(count)
        time.sleep(div)
        clear_output()
    print("searching on Twitter...")

#使ってみる

アルゴリズムの性質上、確実に感染者数の速報をキャッチできるとは限らないので、実際に動かしてみてパラメータを調整しています。（上のコードでは既に調整した値を使っています）

なお、以下は7/19の実行結果になります。

####結果その１

（item=30で実行しています）

この時、LINEには以下の通知が届きました。

continue?の入力欄にyを入力すれば引き続き検索してくれますが、この日の東京都の発表感染者数は188人で合っているので大丈夫です。
（仮に引き続き検索させたとしても、pre_mode=188になっているので繰り返し通知が来ることはありません。）

####結果その２

（item=100で実行しています）

waitingの時はこのように表示されます。
RESULTを見ると分かりますが、50%を超えている候補がないので通知は来ません。
逆に、感染者数の発表から一定時間経つと正しい数字でも割合は下がることも分かります。

#今後の課題

テスト結果を受けて、今後の課題を挙げておきます。

感染者数の速報が出た瞬間の検知はまだできていない（7/20追記にて成功）
rate=0.5としているが、時間帯によっては誤った値を速報値として検出しかねない

これらの問題点については、引き続きこのプログラムを毎日走らせてテストしながら確認したいと思います。

###余談
感染者数の速報値については、わざわざ雑多なツイートから多数決を取らなくても、公式の報道機関のツイートなどを指定すれば取得できそう（しかも正確）ですが、どこが最速でツイートするかよく分からないのでこういう方法にしてみました。
ちなみに7/18にもテストしたのですが、東京都の感染者が290人という大きな数字だったこともあってか、発表直後は"290"のツイート割合が80%を超えていました。それを受けて7/19は初めrate=0.8として動かしていましたが、速報値が出ても通知が届かず失敗。rateを下げて調整した、という次第でした。
日によって話題になる度合いが違うのが難しいポイントですが、「多く呟かれた数字を拾う」という単純なアルゴリズムでどこまで正確な通知ができるようになるか、今後が楽しみです。
そういうわけで、実用向きというよりは、個人開発としての興味追求も含んでいるかもしれません笑

##7/20 追記

7/20にも動かしたところ、速報が出たタイミングでの通知に成功しました。

以下がLINEの画面。

そして、NHK公式の速報ツイートがこちら。

LINE通知が届いたのと同じ13:54です。一応これより数分早く速報値を出している報道機関も見つけましたが、誤作動を防ぐためにもrate=0.5として、数分の遅れは許容することにします。

そういうわけで、私は東京都の感染者数の確認にいちいちTwitterを開く必要がなくなりました。無事目標達成です！