当前位置：首页 > wzjs >正文

我想做一个网站本地信息发布平台

wzjs 2025/9/21 23:00:26

我想做一个网站,本地信息发布平台,欧美电商网站,手机网页布局摘要本文分享两个基于Python的Excel数据净化脚本，通过多进程并行技术清除工作表内不可见字符、批注、单元格样式等冗余内容，利用OpenPyXL实现底层操作，结合tqdm进度条和进程级任务分配，可快速处理百万级单元格数据。适用于数据分…

摘要

本文分享两个基于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 partialdef clean_invisible_chars(text):"""清除字符串中的不可见字符"""if not isinstance(text, str):return textreturn re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]', '', text)def process_cell(cell, no_style, no_fill, cleared_list):"""处理单个单元格的并行化函数"""# 清除单元格所有样式cell.fill = no_fillcell.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 = Nonecleared_list.append('批注')# 清除样式cell.style = no_stylecleared_list.append('单元格样式')def process_sheet(args):"""处理整个工作表的并行化函数"""input_file, output_file, sheet_name = argscleared_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_rowtotal_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_fillcleared_items.add('背景填充色')cell.font = Font(name='Calibri', size=11, bold=False, italic=False)cell.border = Border()cell.number_format = 'General'cell.style = no_stylecleared_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 = Nonecleared_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 shutilshutil.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 nameprint(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 partialdef clean_invisible_chars(text):"""清除字符串中的不可见字符"""if not isinstance(text, str):return textreturn re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]', '', text)def process_sheet(args):"""处理整个工作表的并行化函数"""input_file, output_file, sheet_name, process_idx = argscleared_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_rowtotal_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_fillcell.font = Font(name='Calibri', size=11, bold=False, italic=False)cell.border = Border()cell.number_format = 'General'cell.style = no_stylecleared_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 = Nonecleared_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 shutilshutil.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 nameprint(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)}")