能否不使用handlers基于aiogram实现站点新消息主动推送?
aiogram相关问题解答
站点新增内容自动推送功能的可行性
这个需求完全可以基于aiogram实现,没有框架层面的限制。
实现逻辑也很直接:你可以单独启动一个异步的定时爬取任务,按照你需要的频率请求目标站点,解析内容后和本地存储的历史爬取结果做比对,一旦检测到新增内容,直接调用机器人实例的消息发送接口,推送给所有订阅了该站点通知的用户即可。
需要注意的是爬取逻辑尽量用异步实现(比如搭配aiohttp做网络请求),和aiogram的运行逻辑放在同一个事件循环中,不要写阻塞的同步代码,否则会影响机器人对用户正常指令的响应。定时规则可以直接用asyncio.sleep做简单循环,也可以搭配APScheduler做更灵活的定时配置。
不依赖handlers发送消息的支持情况
aiogram完全支持脱离handlers主动发送消息。
handlers的作用仅仅是响应、处理用户主动发送给机器人的交互事件(比如用户发的指令、点击的按钮回调等),主动推送消息的场景根本不需要经过handler流程。你只要完成Bot实例的初始化,拿到目标用户的有效chat_id,在代码的任意位置(不管是定时爬取任务、业务回调还是独立的逻辑模块)都可以直接调用发送消息的方法完成推送,不需要额外注册任何handler。
下面是最小实现的代码片段参考:
import asyncio import aiohttp from aiogram import Bot, Dispatcher BOT_TOKEN = "替换为你的机器人Token" bot = Bot(token=BOT_TOKEN) dp = Dispatcher() # 提前持久化存储的订阅用户ID集合 SUBSCRIBED_USERS = {123456789, 987654321} last_record = set() async def site_monitor_task(): global last_record async with aiohttp.ClientSession() as session: while True: try: # 请求目标站点、解析内容 async with session.get("替换为目标站点地址") as resp: page_content = await resp.text() # 此处按你的实际需求解析出最新的条目列表 latest_items = parse_content(page_content) new_items = [item for item in latest_items if item["id"] not in last_record] if new_items and last_record: # 遍历新条目推送给所有订阅用户 for item in new_items: send_text = f"检测到站点更新:\n标题:{item['title']}\n链接:{item['url']}" for user_id in SUBSCRIBED_USERS: try: await bot.send_message(chat_id=user_id, text=send_text) except: # 忽略用户拉黑、账号失效等发送失败的情况 pass last_record = {item["id"] for item in latest_items} except Exception as e: print(f"爬取或推送出错: {e}") # 配置爬取间隔,示例为300秒(5分钟) await asyncio.sleep(300) async def main(): # 提前启动站点监控任务 asyncio.create_task(site_monitor_task()) # 启动机器人轮询,你自己的业务handler正常注册在dp上即可 await dp.start_polling(bot) def parse_content(content): # 在这里写你的站点内容解析逻辑 return [] if __name__ == "__main__": asyncio.run(main())
实操提示:主动推送时记得做好异常捕获,个别用户封禁机器人、chat_id失效的问题不要影响整个推送任务和爬取循环。
内容的提问来源于stack exchange,提问作者Raiymbek Zhumatov
相关产品推荐
相关产品推荐

