Python 实战：自动监控数据变化并推送至企业微信外部客户群-编程实验室

QiWe开放平台 · 个人名片
API驱动企微自动化，让开发更高效
核心能力：API 驱动企微自动化，提升开发效率
官方站点：https://www.qiweapi.com
对接通道：进入官方站点联系客服
团队定位：企微生态深度服务，专注 API+RPA 融合技术方案

01. 场景背景

在电商、物流或金融行业，业务人员需要实时掌握关键指标的变化（如：库存跌至警戒线、大额订单产生、系统负载过高等）。相比于被动地打开仪表盘看数据，主动推送至企微外部群能显著提升决策效率。

02. 原创架构设计

我们的逻辑分为三步：

监控端：定时查询数据库或监控 API 接口。
决策端：判断数据变化是否满足“触发阈值”。
推送端：调用企微 API 将格式化后的信息发往目标外部群。

03. 核心代码实现

第一步：数据监控逻辑

这里以监控 Redis 中的库存数据为例：

import redis def check_inventory(sku_id): r = redis.Redis(host='localhost', port=6379, db=0) inventory = int(r.get(f"stock:{sku_id}") or 0) # 阈值判断：少于 10 件即报警 if inventory < 10: return True, inventory return False, inventory

第二步：消息构造与发送

利用我们之前提到的markdown格式，可以让监控信息更具可读性。

import requests import json def push_alert(chat_id, sku_name, stock_count): # 假设已有获取 token 的方法 access_token = get_access_token() url = f"https://qyapi.weixin.qq.com/cgi-bin/externalcontact/groupchat/send?access_token={access_token}" # 构造 Markdown 消息内容 content = f""" ### 🚨 库存预警通知 **商品名称**：<font color=\"info\">{sku_name}</font> **当前库存**：<font color=\"warning\">{stock_count}</font> **处理建议**：请尽快联系供应商补货。 > 来自自动化监控系统 """ payload = { "chat_id": chat_id, "msgtype": "markdown", "markdown": { "content": content } } response = requests.post(url, data=json.dumps(payload)) return response.json()

04. 进阶：如何避免重复推送？

为了防止在库存还没补上的情况下，每分钟都发一次报警，我们需要引入**“告警收敛”**逻辑：

状态记录：在 Redis 中记录该sku_id的报警状态。
时间窗口：设定 1 小时内仅对同一事件发送一次通知。

def should_i_push(sku_id): lock_key = f"alert_lock:{sku_id}" if r.setnx(lock_key, 1): # 设置成功代表当前未在锁定期 r.expire(lock_key, 3600) # 锁定 1 小时 return True return False