
你是不是对下述场景很熟悉:周五下午五点半,正收拾东西准备下班,企业微信群里突然弹出一条消息:
📊 日报 | 2026-06-05今日新增用户:1,283人(↑12.4%)订单量:856笔(↓3.1%)服务器CPU峰值:72%异常告警:0次——以上由监控机器人自动生成
隔壁工位的探过头来:“咱组什么时候来的运维?这日报做得还挺专业。”
我差点没忍住笑——哪有什么运维,就是一个Python脚本加一个企业微信机器人,前后加起来不到50行代码。
但这50行代码,确实让我的日子好过了不少。今天就把这套东西掰开了讲讲,顺便把我踩过的坑也一并交代了。
往期阅读>>>
Python 自动化管理Jenkins的15个实用脚本,提升效率
App2Docker:如何无需编写Dockerfile也可以创建容器镜像
Python 自动化识别Nginx配置并导出为excel文件,提升Nginx管理效率
企业微信群机器人的本质就一个东西:Webhook。
你在群里添加一个机器人,企业微信会给你一个URL。你往这个URL发一个特定格式的POST请求,群里就会收到对应的消息。就这么回事。
不需要申请什么开发者资质,不需要看一堆SDK文档,甚至不需要有服务器——只要你的脚本能发HTTP请求,就能往群里推消息。
创建机器人也很简单:打开企业微信群 → 右上角「…」→「添加群机器人」→「新建」→ 给机器人起个名字 → 完成。然后你会拿到一个类似这样的URL:
https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx这个key就是机器人的唯一身份标识,千万别泄露出去,否则谁都能往你群里发消息。
先装个 requests 库(如果你还没装的话):
pip install requests然后写一个最基础的消息发送函数:
importrequestsimportjsonWEBHOOK_URL = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=你的key"def send_text(content, mentioned_list=None):data = {"msgtype": "text","text": {"content": content,"mentioned_list": mentioned_listor [] } }resp = requests.post(WEBHOOK_URL, json=data)return resp.json()# 测试一下send_text("Hello,我是你的专属运维机器人 🤖")
跑一下,群里应该就收到消息了。对,就这么几行,我当时也不信。
别急,文本消息只是开胃菜。企业微信机器人还支持好几种消息类型,Markdown才是真正的大杀器。
纯文本消息发出来密密麻麻一坨,没人爱看。换成Markdown格式,立刻就不一样了:
def send_markdown(content):data = {"msgtype": "markdown","markdown": {"content": content } }resp = requests.post(WEBHOOK_URL, json=data)return resp.json()# 发一条漂亮的日报report = """## 日报 | 2026-06-05> **新增用户**:<font color="info">1,283人</font>(↑12.4%)> **订单量**:<font color="warning">856笔</font>(↓3.1%)> **服务器CPU峰值**:72%> **异常告警**:<font color="info">0次</font>[查看详细报告](https://your-dashboard.com)"""send_markdown(report)
这里有个小技巧:企业微信的Markdown支持 <font color="info"> 这种写法,info 显示绿色,warning 显示橙色,comment 显示灰色。善用颜色标注,能让你的日报看起来真的很“运维”。
不过要注意,企业微信的Markdown不是完整的Markdown语法,只支持一部分。比如不支持表格、不支持图片内嵌、标题渲染效果也比较有限。别指望它跟GitHub的README一样灵活,当个通知模板用绰绰有余。
光知道怎么发消息没用,关键是接什么脚本。下面这三个场景是我实际在用的,每个都帮我解决过真实问题。
场景一:服务器状态监控
这是最基本的,也是我最先搞起来的。起因是有次线上服务器CPU飙到98%整整一个下午,我愣是刷监控面板才发现。后来我就写了个脚本,每隔几分钟自动检查,有问题直接往群里喊:
importpsutilimportdatetimedef check_server():cpu = psutil.cpu_percent(interval=1)mem = psutil.virtual_memory().percentdisk = psutil.disk_usage('/').percentif cpu>85 or mem>90 or disk>90:alert = f"""## 服务器告警> **时间**:{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}> **CPU使用率**:<font color="warning">{cpu}%</font>> **内存使用率**:<font color="warning">{mem}%</font>> **磁盘使用率**:<font color="warning">{disk}%</font>请相关人员尽快排查! """send_markdown(alert)# 没问题就不说话,安静如鸡
配合crontab每5分钟跑一次,出了问题群里立马就能收到通知。没出问题的话,机器人安安静静的,不会烦你。
场景二:定时数据日报
这个用得太多了。每天固定时间从数据库拉数据,生成日报推到群里。我们组之前都是数据分析师手动做Excel截图发群,后来她请了两周假,没人出日报了,老板急了。我就花了一个下午把这个脚本写出来,从此日报再也没断过:
import pymysqlfrom datetime import datetime, timedeltadef daily_report():conn = pymysql.connect(host='xxx', user='xxx', password='xxx', db='xxx')cursor = conn.cursor()yesterday = (datetime.now() -timedelta(days=1)).strftime('%Y-%m-%d')cursor.execute(""" SELECT COUNT(*), SUM(amount) FROM orders WHERE DATE(created_at) = %s """, (yesterday,))order_count, order_amount = cursor.fetchone()cursor.execute("SELECT COUNT(*) FROM users WHERE DATE(created_at) = %s", (yesterday,))new_users = cursor.fetchone()[0]report = f"""## 业务日报 | {yesterday}> **新增用户**:<font color="info">{new_users}人</font>> **订单数**:{order_count}笔> **订单金额**:<font color="info">¥{order_amount:,.2f}</font>数据截止 {datetime.now().strftime('%H:%M')},如有疑问请联系数据组。 """send_markdown(report)conn.close()
自从搞了这个,产品经理再也不用每天早上@我问“昨天数据出了没”。群里自己看就行。
场景三:脚本异常通知
这个特别实用。我们有个数据清洗脚本,每天凌晨跑,处理前一天的用户行为日志,快的话40分钟,慢的话两三个小时都有。以前我每天起来第一件事就是去服务器看跑完没,后来实在受不了了,就改成让它自己汇报:
import tracebackimport timedef run_etl_job():start = time.time()try:send_text("数据清洗任务开始执行...")process_user_behavior_logs() # 你的核心业务逻辑elapsed = round(time.time() -start, 1)send_markdown(f"""## 数据清洗任务完成> **耗时**:{elapsed}秒> **状态**:<font color="info">成功</font> """)except Exceptionase:elapsed = round(time.time() -start, 1)error_msg = traceback.format_exc()send_markdown(f"""## 数据清洗任务失败> **耗时**:{elapsed}秒> **错误信息**:> ```> {error_msg[:500]}> ```请 <font color="warning">@我</font> 尽快排查! """)
跑完自动汇报,挂了自动报警。自从接了这个,我再也没早起盯过服务器。群里一条消息比啥都靠谱。
坑一:每分钟20条消息限制
这是我踩的第一个坑。当时写了个批量检查脚本,50台服务器每台检查完都发一条消息,结果从第21条开始全部被吞了,群里啥也看不到。
企业微信的规定是:单个机器人每分钟最多发20条消息,超过了会被限流,返回错误码。
解决方案也简单——合并消息。别每台服务器发一条,攒起来发一条汇总:
def check_all_servers(servers):results = []for serverinservers:status = check_one_server(server)results.append(f"> **{server}**:{status}")summary = "##服务器巡检报告\n"+"\n".join(results)send_markdown(summary) # 一条消息搞定
如果你的场景确实需要高频发送(比如实时告警),那就搞多个机器人轮着用,或者引入消息队列做缓冲。
坑二:Markdown内容长度限制
企业微信的Markdown消息有4096字节的限制。注意是字节,不是字符。中文字符在UTF-8下占3个字节,所以实际上你能写的内容比想象的少。
我有一次发巡检报告,服务器多了点,消息太长,直接被截断了,后半截直接没了,而且不会报错,你根本不知道消息被截断了。
所以如果内容可能超限,一定要做截断处理:
def send_markdown_safe(content, max_bytes=3800):"""留点余量,别卡着4096"""encoded = content.encode('utf-8')if len(encoded) >max_bytes:content = encoded[:max_bytes].decode('utf-8', errors='ignore')content += "\n\n...(内容过长已截断)"return send_markdown(content)
坑三:Webhook地址泄露
这个事儿说起来有点丢人。我之前把Webhook URL硬编码在脚本里,然后脚本不小心推到了公开的GitHub仓库。没过两天,群里开始收到莫名其妙的广告消息。
所以,Webhook地址一定要放环境变量或者配置文件里,别写死在代码中:
import osWEBHOOK_URL = os.environ.get("WECOM_WEBHOOK_URL")if not WEBHOOK_URL:raise ValueError("请设置环境变量 WECOM_WEBHOOK_URL")
如果已经泄露了,在企业微信群设置里可以重置机器人的key,旧的URL会立即失效。
坑四:图片消息的坑
企业微信机器人发图片需要先上传图片拿到 media_id,但Webhook机器人的上传接口只支持文件类型(file),图片类型得走其他途径(比如企业微信的应用消息接口)。
如果你的场景只是想把监控截图之类的发到群里,有个省事的变通方案:把图片传到OSS或者内网图床,然后在Markdown里贴个链接。虽然不能直接在消息里内嵌图片,但至少让人能点过去看。我试过用curl上传图片,结果shell里的引号转义把我搞疯了,最后还是贴链接最省心。
到这一步,你的机器人已经能干活了。但如果你想让它在群里看起来不那么像个“机器”,还可以加点小细节。
加上@提醒功能
普通文本消息支持 mentioned_list 参数,可以直接@群里的人。关键告警的时候特别好用:
send_text("数据库连接池即将耗尽,当前连接数 95/100",mentioned_list=["@all"] # @所有人# 或者指定具体的人 ["zhangsan", "lisi"])
分时段调整消息策略
上班时间和下班时间发不同的内容,别半夜三更往群里推数据:
from datetime import datetimedef should_send():hour = datetime.now().hourweekday = datetime.now().weekday()# 周末不发日报if weekday>= 5:return False# 非工作时间只发紧急告警if hour<9orhour>20:return "critical_only"return True
封装成一个通用的通知模块
当你有好几个项目都要用机器人通知时,最好封装一下,别到处复制粘贴那段 requests.post:
# notifier.pyimport osimport requestsclass WeComBot:def __init__(self, webhook_url=None):self.url = webhook_urloros.environ.get("WECOM_WEBHOOK_URL")def text(self, content, at_all=False):mentioned = ["@all"] ifat_allelse []return self._send({"msgtype": "text","text": {"content": content, "mentioned_list": mentioned} })def markdown(self, content):return self._send({"msgtype": "markdown","markdown": {"content": content} })def _send(self, data):resp = requests.post(self.url, json=data, timeout=10)result = resp.json()if result.get("errcode") != 0:print(f"[WeComBot] 发送失败: {result}")return result# 用起来就这么简单bot = WeComBot()bot.text("部署完成")bot.markdown("## 告警\n> CPU飙到90%了")
这样一个文件丢到项目里,谁都能用,一行代码就能发通知。
| 类型 | 大小限制 | 能不能@人 | 适合干嘛 |
|---|---|---|---|
| text | 2048字节 | ✅ 支持 | 简短通知、告警、@具体的人 |
| markdown | 4096字节 | ❌ 不支持 | 日报、周报、格式化报告 |
| image | 2MB | ❌ | 监控截图(需先上传拿media_id) |
| news | 标题+描述+链接 | ❌ | 推送文章、链接卡片 |
| file | 20MB | ❌ | 发送文件(需先上传拿media_id) |
说实话,这套东西技术含量真不高。但架不住效果好。手上有三四个项目都接了企业微信机器人,日报、告警、部署通知、定时巡检全靠它推。有时候半夜收到一条告警消息,迷迷糊糊打开手机看一眼,心里反而踏实——至少知道系统在替我盯着呢。
在同事眼里,你从一个“写脚本的”变成了一个“有自动化体系的”。别小看这个印象差,年终述职的时候你就明白了。
你们公司用什么消息通知方案?有接到钉钉或者飞书机器人的也可以聊聊。
https://ima.qq.com/wiki/?shareId=f2628818f0874da17b71ffa0e5e8408114e7dbad46f1745bbd1cc1365277631c
