使用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
相关产品推荐
相关产品推荐

