Twitter API v2提取推文报错403/401及定制提取功能实现求助

403 Client Forbidden报错 #

• 你当前使用的search/all端点仅对学术研究权限的开发者账号开放，标准开发者账号仅可调用search/recent端点（最多拉取最近7天的公开推文），将代码中的接口地址替换为https://api.twitter.com/2/tweets/search/recent即可
• 二次确认你使用的Token所属的应用，确实已经绑定到你创建的项目下，且项目已经开通Twitter API v2的访问权限

401 Invalid or expired token报错 #

• 检查你填写的Bearer Token是否有多余的空格、换行符，确保复制完全
• 若Token长时间未使用，可前往开发者后台对应应用的密钥页面重新生成Bearer Token，替换代码中的值即可

印度地区推文筛选 #

直接在查询规则中加入place_country:IN即可，IN是印度的ISO 3166-1 alpha-2国家代码，不需要额外配置其他参数。

拉取前一天全量推文 #

• 代码中加入自动计算前一天时间区间的逻辑，无需手动输入时间参数
• 新增分页循环逻辑，通过接口返回的next_token持续拉取数据，直到没有更多结果为止，注意不要超出你的账号请求配额，标准账号单分钟最多可发起450次请求，单次请求最多返回100条结果

可运行完整代码 #

首先安装依赖：

pip install requests python-dotenv

代码内容：

import os
import requests
from datetime import datetime, timedelta
from dotenv import load_dotenv

# 加载环境变量中的Token，也可以直接替换成你的Bearer Token
load_dotenv()
bearer_token = os.getenv("TWITTER_BEARER_TOKEN")

def create_headers(bearer_token):
    headers = {"Authorization": f"Bearer {bearer_token}"}
    return headers

def create_url(start_date, end_date, max_results = 100):
    search_url = "https://api.twitter.com/2/tweets/search/recent"
    # 无关键词，仅筛选印度地区的公开推文，需要指定语言可以加lang:xx规则
    query_params = {
        'query': 'place_country:IN',
        'start_time': start_date,
        'end_time': end_date,
        'max_results': max_results,
        'expansions': 'author_id,in_reply_to_user_id,geo.place_id',
        'tweet.fields': 'id,text,author_id,in_reply_to_user_id,geo,conversation_id,created_at,lang,public_metrics,referenced_tweets,reply_settings,source',
        'user.fields': 'id,name,username,created_at,description,public_metrics,verified',
        'place.fields': 'full_name,id,country,country_code,geo,name,place_type',
        'next_token': None
    }
    return (search_url, query_params)

def connect_to_endpoint(url, headers, params, next_token = None):
    params['next_token'] = next_token
    response = requests.request("GET", url, headers = headers, params = params)
    print(f"接口响应状态码: {response.status_code}")
    if response.status_code != 200:
        raise Exception(response.status_code, response.text)
    return response.json()

if __name__ == "__main__":
    headers = create_headers(bearer_token)
    # 自动计算前一天的时间区间，格式符合Twitter API要求
    end_time = (datetime.utcnow() - timedelta(days=1)).replace(hour=23, minute=59, second=59, microsecond=0).isoformat() + "Z"
    start_time = (datetime.utcnow() - timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0).isoformat() + "Z"
    max_results = 100

    url, params = create_url(start_time, end_time, max_results)
    all_tweets = []
    next_token = None

    # 分页拉取所有结果
    while True:
        json_response = connect_to_endpoint(url, headers, params, next_token)
        if 'data' in json_response:
            all_tweets.extend(json_response['data'])
        print(f"已拉取推文数量: {len(all_tweets)}")
        # 没有下一页则退出循环
        if 'next_token' not in json_response['meta']:
            break
        next_token = json_response['meta']['next_token']

    # 这里可以自行处理保存逻辑，比如存为JSON或者CSV
    print(f"拉取完成，共获取{len(all_tweets)}条印度地区前一天的推文")

你可以在项目根目录创建.env文件，写入TWITTER_BEARER_TOKEN=你的实际Token即可，也可以直接把代码中的bearer_token变量值替换为你的实际Token。

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