零门槛实战：大促期间自动监控库存与订单预警系统

一、前置准备与环境搭建

在大促期间，运营人员需要实时监控核心商品的库存余量以及订单增量，防止超卖或库存积压。本文将手把手教你使用Python构建一套零成本的自动化监控与告警系统。该系统支持多商品监控，一旦库存低于阈值或订单量异常，立即通过微信推送告警。

你需要准备一台联网的服务器或本地电脑，并安装Python 3.8及以上版本。请勿使用旧版Python，以免依赖库兼容性问题。打开终端（Terminal或CMD），执行以下命令安装核心依赖库：

pip install requests schedule

这里我们使用requests库进行接口请求，使用schedule库进行定时任务调度。这两个库轻量且稳定，非常适合此类脚本。

二、核心配置文件编写

为了方便运营人员修改监控参数，我们将所有配置项提取出来，不硬编码在脚本中。请在项目目录下创建一个名为monitor_config.json的文件，并复制以下完整内容：

```json { "monitor_interval": 60, "wechat_push_key": "YOUR_SCKEY_HERE", "alert_threshold": { "stock_min": 100, "order_max": 500 }, "targets": [ { "sku_id": "10086", "product_name": "爆款手机Pro", "api_url": "https://api.your-shop.com/internal/stock/query" }, { "sku_id": "10010", "product_name": "降噪耳机Max", "api_url": "https://api.your-shop.com/internal/stock/query" } ] } ```

配置参数详解：

• monitor_interval：监控轮询间隔，单位为秒，示例中为60秒检测一次。
• wechat_push_key：微信推送的Key。推荐使用Server酱（ServerChan），去https://sct.ftqq.com/免费申请一个Key填入此处，用于接收微信消息。
• alert_threshold：告警阈值。stock_min表示库存低于此数值触发告警，order_max表示单分钟订单量高于此数值触发异常告警（防刷单或流量激增）。
• targets：监控目标列表。你需要将sku_id替换为你们真实的商品ID，api_url替换为你们后端提供的实时库存查询接口地址。

三、监控脚本核心逻辑实现

在同级目录下创建主脚本promo_monitor.py。这个脚本将包含数据获取、逻辑判断和消息推送三个核心模块。

1. 引入库与初始化

```python import json import time import requests import schedule from datetime import datetime 加载配置文件 def load_config(): with open('monitor_config.json', 'r', encoding='utf-8') as f: return json.load(f) config = load_config() ```

2. 实现微信告警推送功能

这是运营人员接收信息的唯一入口，必须保证高可用。代码如下：

```python def send_wechat_alert(title, content): push_key = config.get('wechat_push_key') if not push_key or push_key == "YOUR_SCKEY_HERE": print("错误：未配置微信Push Key，无法发送消息。") return url = f"https://sctapi.ftqq.com/{push_key}.send" payload = { "title": title, "desp": content } try: response = requests.post(url, data=payload, timeout=5) if response.json().get('code') != 0: print(f"消息推送失败：{response.text}") except Exception as e: print(f"消息推送异常：{str(e)}") ```

3. 商品数据获取与模拟接口

实际场景中，你需要调用后端API。为了确保你能直接运行测试，我在代码中内置了一个模拟模式。如果你有真实接口，请替换fetch_real_data中的逻辑。

```python def fetch_real_data(sku_id, api_url): """ 真实环境：请求后端接口获取库存和实时订单数 请根据你们公司的API文档修改headers和params """ headers = { "Content-Type": "application/json", "Authorization": "Bearer YOUR_INTERNAL_TOKEN" 如果需要鉴权 } params = {"sku_id": sku_id} try: 示例请求，请根据实际响应结构修改解析逻辑 response = requests.get(api_url, headers=headers, params=params, timeout=3) data = response.json() return int(data['stock']), int(data['recent_orders']) 以下是模拟数据逻辑，实际使用请删除 import random mock_stock = random.randint(50, 200) mock_orders = random.randint(10, 100) return mock_stock, mock_orders 模拟逻辑结束 except Exception as e: print(f"请求接口异常，SKU: {sku_id}, 错误: {str(e)}") return None, None ```

4. 核心监控循环逻辑

这是主逻辑，负责遍历配置文件中的商品，进行阈值比对。

```python def check_inventory(): print(f"开始执行监控任务：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") targets = config.get('targets', []) threshold = config.get('alert_threshold', {}) stock_min = threshold.get('stock_min', 0) order_max = threshold.get('order_max', 99999) for item in targets: sku_id = item['sku_id'] name = item['product_name'] api_url = item['api_url'] stock, orders = fetch_real_data(sku_id, api_url) if stock is None: continue 库存过低告警 if stock < stock_min: msg = f"商品：{name}\nSKU：{sku_id}\n当前库存：{stock}\n状态：低于警戒线 {stock_min}！请及时补货。" print(f"触发告警：{msg}") send_wechat_alert("【库存紧急预警】", msg) 订单量异常告警（防刷或爆单） elif orders > order_max: msg = f"商品：{name}\nSKU：{sku_id}\n近1分钟订单：{orders}\n状态：流量异常！请关注系统负载。" print(f"触发告警：{msg}") send_wechat_alert("【订单异常预警】", msg) else: print(f"监控正常：{name} | 库存: {stock} | 订单: {orders}") ```

5. 程序入口

```python def job(): check_inventory() if __name__ == "__main__": 立即执行一次 job() 设置定时任务 interval = config.get('monitor_interval', 60) schedule.every(interval).seconds.do(job) print(f"监控服务已启动，每 {interval} 秒检测一次...") while True: schedule.run_pending() time.sleep(1) ```

四、如何获取监控目标ID（实操细节）

很多运营同学卡在不知道如何获取sku_id和接口地址。这里提供一个通用的抓包方法，无需任何专业抓包工具，直接使用Chrome浏览器即可。

步骤如下：

1. 打开开发者工具：在商品详情页右键选择“检查”，或按下F12键。
2. 进入Network标签：点击顶部的Network选项卡。
3. 筛选请求：在筛选框中输入XHR或Fetch，这通常用于过滤出数据接口请求。
4. 触发操作：刷新页面，或者点击“立即购买”按钮（不要真的下单，点一下就关掉）。
5. 查找数据接口：在列表中寻找包含stock、detail、price或sku关键字的请求。
6. 提取信息：点击该请求，查看Headers中的Request URL（这就是你的api_url），查看Payload或Query String Parameters中通常能找到skuId或goods_id。

五、自动化部署方案

脚本写好后，不能一直挂着命令行窗口。我们需要让它在后台稳定运行。以下是Linux服务器和Windows本机的部署方案。

1. Linux服务器部署（推荐）

在Linux环境下，我们使用nohup命令让脚本在后台运行，并将日志输出到文件，方便排查问题。

执行以下命令：

nohup python3 -u promo_monitor.py > monitor.log 2>&1 &

这条命令的含义是：使用Python3运行脚本，将标准输出和错误输出都重定向到monitor.log文件中，&表示在后台运行。

查看实时日志的命令：

tail -f monitor.log

如果需要停止监控，先查找进程ID：

ps -ef | grep promo_monitor.py

然后使用kill命令结束进程：

kill -9 [进程ID]

2. Windows本地电脑部署

在Windows上，你可以使用“任务计划程序”来定时启动脚本，或者简单地双击运行，但为了防止误关窗口，建议使用start命令最小化运行。

创建一个run.bat文件，内容如下：

```batch @echo off start /min python promo_monitor.py ```

双击这个bat文件，脚本会在一个最小化的CMD窗口中运行。关闭窗口即停止脚本。

六、常见问题处理

在实操过程中，可能会遇到网络波动导致的接口超时。本脚本已在fetch_real_data函数中加入了try-except捕获，防止程序因单次请求失败而崩溃。如果发现monitor.log中频繁出现超时错误，建议在requests.get()中增加timeout参数，或者检查服务器的出网带宽。确保系统时间准确，否则schedule库的定时任务可能会出现偏差。