如何使用Twitter API V2实现推文流？转换V1代码的方法

问题 #

想基于Twitter API V2实现推文流功能，但参考的教程都是API V1的。查官方文档知道要改用tweepy.StreamingClient，但找不到可用的实现示例，需要把原V1代码转换为V2兼容版本，同时附上了自己尝试的V2代码。

原V1代码 #

import tweepy

# API KEYS
bearer_token = ""
consumer_key = ''
consumer_secret = ''
access_token = ''
access_token_secret = ''

# 授权
client = tweepy.Client(consumer_key= consumer_key, consumer_secret= consumer_secret,access_token= access_token,access_token_secret= access_token_secret)

# 无法正常工作的监听类
class Listener(tweepy.Stream):
  
    def on_status(self, status):
        print(status)
        return True


twitter_stream = Listener(
    consumer_key, consumer_secret,
    access_token, access_token_secret
)

twitter_stream.filter(track=['#Bitcoin'])

尝试的V2代码 #

class MyStream(tweepy.StreamingClient):
    # 流连接成功时触发
    def on_connect(self):
        print("Connected")

    def on_tweet(self, tweet):
       print(tweet.data) 
       return


stream = MyStream(bearer_token=bearer_token)
stream.filter(track=['#Bitcoin'])

修正后的可用V2代码 #

import tweepy

# API V2流处理仅需Bearer Token
bearer_token = "你的Bearer Token"

class MyStream(tweepy.StreamingClient):
    def on_connect(self):
        print("已连接到Twitter推文流")

    def on_tweet(self, tweet):
        # 打印推文内容，如需更多字段可扩展
        print(f"推文: {tweet.text}")

    def on_errors(self, errors):
        # 捕获并打印错误信息
        print(f"流错误: {errors}")

# 初始化流客户端
stream = MyStream(bearer_token=bearer_token)

# 添加过滤规则：追踪#Bitcoin标签
# 规则名称可自定义，方便后续管理
stream.add_rules(tweepy.StreamRule(value="#Bitcoin", tag="追踪比特币标签"))

# 启动流，可指定需要返回的额外字段（如创建时间、作者ID）
stream.filter(tweet_fields=["created_at", "author_id"])

# 如需移除规则，可先获取规则ID再删除：
# rules = stream.get_rules()
# if rules.data:
#     rule_ids = [rule.id for rule in rules.data]
#     stream.delete_rules(rule_ids)

关键改动说明 #

• 密钥简化：API V2的流处理不需要consumer key、access token等，仅需Bearer Token即可
• 规则管理：V2不再支持直接通过filter的track参数指定关键词，必须先调用add_rules添加过滤规则
• 字段控制：通过tweet_fields参数可以指定需要返回的额外字段，默认仅返回推文ID和文本
• 错误处理：新增on_errors方法捕获流中的错误，便于调试和问题排查

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