基于Tweepy高效查询特定用户推文的多关键词方法

我有一个Twitter用户列表，希望搜索每个用户时间线中的特定关键词。目前使用全量归档搜索（full archive search），通过from:screen_name语法实现，但该搜索的速率限制为每15分钟300次。请问是否存在更高效的查询方式？

当前实现代码 #

单用户检索函数 #

def curated_user_tweets(screen_name):
    
    query = 'lang:en -is:retweet keyword1 from:' + screen_name + ' OR keyword2 from:' + screen_name \
            + ' OR keyword3 from:' + screen_name + ' OR keyword4 from:' + screen_name
    for response in tweepy.Paginator(client.search_all_tweets, 
                                     query = query, 
                                     user_fields = ['username', 'public_metrics', 'description', 'location'],
                                     tweet_fields = ['created_at', 'geo', 'public_metrics', 'text', 'conversation_id', 'attachments'],
                                     expansions = ['author_id'],
                                     start_time = '2019-05-03T00:00:00Z',
                                     end_time = '2022-07-01T00:00:00Z', max_results=500):
            time.sleep(1)
            tweets.append(response)

批量用户遍历逻辑 #

users = ['@screen_name1', '@screen_name2', '@screen_name3'...]
tweets = []

while True:
    try:
        for user in users:
            curated_user_tweets(user)
            print(len(tweets))
    except tweepy.errors.TweepyException as e:
        time.sleep(60*15)
        print(e)
    except StopIteration:
        break

以下是几种能提升检索效率的实用方式：

1. 合并多用户查询，减少请求次数 #

全量归档搜索支持在单个查询中指定多个from:用户，同时保留关键词过滤逻辑。原本N个用户需要N次请求，现在可按批次合并为少量请求，大幅降低请求量。

示例修改后的批量查询实现：

# 按每10个用户为一批（避免查询字符数超出API限制）
def batch_curated_tweets(user_batch):
    # 生成多用户的from条件：from:user1 OR from:user2...
    from_clauses = " OR ".join([f"from:{user.strip('@')}" for user in user_batch])
    # 生成关键词条件：keyword1 OR keyword2...
    keyword_clauses = " OR ".join(["keyword1", "keyword2", "keyword3", "keyword4"])
    query = f'lang:en -is:retweet ({keyword_clauses}) ({from_clauses})'
    
    for response in tweepy.Paginator(client.search_all_tweets, 
                                     query=query, 
                                     user_fields=['username', 'public_metrics', 'description', 'location'],
                                     tweet_fields=['created_at', 'geo', 'public_metrics', 'text', 'conversation_id', 'attachments'],
                                     expansions=['author_id'],
                                     start_time='2019-05-03T00:00:00Z',
                                     end_time='2022-07-01T00:00:00Z', max_results=500):
        tweets.append(response)

# 拆分用户列表为批次
user_batches = [users[i:i+10] for i in range(0, len(users), 10)]
for batch in user_batches:
    batch_curated_tweets(batch)

注意：Twitter API单条查询最大长度为1024字符，需根据用户名长度调整每批用户数量，避免超出限制。

2. 使用用户ID替代用户名查询 #

API查询中from:user_id比from:username更高效，还能避免用户名变更导致的检索遗漏。可先批量获取用户ID，再用ID构建查询：

# 批量获取用户ID（每批最多100个用户，符合API限制）
def get_user_ids(screen_names):
    user_ids = []
    for batch in [screen_names[i:i+100] for i in range(0, len(screen_names), 100)]:
        users_response = client.get_users(usernames=[u.strip('@') for u in batch])
        user_ids.extend([user.id for user in users_response.data])
    return user_ids

# 后续用ID构建查询：from:12345 OR from:67890...

3. 优化速率控制与异常处理 #

• 移除不必要的time.sleep(1)：API速率限制按15分钟窗口计算，只要总请求数不超300次/15分钟，无需每次分页休眠。可改用Tweepy内置的限流处理，或精准跟踪请求计数。
• 捕获特定限流异常：针对tweepy.errors.TooManyRequests处理，而非泛用的TweepyException，利用API返回的重试时间等待，比固定15分钟更精准。

修改后的异常处理示例：

while True:
    try:
        for batch in user_batches:
            batch_curated_tweets(batch)
            print(len(tweets))
    except tweepy.errors.TooManyRequests as e:
        wait_time = e.response.headers.get('x-rate-limit-reset', 0) - time.time()
        if wait_time > 0:
            time.sleep(wait_time)
        print(f"触发限流，等待{int(wait_time)}秒")
    except tweepy.errors.TweepyException as e:
        print(f"API错误：{e}")
        break

4. 评估用户时间线API的适用性 #

如果目标用户推文数量不超过3200条，可考虑使用client.get_users_tweets（用户时间线API）替代搜索接口：

• 该接口速率限制为每15分钟900次请求（单用户），比全量搜索限制更宽松
• 需要在本地对获取的推文进行关键词过滤，会增加本地计算量
• 缺点是只能获取用户最近3200条推文，无法检索更早内容

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