求助：使用Twitter API v2时遇Tweepy无Stream属性的AttributeError

尝试通过Twitter API v2免费获取推文，但代码无法正常运行，反复触发属性错误。已搜索现有解决方案，但均无效。

错误信息 #

Traceback (most recent call last):
File "filepath\twitter.py", line 21, in
class TweetStreamListener(tweepy.Stream):
AttributeError: module 'tweepy' has no attribute 'Stream'

相关代码 #

import tweepy
import json
import csv
from kafka import KafkaProducer

api_key = 'x'
api_secret = 'y'
access_token = 'z'
access_token_secret = '&'

kafka_bootstrap_server = 'localhost:9092'
kafka_topic = 'twitter_data'

csv_output = 'filepath/twitter-data-from-api.csv'

kafka_producer = KafkaProducer(bootstrap_servers=kafka_bootstrap_server)

class TweetStreamListener(tweepy.Stream):
    def on_status(self, status):
        try:
            tweet_id = status.id
            tweet_text = status.text
            tweet_created = status.created_at.strftime('%Y-%m-%d %H:%M:%S')

            tweet_data = {
                'id': tweet_id,
                'text': tweet_text,
                'created at': tweet_created
            }

            kafka_producer.send(kafka_topic,
                                json.dumps(tweet_data).encode("UTF-8"),
                                api_version=(3, 5, 0))

            with open(csv_output, 'a', newline='', encoding='utf-8') as f:
                writer = csv.writer(f)
                writer.writerows([[tweet_id, tweet_text, tweet_created]])

        except Exception as e:
            print(f"Error: {str(e)}")

def extract_data():
    auth = tweepy.OAuthHandler(api_key, api_secret)
    auth.set_access_token(access_token, access_token_secret)
    api = tweepy.API(auth)

    twitter_stream_listener = TweetStreamListener(api.auth, None)
    stream = tweepy.Stream(auth=api.auth, listener=twitter_stream_listener)

    stream.filter(['Key1', 'Key2', 'Key3'])

    kafka_producer.close()

extract_data()

解决思路 #

1. 版本适配问题 #

你的代码是基于Tweepy v1.x的写法，但如果安装的是Tweepy v2+版本，tweepy.Stream已被移除：

• 针对Twitter API v2，需改用tweepy.StreamingClient
• 若要兼容旧版API v1.1，可使用tweepy.StreamV1

2. 适配API v2的代码修改示例 #

import tweepy
import json
import csv
from kafka import KafkaProducer

api_key = 'x'
api_secret = 'y'
access_token = 'z'
access_token_secret = '&'

kafka_bootstrap_server = 'localhost:9092'
kafka_topic = 'twitter_data'

csv_output = 'filepath/twitter-data-from-api.csv'

kafka_producer = KafkaProducer(bootstrap_servers=kafka_bootstrap_server)

class TweetStreamListener(tweepy.StreamingClient):
    def on_tweet(self, tweet):
        try:
            tweet_id = tweet.id
            tweet_text = tweet.text
            tweet_created = tweet.created_at.strftime('%Y-%m-%d %H:%M:%S')

            tweet_data = {
                'id': tweet_id,
                'text': tweet_text,
                'created at': tweet_created
            }

            kafka_producer.send(kafka_topic,
                                json.dumps(tweet_data).encode("UTF-8"),
                                api_version=(3, 5, 0))

            with open(csv_output, 'a', newline='', encoding='utf-8') as f:
                writer = csv.writer(f)
                writer.writerow([tweet_id, tweet_text, tweet_created])

        except Exception as e:
            print(f"Error: {str(e)}")

def extract_data():
    # API v2需使用Bearer Token，在Twitter开发者平台生成
    bearer_token = '你的Bearer Token'
    stream_client = TweetStreamListener(bearer_token=bearer_token)

    # 添加关键词跟踪规则
    rule = tweepy.StreamRule("Key1 OR Key2 OR Key3")
    stream_client.add_rules(rule)

    # 启动流并指定所需字段
    stream_client.filter(tweet_fields=["created_at"])

    kafka_producer.close()

extract_data()

3. 关键注意事项 #

• API v2必须使用Bearer Token认证，需在Twitter开发者平台的项目中生成
• 回调方法从on_status改为on_tweet，推文字段结构与v1不同
• 启动流时需显式指定需要的字段（如tweet_fields=["created_at"]），默认仅返回id和text
• 若坚持使用旧版Stream类，可安装Tweepy v1.x：pip install tweepy==1.10.0，但该版本仅支持API v1.1，免费访问限制更多

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