#!/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.base_tab import BaseTab from gui.widgets import FileSelector, LogText, GuiTextHandler, DeleteProgressWindow, CheckboxTreeview 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 import tempfile class MaterialValidationTab(BaseTab): """物料校验标签页""" def __init__(self, parent, config: ConfigManager, session_manager, main_window=None): """ 初始化物料校验标签页 Args: parent: 父容器 config: 配置管理器 session_manager: SessionManager 实例，用于权限控制 main_window: 主窗口引用，用于获取共享的 Production ID """ super().__init__(parent, config, main_window) self.session_manager = session_manager self.validating = False self.validation_results = None self.material_records_cache = None # 缓存完整的物料记录 self.shared_production_ids = [] # 共享的 Production ID 列表 self.hidden_items = set() # Track hidden item IDs for User mode # 负责人筛选相关 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 Exception as e: self.logger.debug(f"初始化日志消息失败: {e}") 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, 10, 10, 10) # 左、上、右、下 - 底部padding减小 ) 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=1, column=0, sticky="w", padx=5) # 右侧：Production ID 数据源选择区域 self.source_frame = ttk.Frame(source_group) self.source_frame.grid(row=0, column=1, rowspan=2, sticky="nsew", padx=(20, 0)) # Production ID 数据源选择 self.production_id_source_var = tk.StringVar(value="shared") ttk.Radiobutton( self.source_frame, text="使用文件:", variable=self.production_id_source_var, value="file", command=self._on_production_id_source_changed ).pack(side=tk.LEFT) ttk.Radiobutton( self.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) # 文件选择器（放在 source_group 下方，当选择文件时显示） self.db_filtered_production_id_selector = FileSelector( source_group, label_text="", file_type="file", file_types=[("文本文件", "*.txt"), ("所有文件", "*.*")], initial_dir="D:/python/playwrite/", ) # 共享 Production ID 提示标签 self.shared_ids_info_label = ttk.Label( source_group, text="", foreground="blue" ) # 初始时不显示 else: # 普通用户：默认使用 database_filtered 模式 self.source_mode = tk.StringVar(value="database_filtered") # 控制按钮 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) # 全选、取消全选、确认删除、执行删除按钮 - 仅管理员显示在顶部控制面板 if self.session_manager.is_admin(): # 全选按钮 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.execute_delete_button = ttk.Button( button_frame, text="执行删除", command=self.start_delete_execution, state=tk.DISABLED ) self.execute_delete_button.pack(side=tk.LEFT, padx=5) # Admin 用户：显示 dryrun 复选框 if self.session_manager.is_admin(): self.dryrun_var = tk.BooleanVar(value=False) self.dryrun_checkbox = ttk.Checkbutton( button_frame, text="预览模式 (不保存)", variable=self.dryrun_var ) self.dryrun_checkbox.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): """创建结果表格""" # 检查是否为非管理员用户 is_user_only = not self.session_manager.is_admin() if is_user_only: # User 模式：创建两列布局（表格 + 按钮） # 主容器 main_result_container = ttk.Frame(parent) main_result_container.grid(row=0, column=0, sticky="nsew") parent.rowconfigure(0, weight=1) parent.columnconfigure(0, weight=1) # 表格容器 table_container = ttk.Frame(main_result_container) table_container.grid(row=0, column=0, sticky="nsew") main_result_container.rowconfigure(0, weight=1) main_result_container.columnconfigure(0, weight=1) # 创建表格 columns = ("选择", "材料名称", "材料代码", "规格", "型号", "负责人") self.tree = CheckboxTreeview( table_container, 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(table_container, orient=tk.VERTICAL, command=self.tree.yview) scrollbar_x = ttk.Scrollbar(table_container, 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") table_container.rowconfigure(0, weight=1) table_container.columnconfigure(0, weight=1) # 创建筛选按钮（右侧） self._create_user_filter_buttons(main_result_container) else: # Admin 模式：原布局 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_user_filter_buttons(self, parent): """创建 User 用户的筛选按钮""" button_frame = ttk.Frame(parent) button_frame.grid(row=0, column=1, sticky="ns", padx=(10, 0)) self.btn_hide_checked = ttk.Button( button_frame, text="隐藏勾选", command=self._hide_checked_items, state=tk.DISABLED ) self.btn_hide_checked.pack(fill=tk.X, pady=2) self.btn_show_all = ttk.Button( button_frame, text="显示全部", command=self._show_all_items, state=tk.DISABLED ) self.btn_show_all.pack(fill=tk.X, pady=2) # 分隔线 ttk.Separator(button_frame, orient='horizontal').pack(fill=tk.X, pady=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(fill=tk.X, pady=2) # 取消全选按钮 self.deselect_all_button = ttk.Button( button_frame, text="取消全选", command=lambda: self._select_all(check=False), state=tk.DISABLED ) self.deselect_all_button.pack(fill=tk.X, pady=2) # 确认删除按钮 self.confirm_delete_button = ttk.Button( button_frame, text="确认删除", command=self.confirm_deletion, state=tk.DISABLED ) self.confirm_delete_button.pack(fill=tk.X, pady=2) # 执行删除按钮 self.execute_delete_button = ttk.Button( button_frame, text="执行删除", command=self.start_delete_execution, state=tk.DISABLED ) self.execute_delete_button.pack(fill=tk.X, pady=2) def _hide_checked_items(self): """隐藏所有已勾选的物料""" checked_items = self.tree.get_checked_items() if not checked_items: self.log_text.info("没有已勾选的物料") return for item in checked_items: self.tree.detach(item) # Remove from view but keep data self.hidden_items.add(item) self.log_text.info(f"已隐藏 {len(checked_items)} 条勾选的物料") self.btn_show_all.config(state=tk.NORMAL) def _show_all_items(self): """显示所有物料（包括已隐藏的）""" if not self.hidden_items: self.log_text.info("没有隐藏的物料") return for item in self.hidden_items: self.tree.move(item, '', 'end') # Restore to end of tree count = len(self.hidden_items) self.hidden_items.clear() self.log_text.info(f"已显示 {count} 条隐藏的物料") 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() if mode == "database_full": # 全表模式：隐藏所有 Production ID 相关控件 if hasattr(self, 'source_frame') and self.source_frame: self.source_frame.grid_forget() if hasattr(self, 'db_filtered_production_id_selector'): self.db_filtered_production_id_selector.grid_forget() if hasattr(self, 'shared_ids_info_label'): self.shared_ids_info_label.grid_forget() elif mode == "database_filtered": # 过滤模式：显示数据源选择 if hasattr(self, 'source_frame') and self.source_frame: self.source_frame.grid(row=0, column=1, rowspan=2, sticky="nsew", padx=(20, 0)) # 根据 Production ID 数据源选择显示对应控件 self._on_production_id_source_changed() 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.grid_forget() self.shared_ids_info_label.grid(row=2, column=0, columnspan=2, sticky="w", pady=(5, 0)) else: # 使用文件 self.shared_ids_info_label.grid_forget() self.db_filtered_production_id_selector.grid(row=2, column=0, columnspan=2, sticky="ew", pady=(5, 0)) 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: # 启用执行删除按钮（如果有 Production ID） if production_ids and hasattr(self, 'execute_delete_button'): self.execute_delete_button.config(state=tk.NORMAL) 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)) # 启用执行删除按钮 if hasattr(self, 'execute_delete_button'): self.execute_delete_button.config(state=tk.NORMAL) 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() # 禁用执行删除按钮 if hasattr(self, 'execute_delete_button'): self.execute_delete_button.config(state=tk.DISABLED) 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 # 主容器：左右两列布局 main_container = ttk.Frame(parent) main_container.pack(fill=tk.BOTH, expand=True) # 左侧：按钮区域（竖向排列） button_frame = ttk.Frame(main_container) button_frame.pack(side=tk.LEFT, fill=tk.Y, padx=(0, 10)) ttk.Button(button_frame, text="全选", command=self._select_all_managers_button).pack(fill=tk.X, pady=2) ttk.Button(button_frame, text="取消全选", command=self._deselect_all_managers_button).pack(fill=tk.X, pady=2) # 右侧：负责人复选框区域（简化为直接使用 Frame） self.filter_frame = ttk.Frame(main_container) self.filter_frame.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) # 保留 filter_canvas 属性设为 None（兼容性考虑） self.filter_canvas = None 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 # 负责人复选框 for idx, manager in enumerate(self.managers): var = tk.BooleanVar(value=True) self.manager_checkboxes[manager] = var row = idx // items_per_column 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.hidden_items.clear() # 禁用筛选按钮（User 模式） if not self.session_manager.is_admin(): if hasattr(self, 'btn_hide_checked'): self.btn_hide_checked.config(state=tk.DISABLED) if hasattr(self, 'btn_show_all'): self.btn_show_all.config(state=tk.DISABLED) # 保存当前筛选状态 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 OSError as e: self.logger.debug(f"清理临时文件失败: {e}") # 更新 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) # 启用执行删除按钮（需要 Production ID） if self.shared_production_ids or ( self.session_manager.is_admin() and hasattr(self, 'db_filtered_production_id_selector') and self.db_filtered_production_id_selector.get() ): self.execute_delete_button.config(state=tk.NORMAL) # 启用筛选按钮（User 模式） if not self.session_manager.is_admin(): if hasattr(self, 'btn_hide_checked'): self.btn_hide_checked.config(state=tk.NORMAL) if hasattr(self, 'btn_show_all'): self.btn_show_all.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 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="物料类型管理") def start_delete_execution(self): """开始执行删除""" # 1. 获取 Production ID production_ids = None is_admin = self.session_manager.is_admin() if is_admin: # Admin 用户：检查数据源选择 if hasattr(self, 'production_id_source_var'): source = self.production_id_source_var.get() if source == "shared": production_ids = self.shared_production_ids else: production_id_file = self.db_filtered_production_id_selector.get() if production_id_file and os.path.exists(production_id_file): from db.production_order_query import read_production_ids production_ids = read_production_ids(production_id_file) else: # 普通用户：使用共享的 Production ID production_ids = self.shared_production_ids # 验证 Production ID if not production_ids: messagebox.showerror("错误", "没有可用的 Production ID\n请先在校验页面获取数据") return # 2. 获取负责人 if is_admin: manager_names = self._get_selected_managers() if not manager_names: messagebox.showwarning("警告", "请至少选择一个负责人") return else: manager_names = [self.session_manager.get_username()] # 3. 获取 dryrun 设置 dryrun = False if is_admin and hasattr(self, 'dryrun_var'): # Admin 用户：使用界面上的 dryrun 复选框 dryrun = self.dryrun_var.get() else: # User 用户：从配置读取 dryrun 设置 dryrun = self.config.get("execution.dryrun", False) # 4. 确认执行 mode_text = "预览模式（不会保存更改）" if dryrun else "正式执行（将保存更改）" manager_text = "、".join(manager_names) confirm_msg = ( f"确认执行删除操作？\n\n" f"模式: {mode_text}\n" f"负责人: {manager_text}\n" f"Production ID 数量: {len(production_ids)}\n\n" f"此操作可能需要较长时间，是否继续？" ) if not messagebox.askyesno("确认执行", confirm_msg): return # 5. 创建进度窗口 self.progress_window = DeleteProgressWindow( self, title="执行删除", managers=manager_text, dryrun=dryrun, on_cancel=self._cancel_delete_execution ) self.log_text.info(f"开始执行删除（模式: {'预览' if dryrun else '正式'}）...") # 6. 启动后台线程执行 delete_thread = threading.Thread( target=self._delete_worker, args=(production_ids, manager_names, dryrun), daemon=True ) delete_thread.start() def _cancel_delete_execution(self): """取消删除执行""" self.log_text.info("用户取消了执行操作") if hasattr(self, 'progress_window') and self.progress_window: self.progress_window.append_log("正在取消...", "warning") def _delete_worker(self, production_ids: list, manager_names: list, dryrun: bool): """ 后台线程执行删除 Args: production_ids: Production ID 列表 manager_names: 负责人列表 dryrun: 是否为预览模式 """ progress_window = self.progress_window cancelled = False try: from utils.discrete_material_plan_cleaner import DiscreteMaterialPlanCleaner from db.production_order_query import query_production_order_numbers # 创建临时文件保存 Production ID with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False, encoding='utf-8') as f: temp_file = f.name f.write('\n'.join(production_ids)) # 进度回调函数 def progress_callback(current: int, total: int, message: str): if progress_window and progress_window.is_cancelled(): return progress_window.update_progress(current, total, message) progress_window.append_log(message, "info") # 创建 Cleaner 实例 cleaner = DiscreteMaterialPlanCleaner( username=self.config.get("erp.username"), password=self.config.get("erp.password"), manager_names=manager_names, headless=self.config.get("erp.headless", True), verbose=True, dryrun=dryrun, progress_callback=progress_callback ) # 执行清理 progress_window.append_log("正在连接 ERP 系统...", "info") cleaner.clean(temp_file) # 检查是否被取消 if progress_window.is_cancelled(): cancelled = True self.after(0, lambda: self._update_log("执行已取消", "WARNING")) else: # 生成报告 report = cleaner.generate_report() # 在主线程中更新 UI self.after(0, lambda: self._delete_complete(report, cleaner.stats)) except Exception as e: error_msg = f"执行过程中发生错误: {str(e)}" self.after(0, lambda: self._update_log(error_msg, "ERROR")) if progress_window: progress_window.append_log(error_msg, "error") progress_window.set_completed() import traceback self.after(0, lambda: self._update_log(traceback.format_exc(), "ERROR")) finally: # 清理临时文件 try: if 'temp_file' in locals() and os.path.exists(temp_file): os.unlink(temp_file) except OSError as e: self.logger.debug(f"清理临时文件失败: {e}") def _delete_complete(self, report: str, stats: dict): """ 删除完成回调 Args: report: Markdown 格式的报告 stats: 统计信息字典 """ # 更新日志 self.log_text.info("执行完成！") self.log_text.info(f"处理订单: {stats.get('processed_orders', 0)}/{stats.get('total_orders', 0)}") self.log_text.info(f"删除物料: {len(stats.get('deleted_materials', []))} 条") self.log_text.info(f"跳过物料: {len(stats.get('skipped_materials', []))} 条") self.log_text.info(f"错误数量: {len(stats.get('errors', []))} 个") # 显示报告 if hasattr(self, 'progress_window') and self.progress_window: self.progress_window.show_report(report)