#!/usr/bin/env python # -*- coding: utf-8 -*- """ 物料校验标签页 校验物料状态并匹配待删除物料。 支持 2 种校验模式： 1. database_full - 数据库全表校验 2. database_filtered - 数据库 ProductionID 过滤校验 """ import os import threading import logging 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, GuiTextHandler from gui.config_manager import ConfigManager from gui.log_config import setup_gui_logging, get_logger 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.sort_column = None # 当前排序列的列标识符 self.sort_direction = None # 'asc', 'desc', 或 None self.sortable_columns = ["选择", "材料名称"] # 可排序的列白名单 self.original_headings = {} # 存储原始列标题文本（不含箭头） # 存储原始列标题（延迟执行以确保标题已设置） self.after(100, self._store_original_headings) # 绑定点击事件 self.bind("", self._on_click) # 绑定表头点击事件 self.bind("", self._on_heading_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) def _store_original_headings(self): """存储原始列标题文本（不含箭头）""" for col in self['columns']: self.original_headings[col] = self.heading(col, 'text') def _get_column_id_from_column_index(self, column_index): """将列索引 ('#1', '#2') 转换为列标识符 Args: column_index: 列索引字符串，如 '#1', '#2' Returns: 列标识符，如 '选择', '材料名称' """ index = int(column_index[1:]) - 1 columns = self['columns'] if 0 <= index < len(columns): return columns[index] return None def _on_heading_click(self, event): """处理表头点击事件，触发排序""" region = self.identify_region(event.x, event.y) if region == "heading": column = self.identify_column(event.x) column_id = self._get_column_id_from_column_index(column) # 仅对可排序列进行排序 if column_id in self.sortable_columns: self._toggle_sort(column_id) def _toggle_sort(self, column_id): """切换指定列的排序状态 Args: column_id: 列标识符（如 '选择', '材料名称'） """ # 确定新的排序方向 if self.sort_column == column_id: # 同一列：asc -> desc -> None if self.sort_direction == 'asc': new_direction = 'desc' elif self.sort_direction == 'desc': new_direction = None else: new_direction = 'asc' else: # 不同列：从升序开始 new_direction = 'asc' # 应用排序 if new_direction: self._sort_by_column(column_id, new_direction) self.sort_column = column_id self.sort_direction = new_direction else: # 清除排序状态 self.sort_column = None self.sort_direction = None # 更新表头显示 self._update_heading_display() def _sort_by_column(self, column_id, direction): """按指定列和方向排序 Args: column_id: 列标识符 direction: 'asc' 或 'desc' """ # 收集所有项目及其数据和复选框状态 items_data = [] for item in self.get_children(): values = self.item(item, "values") checkbox_state = self.checkboxes.get(item, False) items_data.append({ 'item_id': item, 'values': values, 'checked': checkbox_state }) # 根据列和方向排序 if column_id == "选择": # 按复选框状态排序（选中在前，未选中在后） items_data.sort(key=lambda x: x['checked'], reverse=(direction == 'desc')) elif column_id == "材料名称": # 按材料名称排序 items_data.sort( key=lambda x: str(x['values'][1]) if len(x['values']) > 1 else "", reverse=(direction == 'desc') ) # 重新排列项目顺序（使用 detach 和 move 保留项目ID和状态） for item_data in items_data: self.move(item_data['item_id'], '', 'end') def _update_heading_display(self): """更新列标题显示（添加/移除排序箭头）""" for col in self['columns']: original = self.original_headings.get(col, col) if col == self.sort_column: # 添加排序箭头 arrow = " ↑" if self.sort_direction == 'asc' else " ↓" self.heading(col, text=original + arrow) else: # 移除箭头，显示原始标题 self.heading(col, text=original) class MaterialValidationTab(ttk.Frame): """物料校验标签页""" def __init__(self, parent, config: ConfigManager, session_manager, main_window=None): """ 初始化物料校验标签页 Args: parent: 父容器 config: 配置管理器 session_manager: SessionManager 实例，用于权限控制 main_window: 主窗口引用，用于获取共享的 Production ID """ super().__init__(parent) self.config = config self.session_manager = session_manager self.main_window = main_window self.validating = False self.validation_results = None self.material_records_cache = None # 缓存完整的物料记录 self.shared_production_ids = [] # 共享的 Production ID 列表 # 负责人筛选相关 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.logger = get_logger(__name__) self._gui_handler = None # 将在 _create_log_panel 中设置 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") # Production ID 数据源（仅 Admin） if self.session_manager.is_admin(): source_frame = ttk.Frame(parent) source_frame.pack(fill=tk.X, pady=5) # Production ID 数据源选择 self.production_id_source_var = tk.StringVar(value="shared") ttk.Radiobutton( source_frame, text="使用文件:", variable=self.production_id_source_var, value="file", command=self._on_production_id_source_changed ).pack(side=tk.LEFT) ttk.Radiobutton( source_frame, text="使用数据提取页面的 Production ID:", variable=self.production_id_source_var, value="shared", command=self._on_production_id_source_changed ).pack(side=tk.LEFT, padx=10) # 文件选择器 self.db_filtered_production_id_selector = FileSelector( parent, label_text="", file_type="file", file_types=[("文本文件", "*.txt"), ("所有文件", "*.*")], initial_dir="D:/python/playwrite/", ) self.db_filtered_production_id_selector.pack(fill=tk.X) # 共享 Production ID 提示标签 self.shared_ids_info_label = ttk.Label( parent, text="", foreground="blue" ) # 初始时不显示 # 控制按钮 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) # 设置 GUI 日志处理器，将 logging 输出桥接到 LogText 组件 self._gui_handler = GuiTextHandler(self.log_text) self._gui_handler.setFormatter(logging.Formatter( '%(asctime)s [%(levelname)s] %(message)s', datefmt='%Y-%m-%d %H:%M:%S' )) self.logger.addHandler(self._gui_handler) def _on_source_mode_change(self): """数据源模式切换""" # PERMISSION CHECK: 仅管理员用户可以选择数据源模式 if not self.session_manager.is_admin(): return mode = self.source_mode.get() # 根据模式显示 Production ID 数据源选择（仅 Admin） if mode == "database_full": # 全表模式：隐藏 Production ID 选择区域 if hasattr(self, 'source_frame'): self.source_frame.pack_forget() elif mode == "database_filtered": # 过滤模式：显示 Production ID 选择区域 if hasattr(self, 'source_frame'): self.source_frame.pack(fill=tk.X, pady=5) def _on_production_id_source_changed(self): """Production ID 数据源切换回调（仅 Admin 模式）""" # User 模式下没有文件选择器，直接返回 if not self.session_manager.is_admin(): return source = self.production_id_source_var.get() if source == "shared": # 使用共享 Production ID self.db_filtered_production_id_selector.pack_forget() self.shared_ids_info_label.pack(anchor="w", pady=(0, 5)) else: # 使用文件 self.shared_ids_info_label.pack_forget() self.db_filtered_production_id_selector.pack(fill=tk.X) def on_production_ids_updated(self, production_ids: list): """当数据提取页面的 Production ID 更新时调用""" self.shared_production_ids = production_ids is_user_only = not self.session_manager.is_admin() # User 模式：静默更新，不显示任何提示 if is_user_only: return # Admin 模式：显示提示信息 if production_ids: count = len(production_ids) preview = ", ".join(production_ids[:3]) if count > 3: preview += f" ... (共 {count} 个)" self.shared_ids_info_label.config(text=f"📋 {preview}") # 显示提示标签 if hasattr(self, 'shared_ids_label') and not self.shared_ids_label.winfo_ismapped(): self.shared_ids_label.pack(anchor="w", pady=(0, 5)) else: self.shared_ids_info_label.config(text="") # 隐藏提示标签 if hasattr(self, 'shared_ids_label'): if self.shared_ids_label.winfo_ismapped(): self.shared_ids_label.pack_forget() def reload_config(self): """配置更新后重新加载""" # 从主窗口获取最新的 Production ID if self.main_window: production_ids = self.main_window.get_shared_production_ids() self.on_production_ids_updated(production_ids) def _create_manager_filter_area(self, parent): """创建负责人筛选区域""" # PERMISSION CHECK: 非管理员用户不显示筛选区域 if not self.session_manager.is_admin(): # 创建空的 filter_frame 以避免后续代码访问属性时报错 self.filter_frame = ttk.Frame(parent) self.filter_canvas = None 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]: """获取选中的负责人列表 PERMISSION CHECK: 非管理员用户返回当前用户名 """ # PERMISSION CHECK: 非管理员用户直接返回当前用户名 if not self.session_manager.is_admin(): return [self.session_manager.get_username()] # 管理员：从复选框获取选中的负责人 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 = os.path.join( self.config.get("paths.data_dir", "data/"), self.config.get("paths.validation_output", "物料状态校验结果.xlsx"), ) if not output_file: messagebox.showerror("错误", "请先在设置页面配置输出文件路径") return mode = self.source_mode.get() input_file = None production_id_file = None production_ids_list = None # 新增：用于传递共享的 Production ID 列表 is_user_only = not self.session_manager.is_admin() # 根据模式验证输入文件 if mode == "database_full": # 无需输入文件 pass elif mode == "database_filtered": if is_user_only: # User 模式：自动使用共享的 Production ID production_ids_list = self.shared_production_ids if not production_ids_list: messagebox.showerror("错误", "没有可用的共享 Production ID\n请在数据提取页面输入 Production ID 后再试") return self.log_text.info(f"使用数据提取页面的 Production ID（{len(production_ids_list)} 个）") else: # Admin 模式：检查数据源选择 source = self.production_id_source_var.get() if source == "shared": # 使用共享的 Production ID production_ids_list = self.shared_production_ids if not production_ids_list: messagebox.showerror("错误", "没有可用的共享 Production ID\n请在数据提取页面输入 Production ID 后再试") return self.log_text.info(f"使用共享的 Production ID（{len(production_ids_list)} 个）") else: # 使用文件 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, production_ids_list), daemon=True, ) validation_thread.start() def _validation_worker_enhanced( self, mode: str, input_file: str, production_id_file: str, output_file: str, production_ids_list: list = None ): """增强的校验工作线程（使用完整记录模式）""" import tempfile temp_production_id_file = None try: # 如果提供了共享的 Production ID 列表，创建临时文件 if production_ids_list: with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False, encoding='utf-8') as f: temp_production_id_file = f.name f.write('\n'.join(production_ids_list)) production_id_file = temp_production_id_file self._update_log(f"使用共享的 Production ID 列表（{len(production_ids_list)} 个）", "INFO") # 导入校验器 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 and len(results) > 0: self._update_log(f"校验完成，结果已保存到：{output_file}", "SUCCESS") # 加载结果显示（带删除状态） self._load_results_with_deletion_status(result_file) elif results is not None and len(results) == 0: # 已经在 validator 中输出详细错误信息，这里只做简单提示 self._update_log("校验失败：未找到物料记录，请查看上方日志了解详细原因", "ERROR") 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: # 清理临时文件 if temp_production_id_file and os.path.exists(temp_production_id_file): try: os.unlink(temp_production_id_file) except: pass # 更新 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]): """恢复负责人筛选状态 PERMISSION CHECK: 非管理员用户直接应用筛选（不需要操作复选框） """ try: # PERMISSION CHECK: 非管理员用户直接应用筛选 if not self.session_manager.is_admin(): self._apply_manager_filter() return # 管理员：恢复复选框状态 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"): """ 线程安全的日志更新（兼容接口） 通过统一的 logging 系统输出日志，自动同时输出到控制台和 GUI。 Args: message: 日志消息 level: 日志级别 (INFO, SUCCESS, WARNING, ERROR, DEBUG) """ # 将自定义级别映射到 logging 级别 level_upper = level.upper() if level_upper == "SUCCESS": # SUCCESS 映射到 INFO，但在 UI 中仍显示为 SUCCESS self.logger.info(message) else: # 其他级别直接映射 log_level = getattr(logging, level_upper, logging.INFO) self.logger.log(log_level, message) def export_results(self): """导出结果到 Excel""" # 从配置获取输出文件路径 data_dir = self.config.get("paths.data_dir", "data/") validation_filename = self.config.get("paths.validation_output", "物料状态校验结果.xlsx") output_file = os.path.join(data_dir, validation_filename) # 如果配置的文件不存在，提示用户选择位置 if not os.path.exists(data_dir): output_dir = filedialog.askdirectory( title="选择输出目录", initialdir=data_dir ) if output_dir: self.config.set("paths.data_dir", output_dir) # 重新读取配置并构建输出文件路径 data_dir = output_dir output_file = os.path.join(data_dir, validation_filename) else: return # 如果文件已存在，提示用户覆盖 if os.path.exists(output_file): if not messagebox.askyesno("确认覆盖", f"文件已存在：{output_file}\n是否覆盖？"): 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="物料类型管理")