如何写快递数据整理脚本

我来帮你编写一个快递数据整理脚本,以下是一个通用的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

主要功能

• 手机号清洗：去除特殊字符，统一格式
• 地址解析：自动提取省市县三级信息
• 快递分类：按公司名称自动分类
• 时间处理：统一时间格式，提取年月日等信息
• 数据验证：检查手机号和单号的有效性
• 异常检测：发现数据中的异常值
• 统计分析：生成数据分布统计

自定义扩展

根据你的具体需求,可以：

• 修改正则表达式适应不同的地址格式
• 添加新的数据清洗规则
• 增加更多的数据验证条件
• 添加数据可视化功能

需要我针对你的具体数据格式做调整吗？