Python Tweepy流初始化报错求助：代码问题排查与修复

问题描述 #

我在Python中使用Tweepy初始化流处理代码时遇到错误，代码如下：

class MyStreamListener(tweepy.Stream):
    def __init__(self, consumer_key, consumer_secret, access_token, access_token_secret, asset_name):
        auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
        auth.set_access_token(access_token, access_token_secret)
        super().__init__(auth=auth)
        self.asset_name = asset_name

    def on_status(self, status):
        tweet = status.text
        if self.asset_name.lower() in tweet.lower():
            tweet = preprocess_tweet(tweet)
            sentiment = analyze_sentiment(tweet)
            store_tweet(tweet, sentiment)

# Pass OAuth details to tweepy's OAuth handler
auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_token_secret)

# Ask user to input asset name to filter tweets by
asset_name = input("Enter the asset name to filter tweets by: ")

# Set up streaming connection to Twitter API
myStreamListener = MyStreamListener(consumer_key, consumer_secret, access_token, access_token_secret, asset_name)
myStream = tweepy.Stream(auth=myStreamListener.auth, listener=myStreamListener)

# Filter tweets by asset name
myStream.filter(track=[asset_name])

收到的错误信息：

TypeError: init() got multiple values for argument 'auth'

错误原因 #

问题出在Tweepy版本适配和类继承逻辑的混淆：

1. MyStreamListener继承了tweepy.Stream，而tweepy.Stream的构造函数本身已包含auth参数，后续代码又重复创建tweepy.Stream实例并传入auth，导致参数重复传递。
2. 代码混淆了Tweepy旧版（v3.x）和新版（v4+）的API用法：新版Stream类已整合监听器功能，无需额外创建Stream实例绑定监听器。

修复思路 #

方案1：适配Tweepy v4+版本 #

既然MyStreamListener已继承tweepy.Stream，直接用该子类实例启动流即可，无需额外创建tweepy.Stream对象：

class MyStreamListener(tweepy.Stream):
    def __init__(self, consumer_key, consumer_secret, access_token, access_token_secret, asset_name):
        auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
        auth.set_access_token(access_token, access_token_secret)
        super().__init__(auth=auth)
        self.asset_name = asset_name

    def on_status(self, status):
        tweet = status.text
        if self.asset_name.lower() in tweet.lower():
            tweet = preprocess_tweet(tweet)
            sentiment = analyze_sentiment(tweet)
            store_tweet(tweet, sentiment)

# Ask user to input asset name to filter tweets by
asset_name = input("Enter the asset name to filter tweets by: ")

# 直接使用子类实例启动流
myStream = MyStreamListener(consumer_key, consumer_secret, access_token, access_token_secret, asset_name)
myStream.filter(track=[asset_name])

方案2：适配Tweepy v3.x版本 #

若使用旧版Tweepy，需让监听器继承tweepy.StreamListener，再配合tweepy.Stream实例化：

class MyStreamListener(tweepy.StreamListener):
    def __init__(self, asset_name):
        super().__init__()
        self.asset_name = asset_name

    def on_status(self, status):
        tweet = status.text
        if self.asset_name.lower() in tweet.lower():
            tweet = preprocess_tweet(tweet)
            sentiment = analyze_sentiment(tweet)
            store_tweet(tweet, sentiment)

# Pass OAuth details to tweepy's OAuth handler
auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_token_secret)

# Ask user to input asset name to filter tweets by
asset_name = input("Enter the asset name to filter tweets by: ")

# Set up streaming connection to Twitter API
myStreamListener = MyStreamListener(asset_name)
myStream = tweepy.Stream(auth=auth, listener=myStreamListener)

# Filter tweets by asset name
myStream.filter(track=[asset_name])

额外注意事项 #

• 执行pip show tweepy确认版本，选择对应适配写法。
• 确保preprocess_tweet、analyze_sentiment、store_tweet三个自定义函数已正确实现。

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