python

关注公众号 jb51net

关闭
首页 > 脚本专栏 > python > Python数据处理

Python数据自动化处理之数据清洗和报表生成详解

作者:叫我辉哥e1

这篇文章主要介绍了一个全自动数据处理流水线的Python实现方案,帮助用户解决日常数据处理痛点,文中的示例代码讲解详细,感兴趣的小伙伴可以了解下

前面我们已经搞定了:看板、AI助手、邮件机器人。但还有一个超级痛点数据处理

很多同学每天都要:

今天就用极简代码,带你实现一套全自动数据处理流水线:自动读取原始数据 → 自动清洗 → 自动生成报表 → 自动美化 → 自动保存,真正实现“一键出报表”。

一、本次你能学到什么

用 Python 自动读取多种格式数据(Excel/CSV)

自动数据清洗(去重、补全、格式转换、异常值处理)

自动生成统计指标(求和、平均、最大值、最小值)

自动插入图表(柱状图、折线图、饼图)

一键美化报表(字体、颜色、边框、对齐)

做成可复用的工具,下次直接用

二、前期准备(1 分钟搞定)

1. 安装依赖

打开命令行执行:

pip install pandas openpyxl xlsxwriter python-dotenv

2. 准备测试数据

新建一个 raw_data 文件夹,放一些测试用的 Excel/CSV 文件(模拟从系统导出的原始数据)。

三、核心代码:全自动数据处理流水线(完整可跑)

新建文件:data_processor.py

# -*- coding: utf-8 -*-
import pandas as pd
import os
from datetime import datetime
from dotenv import load_dotenv

# 加载配置
load_dotenv()

# 路径配置
RAW_DATA_PATH = "raw_data"
OUTPUT_PATH = "processed_reports"
if not os.path.exists(OUTPUT_PATH):
    os.mkdir(OUTPUT_PATH)

# ---------------------------
# 1. 读取原始数据
# ---------------------------
def read_raw_data(file_path):
    """自动识别文件格式并读取"""
    ext = os.path.splitext(file_path)[1].lower()
    if ext == ".xlsx":
        return pd.read_excel(file_path)
    elif ext == ".csv":
        return pd.read_csv(file_path)
    else:
        raise ValueError(f"不支持的文件格式:{ext}")

# ---------------------------
# 2. 自动数据清洗
# ---------------------------
def clean_data(df):
    """
    通用数据清洗函数:
    1. 去重
    2. 补全缺失值
    3. 格式转换
    4. 异常值处理
    """
    # 复制一份,避免修改原数据
    df_clean = df.copy()
    
    # 1. 去重
    before_count = len(df_clean)
    df_clean = df_clean.drop_duplicates()
    after_count = len(df_clean)
    print(f"🧹 去重:删除了 {before_count - after_count} 条重复数据")
    
    # 2. 补全缺失值
    # 数值型列用平均值填充
    numeric_cols = df_clean.select_dtypes(include=['number']).columns
    df_clean[numeric_cols] = df_clean[numeric_cols].fillna(df_clean[numeric_cols].mean())
    # 文本型列用"未知"填充
    text_cols = df_clean.select_dtypes(include=['object']).columns
    df_clean[text_cols] = df_clean[text_cols].fillna("未知")
    print("🧹 补全缺失值:已处理所有空值")
    
    # 3. 格式转换(日期列统一格式)
    date_cols = [col for col in df_clean.columns if 'date' in col.lower() or '时间' in col]
    for col in date_cols:
        try:
            df_clean[col] = pd.to_datetime(df_clean[col]).dt.strftime('%Y-%m-%d')
        except:
            pass
    print("🧹 格式转换:日期列已统一格式")
    
    # 4. 异常值处理(简单的3σ原则)
    for col in numeric_cols:
        mean = df_clean[col].mean()
        std = df_clean[col].std()
        lower = mean - 3 * std
        upper = mean + 3 * std
        # 异常值用中位数替换
        median = df_clean[col].median()
        df_clean[col] = df_clean[col].apply(lambda x: median if x < lower or x > upper else x)
    print("🧹 异常值处理:已处理数值型列的异常值")
    
    return df_clean

# ---------------------------
# 3. 自动生成统计指标
# ---------------------------
def generate_stats(df, group_col, value_col):
    """
    生成统计报表:
    按 group_col 分组,统计 value_col 的 总和、平均值、最大值、最小值
    """
    stats = df.groupby(group_col).agg(
        总和=(value_col, 'sum'),
        平均值=(value_col, 'mean'),
        最大值=(value_col, 'max'),
        最小值=(value_col, 'min'),
        数量=(value_col, 'count')
    ).reset_index()
    
    # 格式化数值(保留2位小数)
    numeric_cols = ['总和', '平均值', '最大值', '最小值']
    stats[numeric_cols] = stats[numeric_cols].round(2)
    
    print(f"📊 统计完成:按 {group_col} 分组,统计 {value_col}")
    return stats

# ---------------------------
# 4. 一键美化报表(Excel)
# ---------------------------
def beautify_excel(file_path, sheet_names):
    """
    美化Excel报表:
    1. 设置表头样式
    2. 设置列宽
    3. 设置边框
    4. 设置对齐
    """
    from openpyxl import load_workbook
    from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
    
    wb = load_workbook(file_path)
    
    # 定义样式
    header_font = Font(name='微软雅黑', size=12, bold=True, color='FFFFFF')
    header_fill = PatternFill(start_color='165DFF', end_color='165DFF', fill_type='solid')
    header_alignment = Alignment(horizontal='center', vertical='center')
    cell_font = Font(name='微软雅黑', size=10)
    cell_alignment = Alignment(horizontal='center', vertical='center')
    thin_border = Border(
        left=Side(style='thin'),
        right=Side(style='thin'),
        top=Side(style='thin'),
        bottom=Side(style='thin')
    )
    
    for sheet_name in sheet_names:
        if sheet_name not in wb.sheetnames:
            continue
        ws = wb[sheet_name]
        
        # 设置表头
        for cell in ws[1]:
            cell.font = header_font
            cell.fill = header_fill
            cell.alignment = header_alignment
            cell.border = thin_border
        
        # 设置数据行
        for row in ws.iter_rows(min_row=2):
            for cell in row:
                cell.font = cell_font
                cell.alignment = cell_alignment
                cell.border = thin_border
        
        # 自动调整列宽
        for column in ws.columns:
            max_length = 0
            column_letter = column[0].column_letter
            for cell in column:
                try:
                    if len(str(cell.value)) > max_length:
                        max_length = len(str(cell.value))
                except:
                    pass
            adjusted_width = min(max_length + 2, 50)
            ws.column_dimensions[column_letter].width = adjusted_width
    
    wb.save(file_path)
    print("🎨 美化完成:Excel报表已一键美化")

# ---------------------------
# 5. 自动插入图表
# ---------------------------
def add_charts_to_excel(file_path, stats_df, sheet_name, chart_type='bar'):
    """
    在Excel中插入图表
    """
    import xlsxwriter
    
    # 用xlsxwriter重新写入(方便插入图表)
    temp_path = file_path.replace('.xlsx', '_temp.xlsx')
    with pd.ExcelWriter(temp_path, engine='xlsxwriter') as writer:
        stats_df.to_excel(writer, sheet_name=sheet_name, index=False)
        
        workbook = writer.book
        worksheet = writer.sheets[sheet_name]
        
        # 创建图表
        if chart_type == 'bar':
            chart = workbook.add_chart({'type': 'column'})
        elif chart_type == 'line':
            chart = workbook.add_chart({'type': 'line'})
        elif chart_type == 'pie':
            chart = workbook.add_chart({'type': 'pie'})
        else:
            chart = workbook.add_chart({'type': 'column'})
        
        # 配置图表数据
        max_row = len(stats_df)
        chart.add_series({
            'name': [sheet_name, 0, 1],
            'categories': [sheet_name, 1, 0, max_row, 0],
            'values': [sheet_name, 1, 1, max_row, 1],
        })
        
        # 设置图表标题
        chart.set_title({'name': f'{stats_df.columns[0]} vs {stats_df.columns[1]}'})
        chart.set_x_axis({'name': stats_df.columns[0]})
        chart.set_y_axis({'name': stats_df.columns[1]})
        
        # 插入图表
        worksheet.insert_chart('F2', chart)
    
    # 替换原文件
    os.replace(temp_path, file_path)
    print("📈 图表插入完成")

# ---------------------------
# 主函数:一键处理数据
# ---------------------------
def run_data_processor(file_name, group_col, value_col, chart_type='bar'):
    print("🚀 数据处理流水线启动...")
    
    # 1. 读取数据
    file_path = os.path.join(RAW_DATA_PATH, file_name)
    if not os.path.exists(file_path):
        print(f"❌ 文件不存在:{file_path}")
        return
    df = read_raw_data(file_path)
    print(f"✅ 读取成功:共 {len(df)} 条数据")
    
    # 2. 清洗数据
    df_clean = clean_data(df)
    
    # 3. 生成统计
    stats = generate_stats(df_clean, group_col, value_col)
    
    # 4. 保存结果
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    output_file = os.path.join(OUTPUT_PATH, f'报表_{timestamp}.xlsx')
    
    # 写入多个sheet
    with pd.ExcelWriter(output_file, engine='openpyxl') as writer:
        df_clean.to_excel(writer, sheet_name='清洗后数据', index=False)
        stats.to_excel(writer, sheet_name='统计报表', index=False)
    
    # 5. 插入图表
    add_charts_to_excel(output_file, stats, '统计报表', chart_type)
    
    # 6. 美化报表
    beautify_excel(output_file, ['清洗后数据', '统计报表'])
    
    print(f"🎉 全部完成!报表已保存至:{output_file}")
    return output_file

if __name__ == "__main__":
    # 示例:处理销售数据,按"产品"分组,统计"销售额",插入柱状图
    # 你需要先在 raw_data 文件夹放一个测试文件
    # run_data_processor('销售数据.xlsx', '产品', '销售额', 'bar')
    print("请在代码中配置你的数据文件和统计字段后运行!")

新建 .env 文件(可选):

# 可以在这里配置默认参数
DEFAULT_GROUP_COL=产品
DEFAULT_VALUE_COL=销售额
DEFAULT_CHART_TYPE=bar

四、运行效果(直接看得到)

运行后你会立刻看到:

1.控制台显示每一步的处理进度

2.processed_reports 文件夹生成一个带时间戳的 Excel 文件

3.Excel 包含两个 sheet:

4.报表已经一键美化:蓝色表头、统一字体、自动列宽、居中对齐

你以后只需要:把原始数据丢进 raw_data → 改一下代码里的字段名 → 运行 → 拿报表

五、可以直接扩展的进阶功能

你可以在这篇基础上随便加,非常适合写进博客:

六、总结

本篇我们实现了企业级全自动数据处理流水线,从 0 到 1 完成:原始数据读取 → 自动清洗(去重/补全/格式转换)→ 自动统计 → 自动插入图表 → 一键美化。

代码轻量、稳定、可复用,新手也能直接跑起来,每天帮你节省大量处理数据的时间。

这也是办公自动化体系里非常核心的一环:数据处理是基础,看板负责展示,AI 助手负责交互,邮件机器人负责分发,真正实现全流程自动化。

到此这篇关于Python数据自动化处理之数据清洗和报表生成详解的文章就介绍到这了,更多相关Python数据处理内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

您可能感兴趣的文章:
阅读全文