You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何修复Tweepy报错:TypeError: StreamingClient.add_rules()缺少参数'add'

问题:Tweepy流式获取推文时出现TypeError: StreamingClient.add_rules() missing 1 required positional argument: 'add'错误

运行代码时触发错误:

TypeError: StreamingClient.add_rules() missing 1 required positional argument: 'add'

用户参考的官方示例代码:

streaming_client.add_rules(tweepy.StreamRule("Tweepy"))
streaming_client.filter()

用户的原始代码:

from multiprocessing.connection import Listener
import tweepy
from dotenv import load_dotenv
import os

load_dotenv()

# twitter api consumer key, consumer secret, access token, access secret.
btoken=os.getenv('btoken')

streamingClient = tweepy.StreamingClient(f"{btoken}")

class client(tweepy.StreamingClient):
    
    def on_data(self, data):
        print(data.text)
        return(True)

    def on_error(self, status):
        print(status)
        # Stops stream if rate limit is reached (status code 420)
        if status == 420:
            return False
        if status == 401:
            print("Authentication Error")
            return False

# Placeholder tracking keyword
trackKey = "walmart"
    
rule = tweepy.StreamRule((f"{trackKey} lang:en -is:retweet place_country:US"))

client.add_rules(rule)

client.filter

使用版本:tweepy v4.10.1


错误原因及修复方案

核心错误

你直接调用了类client的方法,而非类的实例。add_rules是实例方法,需要通过子类的实例来调用,直接操作类会导致方法缺少必要的实例参数(错误提示中的add实际是方法定义里的self参数)。

同时代码还存在其他问题:

  • 定义了streamingClient实例但未使用
  • on_data方法中data是JSON字符串,无法直接访问data.text
  • client.filter仅引用方法,未执行调用(缺少括号)

修正后的代码

import tweepy
from dotenv import load_dotenv
import os

load_dotenv()

btoken = os.getenv('btoken')

# 自定义StreamingClient子类
class CustomStreamingClient(tweepy.StreamingClient):
    def on_tweet(self, tweet):
        # 使用on_tweet方法直接获取推文对象,无需手动解析JSON
        print(tweet.text)
        return True

    def on_error(self, status):
        print(status)
        if status == 420:
            return False
        if status == 401:
            print("Authentication Error")
            return False

# 创建子类的实例
streaming_client = CustomStreamingClient(bearer_token=btoken)

# 定义跟踪规则
trackKey = "walmart"
rule = tweepy.StreamRule(f"{trackKey} lang:en -is:retweet place_country:US")

# 通过实例添加规则
streaming_client.add_rules(rule)

# 启动流(执行filter方法)
streaming_client.filter(tweet_fields=["text"])

关键修正点

  1. 创建自定义子类的实例streaming_client,通过实例调用所有方法
  2. 使用tweepy.StreamingClient推荐的on_tweet方法替代on_data,简化推文内容获取
  3. 移除未使用的Listener导入
  4. 正确执行filter方法(添加括号)

内容的提问来源于stack exchange,提问作者Steven Allen

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.19 19:15:53