我来帮你编写一个快递数据整理脚本,以下是一个通用的Python脚本，可以根据实际需求调整：
import pandas as pd
import re
from datetime import datetime
import os
class ExpressDataProcessor:
def __init__(self, file_path):
"""
初始化快递数据处理器
:param file_path: 快递数据文件路径（支持CSV/Excel）
"""
self.file_path = file_path
self.data = None
def load_data(self):
"""加载数据文件"""
try:
if self.file_path.endswith('.csv'):
self.data = pd.read_csv(self.file_path, encoding='utf-8')
elif self.file_path.endswith(('.xlsx', '.xls')):
self.data = pd.read_excel(self.file_path)
print(f"成功加载 {len(self.data)} 条记录")
return True
except Exception as e:
print(f"加载文件失败: {e}")
return False
def clean_phone_numbers(self, column_name='phone'):
"""清洗手机号码格式"""
# 去除空格和特殊字符
self.data[column_name] = self.data[column_name].astype(str).str.replace(r'\D', '', regex=True)
# 如果手机号以86开头，去除86
self.data[column_name] = self.data[column_name].str.replace(r'^86', '', regex=True)
# 确保手机号为11位
self.data.loc[self.data[column_name].str.len() != 11, column_name] = '无效号码'
return self
def extract_address_components(self, address_column='address'):
"""提取地址中的省市县信息"""
# 假设地址格式为：省/市/区/详细地址
patterns = {
'province': r'^([\u4e00-\u9fa5]{2,4}(?:省|自治区|特别行政区))',
'city': r'(?:省|自治区)([\u4e00-\u9fa5]{2,6}(?:市|州|地区))',
'district': r'(?:市|州|地区)([\u4e00-\u9fa5]{2,6}(?:区|县|市|旗))'
}
for component, pattern in patterns.items():
self.data[component] = self.data[address_column].str.extract(pattern)
# 提取详细地址（去除省市县后的剩余部分）
for idx, row in self.data.iterrows():
addr = row[address_column]
for col in ['province', 'city', 'district']:
if pd.notna(row[col]) and row[col] in addr:
addr = addr.replace(row[col], '', 1)
self.data.at[idx, 'detail_address'] = addr.strip()
return self
def categorize_express_type(self, express_column='express_company'):
"""对快递公司进行分类"""
express_categories = {
'顺丰': ['SF', '顺丰'],
'京东': ['JD', '京东'],
'中通': ['ZTO', '中通'],
'圆通': ['YTO', '圆通'],
'韵达': ['YD', '韵达'],
'申通': ['STO', '申通'],
'邮政': ['EMS', '中国邮政', '邮政']
}
def categorize(name):
if pd.isna(name):
return '未知'
name = str(name).upper()
for category, keywords in express_categories.items():
for keyword in keywords:
if keyword.upper() in name:
return category
return '其他'
self.data['express_type'] = self.data[express_column].apply(categorize)
return self
def process_datetime(self, datetime_column='create_time'):
"""处理时间格式"""
# 统一时间格式
self.data[datetime_column] = pd.to_datetime(self.data[datetime_column], errors='coerce')
self.data['year'] = self.data[datetime_column].dt.year
self.data['month'] = self.data[datetime_column].dt.month
self.data['day'] = self.data[datetime_column].dt.day
self.data['weekday'] = self.data[datetime_column].dt.day_name() # 星期几
return self
def validate_express_number(self, number_column='express_number'):
"""验证快递单号格式（示例）"""
def validate(number):
# 不同快递公司单号格式不同，这里提供通用验证
if pd.isna(number):
return False
number = str(number).strip()
# 多数快递单号为12-14位数字
return bool(re.match(r'^\d{12,14}$', number))
self.data['number_valid'] = self.data[number_column].apply(validate)
return self
def handle_outliers(self, weight_column='weight'):
"""处理异常数据（如重量异常）"""
# 将重量列转换为数值类型
if weight_column in self.data.columns:
self.data[weight_column] = pd.to_numeric(self.data[weight_column], errors='coerce')
# 标记异常值（例如重量为0或负值或超过100kg）
self.data['weight_abnormal'] = (self.data[weight_column] <= 0) | (self.data[weight_column] > 100)
return self
def summary_statistics(self):
"""生成数据统计信息"""
stats = {}
stats['total_records'] = len(self.data)
stats['invalid_phones'] = len(self.data[self.data.get('phone', '') == '无效号码'])
stats['express_type_distribution'] = self.data.get('express_type', pd.Series()).value_counts().to_dict()
# 统计日期范围
if 'create_time' in self.data.columns:
stats['date_range'] = {
'start': str(self.data['create_time'].min()),
'end': str(self.data['create_time'].max())
}
return stats
def save_processed_data(self, output_path):
"""保存处理后的数据"""
try:
if output_path.endswith('.csv'):
self.data.to_csv(output_path, index=False, encoding='utf-8-sig')
elif output_path.endswith('.xlsx'):
self.data.to_excel(output_path, index=False)
print(f"处理后的数据已保存至: {output_path}")
return True
except Exception as e:
print(f"保存文件失败: {e}")
return False
# 使用示例
def main():
# 假设你的快递数据文件是CSV格式
input_file = 'express_data.csv'
# 创建处理器实例
processor = ExpressDataProcessor(input_file)
# 加载数据
if processor.load_data():
# 执行数据处理流程
(processor
.clean_phone_numbers('phone') # 清洗手机号
.categorize_express_type('company') # 分类快递公司
.extract_address_components('address') # 提取地址组件
.process_datetime('create_time') # 处理时间
.validate_express_number('tracking_no') # 验证单号
)
# 查看数据概览
print("\n=== 数据预览 ===")
print(processor.data.head())
# 生成统计信息
stats = processor.summary_statistics()
print("\n=== 数据统计 ===")
for key, value in stats.items():
print(f"{key}: {value}")
# 保存处理后的数据
processor.save_processed_data('processed_express_data.csv')
if __name__ == "__main__":
main()
使用说明：
安装依赖
pip install pandas openpyxl
数据格式要求
你的快递数据文件应包含以下字段（示例）：
phone,address,express_company,express_number,create_time,weight 13800138000,北京市海淀区中关村大街1号,顺丰速运,SF1234567890,2024-01-01 10:30:00,2.5
主要功能
- 手机号清洗：去除特殊字符，统一格式
- 地址解析：自动提取省市县三级信息
- 快递分类：按公司名称自动分类
- 时间处理：统一时间格式，提取年月日等信息
- 数据验证：检查手机号和单号的有效性
- 异常检测：发现数据中的异常值
- 统计分析：生成数据分布统计
自定义扩展
根据你的具体需求,可以：
- 修改正则表达式适应不同的地址格式
- 添加新的数据清洗规则
- 增加更多的数据验证条件
- 添加数据可视化功能
需要我针对你的具体数据格式做调整吗？
