Tweepy技术问询：如何实现推文删除通知及轮询代码故障排查

我明白你现在的困境——Twitter不再提供已删除推文的JSON数据后，靠轮询来检测推文是否删除的代码跑不起来，而且现在的逻辑是监听器触发时拿本地计数和API查询值比对却没效果。咱们一步步来拆解可能的问题：

1. 先确认API查询的准确性 #

这是最基础的环节，很多时候问题出在这里：

• 首先检查你调用的API版本和端点是否正确。Twitter v2 API和v1.1的返回结构差异很大，当推文已删除时，v2会返回404状态码，而不是包含删除标记的JSON。你是不是没正确判断状态码，反而去解析了不存在的字段？举个简单的示例：

  import requests
  
  def check_tweet_exists(tweet_id, bearer_token):
      headers = {"Authorization": f"Bearer {bearer_token}"}
      response = requests.get(f"https://api.twitter.com/2/tweets/{tweet_id}", headers=headers)
      if response.status_code == 404:
          return False  # 推文已删除
      elif response.status_code == 200:
          return True   # 推文仍存在
      else:
          # 处理权限不足（403）、限流（429）等其他情况
          print(f"API请求异常: {response.status_code}")
          return None
  

• 还要确认API权限：如果目标是私有推文，即使它存在，你没有访问权限也会拿到403，别把这种情况误判成推文删除。

2. 检查本地计数变量的同步逻辑 #

监听器触发时显示的本地计数，很可能和实际状态不同步：

• 如果用了多线程（比如Tweepy的Stream是跑在单独线程里的），本地计数变量的修改必须保证线程安全，不然会出现计数混乱。比如用锁来保护变量更新：

  import threading
  import tweepy
  
  count_lock = threading.Lock()
  local_tweet_count = 0
  
  class MyStream(tweepy.StreamingClient):
      def on_tweet(self, tweet):
          global local_tweet_count
          with count_lock:
              local_tweet_count += 1
          # 触发比对逻辑
          self.compare_counts(tweet.id)
  
      def compare_counts(self, tweet_id):
          # 调用上面的API检查函数
          tweet_exists = check_tweet_exists(tweet_id, "你的Bearer Token")
          with count_lock:
              print(f"本地计数: {local_tweet_count}")
          # 这里建议直接检查推文是否存在，而不是只比对计数
          if not tweet_exists:
              print(f"推文{tweet_id}已删除")
  

• 另外，本地计数是不是只统计了收到的推文？如果有推文被删除，但你的本地计数没做减法，那比对自然没意义——其实比对计数本身就不是检测删除的可靠方式，因为计数变化可能来自新的互动（转发、点赞），和删除无关，应该直接检查推文是否存在。

3. 轮询时机与缓存问题 #

如果你的轮询是在监听器触发后立刻执行，可能遇到API缓存的坑：

• Twitter API会有一定的缓存机制，推文刚被删除时，API可能还返回“存在”的状态，导致误判。建议给轮询加个短暂延迟，或者增加重试逻辑。
• 也可以调整轮询策略：不要每收到一条推文就立刻检查，而是固定时间间隔批量检查待监控的推文列表，这样更高效也更准确。

4. 补上缺失的错误处理 #

很多时候代码跑不起来是因为没处理异常：

• 网络波动、API限流、令牌过期这些情况都会导致API请求失败，如果没有捕获这些异常，代码可能直接崩溃，或者拿到错误的结果影响比对。一定要加上异常捕获：

  def check_tweet_exists(tweet_id, bearer_token):
      headers = {"Authorization": f"Bearer {bearer_token}"}
      try:
          response = requests.get(f"https://api.twitter.com/2/tweets/{tweet_id}", headers=headers)
          response.raise_for_status()  # 主动抛出HTTP错误
          return True
      except requests.exceptions.HTTPError as e:
          if e.response.status_code == 404:
              return False
          else:
              print(f"API请求错误: {e}")
              return None
      except requests.exceptions.RequestException as e:
          print(f"网络异常: {e}")
          return None
  

如果能把你的具体代码片段（比如监听器触发逻辑、API查询部分）贴出来，能更精准地定位问题哦！

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