Tweepy StreamingClient添加用户列表监听实时推文报400错误如何解决

你的代码返回400是拼写错误+API参数理解错误导致的：

• 变量名拼写错误：定义token时写的是brearer_token，实际传入StreamingClient时用的是bearer_token，变量名不匹配会直接触发鉴权失败
• 参数完全用错：filter()方法的user_fields参数作用是指定接口返回的用户数据要携带哪些附加字段（比如用户简介、所在地、粉丝数等），和指定监听用户没有任何关系。

Twitter API v2的流式监听是通过匹配规则实现的，要监听指定用户的推文，需要先给流添加格式为from:用户数字ID的匹配规则，单条规则最多支持绑定500个用户ID，刚好适配你500+用户的监听需求，超过500个拆分多条规则即可。

import tweepy

bearer_token = "替换成你自己的Bearer Token"

class TweetListener(tweepy.StreamingClient):
    def on_data(self, raw_data):
        print(raw_data)
        # 如果要直接拿到解析好的推文对象，可以重写on_tweet方法：
        # def on_tweet(self, tweet):
        #     print(tweet.text, tweet.author_id)
        return super().on_data(raw_data)

if __name__ == "__main__":
    stream = TweetListener(bearer_token=bearer_token)
    # 第一步：清空之前残留的旧规则，避免重复匹配
    existing_rules = stream.get_rules().data
    if existing_rules:
        stream.delete_rules(rule_ids=[rule.id for rule in existing_rules])
    
    # 替换成你要监听的用户数字ID列表，注意不能是@用户名，必须是数字ID
    user_list = ['123456', '789012'] # 你的500+用户ID放这
    # 每500个用户拆成一组，拼匹配规则
    chunk_size = 500
    for i in range(0, len(user_list), chunk_size):
        chunk = user_list[i:i+chunk_size]
        rule = tweepy.StreamRule(" OR ".join([f"from:{uid}" for uid in chunk]))
        stream.add_rules(rule)
    
    # 启动流，按需传参指定要返回的附加字段即可
    stream.filter(
        # 需要返回用户用户名、简介时才配置user_fields
        # user_fields=["username", "description"],
        # 需要返回推文发布时间、互动数据时才配置tweet_fields
        # tweet_fields=["created_at", "public_metrics"]
    )

• 必须使用用户的数字ID作为监听依据，不能直接传@后面的昵称，如果只有昵称，需要先调用get_users(usernames=昵称列表)接口批量转换为数字ID
• 普通开发者账号的流式规则上限为25条，按单条规则500个用户计算，最多可同时监听12500个账号，完全覆盖500+用户的需求
• 启动filter()时如果不需要额外的返回字段，不需要传多余参数，不要把用户列表传给user_fields

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