Python调用GET statuses/user_timeline API获取推文数量不足的问题咨询
嘿,这个问题我之前也碰到过!其实是因为statuses/user_timeline API默认每次只返回20条,而且要通过分页机制才能拿到最多3200条推文。我给你梳理下具体的解决步骤和代码示例:
核心思路
要拿到更多推文,你需要抓住两个关键:
- 每次请求都把
count参数设为200(这是该API单次请求的最大返回条数,默认是20),最大化单次请求的数据量。 - 使用
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
相关产品推荐
相关产品推荐

