You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

使用tweepy获取用户最新推文的代码如何实现24小时实时监听新内容

实现目标用户新推文自动检测的落地方案

你当前的代码不需要做根本性重构,有两种成熟的实现路径可以支撑7*24小时不间断运行,匹配新推文自动打印的需求:

方案1:基于现有轮询逻辑改造(最易部署、稳定性最高)

这是最适合新手的实现方式,不需要依赖长连接,完全兼容你现在已经写好的接口调用逻辑,核心思路是通过对比推文ID识别新内容:

  • Twitter的推文ID是按发布时间单调递增的,越晚发布的推文ID数值越大
  • 程序启动时先拉一次目标用户的最新推文,把这条推文的唯一ID存为全局变量last_seen_id
  • 按照符合接口限额的固定间隔,重复拉取目标用户的时间线,筛选出所有ID比last_seen_id大的内容——这些就是用户刚发布、你还没打印过的新推文
  • 每次打印完新推文,同步更新last_seen_id为这批新推文里的最大ID,作为下一次检测的对比基准

改造后的完整可运行代码参考:

import tweepy
import time

# 填入你自己的Twitter开发者接口凭证
ckey = ''
csecret = ''
atoken = ''
asecret = ''

auth = tweepy.OAuthHandler(ckey, csecret)
auth.set_access_token(atoken, asecret)
# 开启自动限流等待,请求超限额时自动暂停等待重置,不会直接报错中断
api = tweepy.API(auth, wait_on_rate_limit=True)
TARGET_USER = 'DropSentry'

# 初始化首次拉取,记录当前的最新推文ID
def get_initial_last_id():
    recent_timeline = api.user_timeline(
        screen_name=TARGET_USER,
        count=1,
        include_rts=True,
        tweet_mode='extended'
    )
    return recent_timeline[0].id

last_seen_id = get_initial_last_id()
print(f"监控已启动,将持续追踪用户 {TARGET_USER} 的新发布内容")

# 永久循环检测
while True:
    try:
        # 拉取比上次记录ID更新的所有推文,避免冗余拉取旧内容
        newly_published = api.user_timeline(
            screen_name=TARGET_USER,
            count=10,
            include_rts=True,
            since_id=last_seen_id,
            tweet_mode='extended'
        )
        # 按发布时间从早到晚遍历新推文,避免倒序打印逻辑混乱
        for tweet in reversed(newly_published):
            print(f"\n--- 检测到新推文 ---")
            print(f"发布时间:{tweet.created_at}")
            print(f"推文内容:{tweet.full_text}")
            # 更新已读的最大ID
            last_seen_id = max(last_seen_id, tweet.id)
    except Exception as err:
        print(f"本次拉取出现异常:{err},将在15秒后重试")
    # 间隔15分钟拉取一次,完全符合普通开发者账号的接口频率限制,不会被封禁
    time.sleep(15 * 60)

方案2:基于tweepy Stream流接口实现实时推送

如果你需要更低延迟的检测效果,可以用tweepy内置的流式监听接口替代轮询:

  • 这个接口通过和Twitter服务器建立长连接,只要目标用户发布新推文,服务器会在几秒内把内容推送到你的本地程序,不需要你定时主动拉取
  • 缺点是长连接对网络稳定性要求高,遇到网络波动就会断开,需要自己写自动重连的逻辑;且免费开发者账号的流式接口有监听规则的数量限制,部署复杂度比轮询方案高

7*24稳定运行的注意事项

  • 不要把程序长期挂在个人家用电脑上,关机、断网都会中断运行,可部署在低功耗的云服务器、轻量容器平台上,长期稳定在线
  • 必须在循环逻辑里加全链路异常捕获,遇到临时网络中断、接口返回临时错误时自动重试,不要让程序直接崩溃退出
  • 不要为了追求极低延迟把轮询间隔设的过短,普通免费开发者账号调用user_timeline接口的限额是每15分钟1500次,15分钟拉取一次的频率远低于限额,完全不会触发风控

内容的提问来源于stack exchange,提问作者Haider

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.10.22 11:36:04