tweepyを使わず、Pythonの標準モジュールであるrequestsを使って直接Twitter APIのエンドポイントに接続してツイートを取得する方法を紹介します。
個人的には、tweepyでラップされている書き方よりも、こちらの方が公式ドキュメントに即して書けるのでおすすめです。
今回のツイートを検索するエンドポイントに関する公式ドキュメントはこちらです。
なお、この記事では、Twitter APIがすでに使える状態になっていることを想定しています。
必要なモジュールをインポート
from datetime import datetime #日付データを扱う
import pandas as pd #DataFrameを扱う
import requests #HTTP通信をする
import json #JSONデータを使用する
import time #時間を扱う
検索条件を指定
- 検索キーワードの作り方はこちらを参照してください。
- 期間の
start_timeは最大でも1週間前を指定してください。Twitter APIの仕様なのでしょうがないです。ちなみにこれ以上検索したい場合はAcademic Research Accessを申請してフルアーカイブサーチを使う必要があります。
# 検索キーワード
key_word="Python OR python -is:retweet"
# 期間
start_time="2023-01-15T00:00:00.000Z"
end_time="2023-01-16T00:00:00.000Z"
Bearer Tokenを指定
APIの認証（OAtuth 2.0認証）に必要となるトークンです。Twitter Developer Portalより取得できます。
bearer_token = ""
ここまでは各自で入力が必要ですが、この後はコピペでいけます。
リクエストを送るのに必要な情報を整理
-
search_urlが今回仕様するエンドポイントです。 -
max_resultsは100が最大です。 -
tweet.fieldsにはツイートの内容以外に取得したい情報を指定します。詳しくは公式ドキュメント参照。
search_url = "https://api.twitter.com/2/tweets/search/recent"
headers={'Authorization':f"Bearer {bearer_token}"}
query_params = {
"query":key_word,
"start_time":start_time,
"end_time":end_time,
"max_results":100,
"tweet.fields": ["created_at,author_id"],
}
APIのリクエストを送って結果を取得する
いよいよリクエストを送っていきます。
Twitter APIは複数のページにわたって検索結果がまとまっているのですが、最初のレスポンスではその内最初のページ（max_resultで指定した100件以内分）しか返してくれません。
次のページを取得するには、レスポンス内にあるnext_tokenをパラメータに含めて再度リクエストを送り直す必要があります。
以下ではその処理をWhile文で実装しています。
また、Twitter APIは15分で150requestsの制限があり、それを超えるとエラーを返してくるので、それを防ぐためにcountでリクエスト回数を記録して、140に達したら15分ほど休ませています。
data_list = [] #ここに結果を格納していきます
count = 0 #リクエストを何回送ったか
next_token = None
while True:
query_params["next_token"] = next_token
response = requests.get(search_url, headers=headers, params=query_params)
if response.status_code != 200: #エラーの時に詳細を表示
raise Exception(response.status_code, response.text)
res_json = response.json()
data_list.extend(res_json["data"])
print(res_json["meta"])
if len(res_json["meta"]) == 3: #next_tokenがなくなったら
break
else:
next_token = res_json["meta"]["next_token"]
count += 1
if count == 140: #150でerror
print('waiting')
time.sleep(60 * 15)
count = 0
結果をDataFrameに変換してcsvで保存
カレントディレクトリに検索語と期間の名前がついたcsvファイルで保存されます。
要らなそうな列を落として日付の昇順に直しています。
Twitter APIは直近からさかのぼって取得していくのでデフォルトだと降順になっています。
df = pd.DataFrame(data_list).drop(columns=["edit_history_tweet_ids","id"]).sort_values("created_at")
period = f"{pd.to_datetime(start_time).strftime('%Y%m%d')}_{pd.to_datetime(end_time).strftime('%Y%m%d')}"
filename = f"./{key_word}_{period}.csv"
filename = filename.replace(' ','')
df.to_csv(filename)