Tweepy查询异常：多关键词推文数量不足且脚本无法完成单日任务

我来帮你拆解下这个问题，你遇到的这两个状况在使用Tweepy调用Twitter API时其实挺常见的，咱们一步步来排查解决：

一、多关键词查询结果远少于单个/少量关键词的原因及解决 #

Twitter的搜索API本身有几个限制会导致这种情况：

• 查询字符串长度限制：如果把所有加密货币关键词用OR拼接后，字符串长度超过API允许的上限（当前是512字符），API会自动截断无效部分，导致部分关键词没被纳入查询。
• 结果去重与优先级：当多个关键词的推文有重叠时，API会自动去重；而且热门关键词的结果可能会挤压冷门关键词的展示量，最终返回的总数看起来更少。
• API的结果返回限制：单次搜索API最多返回100条结果，如果多关键词的结果分散在不同时间点，单次分页可能没法覆盖全部。

解决办法： #

• 拆分关键词组，分批查询：把你的加密货币列表分成几个小批次（比如每5-10个关键词一组），分别执行查询，最后把所有结果合并去重。
• 优化查询语法：确保用OR正确连接关键词，比如"bitcoin OR ethereum OR solana"，同时避免重复或者过于相似的关键词。
• 增加分页遍历：不要只取第一页结果，用Cursor或者手动循环调用search_tweets，通过max_id参数逐步回溯更早的推文。

二、脚本无法完成执行（无法遍历完前天全天推文）的问题排查 #

这个问题大概率和终止条件判断、API错误处理或者速率限制有关：

• 时间范围判断错误：你说要从23:59:59倒序查至00:00:00，可能你的代码里没有正确判断获取到的推文时间是否已经早于目标起始时间，导致循环一直运行或者提前中断。
• 未处理API异常：如果调用API时遇到超时、速率限制错误，脚本直接崩溃，没法继续执行。
• 速率限制触发后未重试：Twitter API有严格的调用速率限制，如果你没有处理RateLimitError，脚本会在触发限制后停止。

解决办法： #

1. 明确时间终止条件：
   在循环中，每次获取推文后，检查最早的那条推文的创建时间，如果早于前天的00:00:00，就终止循环。示例代码：

   import datetime
   
   target_date = datetime.date.today() - datetime.timedelta(days=2)
   start_time = datetime.datetime.combine(target_date, datetime.time.min)
   end_time = datetime.datetime.combine(target_date, datetime.time.max)
   
   # 用Cursor遍历推文
   for tweet in tweepy.Cursor(api.search_tweets, q=query, until=target_date.strftime("%Y-%m-%d"), tweet_mode="extended").items():
       tweet_time = tweet.created_at
       if tweet_time < start_time:
           break
       # 这里添加处理推文的逻辑
   

   注意until参数的格式是YYYY-MM-DD，它会返回该日期之前的推文，结合时间判断能更精准控制范围。

2. 添加异常处理与速率限制重试：
   用try-except捕获Tweepy的异常，遇到速率限制时，等待API重置时间后再继续：

   import time
   from tweepy.errors import RateLimitError
   
   while True:
       try:
           # 执行你的查询和处理逻辑
           break
       except RateLimitError:
           # 获取API重置时间
           reset_time = api.rate_limit_status()['resources']['search']['/search/tweets']['reset']
           wait_time = reset_time - time.time() + 10  # 多等10秒避免提前调用
           print(f"触发速率限制，等待{wait_time:.0f}秒...")
           time.sleep(wait_time)
       except Exception as e:
           print(f"遇到其他错误：{str(e)}，暂停10秒后重试")
           time.sleep(10)
   

3. 手动控制分页：
   如果用Cursor有问题，可以手动用max_id来分页，每次更新max_id为当前页最早推文的ID减1，直到获取到的推文时间超出目标范围：

   max_id = None
   while True:
       tweets = api.search_tweets(q=query, count=100, max_id=max_id, until=target_date.strftime("%Y-%m-%d"), tweet_mode="extended")
       if not tweets:
           break
       # 处理推文
       earliest_tweet_time = tweets[-1].created_at
       if earliest_tweet_time < start_time:
           break
       max_id = tweets[-1].id - 1
   

总结 #

先把关键词拆分分批查询并合并结果；然后优化脚本的时间判断逻辑，加上异常处理和速率限制重试，应该就能解决这两个问题了。

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