概要
Twitter search APIを叩いて、ある期間の2hDTMを検索できないか試してました pic.twitter.com/cs8XagzAJz
— Kurene (@_kurene) 2020年4月29日
Twitter上のツイートやメディアを手元で一気に検索したいと思い、Twitter Search API を叩き、キーワードと期間を入力としてツイートを収集するスクリプトを作りました。
実装
※dateutil.parserをpip等でインストールしておく必要があります。
コード
import json import pickle import datetime from dateutil.parser import parse as dateutil_parser from requests_oauthlib import OAuth1Session # アクセストークン等を設定 config = { "CONSUMER_KEY": "...", "CONSUMER_SECRET": "...", "ACCESS_TOKEN": "...", "ACCESS_TOKEN_SECRET": "..." } # ツイート保存用関数 def pickle_dump(obj, path): with open(path, mode='wb') as f: pickle.dump(obj,f) # 保存したツイートをロードする関数 def pickle_load(path): with open(path, mode='rb') as f: data = pickle.load(f) return data class TwitterApi: def __init__(self, config): self.start_session(**config) # search api の url self.url_search = "https://api.twitter.com/1.1/search/tweets.json?tweet_mode=extended" def start_session(self, CONSUMER_KEY, CONSUMER_SECRET, ACCESS_TOKEN, ACCESS_TOKEN_SECRET): # OAuth認証 self.api = OAuth1Session(CONSUMER_KEY, CONSUMER_SECRET, ACCESS_TOKEN, ACCESS_TOKEN_SECRET) def search(self, search_word,# 検索ワード since=None, # ツイート期間の絞りこみ（いつからか、日本時間） until=None, # ツイート期間の絞りこみ（いつまでか、日本時間） max_tweets = 1000 #最大検索数 ): def print_tweet(tweet): print("====================================") print( tweet["user"].get("name"), "@", tweet["user"].get("screen_name") ) print( tweet.get("full_text").replace("\n", " ") ) print( "Date:", tweet["created_at"]) print( "FV:", tweet.get("favorite_count")) print( "RT:", tweet.get("retweet_count")) print() def perser(index, tweet, hours=9): #hours=9 で日本時間への変換となっている dt = dateutil_parser(tweet.get("created_at")) + datetime.timedelta(hours=hours) tweet["created_at"] = dt return tweet, dt # 日付関連の変数 since_dt = dateutil_parser(since+"+00:00") until_params = dateutil_parser(until+"+00:00").strftime('%Y-%m-%d_%H:%M:%S')+"_JST" max_api_call = max_tweets // 100 + 1 # APIを叩く回数 # 検索実行 tweets = [] max_id = None index = 0 params = {"q": search_word, "count": 100, "result_type": "recent", "exclude": "retweets", "until": until_params} for n in range(max_api_call): # APIを叩く（最大100件） req = self.api.get(self.url_search, params=params) if req.status_code == 200: #検索結果を得られた場合 #検索を取得 timeline = json.loads(req.text)["statuses"] if len(timeline) == 0: break for k, tweet in enumerate(timeline): max_id = int(tweet["id"]) - 1 # ツイートの日時を取得 tweet_mod, dt = perser(index, tweet) # ツイートが期間内のものか確認 if dt >= since_dt: tweets.append(tweet_mod) print_tweet(tweet_mod) index += 1 params["max_id"] = max_id else: max_id = None print() if max_id is None: break else: print("Error") return tweets if __name__=='__main__': # 1. インスタンス作成 twitter_api = TwitterApi(config) # 2. 検索を実行し、結果（各ツイート：辞書型）が格納されたリストを得る tweets = twitter_api.search( "検索ワードを与えてください", since="2020-05-01 00:00:00", until="2020-05-03 15:00:00" ) # 3. リストを保存 pickle_dump(tweets, "twitter_search_api.pickle") import code console = code.InteractiveConsole(locals=locals()) console.interact()
ポイント
OAuth認証
Twitter APIを利用するには、Twitter API のアカウント登録を行う必要があります。
英語での文章入力等、少々面倒な作業がありますが、Web上に情報があるため、ググりながらトライしてみてください。
ツイート数が101件以上の検索
APIを叩いて一度に取得できるツイートの最大数は100件です。
そのため、101件以上のツイートを取得するには複数回APIを叩く必要があります。
そこで、"max_id"をいうパラメタを指使います。すでに検索済みのツイートのidで最も小さいものを"max_id"にセットすることで、検索済みのツイートを除外することができます。
注意
以上のスクリプトで日時を指定して検索できますが、Twitter Search APIの仕様上、１週間以上前のツイートは検索できませんのでご了承ください。
長期間のデータを集める場合は、定期的にクロールする感じの使い方になりそうです。
