PythonとTwitter API v2でリアルタイムにツイートを取得する

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()