跳转至

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:

# secrets.yaml
telegram_bot_token: "123456:ABC-DEF-..."
telegram_chat_id: 123456789

调用时支持 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 组件。

# configuration.yaml
tts:
  - platform: xiaomi_miot
    name: xiaomi_tts
    model: xiaomi_miot
    service: play

调用:

- 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

调用:

- service: notify.email_admin
  data:
    message: |
      <b>门磁报警</b>
      <p>大门被打开,请确认。</p>

坑:① 大多数邮箱用授权码而非登录密码,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 用模板转义。

# 消息截断 + Markdown 转义(模板里)
message: "{{ log_text | truncate(3800) | replace('_', '\\_') }}"

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。

相关文章