tweepy4.0以降のリアルタイムのストリーム取得

はじめに

自分への備忘録として書きます。

tweepy 4.0以降でStreamListenerがStreamにマージされ、過去のサンプルコードが動かなくなったので、書き換えました。

環境

Python　3.10.1
tweepy　4.4.0

機能

自分のタイムラインからリアルタイムにツイートを取得して、"検索文"が含まれていた場合のみコンソールに表示する
#tweepy 4.0以降に対応したプログラム

test.py

import tweepy
import settings 

CK = settings.CONSUMER_KEY
CS = settings.CONSUMER_SECRET
AK = settings.ACCESS_TOKEN
AS = settings.ACCESS_TOKEN_SECRET
word = "検索文"
 
auth = tweepy.OAuthHandler(CK, CS)
auth.set_access_token(AK, AS)
api = tweepy.API(auth)
 
print('実行中')

class IDPrinter(tweepy.Stream):
    def on_status(self, status):
        print(status.id)
        print(status.user.screen_name)
        print(status.user.name)
        print(status.text)

printer = IDPrinter(CK,CS,AK,AS)
printer.filter(track=[word])
printer.sample()

tweepy 3.6頃までは以下のコードの書き方が公式サンプルとして紹介されていましたが、
4.0でStreamListenerがStreamにマージされたので、そのままコピーでは使えなくなりました。
そのため、上記ソースのように書き換えたところ動くようになりました。

old.py

class MyStreamListener(tweepy.StreamListener):

    def on_status(self, status):
        print(status.text)

myStreamListener = MyStreamListener()
myStream = tweepy.Stream(auth = api.auth, listener=myStreamListener)

3.6までのやり方
1.StreamListener を継承するクラスを作成する
2.そのクラスを使用して、 Stream オブジェクトを作成する
3.Stream を使用してTwitter APIに接続します