从 0 到 1 打造一个工业级 Excel 拆分合并工具:Python 工程化实战全记录

┌─────────────────────────────────────────┐│ 表现层：Tkinter GUI（excel_split_merge_tool.py）│├─────────────────────────────────────────┤│ 业务逻辑层：Excel 引擎（excel_engine.py） │├─────────────────────────────────────────┤│ 数据层：Pandas + Openpyxl │├─────────────────────────────────────────┤│ 基础设施层：配置管理、日志、异常处理 │└─────────────────────────────────────────┘

import pandas as pddef split_by_column_basic(file_path, column_name, output_dir): ”””基础版：按列拆分 Excel””” df = pd.read_excel(file_path) grouped = df.groupby(column_name) for name, group in grouped: safe_name = str(name).replace('/', '').replace('\\', '') safe_name = safe_name.replace('*', '').replace('?', '').replace(':', '') output_file = f”{output_dir}/{safe_name}.xlsx” group.to_excel(output_file, index=False)

def split_by_column_chunked(self, file_path, column_name, output_dir): ”””工程版：分块读取，内存占用稳定在 800MB 左右”””# 1. 分块读取，避免内存爆炸 chunk_iter = pd.read_excel(file_path, chunksize=100_000)# 2. 用字典缓存每个分组的数据 groups = {} total_rows = 0 for chunk in chunk_iter: total_rows += len(chunk) for key, group in chunk.groupby(column_name): if key not in groups: groups[key] = [] groups[key].append(group)# 3. 每积累一定量就落盘，避免内存堆积 if sum(len(g) for g in groups[key]) > 50_000: self._flush_group(key, groups[key], output_dir) groups[key] = []# 4. 处理剩余数据 for key, chunks in groups.items(): if chunks: self._flush_group(key, chunks, output_dir)

def split_to_multiple_sheets(self, file_path, column_name, output_file): ”””拆分到多个 Sheet，需处理 Sheet 名非法字符””” df = pd.read_excel(file_path) grouped = df.groupby(column_name) with pd.ExcelWriter(output_file, engine='openpyxl') as writer: for name, group in grouped:# Sheet 名最长 31 字符 safe_name = str(name)[:31]# 移除非法字符 \ / : * ? [ ] safe_name = re.sub(r'[\\/:*?\[\]]', '_', safe_name) group.to_excel(writer, sheet_name=safe_name, index=False)

# ❌ 错误写法：UI 主线程被阻塞，进度条不动，窗口”假死”def on_split_click(): split_by_column_chunked(...)# 这一行会阻塞主线程几分钟 update_progress(100)

import tkinter as tkfrom tkinter import ttkimport threadingimport queueclass ExcelToolGUI: def __init__(self, root): self.root = root self.progress_var = tk.DoubleVar() self.task_queue = queue.Queue()# 进度条 self.progress_bar = ttk.Progressbar( root, variable=self.progress_var, maximum=100 ) self.progress_bar.pack() def start_split(self): ”””启动拆分任务”””# 禁用按钮，防止重复点击 self.start_btn.config(state=tk.DISABLED) self.progress_var.set(0)# 启动后台线程 self.worker_thread = threading.Thread( target=self._split_worker,  daemon=True# 守护线程：主程序退出时自动终止 ) self.worker_thread.start()# 主线程开始轮询队列 self._check_queue() def _split_worker(self): ”””后台线程：执行实际拆分工作””” try:# 模拟分块处理 for i, chunk_result in enumerate(self.engine.process_chunks()):# 把进度放进队列 progress = (i / total_chunks) * 100 self.task_queue.put((”progress”, progress)) self.task_queue.put((”done”, ”拆分完成”)) except Exception as e: self.task_queue.put((”error”, str(e))) def _check_queue(self): ”””主线程：定时检查队列，安全更新 UI””” try: status, data = self.task_queue.get_nowait() if status == ”progress”: self.progress_var.set(data) elif status == ”done”: self.start_btn.config(state=tk.NORMAL) self.progress_var.set(100) elif status == ”error”: messagebox.showerror(”错误”, data) self.start_btn.config(state=tk.NORMAL) except queue.Empty: pass# 继续轮询（关键：用 after 调度，不阻塞事件循环） self.root.after(100, self._check_queue)

# 指定 dtype，大幅降低内存df = pd.read_excel( file_path,  chunksize=100_000, dtype={ 'order_id': 'int32', 'product_name': 'category',# 低基数字符串用 category 'price': 'float32', 'quantity': 'int16' })

def split_by_column_safe(self, file_path, column_name, output_dir): ”””带完整异常处理的拆分函数””” try:# 1. 文件存在性检查 if not os.path.exists(file_path): raise FileNotFoundError(f”文件不存在: {file_path}”)# 2. 读取 Excel try: df = pd.read_excel(file_path) except pd.errors.EmptyDataError: raise ValueError(”Excel 文件为空”) except Exception as e: raise RuntimeError(f”读取 Excel 失败: {e}”)# 3. 列名校验 if column_name not in df.columns: available = ', '.join(df.columns) raise KeyError(f”列 '{column_name}' 不存在。可用列: {available}”)# 4. 数据类型校验 if df[column_name].isna().sum() > 0: self.logger.warning( f”列 '{column_name}' 有 {df[column_name].isna().sum()} 个空值，将被忽略” )# 5. 执行拆分 grouped = df.groupby(column_name) success_count = 0 for name, group in grouped: try: safe_name = self._sanitize_filename(str(name)) output_path = os.path.join(output_dir, f”{safe_name}.xlsx”) group.to_excel(output_path, index=False) success_count += 1 except Exception as e: self.logger.error(f”拆分 '{name}' 失败: {e}”) continue# 单个分组失败不影响其他分组 return success_count except Exception as e:# 统一异常处理 self.logger.error(f”拆分任务失败: {e}”) raise finally:# 清理资源 self.progress_var.set(0)

# 单文件 + 无控制台窗口（GUI 程序标配）pyinstaller -F -w --name=”Excel拆分合并工具” excel_split_merge_tool.py

# build_exe.pyimport subprocessimport osdef build_exe(): ”””一键打包脚本””” cmd = [ ”pyinstaller”, ”--onefile”,# 单文件 ”--windowed”,# 无控制台 ”--name=Excel拆分合并工具”, ”--icon=app.ico”,# 自定义图标 ”--hidden-import=pandas”, ”--hidden-import=openpyxl”, ”--hidden-import=tkinter”, ”--hidden-import=tkinter.ttk”, ”--clean”,# 清理临时文件 ”excel_split_merge_tool.py” ]# 如果图标不存在，移除相关参数 if not os.path.exists(”app.ico”): cmd = [arg for arg in cmd if ”app.ico” not in arg] subprocess.run(cmd, check=True)if __name__ == ”__main__”: build_exe()

import sysimport osdef get_resource_path(relative_path): ”””适配打包前后的资源路径””” if getattr(sys, 'frozen', False):# 打包后：使用 PyInstaller 的临时目录 base_path = sys._MEIPASS else:# 开发环境：使用当前目录 base_path = os.path.abspath(”.”) return os.path.join(base_path, relative_path)

任何”读取文件 → 分组/聚合 → 批量输出”的场景都能套用：┌────────────────────────────────────────────┐│ 场景 │ 分组键 │├────────────────────────────────────────────┤│ Excel 按列拆分 │ 某列的值 ││ 日志文件拆分 │ 日期/级别/IP ││ CSV 数据分片 │ 哈希值/行数范围 ││ 配置文件合并 │ 模块名/环境名 ││ 图片按标签分类 │ EXIF 标签 │└────────────────────────────────────────────┘

excel_tool/├── excel_split_merge_tool.py# GUI 主程序（449 行）├── excel_engine.py# 核心引擎（292 行）├── build_exe.py# 打包脚本（48 行）├── config.json# 配置文件（自动记忆上次路径）├── requirements.txt# 依赖清单├── README.md# 使用文档└── test_data.xlsx# 测试数据