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

机器人告警让 Python 替你喊人本文基于 Python 3.9涉及企业微信/钉钉 Webhook。阅读时间约 10 分钟。周末的静默灾难周六晚上阿明正在家陪娃搭积木手机突然响了。是老板“阿明系统是不是挂了客户说登不上。”阿明心里一紧打开 VPN 连公司内网果然——数据库连接池满了服务全崩了。他花了两个小时抢修恢复后一看日志——问题在周六凌晨 3 点就出现了只是没人发现。周一早上阿明顶着黑眼圈到公司被老板叫进办公室“为什么周末出问题了没人知道”阿明委屈“我在家带娃呢不可能 24 小时盯着屏幕……”那就让系统自己’喊人’啊。老板皱眉。阿明找到老张“张哥系统出问题了能自动发消息到微信/钉钉吗”老张“能。让 Python 当你的’看门大爷’出事了替你所有人。”第一步认识 Webhook——机器人的电话号码“阿明你知道怎么让程序往微信群里发消息吗”“不知道……”“用Webhook。你可以把它理解为机器人的’电话号码’——你往这个地址发一条 HTTP 请求机器人就在群里喊一嗓子。”老张在白纸上画了个图你的 Python 脚本 → 发送 HTTP POST 请求 → 企业微信/钉钉服务器 → 群里出现消息“不需要登录微信不需要扫码只要一个 URL就能发消息。”第二步企业微信机器人“先教你企业微信的钉钉类似。”创建机器人打开企业微信群聊右键群设置 → 添加群机器人选择【新建机器人】取个名字比如告警小助手复制Webhook 地址“Webhook 地址长这样”https://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyxxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx“这个key就是机器人的身份标识别泄露不然别人也能往你们群里发消息。”发送文本消息importrequestsimportjson WEBHOOK_URLhttps://qyapi.weixin.qq.com/cgi-bin/webhook/send?key你的keydefsend_text(content,mentioned_listNone):发送文本消息到企业微信data{msgtype:text,text:{content:content,mentioned_list:mentioned_listor[],# 指定人}}responserequests.post(WEBHOOK_URL,datajson.dumps(data,ensure_asciiFalse).encode(utf-8),headers{Content-Type:application/json})resultresponse.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}}responserequests.post(WEBHOOK_URL,datajson.dumps(data,ensure_asciiFalse).encode(utf-8),headers{Content-Type:application/json})returnresponse.json()# 发送格式化的告警信息markdown_content## 系统告警 **告警时间**: 2024-01-22 03:15:00 **告警级别**: font colorred严重/font **异常信息**: - 服务: user-service - 错误: 数据库连接池耗尽 - 影响: 用户无法登录 **处理建议**: 1. 检查数据库连接数 2. 重启连接池 3. 查看慢查询日志 [点击查看详情](http://grafana.company.com/d/xxx) send_markdown(markdown_content)“企业微信的 Markdown 支持有限但基本的加粗、颜色、链接都够用。”“font colorred严重/font是标红让关键信息一眼看到。”第四步发送卡片消息最专业“如果要发得像个正经告警系统用卡片消息——带标题、内容、跳转链接。”defsend_card(title,description,url,btntxt查看详情):发送文本卡片data{msgtype:textcard,textcard:{title:title,description:description,url:url,btntxt:btntxt}}responserequests.post(WEBHOOK_URL,datajson.dumps(data,ensure_asciiFalse).encode(utf-8),headers{Content-Type:application/json})returnresponse.json()# 发送告警卡片send_card(title 数据库连接池告警,descriptiondiv class\gray\2024-01-22 03:15:00/div div class\normal\服务: user-service/divdiv class\highlight\数据库连接数: 198/200 (99%)/div,urlhttp://grafana.company.com/d/xxx,btntxt立即查看)“卡片消息在群里显示得很规整点击按钮直接跳转到监控页面。”第五步钉钉机器人“钉钉机器人的逻辑一模一样只是 API 地址和格式略有不同。”创建机器人打开钉钉群 → 群设置 → 智能群助手添加机器人 → 自定义通过 Webhook 接入复制 Webhook 地址发送消息importrequestsimportjsonimporthmacimporthashlibimportbase64importurllib.parseimporttime DINGTALK_WEBHOOKhttps://oapi.dingtalk.com/robot/send?access_token你的tokenDINGTALK_SECRET你的加签密钥# 如果开启了加签defgenerate_sign(secret):生成钉钉加签timestampstr(round(time.time()*1000))secret_encsecret.encode(utf-8)string_to_signf{timestamp}\n{secret}string_to_sign_encstring_to_sign.encode(utf-8)hmac_codehmac.new(secret_enc,string_to_sign_enc,digestmodhashlib.sha256).digest()signurllib.parse.quote_plus(base64.b64encode(hmac_code))returntimestamp,signdefsend_dingtalk_text(content,at_mobilesNone):发送钉钉文本消息timestamp,signgenerate_sign(DINGTALK_SECRET)urlf{DINGTALK_WEBHOOK}timestamp{timestamp}sign{sign}data{msgtype:text,text:{content:content},at:{atMobiles:at_mobilesor[],isAtAll:False}}responserequests.post(url,jsondata)returnresponse.json()# 测试send_dingtalk_text( 钉钉告警测试系统异常请处理,at_mobiles[13800138000])“钉钉如果开启了’加签’安全设置需要额外计算签名。没开启的话直接 POST 到 Webhook 地址就行。”第六步实战——异常自动告警“来把告警和异常处理结合起来做一个真正能用的监控脚本。”importrequestsimportjsonimportloggingfromdatetimeimportdatetime# 配置WEBHOOK_URLhttps://qyapi.weixin.qq.com/cgi-bin/webhook/send?key你的keyCHECK_URLhttp://your-service.com/health# 健康检查接口# 日志logging.basicConfig(filenamemonitor.log,levellogging.INFO,format%(asctime)s - %(message)s)defsend_alert(title,message,levelwarning):发送告警到企业微信colorrediflevelcriticalelseorangeiflevelwarningelsebluemarkdownf##{title} **时间**:{datetime.now().strftime(%Y-%m-%d %H:%M:%S)} **级别**: font color{color}{level.upper()}/font{message}data{msgtype:markdown,markdown:{content:markdown}}try:responserequests.post(WEBHOOK_URL,datajson.dumps(data,ensure_asciiFalse).encode(utf-8),headers{Content-Type:application/json},timeout10)logging.info(f告警已发送:{title})returnresponse.json()exceptExceptionase:logging.error(f告警发送失败:{e})returnNonedefcheck_service():检查服务状态try:responserequests.get(CHECK_URL,timeout5)ifresponse.status_code!200:send_alert(title 服务异常,messagef健康检查返回非 200 状态码:{response.status_code},levelcritical)returnFalsedataresponse.json()ifdata.get(status)!ok:send_alert(title⚠️ 服务告警,messagef服务状态异常:{data},levelwarning)returnFalselogging.info(服务检查正常)returnTrueexceptrequests.exceptions.Timeout:send_alert(title 服务超时,message健康检查请求超时服务可能无响应,levelcritical)returnFalseexceptExceptionase:send_alert(title 检查失败,messagef健康检查异常:{str(e)},levelcritical)returnFalse# 执行检查if__name____main__:print( 开始服务检查...)check_service()“这段代码干了啥”“1. 访问服务的健康检查接口2. 如果状态码不是 200或者返回内容异常就发告警到企业微信3. 捕获超时、异常等各种错误情况4. 记录日志方便事后排查”“配合上一篇学的定时任务你可以设每 5 分钟检查一次出问题立刻在群里喊人。”踩坑提醒机器人告警的坑坑 1消息频率限制“企业微信和钉钉都有频率限制发太猛会被禁言。”“企业微信每个机器人最多 20 条/秒钉钉每个机器人最多 20 条/秒”解决办法合并多条告警为一条加冷却时间同样的错误 10 分钟内只报一次分级告警警告发邮件严重才发群坑 2Webhook 泄露“Webhook 地址就是机器人的’钥匙’泄露了别人就能往你们群里发消息。”解决办法别把 Webhook 地址硬编码在代码里放环境变量定期更换 Webhook开启 IP 白名单importos WEBHOOK_URLos.environ.get(WECHAT_WEBHOOK_URL)坑 3网络不通“公司内网可能访问不了外网的企业微信/钉钉服务器。”解决办法用公司内网的 IM 系统配置代理或者用邮件告警做备选坑 4告警风暴“系统挂了每分钟报一次警群里 100 条消息大家直接屏蔽群。”解决办法同一个问题5 分钟内只报一次恢复后发送’已恢复’通知汇总告警把同类问题合并成一条# 简单的冷却机制last_alert_time{}defsend_alert_with_cooldown(alert_id,title,message,cooldown300):带冷却的告警nowtime.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、爬虫、可视化、机器人告警打造一个完整的个人自动化工作流。阿明终于实现了代码在跑我在喝咖啡。记住机器人告警就像看门大爷——平时睡觉静默有风吹草动立刻吹哨子发消息。你不需要亲自站岗只需要告诉他什么情况下吹哨子。你们公司用什么方式告警邮件微信钉钉还是靠人盯着欢迎在评论区吐槽。