【有源码】工程师如何优雅的将现场的报警推送到钉钉群?(附下载)-夜雨聆风

【有源码】工程师如何优雅的将现场的报警推送到钉钉群?(附下载)

JZGKCHINA

工控技术分享平台

各位伟大的工程师们，我们是不是经常为了设备突然的报警，或者值经常一闪而过无法捕捉故障，而需要查阅各种历史曲线或者监控很久？现在有一种很简单的方法，可以将报警信息推送出来的方法；大家想学习么？

利用钉钉群机器人实现软件使用监控是目前性价比最高的方案。无需搭建服务器，几行代码即可实现。教程主要分为钉钉设置和代码两部分。

本文介绍一个实现的标准的钉钉机器人信息告警模块，并包含通过ModbusTCP实现的案例。

1. 多线程：发送消息绝不卡顿主程序，发不出去也不报错（静默失败）。

2. 信息全：告警包含了告警对象、告警时间、告警类型、Tag地址、数据类型、异常详情、负责人、运维文档、处理建议等等；

3. 样式美：支持 Markdown 格式，让报错变红色，成功变绿色，比纯文本直观。

一、

钉钉设置

1. 建群：

在钉钉上发起群聊，选人建群（只有你一个人的群也可以）。

2. 添加机器人：

· 进入群聊界面 -> 点击右上角设置按钮

· -> 机器人

-> 添加机器人

· -> 点击 “+” 号。

· 选择 “自定义”

3. 配置机器人：

· 机器人名字：随便填，比如“监控助手”。

· 安全设置（关键步骤）：勾选 “加签”。

· 加签：会获得一串号，等下有用。·

4. 复制链接：

· 点击完成，你会得到一个 Webhook 地址。

· 格式类似：https://oapi.dingtalk.com/robot/send?access_token=xxxxxx

· 点击复制，保存好这个链接，一会要用。

二

代码实现（python版本）

def send_ding_alert(plc_name, alert_type, alert_content, is_at_all=False):"""    发送钉钉告警    """    alert_unique_id = f"{plc_name}_{alert_type}_{alert_content.get('register_addr', 'unknown')}"if alert_unique_id in SENT_ALERT_IDS:print(f"告警去重：{alert_unique_id} 已发送过，本次跳过")return True    plc_info = PLC_CONFIG.get(plc_name, {})    at_mobiles = plc_info.get("ding_at_mobiles", [])    contact = plc_info.get("contact", "未知负责人")    maintenance_doc = plc_info.get("maintenance_doc", "无")    alert_style = {"连接异常": {"emoji": "🔴", "color": "#FF4D4F", "level": "紧急"},"数值超限": {"emoji": "🟡", "color": "#FAAD14", "level": "重要"},"读取Tags异常": {"emoji": "🟠", "color": "#FA8C16", "level": "一般"},"未知异常": {"emoji": "🟤", "color": "#722ED1", "level": "一般"},"恢复通知": {"emoji": "🟢", "color": "#52C41A", "level": "正常"}    }    style = alert_style.get(alert_type, {"emoji": "⚠️", "color": "#333333", "level": "未知"})    markdown_text = f"""### {style['emoji']} PLC{style['level']}告警                        > ** 告警对象**：{plc_name}                        > ** 告警时间**：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}                        > ** 告警类型**：<font color="{style['color']}">{alert_type}</font>                        > ** PLC IP**：{alert_content.get('plc_ip', plc_info.get('ip', '未知'))}                        > ** Tag地址**：{alert_content.get('register_addr', '未知')}                        > ** 数据类型**：{alert_content.get('register_type', '未知')}                        > **❌ 异常详情**：{alert_content.get('error_msg', '无')}                        > **👤 负责人**：{contact}                        > **📖 运维文档**：[点击查看]({maintenance_doc})                        > **💡 处理建议**：{get_troubleshooting_suggestion(alert_type, plc_name, alert_content)}                    """if alert_type == "数值超限":        reg_config = THRESHOLD_CONFIG.get(plc_name, {}).get(alert_content.get('register_addr', 0), {})        markdown_text += f"> ** 当前值**：{alert_content.get('current_value', '未知')} {reg_config.get('unit', '')} \n> ** 阈值范围**：{reg_config.get('min', '未知')} ~ {reg_config.get('max', '未知')} {reg_config.get('unit', '')}"    headers = {"Content-Type": "application/json;charset=utf-8"}    data = {"msgtype": "markdown","markdown": {"title": f"【{plc_name}】{alert_type}", "text": markdown_text},"at": {"atMobiles": at_mobiles, "isAtAll": is_at_all}    }    timestamp = str(round(time.time() * 1000))    url = DING_WEBHOOKif DING_SECRET:        secret_enc = DING_SECRET.encode('utf-8')        string_to_sign = f"{timestamp}\n{DING_SECRET}"        hmac_code = hmac.new(secret_enc, string_to_sign.encode('utf-8'), hashlib.sha256).digest()        sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))        url = f"{url}&timestamp={timestamp}&sign={sign}"    try:        response = requests.post(url, json=data, headers=headers, timeout=10)        result = response.json()if result.get("errcode") == 0:print(f"钉钉告警发送成功：{alert_unique_id}")            SENT_ALERT_IDS.add(alert_unique_id)return Trueelse:print(f"钉钉告警发送失败：{result}")return False    except Exception as e:print(f"钉钉告警发送异常：{e}")return False

其中负责人和号码的配置，可以在群中直接@负责人

三

如何在其他项目中使用

以下是调用代码；

error_msg = f"Modbus读取失败，错误码：{response}"ALERT_STATUS[f"{plc_name}_{register_addr}"] = {"type": "Modbus异常", "start_time": time.time()}send_ding_alert(plc_name, "Modbus异常", {"register_addr": register_addr, "error_msg": error_msg})

如果已经封装好了，可以使用模块名. send_ding_alert();

四

后记

案例代码中很多固定配置，后续可以根据自己需要修改成从json、xml、ini数据库等地方读取配置信息，也可以采集MES系统、EAM、EMS、ERP等系统告警信息，通过不同的告警模板进行告警通知。这只是个抛砖引玉的案例。大家有兴趣可以通过多种途径实现；目前测试比较好用的有飞书、钉钉、企业微信等常用沟通工具。

如果需要获取源码可以和剑指工控技术群管理员索取。

项目下载