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

能否不使用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

相关产品推荐
方舟 Agent Plan

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

最近更新时间:2026.08.30 08:18:17