如何写订单状态更新脚本

本文目录导读：

1. MySQL 存储过程方案
2. Python 脚本方案
3. Node.js 脚本方案
4. 监控统计脚本
5. 数据库表结构
6. 使用建议

我来为您详细介绍如何编写订单状态更新脚本,包含多种实现方案。

MySQL 存储过程方案

DELIMITER $$
CREATE PROCEDURE update_order_status(
    IN p_order_id VARCHAR(50),
    IN p_new_status VARCHAR(20),
    IN p_user_id VARCHAR(50),
    OUT p_result VARCHAR(200)
)
BEGIN
    DECLARE EXIT HANDLER FOR SQLEXCEPTION 
    BEGIN
        ROLLBACK;
        SET p_result = '更新失败：数据库异常';
    END;
    START TRANSACTION;
    -- 验证订单是否存在
    IF NOT EXISTS (SELECT 1 FROM orders WHERE order_id = p_order_id) THEN
        SET p_result = '更新失败：订单不存在';
        ROLLBACK;
    ELSE
        -- 记录状态变更历史
        INSERT INTO order_status_history (
            order_id, 
            old_status, 
            new_status, 
            operator, 
            created_at
        )
        SELECT 
            p_order_id,
            status,
            p_new_status,
            p_user_id,
            NOW()
        FROM orders 
        WHERE order_id = p_order_id;
        -- 更新订单状态
        UPDATE orders 
        SET status = p_new_status,
            update_time = NOW(),
            update_user = p_user_id
        WHERE order_id = p_order_id;
        -- 特殊状态处理
        CASE p_new_status
            WHEN 'shipped' THEN
                UPDATE orders SET shipping_time = NOW() 
                WHERE order_id = p_order_id;
            WHEN 'completed' THEN
                UPDATE orders SET completed_time = NOW() 
                WHERE order_id = p_order_id;
            WHEN 'cancelled' THEN
                UPDATE orders SET cancel_time = NOW() 
                WHERE order_id = p_order_id;
        END CASE;
        COMMIT;
        SET p_result = '更新成功';
    END IF;
END$$
DELIMITER ;

Python 脚本方案

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import pymysql
import logging
from datetime import datetime
from typing import Optional
class OrderStatusUpdater:
    """订单状态更新器"""
    # 定义状态流转规则
    STATUS_FLOW = {
        'pending': ['confirmed', 'cancelled'],
        'confirmed': ['processing', 'cancelled'],
        'processing': ['shipped', 'cancelled'],
        'shipped': ['delivered', 'cancelled'],
        'delivered': ['completed', 'returning'],
        'returning': ['refunded'],
        'refunded': [],  # 最终状态
        'completed': [],  # 最终状态
        'cancelled': []   # 最终状态
    }
    def __init__(self, db_config: dict):
        self.db_config = db_config
        self.setup_logging()
    def setup_logging(self):
        """配置日志"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('order_update.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
    def get_db_connection(self):
        """获取数据库连接"""
        return pymysql.connect(**self.db_config)
    def validate_status_transition(self, old_status: str, new_status: str) -> bool:
        """验证状态流转是否合法"""
        if old_status not in self.STATUS_FLOW:
            return False
        return new_status in self.STATUS_FLOW[old_status]
    def update_order_status(
        self, 
        order_id: str, 
        new_status: str, 
        operator: str = 'system',
        remark: Optional[str] = None
    ) -> dict:
        """
        更新订单状态
        Args:
            order_id: 订单ID
            new_status: 新状态
            operator: 操作人
            remark: 备注
        Returns:
            dict: 操作结果
        """
        conn = None
        try:
            conn = self.get_db_connection()
            cursor = conn.cursor()
            # 查询当前订单信息
            cursor.execute(
                "SELECT id, status, order_number FROM orders WHERE order_id = %s",
                (order_id,)
            )
            order = cursor.fetchone()
            if not order:
                return {'success': False, 'message': '订单不存在'}
            _, old_status, order_number = order
            # 验证状态流转
            if not self.validate_status_transition(old_status, new_status):
                return {
                    'success': False, 
                    'message': f'不允许的状态流转: {old_status} -> {new_status}'
                }
            # 开始事务
            conn.begin()
            # 更新订单状态
            update_time = datetime.now()
            cursor.execute(
                """UPDATE orders 
                   SET status = %s, 
                       update_time = %s,
                       update_user = %s
                   WHERE order_id = %s""",
                (new_status, update_time, operator, order_id)
            )
            # 记录状态变更历史
            cursor.execute(
                """INSERT INTO order_status_history 
                   (order_id, old_status, new_status, operator, remark, created_at)
                   VALUES (%s, %s, %s, %s, %s, %s)""",
                (order_id, old_status, new_status, operator, remark, update_time)
            )
            # 特殊状态处理
            if new_status == 'shipped':
                cursor.execute(
                    "UPDATE orders SET shipping_time = %s WHERE order_id = %s",
                    (update_time, order_id)
                )
            elif new_status == 'completed':
                cursor.execute(
                    "UPDATE orders SET completed_time = %s WHERE order_id = %s",
                    (update_time, order_id)
                )
            conn.commit()
            self.logger.info(
                f"订单 {order_number} 状态更新成功: {old_status} -> {new_status}"
            )
            return {
                'success': True, 
                'message': '更新成功',
                'data': {
                    'order_id': order_id,
                    'order_number': order_number,
                    'old_status': old_status,
                    'new_status': new_status,
                    'update_time': update_time.isoformat()
                }
            }
        except Exception as e:
            if conn:
                conn.rollback()
            self.logger.error(f"更新订单状态失败: {str(e)}")
            return {'success': False, 'message': f'更新失败: {str(e)}'}
        finally:
            if conn:
                conn.close()
    def batch_update_status(
        self, 
        order_ids: list, 
        new_status: str, 
        operator: str = 'system'
    ) -> dict:
        """
        批量更新订单状态
        Args:
            order_ids: 订单ID列表
            new_status: 新状态
            operator: 操作人
        Returns:
            dict: 批量操作结果
        """
        results = {
            'success_count': 0,
            'fail_count': 0,
            'details': []
        }
        for order_id in order_ids:
            result = self.update_order_status(order_id, new_status, operator)
            if result['success']:
                results['success_count'] += 1
            else:
                results['fail_count'] += 1
            results['details'].append(result)
        return results
# 使用示例
if __name__ == '__main__':
    # 数据库配置
    db_config = {
        'host': 'localhost',
        'user': 'root',
        'password': 'password',
        'database': 'order_db',
        'charset': 'utf8mb4'
    }
    # 创建更新器
    updater = OrderStatusUpdater(db_config)
    # 单个订单更新
    result = updater.update_order_status(
        order_id='ORD2024001',
        new_status='shipped',
        operator='admin',
        remark='快递已发出'
    )
    print(result)
    # 批量更新
    order_ids = ['ORD2024002', 'ORD2024003', 'ORD2024004']
    batch_result = updater.batch_update_status(
        order_ids=order_ids,
        new_status='completed',
        operator='system'
    )
    print(batch_result)

Node.js 脚本方案

// orderStatusUpdater.js
const mysql = require('mysql2/promise');
const { v4: uuidv4 } = require('uuid');
class OrderStatusUpdater {
    constructor(dbConfig) {
        this.dbConfig = {
            host: dbConfig.host || 'localhost',
            user: dbConfig.user,
            password: dbConfig.password,
            database: dbConfig.database,
            waitForConnections: true,
            connectionLimit: 10,
            queueLimit: 0
        };
        // 状态流转规则
        this.statusFlow = {
            'pending': ['confirmed', 'cancelled'],
            'confirmed': ['processing', 'cancelled'],
            'processing': ['shipped', 'cancelled'],
            'shipped': ['delivered', 'returning'],
            'delivered': ['completed', 'returning'],
            'returning': ['refunded'],
            'refunded': [],
            'completed': [],
            'cancelled': []
        };
        this.pool = mysql.createPool(this.dbConfig);
    }
    validateStatusTransition(oldStatus, newStatus) {
        return this.statusFlow[oldStatus]?.includes(newStatus) || false;
    }
    async updateOrderStatus(orderId, newStatus, operator = 'system', remark = '') {
        const connection = await this.pool.getConnection();
        try {
            await connection.beginTransaction();
            // 查询订单信息
            const [orders] = await connection.execute(
                'SELECT id, status, order_number FROM orders WHERE order_id = ?',
                [orderId]
            );
            if (orders.length === 0) {
                throw new Error('订单不存在');
            }
            const { status: oldStatus, order_number: orderNumber } = orders[0];
            // 验证状态流转
            if (!this.validateStatusTransition(oldStatus, newStatus)) {
                throw new Error(`不允许的状态流转: ${oldStatus} -> ${newStatus}`);
            }
            // 更新订单状态
            const updateTime = new Date();
            await connection.execute(
                'UPDATE orders SET status = ?, update_time = ?, update_user = ? WHERE order_id = ?',
                [newStatus, updateTime, operator, orderId]
            );
            // 记录状态变更历史
            if (typeof remark !== 'string') {
                remark = '';
            }
            const historyId = uuidv4();
            await connection.execute(
                'INSERT INTO order_status_history (id, order_id, old_status, new_status, operator, remark, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)',
                [historyId, orderId, oldStatus, newStatus, operator, remark, updateTime]
            );
            // 特殊状态处理
            if (newStatus === 'shipped') {
                await connection.execute(
                    'UPDATE orders SET shipping_time = ? WHERE order_id = ?',
                    [updateTime, orderId]
                );
            } else if (newStatus === 'completed') {
                await connection.execute(
                    'UPDATE orders SET completed_time = ? WHERE order_id = ?',
                    [updateTime, orderId]
                );
            }
            await connection.commit();
            console.log(`订单 ${orderNumber} 状态更新成功: ${oldStatus} -> ${newStatus}`);
            return {
                success: true,
                message: '更新成功',
                data: {
                    orderId,
                    orderNumber,
                    oldStatus,
                    newStatus,
                    updateTime: updateTime.toISOString()
                }
            };
        } catch (error) {
            await connection.rollback();
            console.error('更新订单状态失败:', error.message);
            return {
                success: false,
                message: error.message
            };
        } finally {
            connection.release();
        }
    }
    async batchUpdateStatus(orderIds, newStatus, operator = 'system') {
        const results = {
            successCount: 0,
            failCount: 0,
            details: []
        };
        for (const orderId of orderIds) {
            const result = await this.updateOrderStatus(orderId, newStatus, operator);
            if (result.success) {
                results.successCount++;
            } else {
                results.failCount++;
            }
            results.details.push(result);
        }
        return results;
    }
    async close() {
        await this.pool.end();
    }
}
// 使用示例
async function main() {
    const updater = new OrderStatusUpdater({
        host: 'localhost',
        user: 'root',
        password: 'password',
        database: 'order_db'
    });
    try {
        // 单个订单更新
        const result = await updater.updateOrderStatus(
            'ORD2024001',
            'shipped',
            'admin',
            '快递已发出'
        );
        console.log('更新结果:', result);
        // 批量更新
        const orderIds = ['ORD2024002', 'ORD2024003', 'ORD2024004'];
        const batchResult = await updater.batchUpdateStatus(
            orderIds,
            'completed',
            'system'
        );
        console.log('批量更新结果:', batchResult);
    } finally {
        await updater.close();
    }
}
main().catch(console.error);

监控统计脚本

# order_status_monitor.py
import pymysql
import pandas as pd
from datetime import datetime, timedelta
import matplotlib.pyplot as plt
class OrderStatusMonitor:
    """订单状态监控"""
    def __init__(self, db_config):
        self.conn = pymysql.connect(**db_config)
    def get_status_statistics(self, days=30):
        """获取状态统计数据"""
        start_date = datetime.now() - timedelta(days=days)
        query = """
            SELECT 
                DATE(created_at) as date,
                status,
                COUNT(*) as count,
                SUM(total_amount) as total_amount
            FROM orders
            WHERE created_at >= %s
            GROUP BY DATE(created_at), status
            ORDER BY date, status
        """
        df = pd.read_sql(query, self.conn, params=(start_date,))
        return df
    def get_status_change_analysis(self, days=30):
        """获取状态变更分析"""
        start_date = datetime.now() - timedelta(days=days)
        query = """
            SELECT 
                osh.old_status,
                osh.new_status,
                COUNT(*) as count,
                AVG(TIMESTAMPDIFF(HOUR, o.created_at, osh.created_at)) as avg_hours
            FROM order_status_history osh
            JOIN orders o ON osh.order_id = o.order_id
            WHERE osh.created_at >= %s
            GROUP BY osh.old_status, osh.new_status
            ORDER BY count DESC
        """
        df = pd.read_sql(query, self.conn, params=(start_date,))
        return df
    def plot_status_distribution(self, days=30):
        """绘制状态分布图"""
        df = self.get_status_statistics(days)
        # 透视表
        pivot_df = df.pivot_table(
            index='date', 
            columns='status', 
            values='count', 
            fill_value=0
        )
        # 绘图
        plt.figure(figsize=(12, 6))
        pivot_df.plot(kind='bar', stacked=True)
        plt.title(f'订单状态分布（近{days}天）')
        plt.xlabel('日期')
        plt.ylabel('订单数量')
        plt.legend(title='状态')
        plt.tight_layout()
        plt.savefig('order_status_distribution.png')
        plt.close()
    def close(self):
        self.conn.close()
# 使用示例
if __name__ == '__main__':
    db_config = {
        'host': 'localhost',
        'user': 'root',
        'password': 'password',
        'database': 'order_db',
        'charset': 'utf8mb4'
    }
    monitor = OrderStatusMonitor(db_config)
    # 获取状态统计
    stats = monitor.get_status_statistics(30)
    print("状态统计:")
    print(stats)
    # 获取状态变更分析
    analysis = monitor.get_status_change_analysis(30)
    print("\n状态变更分析:")
    print(analysis)
    # 生成图表
    monitor.plot_status_distribution(30)
    monitor.close()

数据库表结构

-- 订单表
CREATE TABLE `orders` (
    `id` bigint(20) NOT NULL AUTO_INCREMENT,
    `order_id` varchar(50) NOT NULL COMMENT '订单唯一标识',
    `order_number` varchar(50) NOT NULL COMMENT '订单编号',
    `user_id` varchar(50) NOT NULL COMMENT '用户ID',
    `total_amount` decimal(10,2) NOT NULL DEFAULT '0.00' COMMENT '订单总金额',
    `status` varchar(20) NOT NULL DEFAULT 'pending' COMMENT '订单状态',
    `shipping_time` datetime DEFAULT NULL COMMENT '发货时间',
    `completed_time` datetime DEFAULT NULL COMMENT '完成时间',
    `cancel_time` datetime DEFAULT NULL COMMENT '取消时间',
    `update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
    `update_user` varchar(50) DEFAULT NULL COMMENT '更新人',
    `created_at` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
    PRIMARY KEY (`id`),
    UNIQUE KEY `uk_order_id` (`order_id`),
    UNIQUE KEY `uk_order_number` (`order_number`),
    KEY `idx_user_id` (`user_id`),
    KEY `idx_status` (`status`),
    KEY `idx_created_at` (`created_at`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='订单表';
-- 订单状态变更历史表
CREATE TABLE `order_status_history` (
    `id` bigint(20) NOT NULL AUTO_INCREMENT,
    `order_id` varchar(50) NOT NULL COMMENT '订单ID',
    `old_status` varchar(20) NOT NULL COMMENT '原状态',
    `new_status` varchar(20) NOT NULL COMMENT '新状态',
    `operator` varchar(50) NOT NULL COMMENT '操作人',
    `remark` varchar(500) DEFAULT NULL COMMENT '备注',
    `created_at` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
    PRIMARY KEY (`id`),
    KEY `idx_order_id` (`order_id`),
    KEY `idx_created_at` (`created_at`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='订单状态变更历史表';

使用建议

1. 事务处理：务必使用事务，确保数据一致性
2. 状态验证：验证状态流转的合法性
3. 审计日志：记录每次状态变更的详细信息
4. 异常处理：完善错误处理和回滚机制
5. 监控告警：设置关键状态的监控和告警

根据实际需求选择合适的方案,建议在生产环境中使用存储过程或Python脚本方案，并配合完善的监控机制。