今回はTwitterで過去全期間のツイートを対象にした検索が可能な「Search Tweets: Full-Archive API」をPythonを使って叩く方法についてまとめておきます。
基本的に過去全期間のツイートが検索できる「Search Tweets: Full-Archive API」はPremium区分のAPIなので課金(149$/月)しないと使うことができませんが、v1.1のAPIであれば少しだけ使用することができます。
ツイッターAPIの種類やツイート検索APIの種類・使い分けについては過去記事参照↓
Twitterで過去ツイートを収集するためのAPIの種類と使い分けについて
Contents
SandboxでFull-Archive APIをPythonで使ってみる
SandboxのFull-Archive APIを有効化する
https://developer.twitter.com/en/account/environments にアクセスします。v1.1であれば以下のような画面が表示されています。
これで適当なdev名を入れて有効化します、(このdev名は後で必要になります)
ちなみにsandbox時の「Full-Archive API」「30days-search API」それぞれのアクセスリソースについては以下で確認できます。
Search Tweets:Full-Archive APIの仕様
公式ドキュメント↓
まずはリクエストのヘッダーに乗せるパラメーターについては以下参照：
基本的にはパラメーターに乗せるのはこの4つだと思います。
| パラメーター | 概要 |
|---|---|
| query | 検索クエリ |
| maxResults | 検索上限数(sandboxはmax100、premiumは500) |
| next | ネクストページトークン |
| fromDate | 開始期間(例：201901311500) |
| toDate | 終了期間(例：201901311500) |
そして、queryに載せるTwitterの検索クエリのオペレーターについては下記参照
https://developer.twitter.com/en/docs/twitter-api/premium/search-api/guides/operators
詳しくはドキュメントに書いてありますが、オペレータを用いることで特定のアカウントのツイートのみやリツイートの除外、特定の国や地域に絞ったツイートなど、検索条件を付けたツイート検索が可能になります。
メモ：自分が過去に使ったオペレーターによる検索クエリ
例) 特定のアカウント(@hogehoge) の過去の全てのツイートを検索したい場合
keyword = 'hogehoge from:hogehoge' # リツイート・リプライ全てを含む keyword = 'twitter.com/hogehoge/status/ from:hogehoge -rt' # ツイートのみを取得 keyword = 'hogehoge -rt' # ツイートのみを取得
検索結果がmaxで指定した検索上限数以上の場合はAPIのレスポンスにnextpagetokenが発行されているので、それをパラメーターに組み込むことで続きの結果を取得することが可能です。
PythonでFull-Archive APIを叩くサンプルコード
サンプルコードは↓のようになります。
import json
from requests_oauthlib import OAuth1Session
import pandas as pd
import traceback
import time
# v1.1のAPI鍵(v2のものはNG)
CK ='xxx'
CS ='xxx'
AT ='xxx'
ATS ='xxx'
twitter = OAuth1Session(CK, CS, AT, ATS)
url = 'https://api.twitter.com/1.1/tweets/search/fullarchive/<devname>.json'
# Enedpointへ渡すパラメーター
keyword = 'コロナ ワクチン'
num = 0
params ={
'query' : keyword , # 検索キーワード
'maxResults': 100 , # 取得するtweet数
'fromDate' : 201901311500,
'toDate' : 202208121200
}
test_data = []
req = twitter.get(url, params = params)
if req.status_code == 200:
res = json.loads(req.text)
test = res['results'][0]
print(test['created_at'])
for tweet in res['results']:
account = tweet['user']['screen_name']
tweet_id = tweet['id']
try:
text_ = tweet['full_text']
except:
text_ = tweet['text']
# print('@'+account, tweet['created_at'], f'https://twitter.com/{account}/status/{tweet_id}', text_)
test_data.append(['@'+account, tweet['created_at'], f'https://twitter.com/{account}/status/{tweet_id}', text_])
else:
print("Failed: %d" % req.status_code)
print(req.text)
# データフレームに変換
test_data2 = pd.DataFrame(test_data, columns=["account", "date",'source', "tweet"])
# CSVに出力
test_data2.to_csv('sample_tweet.csv', encoding='shift-jis',errors='ignore')
2ぺージ目以降のデータが欲しい場合は以下のようにネクストページトークンを載せます。
params ={
'query' : keyword , # 検索キーワード
'maxResults': 100 , # 取得するtweet数
'next':res['next'],
'fromDate' : 201901311500,
'toDate' : 202208121200
}
※ ここの変数res[‘next’]はサンプルコードの変数resを想定しています。
コメント
[…] 参照：【Python】TwitterのAPIを使って過去全期間のツイートをスクレイピング […]