使用Twitter Streaming API仅获取指定用户原创推文的问题求助

核心问题列表 #

• on_status方法未触发的根本原因：你重写了StreamListener的on_data方法，Tweepy默认的on_data逻辑会解析原始数据并调用on_status，自定义on_data后如果不手动调用on_status，该方法永远不会执行。
• follow参数的原生特性：Twitter Streaming API的follow过滤规则本身就会返回四类内容：目标用户发布的推文、目标用户转发的内容、其他用户回复目标用户的内容、其他用户@目标用户的内容，原生不会只返回目标用户自己发的内容，必须自行过滤。
• on_status方法参数错误：Tweepy规定on_status仅接收status一个入参，你额外添加了accounts参数，就算方法被触发也会直接抛出参数不匹配的异常。同时你的判断逻辑status.user.id_str != accounts是字符串和列表对比，永远为真，所有内容都会被过滤。
• 逻辑冲突：代码中重复定义了stream_tweets函数，类外定义的版本错误使用了track参数（track是关键词匹配规则，传用户ID完全不生效），且on_exception中调用的是这个错误的外部函数，会导致异常后逻辑完全走偏。
• 冗余代码：fetched_tweets_filename属性中单独声明的topic变量无任何作用，属于冗余代码。

修复后的实现代码 #

import tweepy
import datetime

# 替换为你的API凭证
api_key = "你的API_KEY"
api_secret_key = "你的API_SECRET_KEY"
access_token = "你的ACCESS_TOKEN"
access_secret_token = "你的ACCESS_SECRET_TOKEN"

topic = "tweets"
# 替换为你要监听的用户ID字符串列表
accounts = ['user_id1', 'user_id2']

class TwitterStreamer():
    def __init__(self, accounts):
        # 把要监听的用户ID列表存到实例里，方便后续过滤
        self.target_accounts = set(accounts)

    def stream_tweets(self, topic):
        listener = StreamListener(topic, self.target_accounts)
        auth = tweepy.OAuthHandler(api_key, api_secret_key)
        auth.set_access_token(access_token, access_secret_token)
        stream = tweepy.Stream(auth, listener)
        # 用follow参数监听目标用户
        stream.filter(follow=list(self.target_accounts))


class StreamListener(tweepy.StreamListener):
    def __init__(self, file_prefix, target_accounts):
        self.prefix = file_prefix
        # 接收要监听的用户ID集合，判断的时候O(1)效率更高
        self.target_accounts = target_accounts
    
    @property
    def fetched_tweets_filename(self):
        date = datetime.datetime.now().strftime("%Y-%m-%d")
        return f"{self.prefix}_{date}.txt"    
    
    def on_status(self, status):
        # 只保留目标用户自己发布的内容
        if status.user.id_str not in self.target_accounts:
            return
        # 输出+存储符合要求的推文
        tweet_data = status._json
        print(tweet_data['text'])
        with open(self.fetched_tweets_filename, 'a', encoding='utf-8') as tf:
            tf.write(str(tweet_data) + '\n')
        return True
        
    def on_exception(self, exception):
        print('exception', exception)
        # 异常后重启流
        twitter_streamer.stream_tweets(topic)       

    def on_error(self, status_code):
        print(f"Error code: {status_code}")
        # 420是速率限制，断开连接避免被封禁
        if status_code == 420:
            return False

if __name__ == '__main__':
    twitter_streamer = TwitterStreamer(accounts)
    twitter_streamer.stream_tweets(topic)

修复说明 #

1. 删除了自定义的on_data方法，保留默认逻辑自动调用on_status，存储逻辑移到on_status里只处理符合要求的推文，减少不必要的IO操作。
2. 把目标用户ID列表传入StreamListener实例，修正on_status的参数和判断逻辑，仅保留发布者是目标用户的内容。
3. 删除了重复定义的外部stream_tweets函数，统一走类内的逻辑，避免规则冲突。
4. 新增on_error方法处理速率限制等异常情况，避免账号被平台限制。

内容的提问来源于stack exchange，提问作者Data_Science_110