一键净化Excel数据：高性能Python脚本实现多核并行清理

摘要

本文分享两个基于Python的Excel数据净化脚本，通过多进程并行技术清除工作表内不可见字符、批注、单元格样式等冗余内容，利用OpenPyXL实现底层操作，结合tqdm进度条和进程级任务分配，可快速处理百万级单元格数据。适用于数据分析预处理、跨系统数据迁移等场景。

脚本一：并行处理统一(单)进度条版本

架构设计：

源码：

import openpyxl
from openpyxl.styles import NamedStyle, Font, Border, PatternFill
from openpyxl.formatting import Rule
import re
from tqdm import tqdm
import multiprocessing
from functools import partial

def clean_invisible_chars(text):
    """清除字符串中的不可见字符"""
    if not isinstance(text, str):
        return text
    return re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]', '', text)

def process_cell(cell, no_style, no_fill, cleared_list):
    """处理单个单元格的并行化函数"""
    # 清除单元格所有样式
    cell.fill = no_fill
    cell.font = Font(name='Calibri', size=11, bold=False, italic=False)
    cell.border = Border()
    cell.number_format = 'General'
    
    # 清除不可见字符
    if cell.value and isinstance(cell.value, str):
        cell.value = clean_invisible_chars(cell.value)
        cleared_list.append('不可见字符')
        
    # 清除批注
    if cell.comment:
        cell.comment = None
        cleared_list.append('批注')
        
    # 清除样式
    cell.style = no_style
    cleared_list.append('单元格样式')

def process_sheet(args):
    """处理整个工作表的并行化函数"""
    input_file, output_file, sheet_name = args
    cleared_items = set()
    
    # 每个进程独立处理一个完整的工作表
    wb = openpyxl.load_workbook(input_file)
    ws = wb[sheet_name]
    no_style = NamedStyle(name="Normal")
    no_fill = PatternFill(fill_type=None)
    
    # 清除所有条件格式
    ws.conditional_formatting = []
    cleared_items.add('条件格式填充色')
    
    total_rows = ws.max_row
    total_cols = ws.max_column
    
    # 使用tqdm显示进度
    for row in tqdm(ws.iter_rows(), total=total_rows, desc=f"处理 {sheet_name}"):
        for cell in row:
            # 清除单元格所有样式
            cell.fill = no_fill
            cleared_items.add('背景填充色')
            cell.font = Font(name='Calibri', size=11, bold=False, italic=False)
            cell.border = Border()
            cell.number_format = 'General'
            cell.style = no_style
            cleared_items.add('单元格样式')
            
            # 清除不可见字符
            if cell.value and isinstance(cell.value, str):
                cell.value = clean_invisible_chars(cell.value)
                cleared_items.add('不可见字符')
                
            # 清除批注
            if cell.comment:
                cell.comment = None
                cleared_items.add('批注')
    
    # 保存临时文件
    temp_file = f"temp_{sheet_name}.xlsx"
    wb.save(temp_file)
    
    return (temp_file, sheet_name, list(cleared_items))

def clear_all_and_save(input_file, output_file, sheet_name, num_processes=None):
    """
    多进程并行清除工作表中的:
    1. 不可见字符
    2. 批注
    3. 单元格样式
    """
    if num_processes is None:
        num_processes = multiprocessing.cpu_count()
    
    print(f"使用 {num_processes} 个进程并行处理...")
    
    # 准备参数 (这里可以扩展为处理多个sheet)
    args = [(input_file, output_file, sheet_name)]
    
    # 创建进程池
    with multiprocessing.Pool(processes=num_processes) as pool:
        results = list(tqdm(pool.imap(process_sheet, args), total=len(args), desc="总进度"))
    
    # 合并处理结果
    cleared_items = set()
    for temp_file, sheet_name, items in results:
        cleared_items.update(items)
        # 这里可以添加合并多个临时文件的逻辑
    
    # 保存最终工作簿 (简化处理，直接使用第一个结果)
    import shutil
    shutil.move(results[0][0], output_file)
    
    cleared_text = "\n".join(f"  ✔ 清除-{item}" for item in cleared_items)
    print("已完成：\n"+cleared_text+f"\n并保存到 {output_file}")

if __name__ == '__main__':
    input_excel_file = '测试文件.xlsx' # 原始文件
    output_excel_file = '清除样式_测试文件_并行版.xlsx' # 生成文件
    sheet_to_clean = 'sheet1' # sheet name
    
    print(f"输入文件: {input_excel_file}")
    print(f"输出文件: {output_excel_file}")
    print(f"目标工作表: {sheet_to_clean}")
    
    try:
        clear_all_and_save(input_excel_file, output_excel_file, sheet_to_clean)
    except Exception as e:
        print(f"处理过程中发生错误: {str(e)}")

脚本二：多核独立进度条版本

架构设计：

源码：

import openpyxl
from openpyxl.styles import NamedStyle, Font, Border, PatternFill
from openpyxl.formatting import Rule
import re
from tqdm import tqdm
import multiprocessing
from functools import partial

def clean_invisible_chars(text):
    """清除字符串中的不可见字符"""
    if not isinstance(text, str):
        return text
    return re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]', '', text)

def process_sheet(args):
    """处理整个工作表的并行化函数"""
    input_file, output_file, sheet_name, process_idx = args
    cleared_items = set()
    
    # 每个进程独立处理一个完整的工作表
    wb = openpyxl.load_workbook(input_file)
    ws = wb[sheet_name]
    no_style = NamedStyle(name="Normal")
    no_fill = PatternFill(fill_type=None)
    
    # 清除所有条件格式
    ws.conditional_formatting = []
    cleared_items.add('条件格式填充色')
    
    total_rows = ws.max_row
    total_cols = ws.max_column
    
    # 使用多行进度条(position参数控制行位置)
    process_idx = args[3]  # 获取进程索引
    for row in tqdm(ws.iter_rows(), total=total_rows, 
                   desc=f"进程{process_idx}", position=process_idx+1):
        for cell in row:
            # 清除单元格所有样式
            cell.fill = no_fill
            cell.font = Font(name='Calibri', size=11, bold=False, italic=False)
            cell.border = Border()
            cell.number_format = 'General'
            cell.style = no_style
            cleared_items.update(['背景填充色', '单元格样式'])
            
            # 清除不可见字符
            if cell.value and isinstance(cell.value, str):
                cell.value = clean_invisible_chars(cell.value)
                cleared_items.add('不可见字符')
                
            # 清除批注
            if cell.comment:
                cell.comment = None
                cleared_items.add('批注')
    
    # 保存临时文件
    temp_file = f"temp_{sheet_name}.xlsx"
    wb.save(temp_file)
    
    return (temp_file, sheet_name, list(cleared_items))

def clear_all_and_save(input_file, output_file, sheet_name, num_processes=None):
    """
    多进程并行清除工作表中的:
    1. 不可见字符
    2. 批注
    3. 单元格样式
    """
    if num_processes is None:
        num_processes = multiprocessing.cpu_count()
    
    print(f"使用 {num_processes} 个进程并行处理...")
    
    # 准备参数并添加进程索引
    args = [(input_file, output_file, sheet_name, i) for i in range(num_processes)]
    # 确保至少有一个参数组
    if not args:
        args = [(input_file, output_file, sheet_name, 0)]
    
    # 创建进程池并显示总进度
    with multiprocessing.Pool(processes=num_processes) as pool:
        # 在主进度条下方显示各进程进度
        with tqdm(total=len(args), desc="总进度", position=0) as pbar:
            results = []
            for result in pool.imap(process_sheet, args):
                results.append(result)
                pbar.update()
    
    # 合并处理结果
    cleared_items = set()
    for temp_file, sheet_name, items in results:
        cleared_items.update(items)
        # 这里可以添加合并多个临时文件的逻辑
    
    # 保存最终工作簿 (简化处理，直接使用第一个结果)
    import shutil
    shutil.move(results[0][0], output_file)
    
    cleared_text = "\n".join(f"  ✔ 清除-{item}" for item in cleared_items)
    print("已完成：\n"+cleared_text+f"\n并保存到 {output_file}")

if __name__ == '__main__':
    input_excel_file = '测试文件.xlsx' # 原始文件
    output_excel_file = '清除样式_测试文件_并行版.xlsx' # 生成文件
    sheet_to_clean = 'sheet1' # sheet name
    
    print(f"输入文件: {input_excel_file}")
    print(f"输出文件: {output_excel_file}")
    print(f"目标工作表: {sheet_to_clean}")
    
    try:
        clear_all_and_save(input_excel_file, output_excel_file, sheet_to_clean)
    except Exception as e:
        print(f"处理过程中发生错误: {str(e)}")