Twitter API限额下批量拉取推文指标及定时存储实现咨询

执行以下命令安装所需依赖：
pip install pandas requests schedule

修改后完整代码 #

import json
import pandas as pd
import requests
import schedule
import time
from datetime import datetime

# 按固定大小拆分列表的工具函数
def split_list(lst, chunk_size):
    return [lst[i:i+chunk_size] for i in range(0, len(lst), chunk_size)]

# 接收一批ID生成对应请求URL
def create_url(ids_batch):
    tweet_fields = "tweet.fields=public_metrics"
    ids_str = "ids=" + ",".join(ids_batch)
    url = f"https://api.twitter.com/2/tweets?{ids_str}&{tweet_fields}"
    return url

def bearer_oauth(r):
    # 此处可替换为你自己的Bearer Token
    r.headers["Authorization"] = f"Bearer {'AAAAAAAAAAAAAAAAAAAAAN%2B7QwEAAAAAEG%2BzRZkmZ4HGizsKCG3MkwlaRzY%3DOwuZeaeHbeMM1JDIafd5riA1QdkDabPiELFsguR4Zba9ywzzOQ'}"
    r.headers["User-Agent"] = "v2TweetLookupPython"
    return r

def connect_to_endpoint(url):
    response = requests.request("GET", url, auth=bearer_oauth)
    print(f"请求状态码：{response.status_code}")
    if response.status_code != 200:
        raise Exception(
            f"请求报错：{response.status_code} {response.text}"
        )
    return response.json()

# 读取数据集里的所有推文ID
def load_all_tweet_ids():
    tweets_data_path = 'dataset.txt'
    tweets_data = []
    with open(tweets_data_path, "r", encoding="utf-8") as f:
        for line in f:
            try:
                tweet = json.loads(line)
                tweets_data.append(tweet)
            except:
                continue
    df = pd.DataFrame.from_dict(pd.json_normalize(tweets_data), orient='columns')
    return df['id'].astype(str).tolist()

def main():
    print(f"开始执行任务，当前时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    # 加载所有ID
    all_ids = load_all_tweet_ids()
    # 按接口上限拆分为每100个ID一批
    id_batches = split_list(all_ids, 100)
    all_results = []
    
    # 分批请求
    for idx, batch in enumerate(id_batches):
        print(f"正在请求第{idx+1}/{len(id_batches)}批数据")
        url = create_url(batch)
        json_response = connect_to_endpoint(url)
        all_results.extend(json_response.get('data', []))
        # 避免请求频率过高，可根据自己的API配额调整等待时间
        time.sleep(1)
    
    # 结果存储到txt，文件名带当前日期避免覆盖历史数据
    save_path = f"tweet_metrics_{datetime.now().strftime('%Y%m%d')}.txt"
    with open(save_path, "w", encoding="utf-8") as f:
        for item in all_results:
            f.write(json.dumps(item, ensure_ascii=False) + "\n")
    print(f"任务执行完成，结果已存储到{save_path}")

if __name__ == "__main__":
    # 配置每日零点执行任务
    schedule.every().day.at("00:00").do(main)
    print("定时任务已启动，将在每日零点自动执行")
    # 首次启动可以先执行一次任务测试，不需要可以注释掉该行
    main()
    # 常驻进程等待定时触发
    while True:
        schedule.run_pending()
        time.sleep(60)

核心功能说明 #

• 分批请求：自动将所有推文ID按100个为一组拆分，依次调用接口，完全适配Twitter v2接口的查询上限
• 定时执行：引入schedule库配置每日零点自动触发任务，进程保持后台运行即可持续生效
• 结果存储：每次执行的结果按当天日期命名存储，不会覆盖历史数据，每一行对应一条推文的指标数据
• 异常兼容：读取原始数据集时自动跳过格式错误的行，请求间隔加了1秒等待避免触发接口频率限制，可根据自己的API配额灵活调整间隔时间

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