Home Assistant 多渠道通知实战:统一 notify 抽象 + 告警分级路由 + 5 个推送事故
适用场景:HA 已有 2 个以上通知渠道,想统一管理、按告警等级分流、避免重复刷屏。一句话结论:把多渠道收进一个 notify 路由入口,用 severity + 冷却时间做分级与防风暴,关键告警多渠道同时发。
1. 为什么通知不能只有一个渠道
我入坑 HA 前两年,全家告警只走一个 Telegram 机器人。直到某个深夜,门磁电池耗尽、窗户传感器误报、热水器超温三条告警挤在一起,而 Telegram 恰好因 token 到期全部发不出去——那晚我是在第二天早上才发现厨房水管漏了一夜。
从那天起,我的原则就一条:关键告警至少两个渠道,一个即时、一个保底。即时渠道手机能秒收(Telegram / 企业微信),保底渠道几乎不依赖外网可用性(邮件 / 本地语音播报)。
单渠道的坑不止"失效"一种,我总结成三类:
| 问题 | 表现 | 典型来源 |
|---|---|---|
| 单点失效 | 渠道 token 失效、被墙、被限流,告警全丢 | Telegram token 被回收、企业微信 429 |
| 通道拥堵 | 通知风暴把渠道打爆,重要告警被淹没 | 门磁反复触发,一次发 200 条 |
| 覆盖死角 | 某类事件不适合该渠道 | 烟雾报警适合语音+电话,邮件太慢 |
# 反面教材:只有单渠道,token 一失效全断
telegram_bot:
- platform: polling
api_key: !secret telegram_bot_token
allowed_chat_ids: [!secret telegram_chat_id]
notify:
- platform: telegram_bot
name: telegram_home
chat_id: !secret telegram_chat_id
所以"多渠道"不是堆渠道,而是让不同告警走最合适的渠道。这需要一个统一抽象层。
2. notify 服务架构:统一调用与按 domain 分组
HA 的 notify 是一个 domain(域),每个配置的渠道就是一个 notify 平台的实例,注册为 notify.<name> 服务。你调用 notify.notify 或 notify.<platform_name> 就能发。先把渠道注册全(Telegram + 邮件):
# configuration.yaml
telegram_bot:
- platform: polling
api_key: !secret telegram_bot_token
allowed_chat_ids:
- !secret telegram_chat_id
notify:
- platform: telegram_bot
name: telegram_home # 生成 notify.telegram_home
chat_id: !secret telegram_chat_id
- platform: smtp
name: email_admin # 生成 notify.email_admin
server: smtp.qq.com
port: 465
sender: !secret smtp_user
username: !secret smtp_user
password: !secret smtp_password
recipient: !secret admin_email
encryption: ssl
注册完,你在自动化里用同一个服务接口就能发不同渠道:
- service: notify.telegram_home
data:
title: "大门被打开"
message: "检测到门磁开启,请确认。"
- service: notify.email_admin
data:
message: "大门被打开,请确认。"
但每个自动化都写好几行太啰嗦。真正要做的,是把"发什么、多严重、走哪些渠道"抽象成一个入口。我用一个 script 做路由(第 8 章详细讲),所有自动化只需要调它:
# 自动化只调路由,不直接调渠道
- service: script.notify_router
data:
severity: warning
title: "门磁报警"
message: "大门被打开,请确认。"
这样做三个好处:① 换渠道只改 script 一处;② 分级、冷却、去重都在路由里做,自动化不用重复写;③ 多渠道同时发、失败降级都在一处实现。这就是"多渠道统一抽象"的核心。
完整的自动化长这样——trigger 检测状态变化,模板拼内容,调 script.notify_router:
# automations.yaml
- alias: 门磁打开告警
trigger:
- platform: state
entity_id: binary_sensor.front_door
to: "on"
for: "00:00:30" # 稳定 30 秒才触发,防抖动(第 9 章事故④)
action:
- service: script.notify_router
data:
severity: warning
title: "门磁报警"
message: >-
大门在 {{ now().strftime('%H:%M') }} 被打开。
当前室外温度 {{ states('sensor.outdoor_temp') }} 度。
这里 severity、title、message 就是路由脚本的入参,模板变量(states()、now())在调用侧拼好,路由侧只负责分发。告警逻辑(自动化)和分发逻辑(路由)彻底分离,这是多渠道抽象最有价值的一点。
3. 渠道接入:Telegram 机器人
Telegram 是 HA 最主流的即时渠道,配置最简单,但要先跟 BotFather 拿 token。
前置:找 @BotFather 发 /newbot 拿 token;把机器人拉进私聊或群,发一条消息激活;记下 chat_id(用 @userinfobot 查)。token 和 chat_id 放 secrets.yaml:
调用时支持 Markdown,可以做醒目格式:
- service: notify.telegram_home
data:
title: "🔥 烟雾报警"
message: |
⚠️ **检测到烟雾!**
- 位置:{{ state_attr('sensor.smoke_location', 'friendly_name') }}
- 时间:{{ now().strftime('%H:%M') }}
坑:Telegram 单条消息上限 4096 字符,且 parse_mode 用 Markdown 时,若消息含未转义的 _、*、[ 会报 Can't parse entities。解决办法:关掉 parse_mode,或用模板转义(第 9 章事故⑤展开)。
4. 渠道接入:企业微信群机器人(webhook)
企业微信我用群机器人 webhook,比个人应用号简单——不用申请应用、不用可信 IP,建个群、加个机器人就有 webhook URL。
前置:企业微信 → 群聊 → 右上角 → 添加群机器人 → 复制 Webhook 地址,放进 secrets.yaml。
由于 webhook 是 HTTP POST 且要特定 JSON,我用内置 rest_command 封装(不依赖会随 HA 升级失效的第三方 notify 组件,这是生产级稳妥做法):
# configuration.yaml
rest_command:
wecom_notify:
url: !secret wecom_webhook
method: post
content_type: "application/json"
payload: >-
{"msgtype":"markdown","markdown":{"content":"{{ title }}\n{{ message }}"}}
timeout: 10
再包一个 script 统一入口,用法和 notify 服务对齐:
script:
wecom_send:
alias: 企业微信推送
fields:
title: {}
message: {}
sequence:
- service: rest_command.wecom_notify
data:
title: "{{ title }}"
message: "{{ message }}"
坑:企业微信 markdown 的 content 里不能用 # 做一级标题、连续换行要 \n,且群机器人限流 20 条/分钟,超了返回 errcode 45008。这是第 9 章事故②的根源。
5. 渠道接入:钉钉群机器人(webhook)
钉钉做法和企业微信几乎一样,差异在 JSON 结构。同样走 rest_command:
# configuration.yaml
rest_command:
dingtalk_notify:
url: !secret dingtalk_webhook
method: post
content_type: "application/json"
payload: >-
{"msgtype":"markdown","markdown":{"title":"{{ title }}","text":"### {{ title }}\n{{ message }}"}}
timeout: 10
钉钉 webhook 需要加签(secret)或自定义关键词。我习惯用关键词模式:在 webhook 设置设一个关键词(比如"HA告警"),消息文本里必须包含该关键词,否则返回 errcode 310000 说"keywords not in content"。所以调用时把关键词拼进标题:
script:
dingtalk_send:
alias: 钉钉推送
fields:
title: {}
message: {}
sequence:
- service: rest_command.dingtalk_notify
data:
title: "HA告警 - {{ title }}"
message: "{{ message }}"
坑:钉钉 markdown 场景建议控制文本在 2000 字符内,超长会被截断或报错;加签模式下 secret 变了要重新配置,这也是第 9 章故障的一部分。
6. 渠道接入:小爱同学(TTS 语音播报)
小爱适合做本地即时语音播报——厨房有人、老人没带手机,语音最直接。我用 xiaomi_miot(Miot 小米集成)暴露的小爱音箱 media_player 配合 tts 组件。
调用:
- service: tts.speak
data:
entity_id: tts.xiaomi_tts
media_player_entity_id: media_player.xiaomi_speaker_living
message: "门磁报警,请检查大门"
坑:小爱播报有明显延迟和丢词,消息长了要排队逐句读;消息含英文或数字时读得很怪。所以小爱只适合短消息 + 即时语音,不适合长文本告警。这是第 9 章事故③的根源——我一度以为推送丢了,其实是小爱在排队。
7. 渠道接入:邮件(SMTP 保底)
邮件是保底渠道——不依赖 Telegram 可用性、不受 webhook 限流,失败率最低。用 HA 内置 smtp 平台:
# configuration.yaml
notify:
- platform: smtp
name: email_admin
server: smtp.qq.com
port: 465
sender: !secret smtp_user
username: !secret smtp_user
password: !secret smtp_password # QQ/163 用授权码,不是登录密码
recipient: !secret admin_email
encryption: ssl
调用:
坑:① 大多数邮箱用授权码而非登录密码,QQ 邮箱在"设置→账户"开启 SMTP 拿授权码;② encryption 要匹配端口——465 用 ssl,587 用 tls;③ 邮件送达快则几秒慢则几分钟,不适合秒级即时告警,只做保底。
8. 告警分级路由与防风暴冷却
这是全文核心——把所有渠道收进一个入口,按 severity 分级路由,用冷却时间 + 消息指纹去重防风暴。
先定义分级策略:
| severity | 含义 | 渠道 | 冷却 |
|---|---|---|---|
| critical | 危及生命/财产(烟雾、水浸、燃气) | 全部渠道即时发 + 语音 | 5 分钟 |
| warning | 需尽快处理(门磁、温度超限) | Telegram + 企业微信 | 10 分钟 |
| info | 低优先级(电量、状态变化) | Telegram 仅白天 + 邮件 | 30 分钟 |
用 helper 记录"上次发送时间"和"上次消息指纹":
# configuration.yaml
input_datetime:
last_critical_alert:
name: 上次 critical 告警时间
has_date: true
has_time: true
input_text:
last_alert_fingerprint:
name: 上次消息指纹
路由脚本(自动化只调它,全站唯一出口):
script:
notify_router:
alias: 多渠道通知路由(全站唯一通知出口)
fields:
severity: {description: "critical/warning/info", example: warning}
title: {description: 告警标题}
message: {description: 告警内容}
sequence:
# 冷却 + 去重:冷却期内且消息指纹相同 → 跳过(防风暴)
- choose:
- conditions:
- condition: template
value_template: >
{% set last = states('input_datetime.last_critical_alert') %}
{% set in_cooldown = false %}
{% if last != 'unknown' and last != 'unavailable' %}
{% set in_cooldown = (now() - (last | as_datetime)) < timedelta(minutes=5) %}
{% endif %}
{% set same_msg = (title + '|' + message)[:200] == states('input_text.last_alert_fingerprint') %}
{{ in_cooldown and same_msg }}
sequence:
- stop: "冷却期内重复告警,跳过"
default:
# 按 severity 路由
- choose:
- conditions:
- condition: template
value_template: "{{ severity == 'critical' }}"
sequence:
- service: notify.telegram_home
data: {title: "{{ title }}", message: "{{ message }}"}
- service: script.wecom_send
data: {title: "{{ title }}", message: "{{ message }}"}
- service: tts.speak
data:
entity_id: tts.xiaomi_tts
media_player_entity_id: media_player.xiaomi_speaker_living
message: "{{ title }}"
- conditions:
- condition: template
value_template: "{{ severity == 'warning' }}"
sequence:
- service: notify.telegram_home
data: {title: "{{ title }}", message: "{{ message }}"}
- service: script.wecom_send
data: {title: "{{ title }}", message: "{{ message }}"}
- conditions:
- condition: template
value_template: "{{ severity == 'info' }}"
sequence:
- service: notify.email_admin
data: {message: "{{ title }}: {{ message }}"}
- service: notify.telegram_home
data: {title: "{{ title }}", message: "{{ message }}"}
# 更新冷却与指纹
- service: input_datetime.set_datetime
target: {entity_id: input_datetime.last_critical_alert}
data: {datetime: "{{ now() }}"}
- service: input_text.set_value
target: {entity_id: input_text.last_alert_fingerprint}
data: {value: "{{ (title + '|' + message)[:200] }}"}
这套方案解决三个问题:① 告警分级,critical 才全渠道即时;② 防风暴,冷却期内同指纹不再重复发;③ 去重,门磁反复触发也不会刷屏。
9. 五个真实推送故障复盘
按"症状 → 排查 → 根因 → 修复"讲,全是我踩过的。
事故①:Telegram 突然全部发不出(bot was blocked)
症状:所有 notify.telegram_home 报 TelegramError: Forbidden: bot was blocked by the user。排查:token 没变,但机器人被用户拉黑,或 token 被 BotFather 回收。根因:我换手机号后旧设备登出,Telegram 认为机器人"被停止"。修复:/start 重新激活,把机器人重新拉回群,确认 allowed_chat_ids 里的 chat_id 没变。教训:Telegram 失效不报错于 token,而报"被拉黑",别只查 token。
事故②:企业微信 webhook 报 429 / 45008 限流
症状:通知风暴时企业微信返回 errcode 45008 或 HTTP 429。排查:群里机器人 20 条/分钟上限被打爆。根因:门磁在 5 分钟内翻转 200 次,自动化每次触发都发,没做冷却。修复:路由里加冷却(第 8 章),告警合并成一条。教训:webhook 渠道必须有冷却,否则限流是必然。
事故③:小爱同学推送"丢失"
症状:语音告警要么延迟几十秒,要么干脆没声。排查:tts.speak 调用成功、日志无错,但小爱不响。根因:小爱 TTS 有队列,消息长要排队逐句读;音箱离线时 HA 静默丢弃。修复:小爱只发短消息(≤15 字),加 media_player 在线判断,离线就走 Telegram。教训:语音渠道的"成功"不等于"播了",要验证媒体播放器状态。
事故④:通知风暴刷屏
症状:一晚上收到 300 条"门磁报警",手机被打爆。排查:门磁接触不良,状态在 open/closed 间高频抖动,自动化无冷却。根因:trigger 用 state 触发且无 for 时长,也没去重。修复:trigger 加 for: 30s(稳定后才触发),路由加指纹去重 + 冷却。教训:所有 state 触发的告警自动化,必须加 for 防抖动。
事故⑤:消息体过长被截断 + Markdown 解析报错
症状:Telegram 消息一半没了,或报 Can't parse entities。排查:消息超 4096 字符,或含未转义的 _ * [。根因:告警 message 里拼接了长日志,Markdown 特殊字符未处理。修复:截断到 3800 字符,转义特殊字符。教训:发长消息前先 | truncate,Markdown 用模板转义。
10. FAQ
# 冷却/去重 helpers(第 8 章配套)
input_datetime:
last_critical_alert:
name: 上次 critical 告警时间
has_date: true
has_time: true
input_text:
last_alert_fingerprint:
name: 上次消息指纹
Q1:通知为什么不能只用一个渠道? 单渠道单点故障。Telegram 被拉黑、企业微信限流、小爱离线都会导致告警漏发。至少配一个即时渠道(Telegram/企业微信)加一个保底渠道(邮件),关键告警多渠道同时发。
Q2:通知风暴怎么防?
三招:severity 分级(critical 才全渠道即时,info 合并)、冷却时间(同一告警 N 分钟只发一次)、去重(记录上次消息指纹,重复跳过)。用 input_datetime 记时间戳 + 模板条件实现。
Q3:Telegram 提示 bot was blocked 怎么办?
403 Forbidden 说明机器人被用户拉黑或 token 失效。先 /start 重新激活,确认 token 有效;若被踢出群,重新拉回群并核对 allowed_chat_ids。
Q4:企业微信 webhook 报 429 限流怎么办? 群机器人限 20 条/分钟,超了返回 45008/429。把告警合并成一条、降级到邮件,或把冷却时间调到 ≥60 秒。别在 webhook 里发消息风暴。
Q5:小爱播报延迟或没声怎么排查?
先确认 media_player 在线,再核对 TTS 服务参数。消息过长会排队逐句读,改成 ≤15 字短句。加在线状态判断,离线就走 Telegram。
相关文章
- HA 自动化基础:触发、条件与动作 — 告警自动化的 trigger/condition 写法
- Home Assistant 数据迁移实战 — 迁移后渠道 token 与 secrets 的核对
- MQTT 配置实战 — 传感器告警接入的基础