本文目录导读：
我来帮你编写一个订单数据整理脚本，这里提供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')
这个脚本可以帮你：
- 自动清洗数据
- 标准化字段格式
- 计算统计字段
- 生成数据摘要
- 导出处理后的数据
需要根据你的实际数据格式调整字段名和逻辑。
