处理过程中发生错误:'Openpyx|Writer'object has no attribute 'path'

<think>这个错误发生在`perform_statistical_analysis`方法中，当我们尝试使用`pd.read_excel(writer.path)`时，因为`pd.ExcelWriter`对象并没有`path`属性。实际上，在pandas中，我们可以通过`writer.handles`来获取文件路径，但更推荐的方式是在创建writer时保存文件路径。 解决方案： 1. 在创建`pd.ExcelWriter`对象时，将文件路径保存为实例变量。 2. 在`perform_statistical_analysis`方法中，使用保存的文件路径来读取Excel文件。 修改代码： 在`execute_merge_analysis`方法中，我们创建writer时，将文件路径保存为`self.current_save_path`（或者作为局部变量，但为了在方法间传递，我们保存为实例变量）。 然后，在`perform_statistical_analysis`方法中，使用这个保存的路径来读取Excel文件。 具体修改如下： 在`execute_merge_analysis`方法中： - 将`save_path`保存为实例变量：`self.current_save_path = save_path` - 创建writer：`writer = pd.ExcelWriter(self.current_save_path, engine='openpyxl')` 在`perform_statistical_analysis`方法中，读取文件时使用`self.current_save_path`。 但是注意：在`perform_statistical_analysis`方法中，我们是在写入过程中读取同一个文件，这可能会导致错误，因为文件可能被占用。更好的做法是不要从正在写入的文件中读取，而是直接使用已经加载到内存的DataFrame。 因此，我们可以重构： - 在合并数据时，将合并后的数据（或工作页合并时的多个工作表）保存在内存中，而不是每次都从磁盘读取。 - 或者，在写入统计结果之前，先完成所有合并操作，然后关闭writer，再重新打开文件进行统计分析？但这样效率低。 考虑到性能，我们选择在内存中保存需要统计分析的数据。 具体调整： 1. 对于工作页合并：我们不需要在统计分析时读取整个工作簿，因为每个工作表是独立的。我们可以跳过工作页合并的统计分析，或者只分析第一个工作表？根据需求，用户可能希望分析所有工作表，但这样比较复杂。根据之前的设计，工作页合并方式下，我们只分析第一个工作表。所以我们可以这样： 在`merge_by_worksheet`方法中，我们只保存第一个工作表的数据到内存（或者不保存，因为工作页合并方式下，每个工作表都是独立的，统计分析可能只针对单个表）。但为了统一，我们可以在合并时收集所有数据，但这样内存占用大。 2. 我们改变设计：对于工作页合并，不进行统计分析？或者只分析第一个工作表？在需求中并没有明确。为了简化，我们修改为： 在`perform_statistical_analysis`方法中，如果合并方式是工作页合并，则跳过统计分析？或者只分析第一个工作表？根据之前代码，我们只分析第一个工作表。所以我们可以保留这种方式，但避免从磁盘读取，而是从内存中读取第一个工作表的数据。 修改方案： 在合并过程中，将合并后的数据（或需要分析的数据）保存在内存中，然后传递给`perform_statistical_analysis`方法。 由于三种合并方式生成的数据结构不同，我们需要分别处理： - 工作页合并：每个工作表单独存储，但统计分析只针对第一个工作表。我们可以保存第一个工作表的DataFrame。 - 区块合并：只有一个合并后的DataFrame。 - 按列合并：只有一个合并后的DataFrame。 因此，在合并方法中，我们除了写入Excel，还要将需要分析的数据保存到一个变量中（例如`self.analysis_df`），如果是工作页合并，则保存第一个工作表的DataFrame。 具体步骤： 1. 在`execute_merge_analysis`方法中，初始化一个变量`analysis_df = None`。 2. 在合并方法中： - 工作页合并：在写入每个工作表后，如果是第一个文件，将df保存为`analysis_df`。 - 区块合并：将合并后的DataFrame（merged_df）保存为`analysis_df`。 - 按列合并：将合并后的DataFrame（merged_df）保存为`analysis_df`。 3. 然后，在调用`perform_statistical_analysis`方法时，将`analysis_df`作为参数传入，这样就不需要从磁盘读取了。 这样修改后，`perform_statistical_analysis`方法将直接使用内存中的DataFrame进行统计分析。 修改后的代码片段： 在`execute_merge_analysis`方法中： ```python # ... 省略前面的代码 ... # 步骤3: 根据合并方式执行合并 writer = pd.ExcelWriter(save_path, engine='openpyxl') analysis_df = None # 用于统计分析的数据 if self.merge_method == "worksheet": analysis_df = self.merge_by_worksheet(writer) # 修改merge_by_worksheet返回第一个工作表的DataFrame elif self.merge_method == "block": analysis_df = self.merge_by_block(writer) # 返回合并后的DataFrame elif self.merge_method == "column": analysis_df = self.merge_by_column(writer) # 返回合并后的DataFrame self.update_progress(60) # 步骤4: 执行统计分析 self.perform_statistical_analysis(writer, analysis_df) # 将analysis_df传入 self.update_progress(80) # ... 省略后面的代码 ... ``` 然后，修改各个合并方法，使其返回需要分析的DataFrame。 对于`merge_by_worksheet`方法： - 返回第一个文件的DataFrame 对于`merge_by_block`和`merge_by_column`方法： - 返回合并后的DataFrame 同时，在`perform_statistical_analysis`方法中，不再从文件读取，而是直接使用传入的DataFrame。 注意：如果用户没有选择任何统计选项，则跳过。 修改后的`perform_statistical_analysis`方法： ```python def perform_statistical_analysis(self, writer, analysis_df): """执行统计分析，使用内存中的DataFrame""" if not self.stats_options or analysis_df is None: return # 创建一个新的DataFrame来存储统计结果 stats_results = [] # 使用传入的analysis_df进行统计分析 # ... 后面的代码保持不变，只是不再从文件读取，直接使用analysis_df ... ``` 同时，修改各个合并方法，使其返回相应的DataFrame。 由于修改点较多，下面给出修改后的完整代码（仅针对修改的部分）：</think>这个错误是因为在 `perform_statistical_analysis` 方法中尝试访问 `writer.path` 属性，但 `pd.ExcelWriter` 对象并没有 `path` 属性。让我们修复这个问题并优化代码： ```python import tkinter as tk from tkinter import ttk, filedialog, messagebox, scrolledtext import os import pandas as pd import numpy as np from datetime import datetime import re import threading import openpyxl from openpyxl.utils import get_column_letter class EnhancedFileMergeApp(ttk.Frame): """增强版文件合并统计分析界面""" # ... [其他代码保持不变] ... def execute_merge_analysis(self): """执行合并与分析的核心逻辑""" try: # 步骤1: 收集用户选项 self.collect_options() self.update_progress(10) # 步骤2: 弹出保存对话框 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") default_name = f"合并分析结果_{timestamp}.xlsx" save_path = filedialog.asksaveasfilename( defaultextension=".xlsx", filetypes=[("Excel文件", "*.xlsx")], initialfile=default_name ) if not save_path: self.status_var.set("操作已取消") self.update_progress(0) self.execute_btn.config(state=tk.NORMAL) return self.current_save_path = save_path # 保存路径供后续使用 self.update_progress(20) # 步骤3: 根据合并方式执行合并 with pd.ExcelWriter(save_path, engine='openpyxl') as writer: if self.merge_method == "worksheet": self.merge_by_worksheet(writer) elif self.merge_method == "block": self.merge_by_block(writer) elif self.merge_method == "column": self.merge_by_column(writer) self.update_progress(60) # 步骤4: 执行统计分析 if self.stats_options: # 获取合并后的数据 if self.merge_method == "worksheet": # 对于工作页合并，使用第一个工作表的数据 df = pd.read_excel(save_path, sheet_name=0) else: # 对于其他合并方式，使用"合并数据"工作表 df = pd.read_excel(save_path, sheet_name="合并数据") self.perform_statistical_analysis(writer, df) self.update_progress(100) # 步骤5: 显示结果 self.show_result(save_path) self.status_var.set(f"处理完成! 文件已保存至: {save_path}") except Exception as e: messagebox.showerror("错误", f"处理过程中发生错误: {str(e)}") self.status_var.set(f"错误: {str(e)}") finally: self.execute_btn.config(state=tk.NORMAL) def perform_statistical_analysis(self, writer, df): """执行统计分析""" # 创建一个新的DataFrame来存储统计结果 stats_results = [] # 汇总统计 if "summary" in self.stats_options: summary_stats = df.describe().reset_index() summary_stats.columns = ["统计项"] + list(summary_stats.columns[1:]) stats_results.append(("汇总统计", summary_stats)) # 计数统计 if "count" in self.stats_options: count_stats = pd.DataFrame({ "列名": df.columns, "非空值数量": df.count().values, "唯一值数量": df.nunique().values }) stats_results.append(("计数统计", count_stats)) # 平均值计算 if "avg" in self.stats_options: numeric_cols = df.select_dtypes(include=[np.number]).columns if not numeric_cols.empty: avg_stats = pd.DataFrame({ "列名": numeric_cols, "平均值": [df[col].mean() for col in numeric_cols] }) stats_results.append(("平均值", avg_stats)) # 极值分析 if "minmax" in self.stats_options: numeric_cols = df.select_dtypes(include=[np.number]).columns if not numeric_cols.empty: minmax_stats = pd.DataFrame({ "列名": numeric_cols, "最小值": [df[col].min() for col in numeric_cols], "最大值": [df[col].max() for col in numeric_cols], "极差": [df[col].max() - df[col].min() for col in numeric_cols] }) stats_results.append(("极值分析", minmax_stats)) # 将统计结果写入Excel for sheet_name, stats_df in stats_results: stats_df.to_excel(writer, sheet_name=sheet_name, index=False) def merge_by_worksheet(self, writer): """工作页合并方式""" # 创建一个字典来存储所有工作表数据 all_sheets = {} for i, file_path in enumerate(self.file_paths): try: # 获取文件名（不带扩展名）作为工作表名 sheet_name = os.path.splitext(os.path.basename(file_path))[0] # 读取Excel文件 df = pd.read_excel(file_path) # 排除指定列 if self.exclude_columns: df = self.exclude_specified_columns(df) # 保存数据用于统计分析 all_sheets[sheet_name] = df # 写入新工作表 df.to_excel(writer, sheet_name=sheet_name, index=False) self.status_var.set(f"正在处理: {os.path.basename(file_path)}") progress = 20 + (i / len(self.file_paths)) * 20 self.update_progress(progress) except Exception as e: messagebox.showwarning("警告", f"处理文件 {file_path} 时出错: {str(e)}") continue # 返回第一个工作表的数据用于统计分析 return list(all_sheets.values())[0] if all_sheets else None def merge_by_block(self, writer): """区块合并方式""" merged_df = pd.DataFrame() for i, file_path in enumerate(self.file_paths): try: # 读取Excel文件 df = pd.read_excel(file_path) # 排除指定列 if self.exclude_columns: df = self.exclude_specified_columns(df) # 追加到合并的DataFrame merged_df = pd.concat([merged_df, df], ignore_index=True) self.status_var.set(f"正在合并: {os.path.basename(file_path)}") progress = 20 + (i / len(self.file_paths)) * 20 self.update_progress(progress) except Exception as e: messagebox.showwarning("警告", f"处理文件 {file_path} 时出错: {str(e)}") continue # 写入合并结果 merged_df.to_excel(writer, sheet_name="合并数据", index=False) return merged_df def merge_by_column(self, writer): """按列合并方式""" # 用于存储每个文件的DataFrame dfs = [] column_names = None # 首先读取所有文件 for i, file_path in enumerate(self.file_paths): try: df = pd.read_excel(file_path) # 如果是第一个文件，保存列名 if column_names is None: column_names = df.columns.tolist() # 确保所有文件具有相同的列 elif df.columns.tolist() != column_names: messagebox.showwarning("警告", f"文件 {os.path.basename(file_path)} 的列结构与其他文件不一致") continue # 排除指定列 if self.exclude_columns: df = self.exclude_specified_columns(df) dfs.append(df) self.status_var.set(f"正在读取: {os.path.basename(file_path)}") progress = 20 + (i / len(self.file_paths)) * 20 self.update_progress(progress) except Exception as e: messagebox.showwarning("警告", f"处理文件 {file_path} 时出错: {str(e)}") continue # 如果没有有效文件，直接返回 if not dfs: return None # 创建合并后的DataFrame merged_df = dfs[0].copy() # 对于每个后续文件，按列合并数据 for i in range(1, len(dfs)): for col in merged_df.columns: # 跳过排除的列 if self.column_to_index(col) in self.exclude_columns: continue # 数值列相加 if pd.api.types.is_numeric_dtype(merged_df[col]): merged_df[col] += dfs[i][col] # 字符串列去重合并 elif pd.api.types.is_string_dtype(merged_df[col]): merged_df[col] = merged_df[col].astype(str) + "、" + dfs[i][col].astype(str) # 去重处理 merged_df[col] = merged_df[col].apply( lambda x: "、".join(set(filter(None, x.split("、")))) if x else x ) # 写入合并结果 merged_df.to_excel(writer, sheet_name="合并数据", index=False) return merged_df # ... [其他方法保持不变] ... def column_to_index(self, col_name): """将列名转换为Excel列字母表示""" # 如果已经是字母表示，直接返回 if isinstance(col_name, str) and re.match(r'^[A-Z]+$', col_name): return col_name # 尝试将列名转换为列索引 if hasattr(self, 'column_mapping') and col_name in self.column_mapping: return self.column_mapping[col_name] # 默认返回列名 return col_name def index_to_letter(self, index): """将数字索引转换为Excel列字母""" return get_column_letter(index + 1) # openpyxl的列索引从1开始 # ... [其他方法保持不变] ... # ... [MainApplication类保持不变] ... ``` ## 错误修复和优化说明 ### 1. 修复 `'Openpyxl Writer' object has no attribute 'path'` 错误 - **问题原因**：`pd.ExcelWriter` 对象没有 `path` 属性 - **解决方案**： - 在 `execute_merge_analysis` 方法中保存文件路径到 `self.current_save_path` - 在 `perform_statistical_analysis` 方法中直接使用保存的路径读取文件 - 使用 `with pd.ExcelWriter(...)` 语句确保文件正确关闭 ### 2. 优化统计分析流程 - **重构逻辑**： - 在合并方法中返回合并后的 DataFrame - 将合并后的 DataFrame 直接传递给 `perform_statistical_analysis` 方法 - 避免从正在写入的文件中读取数据 ### 3. 增强数据类型处理 - **字符串处理优化**： ```python merged_df[col] = merged_df[col].apply( lambda x: "、".join(set(filter(None, x.split("、")))) if x else x ) ``` - 使用 `filter(None, ...)` 过滤空字符串 - 使用 `set` 确保唯一性 ### 4. 使用 openpyxl 工具函数 - **列字母转换优化**： ```python from openpyxl.utils import get_column_letter def index_to_letter(self, index): return get_column_letter(index + 1) # openpyxl的列索引从1开始 ``` ### 5. 添加空值检查 - **防止空 DataFrame 错误**： ```python if not numeric_cols.empty: # 执行数值计算 ``` ### 6. 改进错误处理 - **更健壮的错误捕获**： ```python except Exception as e: messagebox.showerror("错误", f"处理过程中发生错误: {str(e)}") self.status_var.set(f"错误: {str(e)}") finally: self.execute_btn.config(state=tk.NORMAL) ``` ## 完整修复后的文件合并统计分析界面代码 ```python import tkinter as tk from tkinter import ttk, filedialog, messagebox, scrolledtext import os import pandas as pd import numpy as np from datetime import datetime import re import threading import openpyxl from openpyxl.utils import get_column_letter class EnhancedFileMergeApp(ttk.Frame): """增强版文件合并统计分析界面""" def __init__(self, master): super().__init__(master) self.pack(fill=tk.BOTH, expand=True) self.file_paths = [] # 存储所有选择的Excel文件路径 self.exclude_columns = [] # 存储不参与合并的列 self.merge_method = "worksheet" # 默认合并方式 self.stats_options = [] # 存储统计选项 self.current_save_path = "" # 当前保存路径 # 创建主容器 main_frame = ttk.Frame(self) main_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=10) # 左侧文件选择区域 left_frame = ttk.LabelFrame(main_frame, text="文件选择") left_frame.pack(side=tk.LEFT, fill=tk.BOTH, expand=False, padx=5, pady=5) # 文件列表框和滚动条 list_frame = ttk.Frame(left_frame) list_frame.pack(fill=tk.BOTH, expand=True, padx=5, pady=5) scrollbar = ttk.Scrollbar(list_frame) scrollbar.pack(side=tk.RIGHT, fill=tk.Y) self.file_listbox = tk.Listbox( list_frame, yscrollcommand=scrollbar.set, height=10, width=40 ) self.file_listbox.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) scrollbar.config(command=self.file_listbox.yview) # 文件操作按钮 btn_frame = ttk.Frame(left_frame) btn_frame.pack(fill=tk.X, padx=5, pady=5) ttk.Button( btn_frame, text="添加文件", command=self.add_files ).pack(side=tk.LEFT, padx=5, fill=tk.X, expand=True) ttk.Button( btn_frame, text="移除选中", command=self.remove_file ).pack(side=tk.LEFT, padx=5, fill=tk.X, expand=True) ttk.Button( btn_frame, text="清空列表", command=self.clear_files ).pack(side=tk.LEFT, padx=5, fill=tk.X, expand=True) # 右侧设置区域 right_frame = ttk.Frame(main_frame) right_frame.pack(side=tk.RIGHT, fill=tk.BOTH, expand=True, padx=5, pady=5) # 合并选项区域 merge_frame = ttk.LabelFrame(right_frame, text="合并选项") merge_frame.pack(fill=tk.X, padx=5, pady=5) # 指定列不合并 exclude_frame = ttk.Frame(merge_frame) exclude_frame.pack(fill=tk.X, padx=5, pady=5) ttk.Label(exclude_frame, text="指定列不合并:").pack(side=tk.LEFT) self.exclude_entry = ttk.Entry(exclude_frame, width=30) self.exclude_entry.pack(side=tk.LEFT, padx=5, fill=tk.X, expand=True) self.exclude_entry.insert(0, "A,B,C") # 默认值 ttk.Button( exclude_frame, text="应用", command=self.update_exclude_columns, width=8 ).pack(side=tk.RIGHT) # 合并方式单选按钮 method_frame = ttk.Frame(merge_frame) method_frame.pack(fill=tk.X, padx=5, pady=5) self.merge_var = tk.StringVar(value="worksheet") ttk.Radiobutton( method_frame, text="工作页合并", variable=self.merge_var, value="worksheet", command=lambda: self.set_merge_method("worksheet") ).pack(side=tk.LEFT, padx=10) ttk.Radiobutton( method_frame, text="区块合并", variable=self.merge_var, value="block", command=lambda: self.set_merge_method("block") ).pack(side=tk.LEFT, padx=10) ttk.Radiobutton( method_frame, text="按列合并", variable=self.merge_var, value="column", command=lambda: self.set_merge_method("column") ).pack(side=tk.LEFT, padx=10) # 合并方式说明标签 self.method_desc = ttk.Label( merge_frame, text="工作页合并：每个Excel文件作为新工作簿中的一个工作表", font=("Arial", 9), foreground="#555555" ) self.method_desc.pack(fill=tk.X, padx=5, pady=(0, 5)) # 统计分析选项 stats_frame = ttk.LabelFrame(right_frame, text="统计分析选项") stats_frame.pack(fill=tk.X, padx=5, pady=5) self.stats_vars = { "summary": tk.BooleanVar(value=True), "count": tk.BooleanVar(value=True), "avg": tk.BooleanVar(value=False), "minmax": tk.BooleanVar(value=False) } ttk.Checkbutton( stats_frame, text="汇总统计", variable=self.stats_vars["summary"] ).grid(row=0, column=0, padx=10, pady=5, sticky="w") ttk.Checkbutton( stats_frame, text="计数统计", variable=self.stats_vars["count"] ).grid(row=0, column=1, padx=10, pady=5, sticky="w") ttk.Checkbutton( stats_frame, text="平均值计算", variable=self.stats_vars["avg"] ).grid(row=1, column=0, padx=10, pady=5, sticky="w") ttk.Checkbutton( stats_frame, text="极值分析", variable=self.stats_vars["minmax"] ).grid(row=1, column=1, padx=10, pady=5, sticky="w") # 执行按钮 execute_frame = ttk.Frame(right_frame) execute_frame.pack(fill=tk.X, padx=5, pady=10) self.execute_btn = ttk.Button( execute_frame, text="开始合并与分析", command=self.start_merge_analysis, style="Accent.TButton", width=20 ) self.execute_btn.pack(pady=5) # 进度条 self.progress_var = tk.DoubleVar() self.progress_bar = ttk.Progressbar( execute_frame, variable=self.progress_var, maximum=100, mode="determinate" ) self.progress_bar.pack(fill=tk.X, pady=5) # 状态标签 self.status_var = tk.StringVar(value="就绪") ttk.Label( execute_frame, textvariable=self.status_var, font=("Arial", 9), foreground="#666666" ).pack() # 配置样式 self.style = ttk.Style() self.style.configure("Accent.TButton", foreground="white", background="#3498db", font=("Arial", 10, "bold"), padding=6) def add_files(self): """添加Excel文件到列表""" files = filedialog.askopenfilenames( filetypes=[("Excel文件", "*.xls *.xlsx *.xlsm")] ) for file in files: if file not in self.file_listbox.get(0, tk.END): self.file_listbox.insert(tk.END, file) self.file_paths.append(file) def remove_file(self): """移除选中的文件""" selected = self.file_listbox.curselection() if selected: index = selected[0] self.file_listbox.delete(index) if index < len(self.file_paths): del self.file_paths[index] def clear_files(self): """清空文件列表""" self.file_listbox.delete(0, tk.END) self.file_paths = [] def update_exclude_columns(self): """更新不参与合并的列""" exclude_str = self.exclude_entry.get().strip() if exclude_str: # 分割并清理列标识 self.exclude_columns = [col.strip().upper() for col in exclude_str.split(',')] messagebox.showinfo("成功", f"已设置不合并列: {', '.join(self.exclude_columns)}") else: self.exclude_columns = [] messagebox.showinfo("提示", "已清空不合并列设置") def set_merge_method(self, method): """设置合并方式并更新描述""" self.merge_method = method if method == "worksheet": self.method_desc.config(text="工作页合并：每个Excel文件作为新工作簿中的一个工作表") elif method == "block": self.method_desc.config(text="区块合并：将每个Excel文件工作页当前数据行合并到同一个工作页中") elif method == "column": self.method_desc.config(text="按列合并：合并列对应的合并单元格数据进行判断（数值相加，字符串去重合并）") def start_merge_analysis(self): """开始合并与分析过程""" if not self.file_paths: messagebox.showwarning("警告", "请至少添加一个Excel文件") return # 禁用按钮防止重复点击 self.execute_btn.config(state=tk.DISABLED) self.progress_var.set(0) self.status_var.set("正在处理...") # 在后台线程中执行合并与分析 threading.Thread(target=self.execute_merge_analysis, daemon=True).start() def execute_merge_analysis(self): """执行合并与分析的核心逻辑""" try: # 步骤1: 收集用户选项 self.collect_options() self.update_progress(10) # 步骤2: 弹出保存对话框 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") default_name = f"合并分析结果_{timestamp}.xlsx" save_path = filedialog.asksaveasfilename( defaultextension=".xlsx", filetypes=[("Excel文件", "*.xlsx")], initialfile=default_name ) if not save_path: self.status_var.set("操作已取消") self.update_progress(0) self.execute_btn.config(state=tk.NORMAL) return self.current_save_path = save_path # 保存路径供后续使用 self.update_progress(20) # 步骤3: 根据合并方式执行合并 with pd.ExcelWriter(save_path, engine='openpyxl') as writer: merged_data = None if self.merge_method == "worksheet": merged_data = self.merge_by_worksheet(writer) elif self.merge_method == "block": merged_data = self.merge_by_block(writer) elif self.merge_method == "column": merged_data = self.merge_by_column(writer) self.update_progress(60) # 步骤4: 执行统计分析 if self.stats_options and merged_data is not None: self.perform_statistical_analysis(writer, merged_data) self.update_progress(100) # 步骤5: 显示结果 self.show_result(save_path) self.status_var.set(f"处理完成! 文件已保存至: {save_path}") except Exception as e: import traceback traceback.print_exc() messagebox.showerror("错误", f"处理过程中发生错误: {str(e)}") self.status_var.set(f"错误: {str(e)}") finally: self.execute_btn.config(state=tk.NORMAL) def collect_options(self): """收集用户选择的统计选项""" self.stats_options = [] for key, var in self.stats_vars.items(): if var.get(): self.stats_options.append(key) def merge_by_worksheet(self, writer): """工作页合并方式""" # 创建一个字典来存储所有工作表数据 all_sheets = {} for i, file_path in enumerate(self.file_paths): try: # 获取文件名（不带扩展名）作为工作表名 sheet_name = os.path.splitext(os.path.basename(file_path))[0] # 读取Excel文件 df = pd.read_excel(file_path) # 排除指定列 if self.exclude_columns: df = self.exclude_specified_columns(df) # 保存数据用于统计分析 all_sheets[sheet_name] = df # 写入新工作表 df.to_excel(writer, sheet_name=sheet_name, index=False) self.status_var.set(f"正在处理: {os.path.basename(file_path)}") progress = 20 + (i / len(self.file_paths)) * 20 self.update_progress(progress) except Exception as e: messagebox.showwarning("警告", f"处理文件 {file_path} 时出错: {str(e)}") continue # 返回第一个工作表的数据用于统计分析 return list(all_sheets.values())[0] if all_sheets else None def merge_by_block(self, writer): """区块合并方式""" merged_df = pd.DataFrame() for i, file_path in enumerate(self.file_paths): try: # 读取Excel文件 df = pd.read_excel(file_path) # 排除指定列 if self.exclude_columns: df = self.exclude_specified_columns(df) # 追加到合并的DataFrame merged_df = pd.concat([merged_df, df], ignore_index=True) self.status_var.set(f"正在合并: {os.path.basename(file_path)}") progress = 20 + (i / len(self.file_paths)) * 20 self.update_progress(progress) except Exception as e: messagebox.showwarning("警告", f"处理文件 {file_path} 时出错: {str(e)}") continue # 写入合并结果 merged_df.to_excel(writer, sheet_name="合并数据", index=False) return merged_df def merge_by_column(self, writer): """按列合并方式""" # 用于存储每个文件的DataFrame dfs = [] column_names = None # 首先读取所有文件 for i, file_path in enumerate(self.file_paths): try: df = pd.read_excel(file_path) # 如果是第一个文件，保存列名 if column_names is None: column_names = df.columns.tolist() # 确保所有文件具有相同的列 elif df.columns.tolist() != column_names: messagebox.showwarning("警告", f"文件 {os.path.basename(file_path)} 的列结构与其他文件不一致") continue # 排除指定列 if self.exclude_columns: df = self.exclude_specified_columns(df) dfs.append(df) self.status_var.set(f"正在读取: {os.path.basename(file_path)}") progress = 20 + (i / len(self.file_paths)) * 20 self.update_progress(progress) except Exception as e: messagebox.showwarning("警告", f"处理文件 {file_path} 时出错: {str(e)}") continue # 如果没有有效文件，直接返回 if not dfs: return None # 创建合并后的DataFrame merged_df = dfs[0].copy() # 对于每个后续文件，按列合并数据 for i in range(1, len(dfs)): for col in merged_df.columns: # 跳过排除的列 if self.column_to_index(col) in self.exclude_columns: continue # 数值列相加 if pd.api.types.is_numeric_dtype(merged_df[col]): merged_df[col] += dfs[i][col] # 字符串列去重合并 elif pd.api.types.is_string_dtype(merged_df[col]): merged_df[col] = merged_df[col].astype(str) + "、" + dfs[i][col].astype(str) # 去重处理 merged_df[col] = merged_df[col].apply( lambda x: "、".join(set(filter(None, x.split("、")))) if x else x ) # 写入合并结果 merged_df.to_excel(writer, sheet_name="合并数据", index=False) return merged_df def exclude_specified_columns(self, df): """排除指定列""" # 创建列索引映射 self.create_column_mapping(df) # 过滤列 columns_to_keep = [ col for col in df.columns if self.column_to_index(col) not in self.exclude_columns ] return df[columns_to_keep] def create_column_mapping(self, df): """创建列名到列字母的映射""" self.column_mapping = {} for i, col in enumerate(df.columns): self.column_mapping[col] = self.index_to_letter(i) def column_to_index(self, col_name): """将列名转换为Excel列字母表示""" # 如果已经是字母表示，直接返回 if isinstance(col_name, str) and re.match(r'^[A-Z]+$', col_name): return col_name # 尝试将列名转换为列索引 if hasattr(self, 'column_mapping') and col_name in self.column_mapping: return self.column_mapping[col_name] # 默认返回列名 return col_name def index_to_letter(self, index): """将数字索引转换为Excel列字母""" return get_column_letter(index + 1) # openpyxl的列索引从1开始 def perform_statistical_analysis(self, writer, df): """执行统计分析""" # 创建一个新的DataFrame来存储统计结果 stats_results = [] # 汇总统计 if "summary" in self.stats_options: summary_stats = df.describe().reset_index() summary_stats.columns = ["统计项"] + list(summary_stats.columns[1:]) stats_results.append(("汇总统计", summary_stats)) # 计数统计 if "count" in self.stats_options: count_stats = pd.DataFrame({ "列名": df.columns, "非空值数量": df.count().values, "唯一值数量": df.nunique().values }) stats_results.append(("计数统计", count_stats)) # 平均值计算 if "avg" in self.stats_options: numeric_cols = df.select_dtypes(include=[np.number]).columns if not numeric_cols.empty: avg_stats = pd.DataFrame({ "列名": numeric_cols, "平均值": [df[col].mean() for col in numeric_cols] }) stats_results.append(("平均值", avg_stats)) # 极值分析 if "minmax" in self.stats_options: numeric_cols = df.select_dtypes(include=[np.number]).columns if not numeric_cols.empty: minmax_stats = pd.DataFrame({ "列名": numeric_cols, "最小值": [df[col].min() for col in numeric_cols], "最大值": [df[col].max() for col in numeric_cols], "极差": [df[col].max() - df[col].min() for col in numeric_cols] }) stats_results.append(("极值分析", minmax_stats)) # 将统计结果写入Excel for sheet_name, stats_df in stats_results: stats_df.to_excel(writer, sheet_name=sheet_name, index=False) def show_result(self, file_path): """显示结果对话框""" result_window = tk.Toplevel(self) result_window.title("合并与分析结果") result_window.geometry("600x400") # 创建文本显示区域 text_frame = ttk.Frame(result_window) text_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=10) text_area = scrolledtext.ScrolledText( text_frame, wrap=tk.WORD, font=("Consolas", 10) ) text_area.pack(fill=tk.BOTH, expand=True) # 添加结果信息 text_area.insert(tk.END, f"文件合并与分析已完成!\n\n") text_area.insert(tk.END, f"结果文件: {file_path}\n") text_area.insert(tk.END, f"合并方式: {self.get_merge_method_name()}\n") text_area.insert(tk.END, f"排除列: {', '.join(self.exclude_columns) if self.exclude_columns else '无'}\n") text_area.insert(tk.END, f"统计选项: {', '.join(self.get_stats_names()) if self.stats_options else '无'}\n\n") # 添加操作说明 text_area.insert(tk.END, "操作说明:\n") text_area.insert(tk.END, "1. 您可以在Excel中打开生成的文件查看详细结果\n") text_area.insert(tk.END, "2. 点击下方按钮可以打开文件所在目录或直接打开文件\n") text_area.config(state=tk.DISABLED) # 设置为只读 # 添加按钮 btn_frame = ttk.Frame(result_window) btn_frame.pack(fill=tk.X, padx=10, pady=10) ttk.Button( btn_frame, text="打开文件目录", command=lambda: self.open_file_directory(file_path) ).pack(side=tk.LEFT, padx=10) ttk.Button( btn_frame, text="打开文件", command=lambda: self.open_file(file_path) ).pack(side=tk.LEFT, padx=10) ttk.Button( btn_frame, text="关闭", command=result_window.destroy ).pack(side=tk.RIGHT, padx=10) def get_merge_method_name(self): """获取合并方式的友好名称""" names = { "worksheet": "工作页合并", "block": "区块合并", "column": "按列合并" } return names.get(self.merge_method, "未知方式") def get_stats_names(self): """获取统计选项的友好名称""" names = { "summary": "汇总统计", "count": "计数统计", "avg": "平均值计算", "minmax": "极值分析" } return [names.get(opt, opt) for opt in self.stats_options] def open_file_directory(self, file_path): """打开文件所在目录""" try: directory = os.path.dirname(file_path) if os.name == 'nt': # Windows os.startfile(directory) elif os.name == 'posix': # macOS, Linux import subprocess subprocess.Popen(['xdg-open', directory]) except Exception as e: messagebox.showerror("错误", f"无法打开目录: {str(e)}") def open_file(self, file_path): """打开文件""" try: if os.name == 'nt': # Windows os.startfile(file_path) elif os.name == 'posix': # macOS, Linux import subprocess subprocess.Popen(['xdg-open', file_path]) except Exception as e: messagebox.showerror("错误", f"无法打开文件: {str(e)}") def update_progress(self, value): """更新进度条""" if value > 100: value = 100 self.progress_var.set(value) self.master.update_idletasks() class MainApplication(tk.Tk): """主应用程序""" def __init__(self): super().__init__() self.title("Excel文件处理工具") self.geometry("800x600") self.minsize(700, 500) # 创建菜单栏 self.create_menu() # 创建主容器 self.container = ttk.Frame(self) self.container.pack(fill=tk.BOTH, expand=True, padx=10, pady=10) self.container.grid_rowconfigure(0, weight=1) self.container.grid_columnconfigure(0, weight=1) # 初始显示文件合并统计分析界面 self.show_frame("文件合并统计分析") def create_menu(self): """创建菜单栏""" menu_bar = tk.Menu(self) # Excel文件处理菜单 excel_menu = tk.Menu(menu_bar, tearoff=0) excel_menu.add_command( label="宏模块加载分析", command=lambda: self.show_frame("宏模块加载分析") ) excel_menu.add_command( label="文件合并统计分析", command=lambda: self.show_frame("文件合并统计分析") ) # 添加到菜单栏 menu_bar.add_cascade(label="Excel文件处理", menu=excel_menu) self.config(menu=menu_bar) def show_frame(self, frame_name): """显示指定的功能界面""" # 清除当前容器内容 for widget in self.container.winfo_children(): widget.destroy() # 创建新界面 if frame_name == "宏模块加载分析": # 这里应该加载您的VBAExecutorApp frame = ttk.Label(self.container, text="宏模块加载分析界面", font=("Arial", 16)) frame.pack(fill=tk.BOTH, expand=True) elif frame_name == "文件合并统计分析": frame = EnhancedFileMergeApp(self.container) else: return frame.pack(fill=tk.BOTH, expand=True) if __name__ == "__main__": app = MainApplication() app.mainloop() ``` ## 关键修复和改进 ### 1. 文件路径处理 - 使用 `self.current_save_path` 保存文件路径 - 使用 `with pd.ExcelWriter(...)` 上下文管理器确保文件正确关闭 - 在统计分析前显式读取文件内容 ### 2. 列名映射优化 ```python def create_column_mapping(self, df): """创建列名到列字母的映射""" self.column_mapping = {} for i, col in enumerate(df.columns): self.column_mapping[col] = self.index_to_letter(i) ``` ### 3. 字符串处理增强 ```python merged_df[col] = merged_df[col].apply( lambda x: "、".join(set(filter(None, x.split("、")))) if x else x ) ``` ### 4. 错误处理增强 ```python except Exception as e: import traceback traceback.print_exc() messagebox.showerror("错误", f"处理过程中发生错误: {str(e)}") ``` ### 5. 空值检查 ```python if not numeric_cols.empty: # 执行数值计算 ``` 这些修复确保了程序能够正确处理Excel文件合并和统计分析，同时提供了更健壮的错误处理和用户体验。