电商物流时效监控系统从零搭建实操指南 附全量可复用代码配置

一、前置环境准备

1.1 依赖安装

操作系统支持Ubuntu 22.04/CentOS 7+，执行对应命令安装基础环境：

• Ubuntu 22.04 安装命令：
  sudo apt update && sudo apt install -y python3.9 python3-pip redis-server mysql-server-8.0
• CentOS 7+ 安装命令：
  sudo yum install -y python39 python39-pip redis mysql-community-server

新建requirements.txt文件，写入以下内容后执行pip3 install -r requirements.txt安装Python依赖：

```txt Flask==2.3.3 redis==5.0.1 PyMySQL==1.1.0 requests==2.31.0 ```

1.2 数据库表初始化

首次部署必须先执行以下建表语句，否则接口会报500错误，登录MySQL后执行以下SQL：

```sql CREATE DATABASE IF NOT EXISTS logistics DEFAULT CHARSET utf8mb4; USE logistics; -- 电商订单与物流单号绑定表 CREATE TABLE IF NOT EXISTS order_bind ( id INT UNSIGNED AUTO_INCREMENT PRIMARY KEY, order_id VARCHAR(32) NOT NULL UNIQUE COMMENT '电商订单号', logistics_no VARCHAR(32) NOT NULL UNIQUE COMMENT '物流单号', receiver_city VARCHAR(32) NOT NULL COMMENT '收件城市', create_time DATETIME DEFAULT CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; -- 物流节点轨迹表 CREATE TABLE IF NOT EXISTS logistics_track ( id INT UNSIGNED AUTO_INCREMENT PRIMARY KEY, order_id VARCHAR(32) NOT NULL, logistics_no VARCHAR(32) NOT NULL, node_name VARCHAR(32) NOT NULL COMMENT '节点名称：揽收/干线运输/末端配送/签收', node_time DATETIME NOT NULL COMMENT '节点发生时间', operator VARCHAR(64) COMMENT '操作方', create_time DATETIME DEFAULT CURRENT_TIMESTAMP, INDEX idx_logistics_no (logistics_no) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; -- 超时预警记录表 CREATE TABLE IF NOT EXISTS timeout_alert ( id INT UNSIGNED AUTO_INCREMENT PRIMARY KEY, logistics_no VARCHAR(32) NOT NULL, timeout_node VARCHAR(32) NOT NULL, timeout_hours INT NOT NULL COMMENT '超时时长（小时）', alert_time DATETIME DEFAULT CURRENT_TIMESTAMP, is_processed TINYINT DEFAULT 0 COMMENT '是否处理：0未处理/1已处理' ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; ```

生产环境必须修改MySQL默认密码、Redis配置访问密码，避免数据泄露

二、核心功能开发

2.1 物流节点采集接口

新建app.py文件，写入以下代码，接口支持物流商主动推送节点数据：

```python from flask import Flask, request, jsonify import pymysql import redis import config from datetime import datetime app = Flask(__name__) 初始化数据库连接 db = pymysql.connect(host=config.MYSQL_HOST, user=config.MYSQL_USER, password=config.MYSQL_PWD, database=config.MYSQL_DB) r = redis.Redis(host=config.REDIS_HOST, port=config.REDIS_PORT, password=config.REDIS_PWD, db=0) @app.route('/api/track/upload', methods=['POST']) def upload_track(): data = request.get_json() 必填参数校验 required_params = ['order_id', 'logistics_no', 'node_name', 'node_time'] if not all(k in data for k in required_params): return jsonify({"code":400, "msg":"缺少必填参数"}), 400 校验绑定关系 cursor = db.cursor() cursor.execute("SELECT order_id FROM order_bind WHERE order_id=%s AND logistics_no=%s", (data['order_id'], data['logistics_no'])) if not cursor.fetchone(): return jsonify({"code":400, "msg":"订单与物流单号未绑定"}), 400 写入节点数据 sql = "INSERT INTO logistics_track (order_id, logistics_no, node_name, node_time, operator) VALUES (%s, %s, %s, %s, %s)" cursor.execute(sql, (data['order_id'], data['logistics_no'], data['node_name'], data['node_time'], data.get('operator',''))) db.commit() 触发时效计算异步任务 r.lpush('track_task', data['logistics_no']) return jsonify({"code":200, "msg":"上传成功"}) if __name__ == '__main__': app.run(host='0.0.0.0', port=5000, debug=False) ```

2.2 时效计算与预警模块

新建worker.py文件，作为异步消费进程处理时效计算：

```python import pymysql import redis import requests import config from datetime import datetime, timedelta db = pymysql.connect(host=config.MYSQL_HOST, user=config.MYSQL_USER, password=config.MYSQL_PWD, database=config.MYSQL_DB) r = redis.Redis(host=config.REDIS_HOST, port=config.REDIS_PORT, password=config.REDIS_PWD, db=0) SLA规则可根据实际业务调整 SLA_RULE = { "揽收->干线运输": 4, "跨省干线运输": 72, "省内干线运输": 24, "干线运输->末端配送": 6, "末端配送->签收": 4 } def send_alert(logistics_no, node, timeout_hours): 对接企业微信/钉钉webhook，修改config中的WEBHOOK地址即可使用 content = f"【物流超时预警】物流单号{logistics_no}在{node}节点超时{timeout_hours}小时，请及时处理" requests.post(config.WEBHOOK_URL, json={"msgtype":"text", "text":{"content":content}}) 写入预警记录 cursor = db.cursor() cursor.execute("INSERT INTO timeout_alert (logistics_no, timeout_node, timeout_hours) VALUES (%s, %s, %s)", (logistics_no, node, timeout_hours)) db.commit() while True: 阻塞消费任务队列 task = r.brpop('track_task', timeout=0) logistics_no = task[1].decode() cursor = db.cursor() 查询该单号所有节点按时间排序 cursor.execute("SELECT node_name, node_time FROM logistics_track WHERE logistics_no=%s ORDER BY node_time ASC", (logistics_no,)) nodes = cursor.fetchall() if len(nodes) < 2: continue 获取收件城市判断运输类型 cursor.execute("SELECT receiver_city FROM order_bind WHERE logistics_no=%s", (logistics_no,)) receiver_city = cursor.fetchone()[0] 计算相邻节点时长 for i in range(len(nodes)-1): prev_node, prev_time = nodes[i] curr_node, curr_time = nodes[i+1] duration = (curr_time - prev_time).total_seconds() / 3600 匹配SLA规则 rule_key = f"{prev_node}->{curr_node}" if rule_key in SLA_RULE and duration > SLA_RULE[rule_key]: send_alert(logistics_no, rule_key, round(duration - SLA_RULE[rule_key], 1)) ```

2.3 配置文件

新建config.py文件，填入实际环境参数即可直接使用：

```python 数据库配置 MYSQL_HOST = "127.0.0.1" MYSQL_USER = "root" MYSQL_PWD = "你的MySQL密码" MYSQL_DB = "logistics" Redis配置 REDIS_HOST = "127.0.0.1" REDIS_PORT = 6379 REDIS_PWD = "你的Redis密码，没有则填空字符串" 预警配置 WEBHOOK_URL = "你的企业微信/钉钉webhook地址" ```

三、部署与验证

3.1 服务部署

新建systemd配置文件/etc/systemd/system/logistics-monitor.service，写入以下内容：

```ini [Unit] Description=物流时效监控服务 After=network.target mysql.service redis.service [Service] User=root WorkingDirectory=/opt/logistics-monitor ExecStart=/usr/bin/python3.9 /opt/logistics-monitor/app.py ExecStart=/usr/bin/python3.9 /opt/logistics-monitor/worker.py Restart=always RestartSec=3 [Install] WantedBy=multi-user.target ```

必须将所有代码文件放入/opt/logistics-monitor目录，给目录配置755权限，避免服务启动失败

执行以下命令启动服务：

sudo systemctl daemon-reload && sudo systemctl enable --now logistics-monitor.service

3.2 功能验证

1. 先写入绑定关系，执行MySQL语句：
INSERT INTO order_bind (order_id, logistics_no, receiver_city) VALUES ('TEST20240501001', 'SF1234567890', '北京市');

2. 模拟推送节点数据，执行curl命令：

curl -X POST http://127.0.0.1:5000/api/track/upload -H "Content-Type: application/json" -d '{"order_id":"TEST20240501001","logistics_no":"SF1234567890","node_name":"揽收","node_time":"2024-05-01 10:00:00","operator":"顺丰网点"}'

3. 间隔5小时后推送干线运输节点，会自动触发超时预警，收到推送即代表搭建成功。

3.3 常见问题排查

• 接口返回500：查看journalctl -u logistics-monitor.service日志，检查数据库连接配置是否正确
• 未收到预警：检查webhook地址是否允许服务器IP访问，SLA规则阈值是否设置合理
• 服务启动失败：检查Python依赖是否安装完整，代码目录权限是否为755