使用Academic Twitter API及AcademicTwitteR批量获取对话推文的效率优化求助

问题背景 #

已通过Academic Twitter API和AcademicTwitteR包爬取Amnesty International UK的全部推文（约3万条），下一步需基于这些推文的conversation_id获取完整对话线程，但当前串行循环查询的方式效率极低：

• AcademicTwitteR仅支持单次传入一个conversation_id，直接传列表会报错
• 单个查询耗时3秒起（推文多的对话耗时更久），加上API速率限制，预估需数天完成
• 测试3个conversation_id功能正常，但急需提速

R语言优化方案 #

1. 先去重conversation_id #

3万条推文中必然存在大量重复的conversation_id，先去重可大幅减少查询次数：

# 从已爬取的ai_t数据中提取去重后的conversation_id
unique_convs <- unique(ai_t$conversation_id)
# 过滤掉NA值（若有的话）
unique_convs <- unique_convs[!is.na(unique_convs)]

2. 并行化查询 #

用furrr包实现并行循环，替代串行for循环，充分利用多核CPU：

library(furrr)
library(academictwitteR)

# 设置并行工作进程数（根据CPU核心数调整，建议不超过API速率限制允许的并发数）
plan(multisession, workers = 4)

# 并行查询每个conversation_id，结果存储为列表
conversations_list <- future_map(unique_convs, function(conv_id) {
  get_all_tweets(
    start_tweets = "2008-01-01T00:00:00Z",
    end_tweets = "2022-11-14T00:00:00Z",
    bearer_token = BearerToken,
    n = Inf,
    conversation_id = conv_id
  )
})

# 将列表合并为单个数据框
conversations <- bind_rows(conversations_list)

3. 优化数据拼接 #

避免用c()循环拼接数据，改用列表存储后一次性合并，减少内存开销和耗时。

Python语言优化方案（更高效） #

Python可直接构造多conversation_id的查询语句，单次API请求即可查询多个对话，大幅减少请求次数：

1. 使用tweepy实现批量查询 #

import tweepy
import pandas as pd

# 初始化学术版API客户端
client = tweepy.Client(
    bearer_token="YOUR_BEARER_TOKEN",
    wait_on_rate_limit=True
)

# 从已爬取的推文数据中提取去重后的conversation_id
# 假设ai_t是已加载的DataFrame，替换为你的数据读取方式
ai_t = pd.read_csv("amnesty_uk_tweets.csv")
unique_convs = ai_t["conversation_id"].dropna().unique().tolist()

# 批量构造查询：每10个conversation_id为一组（可根据API限制调整）
batch_size = 10
all_conversations = []

for i in range(0, len(unique_convs), batch_size):
    batch_convs = unique_convs[i:i+batch_size]
    # 构造OR连接的查询语句
    query = " OR ".join([f"conversation_id:{conv}" for conv in batch_convs])
    
    # 批量获取对话推文
    response = client.search_all_tweets(
        query=query,
        start_time="2008-01-01T00:00:00Z",
        end_time="2022-11-14T00:00:00Z",
        max_results=500,  # 学术版单次最大返回500条
        expansions=["author_id", "referenced_tweets.id"],
        tweet_fields=["created_at", "conversation_id", "public_metrics"]
    )
    
    # 处理返回结果，转为DataFrame
    if response.data:
        tweets_df = pd.DataFrame([tweet.data for tweet in response.data])
        all_conversations.append(tweets_df)

# 合并所有结果
final_conversations = pd.concat(all_conversations, ignore_index=True)

优势说明 #

• 单次请求可查询多个conversation_id，将请求次数从3万+降至3千+（按batch_size=10计算）
• Tweepy自动处理API速率限制，无需手动等待
• 可灵活调整batch_size，平衡查询效率和API限制

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