求助解决Tweepy调用Twitter API时TooManyRequests:429错误

核心原因 #

Twitter学术研究API的search_all_tweets端点有严格限流规则：每15分钟最多调用100次，每次最多返回100条推文（即每15分钟上限10000条）。你的代码未处理限流等待逻辑，直接触发了429错误。

解决方案 #

1. 开启Tweepy自动限流等待（推荐）

Tweepy的Client内置了自动处理限流的机制，初始化时添加wait_on_rate_limit=True参数即可，触发限流时程序会自动等待到窗口重置后继续请求。

修改后的完整代码：

import tweepy
import config_academic
import pandas as pd

# 初始化客户端，开启自动限流等待
client = tweepy.Client(
    bearer_token=config_academic.ACADEMIC_BEARER_TOKEN,
    wait_on_rate_limit=True
)
# 修正搜索query：去掉from:后的空格，否则搜索无效
query = 'from:elonmusk'

start_time = "2010-06-29T00:00:00Z"
end_time = "2022-06-29T00:00:00Z"

all_tweets = []
# 迭代获取所有推文，指定需要的字段方便后续分析
for tweet in tweepy.Paginator(
    client.search_all_tweets,
    query=query,
    max_results=100,
    start_time=start_time,
    end_time=end_time,
    tweet_fields=['created_at', 'text', 'public_metrics']
).flatten():
    # 整理推文数据为字典，便于存入DataFrame
    tweet_data = {
        'tweet_id': tweet.id,
        'created_at': tweet.created_at,
        'content': tweet.text,
        'retweet_count': tweet.public_metrics['retweet_count'],
        'like_count': tweet.public_metrics['like_count']
    }
    all_tweets.append(tweet_data)
    
    # 每积累1000条就保存一次临时文件，防止数据丢失
    if len(all_tweets) % 1000 == 0:
        pd.DataFrame(all_tweets).to_csv('elonmusk_tweets_temp.csv', index=False)
        print(f"已保存{len(all_tweets)}条推文")

# 最终保存全部数据
pd.DataFrame(all_tweets).to_csv('elonmusk_all_tweets.csv', index=False)
print("所有推文获取完成，已保存到elonmusk_all_tweets.csv")

2. 手动处理限流（备选方案）

如果自动等待机制失效，可通过捕获HTTP错误，结合响应头的限流重置时间手动控制等待：

import tweepy
import config_academic
import time
import pandas as pd
from requests.exceptions import HTTPError

client = tweepy.Client(bearer_token=config_academic.ACADEMIC_BEARER_TOKEN)
query = 'from:elonmusk'
start_time = "2010-06-29T00:00:00Z"
end_time = "2022-06-29T00:00:00Z"

all_tweets = []
page_token = None

while True:
    try:
        response = client.search_all_tweets(
            query=query,
            max_results=100,
            start_time=start_time,
            end_time=end_time,
            next_token=page_token,
            tweet_fields=['created_at', 'text', 'public_metrics']
        )
        if response.data:
            for tweet in response.data:
                all_tweets.append({
                    'tweet_id': tweet.id,
                    'created_at': tweet.created_at,
                    'content': tweet.text,
                    'retweet_count': tweet.public_metrics['retweet_count'],
                    'like_count': tweet.public_metrics['like_count']
                })
        
        page_token = response.meta.get('next_token')
        if not page_token:
            break
        
        # 每次请求后短等待，降低限流触发概率
        time.sleep(1)
    
    except HTTPError as e:
        if e.response.status_code == 429:
            # 获取限流重置时间，计算等待时长
            reset_timestamp = int(e.response.headers.get('x-rate-limit-reset', time.time() + 900))
            wait_seconds = reset_timestamp - time.time() + 10  # 多等10秒确保安全
            print(f"触发限流，将等待{int(wait_seconds)}秒")
            time.sleep(wait_seconds)
        else:
            # 其他错误直接抛出
            raise

# 保存最终数据
pd.DataFrame(all_tweets).to_csv('elonmusk_all_tweets.csv', index=False)
print("所有推文获取完成")

关键注意事项 #

• 修正搜索query：原代码中from: elonmusk的空格会导致搜索失效，必须改为from:elonmusk。
• 增量保存数据：获取全部推文耗时较长，定期保存临时文件可避免程序中断丢失数据。
• 按需获取字段：通过tweet_fields指定需要的字段（如创建时间、互动数据），减少冗余传输，也方便后续Pandas分析。
• 权限确认：确保你的学术API密钥已开通search_all_tweets端点的访问权限，该端点仅对学术研究用户开放。

内容的提问来源于stack exchange，提问作者Alex Günsberg