Tweepy监听特定用户推文转Telegram延迟问题排查求助

一、为什么会出现应用级认证的限流效果？ #

首先，你的脚本看似用了用户级OAuth1认证，但实际触发应用级限流，大概率有两个核心原因：

1. 认证未正确生效：可能你的access token/secret配置有误，导致Tweepy实际使用了应用级认证（而非用户级）。你可以通过调用api.rate_limit_status()快速验证：

   rate_limit = api.rate_limit_status()
   print(rate_limit['resources']['statuses']['/statuses/user_timeline'])
   

   如果返回的limit是300，说明确实用的是应用级配额；如果是900，那就是用户级，这时候要关注第二个原因。
2. 无效请求浪费配额：你的脚本每1.5秒就调用一次api.user_timeline，哪怕没有新推文也在重复请求。7.5分钟刚好是300次请求，刚好触达应用级配额上限（如果认证真的出问题的话）。就算是用户级的900次配额，这种无意义的轮询也会快速耗尽配额，导致限流延迟。

另外，你的时间判断逻辑只精确到分钟，直接造成两个问题：

• 同一分钟内的多条推文只会处理第一条，然后强制sleep60秒，错过其他推文
• 新推文发布后，要等到当前分钟匹配才会处理，直接造成最多59秒的延迟，完全不符合你≤1.5秒的要求

二、更优解决方案（最低延迟+避免限流） #

1. 优先切换到Twitter API v2实时流（推荐）

轮询本身就有延迟上限，而且容易触发限流。Twitter API v2提供了User Tweets Stream，可以实时接收目标用户的推文推送，不需要主动轮询，延迟几乎为0，还能彻底解决配额问题。用Tweepy实现的示例代码：

import tweepy
import datetime
from notifiers import get_notifier
from re import sub

TOKEN = '你的telegram token'
USER_ID = '目标用户的telegram ID'
ADMIN_ID = '你的telegram ID'
BEARER_TOKEN = '你的Twitter API v2 Bearer Token'
TARGET_TWITTER_USER_ID = '要监听的Twitter用户ID'

telegram = get_notifier('telegram')

def send_to_telegram(tweet, author):
    # 清理推文链接
    tweet_text = sub(r"https?://t.co[^,\s]+,?", "", tweet.text)
    # 构建用户通知消息
    user_msg = f'⚠️@{author.username} 发布新推文⬇️\n{tweet_text}'
    # 添加媒体链接（如果有）
    if 'media' in tweet.entities:
        for media in tweet.entities['media']:
            user_msg += f'\n{media["url"]}'
    telegram.notify(token=TOKEN, chat_id=USER_ID, message=user_msg)
    
    # 构建管理员通知消息（含时间戳）
    admin_msg = f'⚠️新推文通知\n用户：@{author.username}\n内容：{tweet_text}\n发布时间：{tweet.created_at.strftime("%d.%m.%Y, %H:%M:%S")}\n当前时间：{datetime.datetime.now().strftime("%d.%m.%Y, %H:%M:%S")}'
    if 'media' in tweet.entities:
        for media in tweet.entities['media']:
            admin_msg += f'\n媒体链接：{media["url"]}'
    telegram.notify(token=TOKEN, chat_id=ADMIN_ID, message=admin_msg)

# 定义流处理器
class UserTweetStream(tweepy.StreamingClient):
    def on_tweet(self, tweet):
        # 获取完整推文和作者信息
        tweet_data = tweepy.Client(BEARER_TOKEN).get_tweet(
            tweet.id, 
            tweet_fields=["created_at", "entities"], 
            expansions=["author_id"], 
            user_fields=["username"]
        )
        send_to_telegram(tweet_data.data, tweet_data.includes['users'][0])

# 初始化流
stream = UserTweetStream(bearer_token=BEARER_TOKEN)
# 添加监听规则：只接收目标用户的推文
stream.add_rules(tweepy.StreamRule(f"from:{TARGET_TWITTER_USER_ID}"))
# 启动流（持续运行）
stream.filter(tweet_fields=["created_at", "entities"], expansions=["author_id"], user_fields=["username"])

2. 如果必须用v1.1轮询，优化脚本逻辑

如果暂时无法切换到v2，需要彻底改造轮询逻辑，避免无效请求：

import tweepy
import datetime
import time
from notifiers import get_notifier
from re import sub

TOKEN = '你的telegram token'
USER_ID = '目标用户的telegram ID'
ADMIN_ID = '你的telegram ID'

auth = tweepy.OAuth1UserHandler(
    consumer_key="你的consumer key",
    consumer_secret="你的consumer secret",
    access_token="你的access token",
    access_token_secret="你的access token secret",
)
api = tweepy.API(auth)

# 验证当前配额
rate_limit = api.rate_limit_status()
print("当前user_timeline配额：", rate_limit['resources']['statuses']['/statuses/user_timeline'])

target_user = api.get_user(screen_name='要监听的Twitter用户名')
last_processed_tweet_id = -1  # 记录最后处理的推文ID，避免重复处理
telegram = get_notifier('telegram')

def process_tweet(tweet):
    tweet_text = sub(r"https?://t.co[^,\s]+,?", "", tweet.full_text)
    tweet_time = tweet.created_at.strftime("%d.%m.%Y, %H:%M:%S")
    current_time = datetime.datetime.now().strftime("%d.%m.%Y, %H:%M:%S")
    
    # 合并用户通知，减少Telegram API调用
    user_msg = f'⚠️@{target_user.screen_name} 发布新推文⬇️\n{tweet_text}'
    if hasattr(tweet, 'extended_entities') and 'media' in tweet.extended_entities:
        for media in tweet.extended_entities['media']:
            user_msg += f'\n{media["media_url_https"]}'
    telegram.notify(token=TOKEN, chat_id=USER_ID, message=user_msg)
    
    # 合并管理员通知
    admin_msg = f'⚠️新推文通知\n用户：@{target_user.screen_name}\n内容：{tweet_text}\n发布时间：{tweet_time}\n当前时间：{current_time}'
    if hasattr(tweet, 'extended_entities') and 'media' in tweet.extended_entities:
        for media in tweet.extended_entities['media']:
            admin_msg += f'\n媒体链接：{media["media_url_https"]}'
    telegram.notify(token=TOKEN, chat_id=ADMIN_ID, message=admin_msg)

def start_listening():
    print("开始监听目标用户...")
    global last_processed_tweet_id
    while True:
        try:
            # 只获取比最后处理ID新的推文，避免重复请求旧内容
            tweets = api.user_timeline(
                user_id=target_user.id,
                since_id=last_processed_tweet_id,
                count=20,  # 一次获取多条，避免漏过短时间内的多条推文
                tweet_mode='extended'  # 获取完整长推文，避免截断
            )
            # 按时间升序处理（因为user_timeline是降序返回）
            for tweet in reversed(tweets):
                process_tweet(tweet)
                last_processed_tweet_id = tweet.id
            time.sleep(1.5)
        except tweepy.TooManyRequests:
            print("触发限流，等待15分钟后重试...")
            time.sleep(15*60)
        except Exception as e:
            print(f"发生错误：{str(e)}，等待5秒后重试...")
            time.sleep(5)

start_listening()

三、额外优化建议 #

• 合并Telegram消息：不要多次调用telegram.notify，把相关内容合并成一条消息发送，减少API调用次数，提升处理速度
• 添加异常捕获：处理限流、网络错误等异常情况，避免脚本直接崩溃
• 使用tweet_mode='extended'：确保能获取完整的长推文内容，避免被截断

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