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

Python调用GET statuses/user_timeline API获取推文数量不足的问题咨询

嘿,这个问题我之前也碰到过!其实是因为statuses/user_timeline API默认每次只返回20条,而且要通过分页机制才能拿到最多3200条推文。我给你梳理下具体的解决步骤和代码示例:

核心思路

要拿到更多推文,你需要抓住两个关键:

  1. 每次请求都把count参数设为200(这是该API单次请求的最大返回条数,默认是20),最大化单次请求的数据量。
  2. 使用max_id参数实现分页:每次请求后,取返回的最后一条推文的ID减1,作为下一次请求的max_id,这样就能获取到比这条更早的推文,循环直到没有更多数据或者达到3200条的官方上限。
推荐实现:用Tweepy库(更省心)

Tweepy是Python中专门对接Twitter API的工具库,它封装了速率限制处理、分页逻辑等细节,用起来很方便:

首先安装Tweepy:

pip install tweepy

然后是完整代码:

import tweepy

# 替换成你的API密钥和令牌
API_KEY = "你的API_KEY"
API_SECRET = "你的API_SECRET"
ACCESS_TOKEN = "你的ACCESS_TOKEN"
ACCESS_TOKEN_SECRET = "你的ACCESS_TOKEN_SECRET"

# 完成认证流程
auth = tweepy.OAuthHandler(API_KEY, API_SECRET)
auth.set_access_token(ACCESS_TOKEN, ACCESS_TOKEN_SECRET)

# 初始化API对象,开启自动等待速率限制重置
api = tweepy.API(auth, wait_on_rate_limit=True)

target_screen_name = "你要获取推文的用户用户名"
all_tweets = []

# 第一次请求,获取最近的200条推文
new_tweets = api.user_timeline(screen_name=target_screen_name, count=200)
all_tweets.extend(new_tweets)

# 记录最后一条推文的ID(减1是为了避免重复获取同一条)
oldest_tweet_id = all_tweets[-1].id - 1

# 循环分页获取,直到没有更多推文或达到3200条上限
while len(new_tweets) > 0 and len(all_tweets) < 3200:
    print(f"正在获取早于ID {oldest_tweet_id}的推文...")
    new_tweets = api.user_timeline(
        screen_name=target_screen_name,
        count=200,
        max_id=oldest_tweet_id
    )
    all_tweets.extend(new_tweets)
    # 更新最新的旧推文ID
    oldest_tweet_id = all_tweets[-1].id - 1
    print(f"当前累计获取了 {len(all_tweets)} 条推文")

print(f"最终成功获取到 {len(all_tweets)} 条推文")
关键细节说明
  • wait_on_rate_limit=True:让Tweepy自动处理API的速率限制,触发限制时会自动等待到限制重置,不用自己写复杂的等待逻辑。
  • max_id减1:因为max_id会返回ID小于等于该值的推文,减1可以避免重复获取上一次请求的最后一条推文。
  • 3200条是官方设定的硬上限,即使用户发布了更多推文,也无法通过这个API获取到。
备选方案:用原生Requests库实现

如果不想依赖第三方库,也可以用Requests手动处理分页和速率限制:

import requests
import time

# 替换成你的Bearer Token(OAuth2认证方式)
BEARER_TOKEN = "你的BEARER_TOKEN"
headers = {"Authorization": f"Bearer {BEARER_TOKEN}"}

target_screen_name = "目标用户用户名"
all_tweets = []
max_id = None

while len(all_tweets) < 3200:
    params = {
        "screen_name": target_screen_name,
        "count": 200
    }
    if max_id is not None:
        params["max_id"] = max_id - 1

    response = requests.get(
        "https://api.twitter.com/1.1/statuses/user_timeline.json",
        headers=headers,
        params=params
    )

    # 处理速率限制
    if response.status_code == 429:
        reset_time = int(response.headers.get("x-rate-limit-reset", time.time() + 60))
        wait_seconds = reset_time - time.time()
        print(f"触发速率限制,等待 {wait_seconds:.1f} 秒...")
        time.sleep(wait_seconds + 1)
        continue
    # 处理其他请求错误
    elif response.status_code != 200:
        print(f"请求失败,状态码:{response.status_code}")
        break

    new_tweets = response.json()
    if not new_tweets:
        print("没有更多推文可以获取了")
        break

    all_tweets.extend(new_tweets)
    max_id = new_tweets[-1]["id"]
    print(f"当前累计获取 {len(all_tweets)} 条推文")

print(f"最终获取到 {len(all_tweets)} 条推文")

内容的提问来源于stack exchange,提问作者George Constantine

相关产品推荐
方舟 Agent Plan

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

最近更新时间:2026.05.15 03:43:52