无法将Twitter API V2数据存储至MongoDB，求适配类实现方案

问题修正点 #

• 原代码on_error方法参数名错误：参数定义为tweet_code但判断时用status_code，导致变量未定义，逻辑失效。
• 缺少MongoDB连接初始化代码：tweet_collection实例必须先通过MongoDB客户端创建才能使用。
• 异常处理过于简陋：直接pass会掩盖存储失败的原因，建议捕获并打印异常信息。

完整可行实现代码 #

import tweepy
from pymongo import MongoClient

# 初始化MongoDB连接（替换为你的实际连接配置）
mongo_client = MongoClient("mongodb://localhost:27017/")
twitter_db = mongo_client["twitter_stream_db"]
tweet_collection = twitter_db["raw_tweets"]

class TwitterStreamSaver(tweepy.StreamingClient):
    def on_tweet(self, tweet):
        print(f"捕获推文ID: {tweet.id}")
        try:
            # 将Tweepy Tweet对象转为字典结构插入MongoDB
            tweet_dict = tweet._json
            tweet_collection.insert_one(tweet_dict)
            print("推文已成功存储至MongoDB")
        except Exception as e:
            print(f"存储推文失败: {str(e)}")

    def on_error(self, status_code):
        print(f"流请求错误: {status_code}")
        # 420错误代表速率限制，返回False断开流避免持续触发限制
        if status_code == 420:
            return False

# 初始化流客户端并启动采样流
bearer_token = "你的Twitter API V2 Bearer Token"
stream_client = TwitterStreamSaver(bearer_token)
stream_client.sample()

额外注意事项 #

• 先安装依赖：执行pip install tweepy pymongo确保环境依赖齐全。
• 确认API权限：你的Bearer Token必须拥有Twitter API V2的流访问权限，sample接口需要对应权限才能正常返回数据。
• 自定义过滤规则：如果需要抓取特定关键词/话题的推文，可在启动流前添加规则：

  # 添加关键词过滤规则
  stream_client.add_rules(tweepy.StreamRule("Python"))
  # 启动过滤流（替换sample方法）
  stream_client.filter(tweet_fields=["created_at", "author_id"])
  

• 异常日志优化：生产环境可将异常信息写入日志文件，而非仅打印到控制台。

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