如何用Python的requests实现Twitter API全量搜索分页？

原代码的问题 #

• connect_to_endpoint函数未返回JSON响应，导致main函数无法获取分页所需的next_token
• while循环仅更新了next_token参数，但未重新调用接口获取下一页数据
• 直接追加写入JSON对象会导致最终文件格式无效（多个顶级JSON对象不符合规范）

修复后的完整代码 #

# Extended script for full archive search with Twitter academic API
# based on a sample provided by Twitter
import requests
import json

# 替换为你的Bearer Token
bearer_token = "MYTOKEN"

search_url = "https://api.twitter.com/2/tweets/search/all"

query_params = {
    'query': '#WEURO2022',
    'tweet.fields': 'author_id,conversation_id,created_at',
    'expansions': 'geo.place_id',
    'place.fields': 'contained_within,country,country_code,full_name,geo,id,name,place_type',
    'user.fields': 'created_at,description,entities,id,location,name',
    'start_time': '2022-02-15T00:00:01.000Z',
    'end_time': '2022-09-16T23:59:59.000Z',
    'max_results': '500'
}

def bearer_oauth(r):
    r.headers["Authorization"] = f"Bearer {bearer_token}"
    r.headers["User-Agent"] = "v2FullArchiveSearchPython"
    return r

def connect_to_endpoint(url, params):
    response = requests.get(url, auth=bearer_oauth, params=params)
    if response.status_code != 200:
        raise Exception(f"请求失败: {response.status_code} - {response.text}")
    print("请求成功，获取到数据")
    return response.json()

def save_tweets(all_data):
    # 将所有数据写入JSON文件，保证格式有效
    with open('C:\\Users\\####\\Downloads\\MyTweets.json', 'w', encoding='utf-8') as json_f:
        json.dump(all_data, json_f, ensure_ascii=False, indent=2)
    print("所有数据已成功写入文件")

def main():
    all_tweets = []
    next_token = None
    
    while True:
        # 每次循环更新参数中的next_token
        if next_token:
            query_params['next_token'] = next_token
        
        json_response = connect_to_endpoint(search_url, query_params)
        # 收集当前页的推文数据
        if 'data' in json_response:
            all_tweets.extend(json_response['data'])
        
        # 检查是否有下一页
        next_token = json_response['meta'].get('next_token')
        if not next_token:
            break
        print(f"准备获取下一页，next_token: {next_token}")
    
    # 保存所有收集到的数据
    save_tweets(all_tweets)

if __name__ == "__main__":
    main()

关键修改说明 #

• 让connect_to_endpoint返回JSON响应，供main函数处理分页逻辑
• 使用while True循环，通过next_token的存在与否判断是否继续请求
• 先收集所有推文数据，再一次性写入文件，避免JSON格式错误
• 使用dict.get()方法安全获取next_token，避免因字段不存在引发报错

新手入门建议 #

• 先吃透Twitter API官方文档里的分页规则：每次请求返回的meta字段中，next_token是下一页的唯一凭证，当该字段不存在时，说明已获取完所有数据
• 处理API响应时，优先检查data字段是否存在，避免因无数据返回导致报错
• 注意API调用频率限制，学术版API有对应的请求次数上限，不要短时间内频繁请求
• 可以尝试按行写入单条推文（每条占一行），这样即使中途中断，已获取的数据也能正常读取

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