如何用 Python 发送告警通知到钉钉？

作者：老表

2021 年 12 月 09 日
本文字数：5797 字
阅读完需：约 19 分钟

这是老表 12 月的第 3 篇更文分享～记得关注我，坚持分享编程开发、数据分析、机器学习等学习笔记。

跟老表一起学云服务器开发相关文章（如果是第一次阅读该系列文章， 强烈建议先学习下面文章）：

先导篇：拥有有一台服务器后，我竟然这么酷？

替代项目：10行代码写一个简历页面！

和不安全访问 Say goodbye，手把手教大家如何给域名申请免费 SSL 证书

Linux里的宝塔，真正的宝塔！详细教程

终于有了一个人人可以访问的网站了

如何用 Python 发送告警通知到钉钉？

一、前言

前不久，看到了明哥写的如何用 Python 发送警告通知到企业微信，想起来之前写过用 Pytho 发送指定格式数据到钉钉的服务，本文将之前的代码重构下，变成一个：利用 Python 监控服务器数据，然后有异常就通过钉钉发送给用户。

本项目大纲如下，项目已经开源到 GitHub 啦，大家可以直接点击阅读原文或者浏览器访问：https://github.com/XksA-me/DingdingBot 下载项目。

项目环境说明：

Python 3.6.8 （理论 3.6 及以上肯定可以）
第三方依赖库：
requests 发送 post 请求，发送数据
psutil 获取操作系统运行相关数据
apscheduler 设置定时任务

因为相关依赖较少，你可以直接在本地环境安装使用，也可以创建一个虚拟环境安装使用（Python 虚拟环境推荐使用 pipenv 进行管理）。

进入环境后，输入下面 pip 指令进行安装：

pip3 install requests psutil apscheduler

复制代码

二、开始动手动脑

2.1 创建钉钉机器人

钉钉机器人个人版只针对群聊，所以我们需要先建一个群，打开钉钉，然后创建一个群聊，随便拉 2 个人，创建成功后，可以把这两个好友再移除群聊（有点损？！），点击群设置中的智能群助手。

进入到机器人管理页面，点击添加机器人后的三点按钮，进入机器人选择页面。

我们下滑页面，选择自定义机器人。

给机器人取个名字，然后需要进行安全设置，选择加签方式（数据传输是需要其当作参数，避免安全问题），复制好里面的内容。点击完成，即可完成创建。

复制 Webhook 链接，后面我们就是通过 Python 向这个 url 发送 post 请求进行数据传输，你可以点击下设置说明查看机器人相关功能和配置方法。

如果后面忘记了前面设置的加签密钥或者Webhook地址，可以群管理->智能群助手->点击对应机器人的三点按钮 查看或修改。

这样我们就创建好了钉钉机器人，接下来，我们只需编编写好 Python 代码即可。

2.2 编写一个简单钉钉消息传输助手

2.2.1 计算数字加签内容，为自动发送消息做准备

前面我们设置了加签的安全防护方法，所以我们在进行数据传输之前，首先得先计算下钉钉机器人数字签名内容，钉钉文档上有非常详细的说明哈，并给了各种语言的计算方法，我们直接拿来调试即可。

官方文档地址：https://developers.dingtalk.com/document/robots/customize-robot-security-settings

计算过程你可以不用理解（下面代码），修改secret为你自己的即可。

import hmacimport hashlibimport base64import urllib.parsefrom time import time

'''钉钉机器人数字签名计算'''def get_digest():    # 取毫秒级别时间戳，round(x, n) 取x小数点后n位的结果，默认取整    timestamp = str(round(time() * 1000))    secret = '你自己的加签内容'    secret_enc = secret.encode('utf-8')  # utf-8编码    string_to_sign = '{}\n{}'.format(timestamp, secret)  # 字符串格式化拼接    string_to_sign_enc = string_to_sign.encode('utf-8')  # utf-8编码    hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()  # HmacSHA256算法计算签名    sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))  # Base64编码后进行urlEncode    #  返回时间戳和计算好的编码拼接字符串，后面直接拼接到Webhook即可    return f"&timestamp={timestamp}&sign={sign}"

复制代码

2.2.2 Post 请求发送数据，实现自动发送消息到钉钉

我们直接向我们自己的 Webhook 地址发送 post 请求传输数据即可，这里我使用的 markdown 数据类型，还有很多其他数据格式可以选择，如文本、跳转卡片、信息流卡片等，非常丰富。

大家有兴趣可以前往官网查看：https://developers.dingtalk.com/document/robots/custom-robot-access?spm=ding_open_doc.document.0.0.62846573TCzj7A#topic-2026027

需要注意的是，如果你需要机器人在发送消息的时候还 @指定的人，那么你需要在内容中（下面代码中的@xxxxxx）也加上@指定人的钉钉手机号。

# 简单发送markdown消息def warning_bot():    data = {        "msgtype": "markdown",        "markdown": {            "title": "【简说Python】今日福利",            "text": """### 福利介绍@xxxxxx 你好，我是老表，和我一起学习Python、云服务器开发等编程知识啦。\n>![](https://p3-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/db15766f66e54a228b4c637c19163ecf~tplv-k3u1fbpfcp-zoom-1.image)**[老表的个人博客，已上线](https://python-brief.com/)**\n            """        },        "at": {          "atMobiles": [              "xxxxxx"  # 要@对象的手机号          ],        }    }    # 机器人链接地址，发post请求 向钉钉机器人传递指令    webhook_url = '你的Webhookurl'    # 利用requests发送post请求    req = requests.post(webhook_url+get_digest(), json=data)

复制代码

显示效果如下（上面是聊天消息栏，显示了我们指定的标题；下面是群聊内，显示了 markdown 渲染后的效果），好看，记得点赞（想不到，到这里，就有一千字了～点赞呐转发呐支持下作者～）：

2.3 编写统计系统基本数据的函数

在 Linux 里我们一般利用top指令来查看 CPU 的使用情况，主要看以下几个数据：进程 CPU 使用率、负载情况、虚拟/物理内存使用情况，所以本部分我们将利用 Python 获取相关数据。

这里我们利用psutil，Process and System utilities（进程和系统实用工具），用于检索系统运行的进程和系统使用率（CPU，内存，磁盘，网络，传感器）信息的跨平台库，通过几行代码就可以获取到本地系统相关数据啦～（本文点赞过 50，就更新一期专门介绍psutil的文章）～

import psutil as psuimport os

'''云服务器基础数据服务器已运行时间、负载状态、CPU使用率、运行内存使用率、物理内存使用率'''def get_server_info():    # 获取系统的基本数据    # 服务器已运行时间=现在时间和服务器开启时间之差    run_times = str(timedelta(seconds=int(time())-int(psu.boot_time())))    # 系统负载状态（最近1、5、15分钟）    loadavg = [round(i, 2) for i in os.getloadavg()]    # CPU使用率 测试间隔0.3秒    cpu_in_use = psu.cpu_percent(interval=0.3)    # 系统运行内存使用率    # 内存使用率大于80% 触发报警    vm_in_use = psu.virtual_memory().percent    vm_available = round(psu.virtual_memory().available/(1024**3), 2)    # 系统物理存储使用率    disk_in_use = psu.disk_usage('/').percent    disk_free = round(psu.disk_usage('/').free/(1024**3), 2)        # 还可以添加进程、线程等信息，后面专门安排一篇文章写        base_info = f"""> 您的云服务器已运行-{run_times}，机器负载情况为(最近1、5、15分钟)：{loadavg}- 目前CPU使用率为：{cpu_in_use}%，- 系统运行内存使用率为：{vm_in_use}%，- 剩余可用运行内存为：{vm_available}GiB，- 系统存储内存使用率为：{disk_in_use}%，- 剩余可用存储内存为：{disk_free}GiB<br>**{'机器CPU使用率正常' if cpu_in_use<=80 else '机器CPU使用率过高，可能触发预警'}**"""    return base_info, loadavg, cpu_in_use, vm_in_use, disk_in_use

复制代码

代码注释写的都还比较清楚啦，主要是调用了 psutil 内置的一些方法，我们获取到了操作系统的一些基本数据，另外系统负载情况我们是利用 os 模块的内置方法getloadavg获取的，最后对数据进行了拼接，拼接的时候我们使用到了 f-string，点赞破 50 出一期文章给大家介绍，这里主要给大家额外说下 GB 和 GiB 的区别：

1GB = 1000MB = 1000*1000 KB = 1000*1000*1000 B1GiB = 1024MiB = 1024*1024 KiB = 1024*1024*1024 B# 有机会和大家细聊 有意思

复制代码

2.4 编写一个监控数据的函数

接下来我们还需要写一个监控函数，这里只做了两个简单监控（项目已开源，后面有时间我也会），一个是对负载的情况进行监控，另外一个是对 CPU 使用率进行监控，设置了两个阀值：CPU 使用率 70%和负载（1 分钟和 15 分钟）70%*CPU 数量，当超过阀值就返回对应报警内容，没有问题就返回ok。

截图 @阮一峰老师的博客

'''服务器预警设置本篇先简单点，只设置负载和CPU使用率预警'''def get_warning():    base_info, loadavg, cpu_in_use, vm_in_use, disk_in_use = get_server_info()    # 首先判断服务器负载情况    # 只看近一分钟和近十五分钟情况 应该<= 0.7*CPU数量    loadavg_max = psu.cpu_count() * 0.7    loadavg_max = 0.01   # 测试使用，正式环境请注释掉    if loadavg[0] >= loadavg_max and loadavg[2] >= loadavg_max:        warning1 = f'⚠️<font color="#d30c0c">【警告】</font>您的云服务器当前负载率为(最近1、5、15分钟)-{loadavg}，负载率已达<font color="#d30c0c">{round(loadavg[2]/loadavg_max, 2)*100}%</font>，请及时检查系统是否存在问题，也可以@我，发送：基础信息，查看云服务器基础信息。'        return warning1            if cpu_in_use >= 80:        warning2 = f'⚠️<font color="#d30c0c">【警告】</font>您的云服务器当前CPU使用率为<font color="#d30c0c">{cpu_in_use}%</font>，请及时检查系统是否存在问题，也可以@我，发送：基础信息，查看云服务器基础信息。'        return warning2    return 'ok'

复制代码

本来还想写一个功能的，就是前面说的时时查看数据（@下机器人，给出对应指令就行），但是到这里发现这个功能需要新建一个企业机器人，下次再给大家分享吧～

2.5 写定时任务

在 Linux 上面，我们可以直接利用之前讲过的宝塔面板设置定时任务，可以看Linux里的宝塔，真正的宝塔！详细教程，这里我们自己写定时任务，我们用到了apscheduler这个第三方库，关于这个库又可以写一篇长文介绍了，这里就不展开说了，本文点赞如果破 100，我下周安排（老表在线卑微求赞！）。

from apscheduler.schedulers.blocking import BlockingScheduler

'''1、每天早上9:00 发送服务器情况到钉群'''def every_day_nine():    message = get_server_info()[0]    title = '服务器基础信息'    warning_bot(message, title)

'''2、时时预警（每30秒检测一次）'''def every_seconds_30():    warning = get_warning()    if warning != 'ok':        title = '【⚠️警告】服务器故障'        warning_bot(warning, title)
'''3、@机器人，自动问答设置下一篇安排，需要另外新建一个企业机器人和群聊机器人流程不一样～'''        # 选择BlockingScheduler调度器sched = BlockingScheduler(timezone='Asia/Shanghai')
# job_every_nine 每天早上9点运行一次  日常发送sched.add_job(every_day_nine, 'cron', hour=21, minute=55)
# every_seconds_30 每30s执行一次  数据监控sched.add_job(every_seconds_30, 'interval', seconds=3)
# 启动定时任务sched.start()

复制代码

add_job 设置定时任务中的interval表示周期性触发触发，比如每分钟；cron是 apscheduler 中功能最强的触发器，可以具体到每个月的某个时间触发，比如每天早上 9:00。

2.6 运行，来看看效果啦

目前都属于自动触发，然后发送消息，首先是每日定时早上 9:00 发送服务器基本情况。

其次是每 30s 进行一次服务器数据检测（CPU 使用率和负载），当数据超过阀值时，触发报警，发送消息提醒。

2.7 给程序创建守护进程

经过上面我们完成了功能开发，但是会发现，一旦我们关闭程序，提醒监测服务也会停止，所以我们需要创建一个守护进程来保护我们的进程。

以我自己为例，我们登录宝塔面板后，进入/etc/systemd/system文件夹下，新建一个ding_bot.service文件，并写入下面内容：

[Unit]Description=Dingding Bot service
[Service]Type=forkingExecStart=/usr/bin/python3 /root/Project/Little_project/DingdingBot/scheduler.pyKillMode=processRestart=on-failureRestartSec=3s
[Install]WantedBy=multi-user.target

复制代码

简单解释下Service里设置的含义，Type=forking表示程序启动后，会放到后台运行；ExecStart服务的具体执行指令（执行 scheduler.py 文件即可）；KillMode=process表示服务停止的同时也会杀死程序主进程；Restart=on-failure表示系统发生意外导致程序退出时，程序自动重启。

保存好文件后，我们直接终端内执行下面指令即可开启进程守护，运行后会进入守护进程状态，我们可以按 ctrl+c 退出，不会影响守护进程：

systemctl start ding_bot

复制代码

代码修改后，需要重启守护进程，修改代码才会生效，重启指令如下：

systemctl restart ding_bot

复制代码

如果不想设置这个守护进程了，执行 stop 指令可以停止该 service（程序也会停止），指令如下：

systemctl stop ding_bot

复制代码

关于守护进程 system 其他的相关指令和操作可以自行搜查哈，也可以留言区交流，展开讲又是一篇推文啦～点赞破 50，安排～

三、下期预告

整体效果感觉还不错，嘿嘿嘿，后面有时间会继续优化，大家有什么额外想法和需求，也可以留言说说哈。

另外预计这周还会出一篇相关文章：@机器人输入关键词可以自动回复功能实现，那个需要使用企业机器人，流程稍微麻烦点～尽量本周出吧～另外文中提到的psutil和apscheduler库、守护进程详细介绍，有时间有场景就给大家分享，或者本期推文点赞到 50+50+50～

跟老表一起学云服务器开发相关文章（如果是第一次阅读该系列文章， 强烈建议先学习下面文章）：

先导篇：拥有有一台服务器后，我竟然这么酷？

替代项目：10行代码写一个简历页面！

和不安全访问 Say goodbye，手把手教大家如何给域名申请免费 SSL 证书

Linux里的宝塔，真正的宝塔！详细教程

终于有了一个人人可以访问的网站了

创作不易，大家多点赞留言转发支持～一直在求点赞，其实留言更重要啦，因为留言能更直观的让我看到大家的支持，和大家互动、交流，我想也能触发我的很多创作灵感。

好的，那么下期见，我是爱猫爱技术，更爱思思的老表⁽⁽ଘ( ˙꒳˙ )ଓ⁾⁾

发布于: 1 小时前阅读数: 10

原文链接:【http://xie.infoq.cn/article/0d31f8da82b3191a993e50054】。文章转载请联系作者。

老表

关注

公众号｜简说Python 2018.09.23 加入

【公众号：简说Python】爱猫爱技术，Python终身学习者、数据分析爱好者、Go语言内卷机。

发布

暂无评论

创作场景

如何用 Python 发送告警通知到钉钉？

如何用 Python 发送告警通知到钉钉？

一、前言

二、开始动手动脑

2.1 创建钉钉机器人

2.2 编写一个简单钉钉消息传输助手

2.2.1 计算数字加签内容，为自动发送消息做准备

2.2.2 Post 请求发送数据，实现自动发送消息到钉钉

2.3 编写统计系统基本数据的函数

2.4 编写一个监控数据的函数

2.5 写定时任务

2.6 运行，来看看效果啦

2.7 给程序创建守护进程

三、下期预告

老表

评论