#!/usr/bin/env python # -*- coding: utf-8 -*- """ 物料校验标签页 校验物料状态并匹配待删除物料。 支持 2 种校验模式： 1. database_full - 数据库全表校验 2. database_filtered - 数据库 ProductionID 过滤校验 """ import os import threading import tkinter as tk from tkinter import ttk, messagebox, filedialog, simpledialog from pathlib import Path from io import StringIO from contextlib import redirect_stdout from typing import List, Dict from gui.widgets import FileSelector, LogText from gui.config_manager import ConfigManager from gui.material_type_management_dialog import MaterialTypeManagementDialog from db.materials_to_be_deleted_dao import MaterialsTypeToBeDeletedDAO import pandas as pd class CheckboxTreeview(ttk.Treeview): """支持 checkbox 的 Treeview 组件 使用 Unicode 字符模拟 checkbox： - ☐ 未选中 - ☑ 选中 """ def __init__(self, parent, on_checkbox_change=None, **kwargs): """初始化 CheckboxTreeview Args: parent: 父容器 on_checkbox_change: checkbox 状态改变时的回调函数 **kwargs: 传递给 Treeview 的参数 """ super().__init__(parent, **kwargs) self.checkboxes = {} # item_id -> bool self.checkbox_column = "选择" self.on_checkbox_change = on_checkbox_change # checkbox 状态改变回调 # 绑定点击事件 self.bind("", self._on_click) def _on_click(self, event): """处理点击事件，切换 checkbox 状态""" # 获取点击位置对应的 item 和 column region = self.identify_region(event.x, event.y) if region == "cell": column = self.identify_column(event.x) item = self.identify_row(event.y) # 检查是否点击了 checkbox 列（第一列） if column == "#1" and item: # 切换 checkbox 状态 current_state = self.checkboxes.get(item, False) new_state = not current_state self.set_checked(item, new_state) # 通知父组件 checkbox 状态已改变 if self.on_checkbox_change: self.on_checkbox_change(item, new_state) return "break" # 阻止默认行为 def set_checked(self, item, checked: bool): """设置指定 item 的 checkbox 状态 Args: item: Treeview item ID checked: 是否选中 """ self.checkboxes[item] = checked # 更新显示 checkbox_char = "☑" if checked else "☐" values = list(self.item(item, "values")) if values: values[0] = checkbox_char self.item(item, values=values) def get_checked_items(self) -> list: """获取所有选中的 item Returns: List of item IDs """ return [item for item, checked in self.checkboxes.items() if checked] def check_all(self, checked: bool = True): """全选或取消全选 Args: checked: True 为全选，False 为取消全选 """ for item in self.get_children(): self.set_checked(item, checked) def insert(self, parent, index, values=None, **kwargs): """重写 insert 方法，初始化 checkbox 状态""" if values is None: values = [] # 确保第一个值是 checkbox if not values or values[0] not in ["☐", "☑"]: values = ["☐"] + list(values) item = super().insert(parent, index, values=values, **kwargs) # 初始化 checkbox 状态为未选中 checkbox_char = values[0] if values else "☐" self.checkboxes[item] = (checkbox_char == "☑") return item def delete(self, *items): """重写 delete 方法，清理 checkbox 状态""" for item in items: if item in self.checkboxes: del self.checkboxes[item] super().delete(*items) class MaterialValidationTab(ttk.Frame): """物料校验标签页""" def __init__(self, parent, config: ConfigManager, session_manager): """ 初始化物料校验标签页 Args: parent: 父容器 config: 配置管理器 session_manager: SessionManager 实例，用于权限控制 """ super().__init__(parent) self.config = config self.session_manager = session_manager self.validating = False self.validation_results = None self.material_records_cache = None # 缓存完整的物料记录 # 负责人筛选相关 self.managers: List[str] = [] # 可用负责人列表 self.manager_checkboxes: Dict[str, tk.BooleanVar] = {} # 负责人复选框状态 self.select_all_managers_var: tk.BooleanVar = tk.BooleanVar(value=True) # 全选复选框状态 self.previously_selected_managers: List[str] = [] # 保存筛选状态 self.create_widgets() # 稍后显示就绪消息 try: self.log_text.info("物料校验标签页已就绪") except: pass # 如果窗口还未完全就绪，忽略错误 def create_widgets(self): """创建界面组件""" # 主容器 main_container = ttk.Frame(self) main_container.pack(fill=tk.BOTH, expand=True, padx=10, pady=10) # 上部：控制面板 control_frame = ttk.Frame(main_container) control_frame.pack(fill=tk.X, pady=(0, 10)) # 负责人筛选区域 - 仅管理员可见 # PERMISSION CHECK: 非管理员用户不显示筛选区域 if self.session_manager.is_admin(): manager_filter_frame = ttk.LabelFrame( main_container, text="筛选（按负责人）", padding=10 ) manager_filter_frame.pack(fill=tk.X, pady=(0, 10)) else: # 普通用户：创建一个空容器，避免后续代码出错 manager_filter_frame = ttk.Frame(main_container) # 中部：结果表格 result_frame = ttk.LabelFrame(main_container, text="校验结果", padding=5) result_frame.pack(fill=tk.BOTH, expand=True, pady=(0, 10)) # 下部：日志输出 log_frame = ttk.LabelFrame(main_container, text="日志", padding=5) log_frame.pack(fill=tk.BOTH, expand=True) self._create_control_panel(control_frame) self._create_manager_filter_area(manager_filter_frame) self._create_result_table(result_frame) self._create_log_panel(log_frame) # 初始化文件选择器显示状态 self._on_source_mode_change() def _create_control_panel(self, parent): """创建控制面板""" # 数据来源选择 - 仅管理员可见 # PERMISSION CHECK: 非管理员用户不显示数据来源选项 if self.session_manager.is_admin(): source_group = ttk.LabelFrame(parent, text="数据来源", padding=10) source_group.pack(fill=tk.X, pady=5) self.source_mode = tk.StringVar(value="database_full") # 数据库模式 ttk.Radiobutton( source_group, text="数据库 - 全表校验", variable=self.source_mode, value="database_full", command=self._on_source_mode_change, ).grid(row=0, column=0, sticky="w", padx=5) ttk.Radiobutton( source_group, text="数据库 - ProductionID 过滤", variable=self.source_mode, value="database_filtered", command=self._on_source_mode_change, ).grid(row=0, column=1, sticky="w", padx=5) else: # 普通用户：默认使用 database_filtered 模式 self.source_mode = tk.StringVar(value="database_filtered") # 文件选择 file_group = ttk.LabelFrame(parent, text="文件选择", padding=10) file_group.pack(fill=tk.X, pady=5) # 数据库全表模式 - 无需输入文件 self.db_full_frame = ttk.Frame(file_group) ttk.Label( self.db_full_frame, text="数据库全表模式：将查询 DiscreteMaterialPlanData 表中的所有材料", foreground="gray" ).pack(anchor="w") # 数据库过滤模式 - 需要 ProductionID 文件 self.db_filtered_frame = ttk.Frame(file_group) self.db_filtered_production_id_selector = FileSelector( self.db_filtered_frame, label_text="ProductionID 文件:", file_type="file", file_types=[("文本文件", "*.txt"), ("所有文件", "*.*")], initial_dir="D:/python/playwrite/", ) self.db_filtered_production_id_selector.pack(fill=tk.X) # 输出文件（对于所有用户都可见） self.output_file_selector = FileSelector( file_group, label_text="输出文件:", file_type="file", file_types=[("Excel 文件", "*.xlsx"), ("所有文件", "*.*")], initial_dir=self.config.get("paths.data_dir", "data/"), ) self.output_file_selector.pack(fill=tk.X, pady=(10, 0)) # 设置默认输出 default_output = os.path.join( self.config.get("paths.data_dir", "data/"), self.config.get("paths.validation_output", "物料状态校验结果.xlsx"), ) self.output_file_selector.set(default_output) # 控制按钮 button_frame = ttk.Frame(parent) button_frame.pack(fill=tk.X, pady=10) self.start_button = ttk.Button( button_frame, text="开始校验", command=self.start_validation ) self.start_button.pack(side=tk.LEFT, padx=5) # 全选按钮 self.select_all_button = ttk.Button( button_frame, text="全选", command=lambda: self._select_all(check=True), state=tk.DISABLED ) self.select_all_button.pack(side=tk.LEFT, padx=5) # 取消全选按钮 self.deselect_all_button = ttk.Button( button_frame, text="取消全选", command=lambda: self._select_all(check=False), state=tk.DISABLED ) self.deselect_all_button.pack(side=tk.LEFT, padx=5) # 确认删除按钮 self.confirm_delete_button = ttk.Button( button_frame, text="确认删除", command=self.confirm_deletion, state=tk.DISABLED ) self.confirm_delete_button.pack(side=tk.LEFT, padx=5) self.export_button = ttk.Button( button_frame, text="导出结果", command=self.export_results, state=tk.DISABLED, ) self.export_button.pack(side=tk.LEFT, padx=5) # 类型管理按钮 self.type_management_button = ttk.Button( button_frame, text="类型管理", command=self.open_type_management ) self.type_management_button.pack(side=tk.LEFT, padx=5) def _create_result_table(self, parent): """创建结果表格""" # 新列定义：选择、材料名称、材料代码、规格、型号、负责人 columns = ("选择", "材料名称", "材料代码", "规格", "型号", "负责人") self.tree = CheckboxTreeview( parent, columns=columns, show="headings", height=10, on_checkbox_change=self._sync_checkbox_by_material_code ) # 设置列标题和宽度 self.tree.heading("选择", text="选择") self.tree.heading("材料名称", text="材料名称") self.tree.heading("材料代码", text="材料代码") self.tree.heading("规格", text="规格") self.tree.heading("型号", text="型号") self.tree.heading("负责人", text="负责人（双击编辑）") self.tree.column("选择", width=50, anchor="center") self.tree.column("材料名称", width=250) self.tree.column("材料代码", width=150) self.tree.column("规格", width=150) self.tree.column("型号", width=150) self.tree.column("负责人", width=150) # 添加滚动条 scrollbar_y = ttk.Scrollbar(parent, orient=tk.VERTICAL, command=self.tree.yview) scrollbar_x = ttk.Scrollbar( parent, orient=tk.HORIZONTAL, command=self.tree.xview ) self.tree.configure( yscrollcommand=scrollbar_y.set, xscrollcommand=scrollbar_x.set ) # 布局 self.tree.grid(row=0, column=0, sticky="nsew") scrollbar_y.grid(row=0, column=1, sticky="ns") scrollbar_x.grid(row=1, column=0, sticky="ew") parent.rowconfigure(0, weight=1) parent.columnconfigure(0, weight=1) # 绑定双击事件用于编辑负责人 self.tree.bind("", self._on_cell_double_click) def _create_log_panel(self, parent): """创建日志面板""" self.log_text = LogText(parent, height=8, readonly=True) self.log_text.pack(fill=tk.BOTH, expand=True) def _on_source_mode_change(self): """数据源模式切换""" # PERMISSION CHECK: 普通用户固定为 database_filtered 模式 if not self.session_manager.is_admin(): # 对于普通用户，pack 数据库过滤模式框架 self.db_full_frame.pack_forget() self.db_filtered_frame.pack(fill=tk.X, pady=(0, 5)) return mode = self.source_mode.get() # 隐藏所有文件选择框架 self.db_full_frame.pack_forget() self.db_filtered_frame.pack_forget() # 根据模式显示对应的文件选择器 if mode == "database_full": self.db_full_frame.pack(fill=tk.X, pady=(0, 5)) elif mode == "database_filtered": self.db_filtered_frame.pack(fill=tk.X, pady=(0, 5)) def _create_manager_filter_area(self, parent): """创建负责人筛选区域""" # PERMISSION CHECK: 非管理员用户不显示筛选区域 if not self.session_manager.is_admin(): return # 创建复选框容器 canvas_container = ttk.Frame(parent) canvas_container.pack(fill=tk.BOTH, expand=True) # Canvas和滚动条 self.filter_canvas = tk.Canvas(canvas_container, height=80) scrollbar = ttk.Scrollbar(canvas_container, orient="vertical", command=self.filter_canvas.yview) self.filter_frame = ttk.Frame(self.filter_canvas) self.filter_frame.bind( "", lambda e: self.filter_canvas.configure(scrollregion=self.filter_frame.bbox("all")) ) self.filter_canvas.create_window((0, 0), window=self.filter_frame, anchor="nw") self.filter_canvas.configure(yscrollcommand=scrollbar.set) # 布局 self.filter_canvas.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) scrollbar.pack(side=tk.RIGHT, fill=tk.Y) # 鼠标滚轮支持 def _on_mousewheel(event): self.filter_canvas.yview_scroll(int(-1*(event.delta/120)), "units") self.filter_canvas.bind("", _on_mousewheel) # 快捷按钮 button_frame = ttk.Frame(parent) button_frame.pack(fill=tk.X, pady=(5, 0)) ttk.Button(button_frame, text="全选", command=self._select_all_managers_button).pack(side=tk.LEFT, padx=5) ttk.Button(button_frame, text="取消全选", command=self._deselect_all_managers_button).pack(side=tk.LEFT, padx=5) def _create_manager_checkboxes(self): """创建负责人复选框""" for widget in self.filter_frame.winfo_children(): widget.destroy() self.manager_checkboxes.clear() # PERMISSION CHECK: 非管理员用户只能看到自己的数据 if not self.session_manager.is_admin(): self.managers = [self.session_manager.get_username()] # 不显示复选框，直接显示提示信息 ttk.Label( self.filter_frame, text=f"仅显示您的数据（负责人：{self.session_manager.get_username()}）" ).pack(anchor="w") return # 管理员：从两个表获取负责人列表并合并去重 from db.materials_to_be_deleted_records_dao import MaterialsToBeDeletedDAO record_dao = MaterialsToBeDeletedDAO() record_managers = record_dao.get_managers() type_dao = MaterialsTypeToBeDeletedDAO() type_managers = type_dao.get_managers() # 合并两个表的负责人列表并去重 self.managers = sorted(list(set(record_managers + type_managers))) if not self.managers: ttk.Label(self.filter_frame, text="暂无负责人数据").pack(anchor="w") return items_per_column = 6 # 全选复选框 ttk.Checkbutton( self.filter_frame, text="全选", variable=self.select_all_managers_var, command=self._on_select_all_managers_toggle ).grid(row=0, column=0, sticky="w", padx=5, pady=2) # 负责人复选框 for idx, manager in enumerate(self.managers): var = tk.BooleanVar(value=True) self.manager_checkboxes[manager] = var row = (idx // items_per_column) + 1 col = idx % items_per_column ttk.Checkbutton( self.filter_frame, text=manager, variable=var, command=self._on_manager_checkbox_change ).grid(row=row, column=col, sticky="w", padx=5, pady=2) def _on_select_all_managers_toggle(self): """全选复选框状态改变""" select_all = self.select_all_managers_var.get() for var in self.manager_checkboxes.values(): var.set(select_all) self._apply_manager_filter() def _on_manager_checkbox_change(self): """单个负责人复选框状态改变""" all_selected = all(var.get() for var in self.manager_checkboxes.values()) self.select_all_managers_var.set(all_selected) self._apply_manager_filter() def _select_all_managers_button(self): """全选所有负责人""" self.select_all_managers_var.set(True) for var in self.manager_checkboxes.values(): var.set(True) self._apply_manager_filter() def _deselect_all_managers_button(self): """取消全选""" self.select_all_managers_var.set(False) for var in self.manager_checkboxes.values(): var.set(False) self._apply_manager_filter() def _get_selected_managers(self) -> List[str]: """获取选中的负责人列表""" return [ manager for manager, var in self.manager_checkboxes.items() if var.get() ] def _apply_manager_filter(self): """应用负责人筛选""" if not self.material_records_cache: return # PERMISSION CHECK: 非管理员用户强制筛选到当前用户 if not self.session_manager.is_admin(): selected_managers = [self.session_manager.get_username()] else: selected_managers = self._get_selected_managers() if not selected_managers: # 没有选中任何负责人，清空表格 self._refresh_filtered_results([]) self._update_log("未选择任何负责人，表格已清空", "WARNING") return # 筛选记录：包含选中负责人的记录 + 负责人为空的记录 filtered_records = [ record for record in self.material_records_cache if (record.manager_name in selected_managers or not record.manager_name or record.manager_name.strip() == "") ] # 统计空负责人的记录数 empty_manager_count = sum( 1 for r in filtered_records if not r.manager_name or r.manager_name.strip() == "" ) self._refresh_filtered_results(filtered_records) self._update_log( f"筛选结果：共 {len(filtered_records)} 条记录" f"（其中 {empty_manager_count} 条负责人为空，待编辑）", "INFO" ) def _refresh_filtered_results(self, filtered_records): """刷新表格显示筛选后的结果""" # 清空表格 for item in self.tree.get_children(): self.tree.delete(item) if not filtered_records: return # 从数据库获取已标记删除的记录 from db.materials_to_be_deleted_records_dao import MaterialsToBeDeletedDAO record_dao = MaterialsToBeDeletedDAO() # PERMISSION CHECK: 非管理员用户只获取自己的记录 if not self.session_manager.is_admin(): marked_records = record_dao.get_materials_by_manager(self.session_manager.get_username()) else: marked_records = record_dao.get_all_records() # Build dictionary: MaterialCode -> ManagerName marked_codes_dict = { r['MaterialCode']: r['ManagerName'] for r in marked_records if r.get('MaterialCode') and r.get('ManagerName') } # 填充筛选后的记录 for record in filtered_records: # MaterialValidationResult 对象使用属性访问 material_code = str(record.material_code) # 优先使用 MaterialsToBeDeleted 中的 ManagerName manager_name = marked_codes_dict.get(material_code) is_marked = manager_name is not None # 如果没有在 MaterialsToBeDeleted 中找到，使用记录中的负责人 if not manager_name: manager_name = str(record.manager_name if record.manager_name else "") # 准备显示值 checkbox = "☑" if is_marked else "☐" values = ( checkbox, str(record.material_name), material_code, str(record.specification if record.specification else ""), str(record.model if record.model else ""), manager_name, ) item = self.tree.insert("", tk.END, values=values) # 设置 checkbox 状态 if is_marked: self.tree.set_checked(item, True) # 更新按钮状态 if len(filtered_records) > 0: self.export_button.config(state=tk.NORMAL) def _on_cell_double_click(self, event): """处理单元格双击事件，编辑负责人""" # 获取点击位置 region = self.tree.identify_region(event.x, event.y) if region == "cell": column = self.tree.identify_column(event.x) item = self.tree.identify_row(event.y) # 检查是否点击了"负责人"列（第6列） if column == "#6" and item: values = self.tree.item(item, "values") current_value = values[5] if len(values) > 5 else "" # 弹出编辑对话框 new_value = simpledialog.askstring( "编辑负责人", f"请输入负责人姓名：", initialvalue=current_value ) if new_value is not None: # 用户没有取消 # 更新单元格值 new_values = list(values) new_values[5] = new_value self.tree.item(item, values=new_values) self.log_text.info(f"已更新负责人: {current_value} -> {new_value}") def _sync_checkbox_by_material_code(self, changed_item: str, new_state: bool): """同步相同材料代码的所有记录的选择状态 当用户点击某个记录的 checkbox 时，自动将所有具有相同材料代码的记录 的 checkbox 状态同步更新。 Args: changed_item: 被点击的 item ID new_state: 新的 checkbox 状态 """ # 获取被点击行的材料代码 values = self.tree.item(changed_item, "values") if not values or len(values) <= 2: return material_code = values[2] # 材料代码在第3列（索引2） # 同步所有具有相同材料代码的记录 synced_count = 0 for item in self.tree.get_children(): item_values = self.tree.item(item, "values") if item_values and len(item_values) > 2: if item_values[2] == material_code and item != changed_item: # 只更新状态不同的行，避免重复更新 current_state = self.tree.checkboxes.get(item, False) if current_state != new_state: self.tree.set_checked(item, new_state) synced_count += 1 if synced_count > 0: action = "选中" if new_state else "取消选中" self.log_text.info(f"已同步 {synced_count} 条相同材料代码的记录{action}") def _select_all(self, check: bool = True): """全选或取消全选 Args: check: True 为全选，False 为取消全选 """ self.tree.check_all(check) action = "全选" if check else "取消全选" self.log_text.info(f"{action}完成") def start_validation(self): """开始校验""" # 验证输入 output_file = self.output_file_selector.get() if not output_file: messagebox.showerror("错误", "请指定输出文件路径") return mode = self.source_mode.get() input_file = None production_id_file = None # 根据模式验证输入文件 if mode == "database_full": # 无需输入文件 pass elif mode == "database_filtered": production_id_file = self.db_filtered_production_id_selector.get() if not production_id_file: messagebox.showerror("错误", "请选择 ProductionID 文件") return if not os.path.exists(production_id_file): messagebox.showerror("错误", f"文件不存在：{production_id_file}") return else: messagebox.showerror("错误", f"未知的校验模式: {mode}") return # 确保输出目录存在 output_dir = os.path.dirname(output_file) if output_dir and not os.path.exists(output_dir): os.makedirs(output_dir, exist_ok=True) # 更新 UI 状态 self.validating = True self.start_button.config(state=tk.DISABLED) self.confirm_delete_button.config(state=tk.DISABLED) self.select_all_button.config(state=tk.DISABLED) self.deselect_all_button.config(state=tk.DISABLED) self.log_text.clear() # 保存当前筛选状态 self.previously_selected_managers = self._get_selected_managers() self.log_text.info(f"开始物料校验（模式: {mode}）...") # 清空结果表格 for item in self.tree.get_children(): self.tree.delete(item) # 在后台线程中执行校验 validation_thread = threading.Thread( target=self._validation_worker_enhanced, args=(mode, input_file, production_id_file, output_file), daemon=True, ) validation_thread.start() def _validation_worker_enhanced( self, mode: str, input_file: str, production_id_file: str, output_file: str ): """增强的校验工作线程（使用完整记录模式）""" try: # 导入校验器 from utils.material_status_validator import MaterialStatusValidator # 创建校验器实例 validator = MaterialStatusValidator( username=self.config.get("erp.username"), password=self.config.get("erp.password"), headless=self.config.get("erp.headless", True), verbose=True, ) # 捕获 stdout 输出 captured_output = StringIO() # 根据模式执行增强校验 with redirect_stdout(captured_output): if mode == "database_full": result_file, results = validator.validate_from_database_enhanced( full_table=True, output_file=output_file ) elif mode == "database_filtered": result_file, results = validator.validate_from_database_enhanced( production_id_file=production_id_file, full_table=False, output_file=output_file ) else: raise ValueError(f"未知的校验模式: {mode}") # 缓存结果记录 self.material_records_cache = results # 获取捕获的输出并显示到日志 output_text = captured_output.getvalue() if output_text: for line in output_text.split("\n"): if line.strip(): self._update_log(line, "INFO") if result_file and results: self._update_log(f"校验完成，结果已保存到：{output_file}", "SUCCESS") # 加载结果显示（带删除状态） self._load_results_with_deletion_status(result_file) else: self._update_log("校验失败", "ERROR") except Exception as e: self._update_log(f"校验过程中发生错误：{str(e)}", "ERROR") import traceback self._update_log(traceback.format_exc(), "ERROR") finally: # 更新 UI 状态 self.after(0, self._validation_complete) def _validation_complete(self): """校验完成后的 UI 更新""" self.validating = False self.start_button.config(state=tk.NORMAL) self.select_all_button.config(state=tk.NORMAL) self.deselect_all_button.config(state=tk.NORMAL) self.confirm_delete_button.config(state=tk.NORMAL) # 恢复筛选状态 if hasattr(self, 'previously_selected_managers') and self.previously_selected_managers: self._restore_manager_filter_state(self.previously_selected_managers) def _load_results_with_deletion_status(self, file_path: str): """加载结果并设置 checkbox 选中状态""" try: df = pd.read_excel(file_path) # 在主线程中更新表格 def update_table(): # 从数据库获取已标记删除的 MaterialCode -> ManagerName 映射 from db.materials_to_be_deleted_records_dao import MaterialsToBeDeletedDAO record_dao = MaterialsToBeDeletedDAO() # PERMISSION CHECK: 非管理员用户只获取自己的记录 if not self.session_manager.is_admin(): marked_records = record_dao.get_materials_by_manager(self.session_manager.get_username()) else: marked_records = record_dao.get_all_records() # Build dictionary: MaterialCode -> ManagerName marked_codes_dict = { r['MaterialCode']: r['ManagerName'] for r in marked_records if r.get('MaterialCode') and r.get('ManagerName') } for _, row in df.iterrows(): material_code = str(row.get("材料代码", "")) # 优先使用 MaterialsToBeDeleted 中的 ManagerName manager_name = marked_codes_dict.get(material_code) is_marked = manager_name is not None # 如果没有在 MaterialsToBeDeleted 中找到，使用 Excel 中的负责人 if not manager_name: manager_name = str(row.get("负责人", "")) # 准备显示值 checkbox = "☑" if is_marked else "☐" values = ( checkbox, str(row.get("材料名称", "")), material_code, str(row.get("规格", "")), str(row.get("型号", "")), manager_name, ) item = self.tree.insert("", tk.END, values=values) # 设置 checkbox 状态 if is_marked: self.tree.set_checked(item, True) if len(df) > 0: self.export_button.config(state=tk.NORMAL) self._update_log(f"已加载 {len(df)} 条结果（其中 {len(marked_codes_dict)} 条已标记删除）", "INFO") # 初始化负责人筛选器 self._initialize_manager_filter() self.after(0, update_table) except Exception as e: self._update_log(f"加载结果失败：{str(e)}", "ERROR") def _initialize_manager_filter(self): """加载结果后初始化负责人筛选器""" try: self._create_manager_checkboxes() self._apply_manager_filter() except Exception as e: self._update_log(f"初始化筛选器失败：{str(e)}", "ERROR") def _restore_manager_filter_state(self, selected_managers: List[str]): """恢复负责人筛选状态""" try: for manager in selected_managers: if manager in self.manager_checkboxes: self.manager_checkboxes[manager].set(True) self._apply_manager_filter() except Exception as e: self._update_log(f"恢复筛选状态失败：{str(e)}", "ERROR") def confirm_deletion(self): """确认删除主流程 处理所有行： - 已勾选的行：写入/更新到数据库（负责人不能为空） - 未勾选的行：从数据库中删除 """ if not self.tree.get_children(): messagebox.showwarning("警告", "没有可处理的数据") return # 收集所有数据 to_upsert = [] # 需要写入/更新的记录 to_delete = [] # 需要删除的记录 missing_manager = [] # 缺少负责人的记录 for item in self.tree.get_children(): values = self.tree.item(item, "values") is_checked = self.tree.checkboxes.get(item, False) material_code = values[2] if len(values) > 2 else "" manager_name = values[5] if len(values) > 5 else "" # 跳过材料代码为空的行 if not material_code or not material_code.strip(): continue material_code = material_code.strip() if is_checked: # 已勾选：需要写入/更新 if not manager_name or not manager_name.strip(): missing_manager.append(material_code) else: to_upsert.append({ "material_code": material_code, "manager_name": manager_name.strip() }) else: # 未勾选：需要删除 to_delete.append(material_code) # 验证：已勾选的记录必须有负责人 if missing_manager: msg = f"以下已勾选的记录缺少负责人信息，无法保存：\n\n" + \ "\n".join(missing_manager[:10]) if len(missing_manager) > 10: msg += f"\n... 共 {len(missing_manager)} 条" messagebox.showwarning("警告", msg) return # 构建确认消息 confirm_parts = [] if to_upsert: confirm_parts.append(f"写入/更新 {len(to_upsert)} 条记录") if to_delete: confirm_parts.append(f"删除 {len(to_delete)} 条记录") if not to_upsert and not to_delete: messagebox.showinfo("提示", "没有需要处理的记录") return confirm_msg = "确认以下操作吗？\n\n" + "\n".join(confirm_parts) if not messagebox.askyesno("确认操作", confirm_msg): return # 在后台线程中执行 self._update_log(f"开始处理：写入/更新 {len(to_upsert)} 条，删除 {len(to_delete)} 条...", "INFO") deletion_thread = threading.Thread( target=self._execute_sync_in_background, args=(to_upsert, to_delete), daemon=True, ) deletion_thread.start() def _execute_sync_in_background(self, to_upsert: list, to_delete: list): """后台线程执行同步操作 Args: to_upsert: 需要写入/更新的记录列表 to_delete: 需要删除的材料代码列表 """ try: from db.materials_to_be_deleted_records_dao import MaterialsToBeDeletedDAO dao = MaterialsToBeDeletedDAO() # 执行写入/更新操作 upsert_stats = {'total': 0, 'success': 0, 'failed': 0} if to_upsert: upsert_stats = dao.upsert_batch(to_upsert) # 执行删除操作 delete_count = 0 if to_delete: delete_count = dao.delete_by_material_codes(to_delete) # 组合统计信息 combined_stats = { 'upsert_success': upsert_stats['success'], 'upsert_failed': upsert_stats['failed'], 'deleted': delete_count } self.after(0, lambda: self._sync_complete(combined_stats)) except Exception as e: self.after(0, lambda: self._update_log(f"同步操作失败：{str(e)}", "ERROR")) def _sync_complete(self, stats: dict): """同步完成回调，更新 UI Args: stats: 包含 upsert_success, upsert_failed, deleted 的字典 """ upsert_success = stats.get('upsert_success', 0) upsert_failed = stats.get('upsert_failed', 0) deleted = stats.get('deleted', 0) msg_parts = [] if upsert_success > 0: msg_parts.append(f"写入/更新成功：{upsert_success} 条") if upsert_failed > 0: msg_parts.append(f"写入/更新失败：{upsert_failed} 条") if deleted > 0: msg_parts.append(f"删除成功：{deleted} 条") msg = "同步操作完成！\n\n" + "\n".join(msg_parts) if upsert_failed > 0: messagebox.showwarning("完成（部分失败）", msg) self._update_log(msg, "WARNING") else: messagebox.showinfo("成功", msg) self._update_log(msg, "SUCCESS") # 刷新筛选器和结果（可能添加了新的负责人） self._initialize_manager_filter() def _update_log(self, message: str, level: str = "INFO"): """线程安全的日志更新""" def update(): if level == "INFO": self.log_text.info(message) elif level == "SUCCESS": self.log_text.success(message) elif level == "WARNING": self.log_text.warning(message) elif level == "ERROR": self.log_text.error(message) self.after(0, update) def export_results(self): """导出结果到 Excel""" output_file = self.output_file_selector.get() if not output_file: output_file = filedialog.asksaveasfilename( title="保存结果", defaultextension=".xlsx", filetypes=[("Excel 文件", "*.xlsx"), ("所有文件", "*.*")], ) if not output_file: return try: # 收集表格数据 data = [] for item in self.tree.get_children(): values = self.tree.item(item)["values"] # 转换 checkbox 字符为布尔值 checkbox = values[0] if values else "☐" is_checked = checkbox == "☑" data.append([ "是" if is_checked else "否", # 选择状态 values[1] if len(values) > 1 else "", # 材料名称 values[2] if len(values) > 2 else "", # 材料代码 values[3] if len(values) > 3 else "", # 规格 values[4] if len(values) > 4 else "", # 型号 values[5] if len(values) > 5 else "", # 负责人 ]) if not data: messagebox.showwarning("警告", "没有数据可导出") return # 创建 DataFrame 并保存 df = pd.DataFrame( data, columns=["选择", "材料名称", "材料代码", "规格", "型号", "负责人"] ) df.to_excel(output_file, index=False) messagebox.showinfo("成功", f"结果已导出到：{output_file}") self.log_text.success(f"结果已导出到：{output_file}") except Exception as e: messagebox.showerror("错误", f"导出失败：{str(e)}") self.log_text.error(f"导出失败：{str(e)}") def open_type_management(self): """打开类型管理窗口""" dialog = MaterialTypeManagementDialog(self, self.session_manager, title="物料类型管理")