Twitter API v1.1のリアルタイムのツイート収集が使えなくなったので、API v2に移行した。PythonライブラリのTweepyとTwitter API v2でリアルタイムにツイート収集する方法をまとめておく。
環境
WSL2(Ubuntu20.04)。
$ lsb_release -dr Description: Ubuntu 20.04.5 LTS Release: 20.04 $ python3 -V Python 3.8.10 $ pip3 show tweepy Name: tweepy Version: 4.12.1 ...
Twitter API のBearer Tokenの取得
Twitter API v2でリアルタイムにツイートを収集するために、Twitterの開発者サイトでBearer Tokenを取得しておく。開発者サイトのProjects & Apps > Overviewの画面で、対象アプリの［Keys and tokens］（鍵のアイコン）を開く。この画面でBearer Tokenを生成できる。
Tweepyでリアルタイムツイート取得
TweepyでTwitter API v2を使ってリアルタイムにツイートを収集するには、StreamingClientクラスを使う。v1.1のとき（Tweepy4でリアルタイムツイートを取得する）と同様に、このクラスを継承したクラスを作成する。以下のコードでは、Twitter API v2のsampleメソッドを使って、リアルタイムに1%のツイートを出力する。
import tweepy
# Twitter APIの認証情報
# Twitterの開発者向けページで取得したBearer Tokenを使う
BEARER_TOKEN = 'Bearer Token'
class TwStream(tweepy.StreamingClient):
def on_tweet(self, tweet):
print('-------------------------')
print(tweet.data)
def main():
twstream = TwStream(BEARER_TOKEN)
twstream.sample()
if __name__ == '__main__':
main()
ただ、このコードで取得できるのはツイートのIDやテキストのみで、取得できるデータはv1.1のときと違ってかなり限定されている。v2では、取得する情報を指定する必要がある。
APIで取得対象のデータを指定する
Twitter API v2では、ExpansionsとFieldsを指定することで、さまざまなデータを取得できる。以下のコードでは、ExpansionsとTweetフィールド、Userフィールドを指定。
import tweepy
# Twitter APIの認証情報
# Twitterの開発者向けページで取得したBearer Tokenを使う
BEARER_TOKEN = 'Bearer Token'
class TwStream(tweepy.StreamingClient):
def on_tweet(self, tweet):
print('-------------------------')
print(tweet.data)
def main():
twstream = TwStream(BEARER_TOKEN)
# 取得するExpansions
expansions = ['author_id', 'referenced_tweets.id', 'attachments.media_keys']
# 取得するTweetフィールド
tweet_fields = ['author_id', 'created_at', 'attachments', 'lang', 'public_metrics']
# 取得するUserフィールド
user_fields = ['created_at', 'description', 'location', 'verified']
twstream.sample(expansions=expansions, tweet_fields=tweet_fields, user_fields=user_fields)
if __name__ == '__main__':
main()
上記のように取得データを指定できるが、v1.1のように言語の指定ができない。このままだと、英語やさまざまな言語のツイートが混ざって取得される。filterメソッドを使うと言語の指定ができるが、filterメソッドでは取得できるツイート数に制限があるので、大量のツイートを収集したい場合には使えない。日本語で絞りたい場合は次のようなコードになる。取得したTweetオブジェクトのlangフィールドを参照して、日本語（ja）のみに限定する。
import tweepy
# Twitter APIの認証情報
# Twitterの開発者向けページで取得したBearer Tokenを使う
BEARER_TOKEN = 'Bearer Token'
class TwStream(tweepy.StreamingClient):
def on_tweet(self, tweet):
if tweet.lang == 'ja':
print('-------------------------')
print(tweet.data)
def main():
twstream = TwStream(BEARER_TOKEN)
# 取得するExpansions
expansions = ['author_id', 'referenced_tweets.id', 'attachments.media_keys']
# 取得するTweetフィールド
tweet_fields = ['author_id', 'created_at', 'attachments', 'lang', 'public_metrics']
# 取得するUserフィールド
user_fields = ['created_at', 'description', 'location', 'verified']
twstream.sample(expansions=expansions, tweet_fields=tweet_fields, user_fields=user_fields)
if __name__ == '__main__':
main()
