はじめに

一身上の都合で、緯度・経度からなる手軽なGIS（Geographic Information System）データが欲しいという状況になった。そこで注目したのが、Twitter の位置情報付きのつぶやきである。あるキーワードに該当するような位置情報付きのつぶやきを集めれば、それは立派なGISデータになる。そこで、Twitter API に登録し、できる限り無料枠内で tweet の情報を集める方法を紹介したいと思う。

Twitter API の認証情報の取得

こちらについては、すでに既存の記事に詳しい方法が載っているので、そちらを参考にしてほしい。

上記の記事にもあるが、Tweet の取得に必要なものは、

API Key
API Key Secret
Access Token
Access Token Secret

の四つである。

tweepy のインストール

Twitter API を Python で扱うためのモジュールが tweepy である。インストール方法を以下に示す。公式ドキュメントにあるのは pip と GitHub 経由のインストールであるが、conda 経由でもインストール可能であった。

pip で行う場合

$ pip install tweepy

GitHub 経由で行う場合

$ git clone https://github.com/tweepy/tweepy.git
$ cd tweepy
$ pip install .

conda で行う場合

$ conda install tweepy -c conda-forge

無料枠で位置データを取得するスクリプト

あるキーワードのある１日間のつぶやきを自動取得するスクリプトを以下に示す。部分部分で解説を入れていく。

必要なモジュールのインポートと tweepy アクセス情報の設定

# 本スクリプトで使用するモジュール
import tweepy
from datetime import datetime, timedelta
import pandas as pd
import time

# 最初に取得した API の認証情報
api_key = "********************"
api_key_secret = "********************"
access_token = "********************"
access_token_secret = "********************"

# 認証情報を使用して API にアクセス
auth = tweepy.OAuthHandler(api_key, api_key_secret)
auth.set_access_token(access_token, access_token_secret)
api = tweepy.API(auth)

準備が整ったので、tweet を検索し、位置情報を取得し、データを作っていきたいのだが、当然無料枠では制限がある。制限としては

1回のアクセスで100件までしか取得できない。
15分間で180回のアクセスしか許されない。

の二つである。使ってみた雑感ではあるが、日本語のキーワードであれば、1分間に100件以下の tweet 数になり、扱いやすい。そこで、検索したい時間の1時間ごとに5分間のsleep を挟み、1分間ごとの tweet を取得し、その中から位置情報のあるものだけを抜き取るスクリプトを作りたい。そのための関数を以下の様に定義した。

関数定義

# 開始時間から終了時間まで、１分ごとの時刻を返すジェネレータ
def date_range_1min(start, stop, step = timedelta(minutes=1)):
    current = start
    while current < stop:
        yield current
        current += step

# 開始時間から終了時間まで、１時間ごとの時刻を返すジェネレータ
def date_range_1hour(start, stop, step = timedelta(hours=1)):
    current = start
    while current < stop:
        yield current
        current += step

# 取得した tweets から位置情報のみを取り出しデータフレーム化する関数
def extract_latlong(tweets, df):
    # リストの tweets から一つの tweet を取り出す。
    for tweet in tweets:
        # 位置情報あるもののみを取り出す。 
        if (tweet.place != None):
            # 各種情報の抜き出し
            created_dt = tweet.created_at # tweet 日時
            tweet_id = tweet.id # tweet id
            tweet_text = tweet.text # tweet 内容

            # ごくまれに位置情報の入れ物のみあり、中身が無い場合がある。それを回避する。
            if (tweet.place.bounding_box == None): 
                print("Data is None")
                continue

            # 位置情報の取得と加工（四角形で与えられるので中心座標を計算）
            box = tweet.place.bounding_box.coordinates
            lat = (box[0][0][1] + box[0][1][1] + box[0][2][1] + box[0][3][1]) / 4
            long = (box[0][0][0] + box[0][1][0] + box[0][2][0] + box[0][3][0]) / 4
            
            # 取得した情報をデータフレームに変換
            _df = pd.DataFrame({"date":[created_dt], "id":[tweet_id], "text":[tweet_text], "lat":[lat], "long":[long]})

            # これまでのデータフレームと結合
            df = pd.concat([df, _df], axis=0)
            
    return df

# JST のある date から +dt 分後までの key_word とつぶやかれた tweets を取得
def get_tweet(date, key_word, df, dt):
    # クエリの作成
    s = date
    e = date + timedelta(minutes=dt)
    query = (
        key_word
        + " since:"
        + s.strftime("%Y-%m-%d_%H:%M:%S_JST")
        + " until:"
        + e.strftime("%Y-%m-%d_%H:%M:%S_JST")
    )
    
    # tweetの取得
    try:
        tweets = api.search_tweets(count=100, q=query)
    except:
        # 15分180回の制限が過ぎるとエラーが出るので、15分間待ってから再取得。
        print("Error:TooManyRequests!")
        time.sleep(900)
        tweets = api.search_tweets(count=100, q=query)
    
    # tweet 件数の表示
    print(len(tweets))
    
    # 100件以上の tweet 数だと 100 件までしか取得できてないので、時間を分割し再度実行。（再帰的に get_tweet を実行している。）
    if (len(tweets) == 100):
        print("Warning!:")
        df = get_tweet(s, key_word, df, dt/2)
        s2 = s + timedelta(minutes=dt/2)
        df = get_tweet(s2, key_word, df, dt/2)
    else:
        # 100件未満の場合、位置情報を取得する。
        df = extract_latlong(tweets, df)

    return df

関数が準備できたので、実際に tweet を取得する部分を書いていく。検索したい期間の1時間ごとに5分間の sleep を挟み、1分間ごとの tweet を取得するようにスクリプトを書く。

# 情報を収集するためのデータフレームを定義
df = pd.DataFrame(columns=["date", "id", "text", "lat", "long"])

# 検索の開始日時と終了日時
s = datetime(2022, 5, 25, 0, 0, 0)
e = datetime(2022, 5, 26, 0, 0, 0)

# 検索するキーワード
key_word = "ビール"

# 以下、実際の tweet の取得
for s_a in date_range_1hour(s, e):
    e_a = s_a + timedelta(hours=1)
    print(s_a, e_a)
    
    for date in date_range_1min(s_a, e_a):
        df = get_tweet(date, key_word, df, 1.0)
        
    time.sleep(300)

これで、dfに必要なデータが書き込まれるので、それを保存すれば立派な GIS データである。

出力

# （あまり意味ないが）インデックスを整理
df.reset_index(drop=True, inplace=True)

# CSV に書き出す。
df.to_csv("./20220525_20220526_ビール.csv", index=False)

20220525_20220526_ビール.csv の中身の一部をお見せすると、

となる。検索をかけている時刻は JST だが、date に記載されているのは UTC であることを考慮する必要がある。

おわりに

これであるキーワードに基づいたGISデータの作成をすることができた。（つぶやきなので厳密な意味での地理データではないかもしれないが、広義の意味での「位置情報を伴ったデータ」という意味で考えてほしい。）次はこのデータを用いた可視化手法について記事にしたいと思う。