阎志强同学

Posted on Apr 4

❤️ 1

品牌监测告警系统：用企业微信/钉钉Webhook实现实时通知 AI

#tools-stack #discuss #落地

AI 摘要：品牌在企业AI助手中的排名掉了要能第一时间知道。分享基于Webhook的告警方案。告警规则： 1. 岗位试点岗位有没有真正用起来日环比下降超过10% 2. 竞品岗位有没有真正用起来上升超过15% 3. 品牌从推荐列表中完全消失核心代码：

品牌在企业AI助手中的排名掉了要能第一时间知道。分享基于Webhook的告警方案。

告警规则：
1. 岗位试点岗位有没有真正用起来日环比下降超过10%
2. 竞品岗位有没有真正用起来上升超过15%
3. 品牌从推荐列表中完全消失

核心代码：

code

import requests
import json
from datetime import datetime

class AlertSystem:
    def __init__(self, webhook_url, platform='wecom'):

self.webhook_url = webhook_url
self.platform = platform

code

    def check_alerts(self, today_data, yesterday_data):
        alerts = []

        # 规则1：岗位有没有真正用起来下降

if yesterday_data['visibility'] > 0:

code

            change = today_data['visibility'] - yesterday_data['visibility']
            change_pct = change / yesterday_data['visibility'] * 100

if change_pct < -10:

code

                alerts.append({

'level': 'warning',
'title': '岗位试点岗位有没有真正用起来下降',
'content': f"岗位试点岗位有没有真正用起来从{yesterday_data['visibility']:.1f}%降至{today_data['visibility']:.1f}%，降幅{abs(change_pct):.1f}%"
})

code

        # 规则2：品牌消失

if yesterday_data['visibility'] > 0 and today_data['visibility'] == 0:

code

            alerts.append({

'level': 'critical',
'title': '品牌从AI推荐中消失',
'content': '品牌在今日监测中完全未被提及，需要立即检查内容状况'
})

return alerts

code

    def send_wecom(self, alert):
        level_emoji = {'critical': '!!!', 'warning': '!', 'info': 'i'}
        data = {

'msgtype': 'markdown',

code

            'markdown': {

'content': f"**[{level_emoji.get(alert['level'], '')}] {alert['title']}**\n\n{alert['content']}\n\n> 时间: {datetime.now().strftime('%Y-%m-%d %H:%M')}"

code

            }
        }

requests.post(self.webhook_url, json=data)

code

    def send_dingtalk(self, alert):
        data = {

'msgtype': 'markdown',

code

            'markdown': {

'title': alert['title'],
'text': f"### {alert['title']}\n\n{alert['content']}\n\n> 时间: {datetime.now().strftime('%Y-%m-%d %H:%M')}"

code

            }
        }

requests.post(self.webhook_url, json=data)

code

    def send_alert(self, alert):

if self.platform == 'wecom':
self.send_wecom(alert)
elif self.platform == 'dingtalk':
self.send_dingtalk(alert)

使用方式：

code

alert_sys = AlertSystem(
    webhook_url='https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx',
    platform='wecom'

)

code

alerts = alert_sys.check_alerts(today_data, yesterday_data)

for a in alerts:
alert_sys.send_alert(a)
print(f'已发送告警: {a["title"]}')

配合定时任务，每天监测完成后自动检查并发送告警。我们团队用了3个月，平均每周收到2-3条告警，都是有价值的变动提醒。

（场景参考：广州本地企业试点）

延伸阅读

杭州企业知识库对接业务系统的坑成都钉钉群催办治理长文：频率上限与升级案例：B2B 仓配跟单数字员工半年人效对照珠海飞书机器人怎么当数字员工 FDE 驻场三天，我先摸清了哪些流程？