用python解放右手(九) 机器人告警-让Python替你喊人

机器人告警：让 Python 替你"喊人"

本文基于 Python 3.9+，涉及：企业微信/钉钉 Webhook。阅读时间约 10 分钟。

周末的"静默灾难"

周六晚上，阿明正在家陪娃搭积木，手机突然响了。

是老板：“阿明，系统是不是挂了？客户说登不上。”

阿明心里一紧，打开 VPN 连公司内网，果然——数据库连接池满了，服务全崩了。

他花了两个小时抢修，恢复后一看日志——问题在周六凌晨 3 点就出现了，只是没人发现。

周一早上，阿明顶着黑眼圈到公司，被老板叫进办公室：“为什么周末出问题了没人知道？”

阿明委屈：“我在家带娃呢，不可能 24 小时盯着屏幕……”

"那就让系统自己’喊人’啊。"老板皱眉。

阿明找到老张：“张哥，系统出问题了，能自动发消息到微信/钉钉吗？”

老张：“能。让 Python 当你的’看门大爷’，出事了替你@所有人。”

第一步：认识 Webhook——机器人的"电话号码"

“阿明，你知道怎么让程序往微信群里发消息吗？”

“不知道……”

“用Webhook。你可以把它理解为机器人的’电话号码’——你往这个地址发一条 HTTP 请求，机器人就在群里喊一嗓子。”

老张在白纸上画了个图：

你的 Python 脚本 → 发送 HTTP POST 请求 → 企业微信/钉钉服务器 → 群里出现消息

“不需要登录微信，不需要扫码，只要一个 URL，就能发消息。”

第二步：企业微信机器人

“先教你企业微信的，钉钉类似。”

创建机器人

1. 打开企业微信群聊
2. 右键群设置 → 添加群机器人
3. 选择【新建机器人】，取个名字（比如"告警小助手"）
4. 复制Webhook 地址

“Webhook 地址长这样：”

https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx

“这个key就是机器人的身份标识，别泄露，不然别人也能往你们群里发消息。”

发送文本消息

importrequestsimportjson WEBHOOK_URL="https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=你的key"defsend_text(content,mentioned_list=None):"""发送文本消息到企业微信"""data={"msgtype":"text","text":{"content":content,"mentioned_list":mentioned_listor[],# @指定人}}response=requests.post(WEBHOOK_URL,data=json.dumps(data,ensure_ascii=False).encode("utf-8"),headers={"Content-Type":"application/json"})result=response.json()ifresult.get("errcode")==0:print("✅ 消息发送成功！")else:print(f"❌ 发送失败:{result}")# 测试发送send_text("🚨 告警测试：这是一条来自 Python 的消息")# @所有人send_text("🔥 系统异常！请立即处理！",mentioned_list=["@all"])

“msgtype": "text"是发纯文本，mentioned_list可以 @指定人或 @所有人。”

阿明瞪眼：“就这么简单？”

“就这么简单。”

第三步：发送 Markdown 消息（更美观）

“纯文本太朴素了，咱们用 Markdown 格式，支持加粗、颜色、链接。”

defsend_markdown(content):"""发送 Markdown 消息"""data={"msgtype":"markdown","markdown":{"content":content}}response=requests.post(WEBHOOK_URL,data=json.dumps(data,ensure_ascii=False).encode("utf-8"),headers={"Content-Type":"application/json"})returnresponse.json()# 发送格式化的告警信息markdown_content="""## 🚨 系统告警 > **告警时间**: 2024-01-22 03:15:00 > **告警级别**: <font color='red'>严重</font> **异常信息**: - 服务: `user-service` - 错误: 数据库连接池耗尽 - 影响: 用户无法登录 **处理建议**: 1. 检查数据库连接数 2. 重启连接池 3. 查看慢查询日志 [点击查看详情](http://grafana.company.com/d/xxx) """send_markdown(markdown_content)

“企业微信的 Markdown 支持有限，但基本的加粗、颜色、链接都够用。”

“<font color='red'>严重</font>是标红，让关键信息一眼看到。”

第四步：发送卡片消息（最专业）

“如果要发得像个正经告警系统，用卡片消息——带标题、内容、跳转链接。”

defsend_card(title,description,url,btntxt="查看详情"):"""发送文本卡片"""data={"msgtype":"textcard","textcard":{"title":title,"description":description,"url":url,"btntxt":btntxt}}response=requests.post(WEBHOOK_URL,data=json.dumps(data,ensure_ascii=False).encode("utf-8"),headers={"Content-Type":"application/json"})returnresponse.json()# 发送告警卡片send_card(title="🚨 数据库连接池告警",description="<div class=\"gray\">2024-01-22 03:15:00</div> <div class=\"normal\">服务: user-service</div><div class=\"highlight\">数据库连接数: 198/200 (99%)</div>",url="http://grafana.company.com/d/xxx",btntxt="立即查看")

“卡片消息在群里显示得很规整，点击按钮直接跳转到监控页面。”

第五步：钉钉机器人

“钉钉机器人的逻辑一模一样，只是 API 地址和格式略有不同。”

创建机器人

1. 打开钉钉群 → 群设置 → 智能群助手
2. 添加机器人 → 自定义（通过 Webhook 接入）
3. 复制 Webhook 地址

发送消息

importrequestsimportjsonimporthmacimporthashlibimportbase64importurllib.parseimporttime DINGTALK_WEBHOOK="https://oapi.dingtalk.com/robot/send?access_token=你的token"DINGTALK_SECRET="你的加签密钥"# 如果开启了加签defgenerate_sign(secret):"""生成钉钉加签"""timestamp=str(round(time.time()*1000))secret_enc=secret.encode("utf-8")string_to_sign=f"{timestamp}\n{secret}"string_to_sign_enc=string_to_sign.encode("utf-8")hmac_code=hmac.new(secret_enc,string_to_sign_enc,digestmod=hashlib.sha256).digest()sign=urllib.parse.quote_plus(base64.b64encode(hmac_code))returntimestamp,signdefsend_dingtalk_text(content,at_mobiles=None):"""发送钉钉文本消息"""timestamp,sign=generate_sign(DINGTALK_SECRET)url=f"{DINGTALK_WEBHOOK}&timestamp={timestamp}&sign={sign}"data={"msgtype":"text","text":{"content":content},"at":{"atMobiles":at_mobilesor[],"isAtAll":False}}response=requests.post(url,json=data)returnresponse.json()# 测试send_dingtalk_text("🚨 钉钉告警测试：系统异常，请处理！",at_mobiles=["13800138000"])

“钉钉如果开启了’加签’安全设置，需要额外计算签名。没开启的话，直接 POST 到 Webhook 地址就行。”

第六步：实战——异常自动告警

“来，把告警和异常处理结合起来，做一个真正能用的监控脚本。”

importrequestsimportjsonimportloggingfromdatetimeimportdatetime# 配置WEBHOOK_URL="https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=你的key"CHECK_URL="http://your-service.com/health"# 健康检查接口# 日志logging.basicConfig(filename="monitor.log",level=logging.INFO,format="%(asctime)s - %(message)s")defsend_alert(title,message,level="warning"):"""发送告警到企业微信"""color="red"iflevel=="critical"else"orange"iflevel=="warning"else"blue"markdown=f"""##{title}> **时间**:{datetime.now().strftime("%Y-%m-%d %H:%M:%S")}> **级别**: <font color='{color}'>{level.upper()}</font>{message}"""data={"msgtype":"markdown","markdown":{"content":markdown}}try:response=requests.post(WEBHOOK_URL,data=json.dumps(data,ensure_ascii=False).encode("utf-8"),headers={"Content-Type":"application/json"},timeout=10)logging.info(f"告警已发送:{title}")returnresponse.json()exceptExceptionase:logging.error(f"告警发送失败:{e}")returnNonedefcheck_service():"""检查服务状态"""try:response=requests.get(CHECK_URL,timeout=5)ifresponse.status_code!=200:send_alert(title="🚨 服务异常",message=f"健康检查返回非 200 状态码:{response.status_code}",level="critical")returnFalsedata=response.json()ifdata.get("status")!="ok":send_alert(title="⚠️ 服务告警",message=f"服务状态异常:{data}",level="warning")returnFalselogging.info("服务检查正常")returnTrueexceptrequests.exceptions.Timeout:send_alert(title="🚨 服务超时",message="健康检查请求超时，服务可能无响应",level="critical")returnFalseexceptExceptionase:send_alert(title="🚨 检查失败",message=f"健康检查异常:{str(e)}",level="critical")returnFalse# 执行检查if__name__=="__main__":print("🔍 开始服务检查...")check_service()

“这段代码干了啥？”

“1. 访问服务的健康检查接口
2. 如果状态码不是 200，或者返回内容异常，就发告警到企业微信
3. 捕获超时、异常等各种错误情况
4. 记录日志，方便事后排查”

“配合上一篇学的定时任务，你可以设每 5 分钟检查一次，出问题立刻在群里喊人。”

踩坑提醒：机器人告警的坑

坑 1：消息频率限制

“企业微信和钉钉都有频率限制，发太猛会被禁言。”

“企业微信：每个机器人最多 20 条/秒
钉钉：每个机器人最多 20 条/秒”

"解决办法：

• 合并多条告警为一条
• 加冷却时间（同样的错误 10 分钟内只报一次）
• 分级告警：警告发邮件，严重才发群"

坑 2：Webhook 泄露

“Webhook 地址就是机器人的’钥匙’，泄露了别人就能往你们群里发消息。”

"解决办法：

• 别把 Webhook 地址硬编码在代码里，放环境变量
• 定期更换 Webhook
• 开启 IP 白名单"

importos WEBHOOK_URL=os.environ.get("WECHAT_WEBHOOK_URL")

坑 3：网络不通

“公司内网可能访问不了外网的企业微信/钉钉服务器。”

"解决办法：

• 用公司内网的 IM 系统
• 配置代理
• 或者用邮件告警做备选"

坑 4：告警风暴

“系统挂了，每分钟报一次警，群里 100 条消息，大家直接屏蔽群。”

"解决办法：

• 同一个问题，5 分钟内只报一次
• 恢复后发送’已恢复’通知
• 汇总告警：把同类问题合并成一条"

# 简单的冷却机制last_alert_time={}defsend_alert_with_cooldown(alert_id,title,message,cooldown=300):"""带冷却的告警"""now=time.time()ifalert_idinlast_alert_time:ifnow-last_alert_time[alert_id]<cooldown:return# 冷却中，不发last_alert_time[alert_id]=now send_alert(title,message)

一句话总结

阿明把监控脚本部署到了服务器，设置了每 5 分钟检查一次。

周六凌晨 3 点，数据库连接池满了。3:05，企业微信群里弹出一条消息：

🚨服务异常
时间: 2024-01-22 03:05:00
级别:严重
服务状态异常，请立即处理！

值班同事被@醒，及时处理，避免了事故扩大。

周一，老板在群里说：“这个告警机制不错，推广到其他系统。”

阿明找到老张：“张哥，最后教我一招——怎么把这些全部串起来？”

老张笑了：“下篇，终极篇。”

扩展思考

"机器人告警的应用场景：

• 系统监控：CPU/内存/磁盘超过阈值，自动告警
• 业务监控：订单量突降、支付失败率升高，自动告警
• 数据质量：ETL 任务失败、数据延迟，自动告警
• 安全告警：异常登录、敏感操作，自动告警
• 定时汇报：每天自动发送业务数据摘要到群

“核心逻辑：检测 → 判断 → 通知 → 记录。”

下集预告

下一篇，终极篇。老张带阿明把前面学的所有技能串起来——文件处理、Excel、邮件、定时任务、PDF、爬虫、可视化、机器人告警，打造一个完整的个人自动化工作流。阿明终于实现了"代码在跑，我在喝咖啡"。

记住：机器人告警就像看门大爷——平时睡觉（静默），有风吹草动立刻吹哨子（发消息）。你不需要亲自站岗，只需要告诉他"什么情况下吹哨子"。

你们公司用什么方式告警？邮件？微信？钉钉？还是靠人盯着？欢迎在评论区吐槽。