如何编写订单数据整理脚本

wen 实用脚本 2026-06-28 263

本文目录导读：

如何编写订单数据整理脚本

基础订单数据整理脚本
简单版脚本（适合快速使用）
使用说明

我来帮你编写一个订单数据整理脚本，这里提供Python版本,包含多种常见的数据整理功能：

基础订单数据整理脚本

import pandas as pd
import json
from datetime import datetime
import re
class OrderDataProcessor:
    def __init__(self, file_path=None):
        self.file_path = file_path
        self.orders = None
    def load_data(self, file_path=None):
        """加载订单数据，支持CSV和Excel格式"""
        if file_path:
            self.file_path = file_path
        if self.file_path.endswith('.csv'):
            self.orders = pd.read_csv(self.file_path)
        elif self.file_path.endswith(('.xlsx', '.xls')):
            self.orders = pd.read_excel(self.file_path)
        elif self.file_path.endswith('.json'):
            with open(self.file_path, 'r', encoding='utf-8') as f:
                data = json.load(f)
                self.orders = pd.DataFrame(data)
        else:
            raise ValueError("不支持的文件格式，请使用CSV、Excel或JSON格式")
        return self.orders
    def clean_data(self):
        """数据清洗"""
        if self.orders is None:
            raise ValueError("请先加载数据")
        # 1. 删除空行
        self.orders = self.orders.dropna(how='all')
        # 2. 填充缺失值
        self.orders = self.orders.fillna({
            '订单状态': '待处理',
            '备注': '',
            '数量': 1,
            '金额': 0
        })
        # 3. 删除重复订单（根据订单号去重）
        if '订单号' in self.orders.columns:
            self.orders = self.orders.drop_duplicates(subset=['订单号'])
        # 4. 统一日期格式
        if '订单日期' in self.orders.columns:
            self.orders['订单日期'] = pd.to_datetime(self.orders['订单日期'], errors='coerce')
        return self.orders
    def standardize_fields(self):
        """标准化字段"""
        if self.orders is None:
            raise ValueError("请先加载数据")
        # 1. 统一列名（驼峰命名转下划线）
        column_mapping = {
            'OrderID': '订单号',
            'OrderDate': '订单日期',
            'CustomerName': '客户名称',
            'ProductName': '产品名称',
            'Quantity': '数量',
            'UnitPrice': '单价',
            'TotalAmount': '总金额',
            'Status': '订单状态'
        }
        self.orders = self.orders.rename(columns=column_mapping)
        # 2. 统一电话号码格式
        if '联系电话' in self.orders.columns:
            self.orders['联系电话'] = self.orders['联系电话'].apply(
                lambda x: re.sub(r'[^0-9-]', '', str(x))
            )
        # 3. 统一金额格式（保留两位小数）
        if '金额' in self.orders.columns:
            self.orders['金额'] = pd.to_numeric(self.orders['金额'], errors='coerce').round(2)
        return self.orders
    def calculate_fields(self):
        """计算衍生字段"""
        if self.orders is None:
            raise ValueError("请先加载数据")
        # 1. 计算总金额（如果存在数量和单价）
        if '数量' in self.orders.columns and '单价' in self.orders.columns:
            self.orders['计算金额'] = self.orders['数量'] * self.orders['单价']
        # 2. 添加订单状态分类
        if '订单状态' in self.orders.columns:
            status_map = {
                '待付款': '未完成',
                '已付款': '完成',
                '已发货': '完成',
                '已完成': '完成',
                '已取消': '已取消',
                '退款中': '异常',
                '已退款': '已取消'
            }
            self.orders['状态分类'] = self.orders['订单状态'].map(status_map).fillna('其他')
        # 3. 添加日期维度
        if '订单日期' in self.orders.columns:
            self.orders['年份'] = self.orders['订单日期'].dt.year
            self.orders['月份'] = self.orders['订单日期'].dt.month
            self.orders['季度'] = self.orders['订单日期'].dt.quarter
            self.orders['星期'] = self.orders['订单日期'].dt.day_name()
        return self.orders
    def filter_and_sort(self, status_filter=None, date_range=None):
        """筛选和排序"""
        if self.orders is None:
            raise ValueError("请先加载数据")
        # 筛选状态
        if status_filter:
            if '订单状态' in self.orders.columns:
                self.orders = self.orders[self.orders['订单状态'].isin(status_filter)]
        # 筛选日期范围
        if date_range:
            if '订单日期' in self.orders.columns:
                start_date, end_date = date_range
                self.orders = self.orders[
                    (self.orders['订单日期'] >= start_date) & 
                    (self.orders['订单日期'] <= end_date)
                ]
        # 按日期降序排列
        if '订单日期' in self.orders.columns:
            self.orders = self.orders.sort_values('订单日期', ascending=False)
        return self.orders
    def generate_summary(self):
        """生成数据摘要"""
        if self.orders is None:
            raise ValueError("请先加载数据")
        summary = {
            '总订单数': len(self.orders),
            '总金额': self.orders['金额'].sum() if '金额' in self.orders.columns else 0,
            '平均金额': self.orders['金额'].mean() if '金额' in self.orders.columns else 0,
            '最大金额': self.orders['金额'].max() if '金额' in self.orders.columns else 0,
            '最小金额': self.orders['金额'].min() if '金额' in self.orders.columns else 0,
            '订单状态分布': self.orders['订单状态'].value_counts().to_dict() if '订单状态' in self.orders.columns else {},
            '数据日期范围': f"{self.orders['订单日期'].min()} 至 {self.orders['订单日期'].max()}" if '订单日期' in self.orders.columns else "无日期数据"
        }
        return summary
    def export_data(self, output_path='processed_orders.xlsx', file_type='excel'):
        """导出处理后的数据"""
        if self.orders is None:
            raise ValueError("请先加载数据")
        if file_type == 'excel':
            with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
                # 写入主数据表
                self.orders.to_excel(writer, sheet_name='订单数据', index=False)
                # 写入数据摘要表
                summary = self.generate_summary()
                summary_df = pd.DataFrame([summary])
                summary_df.to_excel(writer, sheet_name='数据摘要', index=False)
                # 写入数据透视表
                if '客户名称' in self.orders.columns and '金额' in self.orders.columns:
                    pivot_table = pd.pivot_table(
                        self.orders,
                        values='金额',
                        index='客户名称',
                        columns='订单状态',
                        aggfunc='sum',
                        fill_value=0
                    )
                    pivot_table.to_excel(writer, sheet_name='客户分析')
        elif file_type == 'csv':
            self.orders.to_csv(output_path.replace('.xlsx', '.csv'), index=False, encoding='utf-8-sig')
        print(f"数据已导出到: {output_path}")
    def process_pipeline(self, file_path=None, output_path='processed_orders.xlsx'):
        """完整的数据处理流水线"""
        print("开始处理订单数据...")
        # 1. 加载数据
        print("1/5 加载数据...")
        self.load_data(file_path)
        # 2. 数据清洗
        print("2/5 数据清洗...")
        self.clean_data()
        # 3. 字段标准化
        print("3/5 字段标准化...")
        self.standardize_fields()
        # 4. 计算衍生字段
        print("4/5 计算衍生字段...")
        self.calculate_fields()
        # 5. 导出数据
        print("5/5 导出数据...")
        self.export_data(output_path)
        # 显示摘要
        summary = self.generate_summary()
        print("\n数据处理完成！摘要如下：")
        for key, value in summary.items():
            print(f"{key}: {value}")
        return self.orders
# 使用示例
if __name__ == "__main__":
    # 创建处理器实例
    processor = OrderDataProcessor()
    # 完整处理流水线
    processor.process_pipeline(
        file_path='orders.csv',  # 原始数据文件
        output_path='processed_orders.xlsx'  # 输出文件
    )
    # 或者分步处理
    # processor.load_data('orders.csv')
    # processor.clean_data()
    # processor.standardize_fields()
    # processor.calculate_fields()
    # processor.filter_and_sort(['已完成', '已付款'], date_range=('2024-01-01', '2024-12-31'))
    # processor.export_data('filtered_orders.xlsx')

简单版脚本（适合快速使用）

import pandas as pd
import os
def quick_order_processing(input_file, output_file='cleaned_orders.xlsx'):
    """快速订单处理函数"""
    # 读取数据
    if input_file.endswith('.csv'):
        df = pd.read_csv(input_file)
    elif input_file.endswith('.xlsx'):
        df = pd.read_excel(input_file)
    else:
        print("不支持的文件格式")
        return
    # 基本清洗
    df = df.dropna(how='all')
    df = df.drop_duplicates()
    # 转换日期
    if '订单日期' in df.columns:
        df['订单日期'] = pd.to_datetime(df['订单日期'])
    # 处理金额
    if '金额' in df.columns:
        df['金额'] = pd.to_numeric(df['金额'], errors='coerce')
    # 导出
    df.to_excel(output_file, index=False)
    print(f"处理完成！共 {len(df)} 条记录")
    print(f"数据已保存到: {output_file}")
    return df
# 使用
# df = quick_order_processing('orders.csv')

使用说明

安装依赖：
```
pip install pandas openpyxl xlrd
```
准备数据：将订单数据保存为CSV或Excel格式

运行脚本：

processor = OrderDataProcessor()
processor.process_pipeline('your_orders.csv')

这个脚本可以帮你：

自动清洗数据
标准化字段格式
计算统计字段
生成数据摘要
导出处理后的数据

需要根据你的实际数据格式调整字段名和逻辑。

本文地址： https://www.bmjup.com/post/11216.html