#!/usr/bin/env python # -*- coding: utf-8 -*- """ 物料校验标签页 校验物料状态并匹配待删除物料。 支持 4 种校验模式： 1. database_full - 数据库全表校验 2. database_filtered - 数据库 ProductionID 过滤校验 3. excel_existing - Excel 现有文件校验 4. excel_full - Excel 完整工作流 """ import os import threading import tkinter as tk from tkinter import ttk, messagebox, filedialog from pathlib import Path from io import StringIO from contextlib import redirect_stdout from gui.widgets import FileSelector, LogText from gui.config_manager import ConfigManager import pandas as pd class MaterialValidationTab(ttk.Frame): """物料校验标签页""" def __init__(self, parent, config: ConfigManager): """ 初始化物料校验标签页 Args: parent: 父容器 config: 配置管理器 """ super().__init__(parent) self.config = config self.validating = False self.validation_results = None 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)) # 中部：结果表格 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_result_table(result_frame) self._create_log_panel(log_frame) # 初始化文件选择器显示状态 self._on_source_mode_change() def _create_control_panel(self, parent): """创建控制面板""" # 数据来源选择 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) # Excel 模式 ttk.Radiobutton( source_group, text="Excel - 现有文件", variable=self.source_mode, value="excel_existing", command=self._on_source_mode_change, ).grid(row=1, column=0, sticky="w", padx=5) ttk.Radiobutton( source_group, text="Excel - 完整工作流", variable=self.source_mode, value="excel_full", command=self._on_source_mode_change, ).grid(row=1, column=1, sticky="w", padx=5) # 文件选择 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) # Excel 现有文件模式 self.excel_existing_frame = ttk.Frame(file_group) self.excel_existing_selector = FileSelector( self.excel_existing_frame, label_text="现有 Excel 文件:", file_type="file", file_types=[("Excel 文件", "*.xlsx"), ("所有文件", "*.*")], initial_dir=self.config.get("paths.data_dir", "data/"), ) self.excel_existing_selector.pack(fill=tk.X) # Excel 完整工作流模式 self.excel_full_frame = ttk.Frame(file_group) self.excel_full_production_id_selector = FileSelector( self.excel_full_frame, label_text="ProductionID 文件:", file_type="file", file_types=[("文本文件", "*.txt"), ("所有文件", "*.*")], initial_dir="D:/python/playwrite/", ) self.excel_full_production_id_selector.pack(fill=tk.X) # 输出文件 output_frame = ttk.Frame(file_group) output_frame.grid(row=1, column=0, columnspan=2, sticky="ew", pady=(10, 0)) self.output_file_selector = FileSelector( output_frame, 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) # 设置默认输出 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.export_button = ttk.Button( button_frame, text="导出结果", command=self.export_results, state=tk.DISABLED, ) self.export_button.pack(side=tk.LEFT, padx=5) def _create_result_table(self, parent): """创建结果表格""" # 创建 Treeview columns = ("材料名称", "匹配的MaterialName", "负责人", "匹配状态") self.tree = ttk.Treeview(parent, columns=columns, show="headings", height=10) # 设置列标题和宽度 self.tree.heading("材料名称", text="材料名称") self.tree.heading("匹配的MaterialName", text="匹配的MaterialName") self.tree.heading("负责人", text="负责人") self.tree.heading("匹配状态", text="匹配状态") self.tree.column("材料名称", width=300) self.tree.column("匹配的MaterialName", width=300) self.tree.column("负责人", width=150) self.tree.column("匹配状态", width=100) # 添加滚动条 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) 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): """数据源模式切换""" mode = self.source_mode.get() # 隐藏所有文件选择框架 self.db_full_frame.grid_remove() self.db_filtered_frame.grid_remove() self.excel_existing_frame.grid_remove() self.excel_full_frame.grid_remove() # 根据模式显示对应的文件选择器 if mode == "database_full": self.db_full_frame.grid(row=0, column=0, columnspan=2, sticky="ew") elif mode == "database_filtered": self.db_filtered_frame.grid(row=0, column=0, columnspan=2, sticky="ew") elif mode == "excel_existing": self.excel_existing_frame.grid(row=0, column=0, columnspan=2, sticky="ew") elif mode == "excel_full": self.excel_full_frame.grid(row=0, column=0, columnspan=2, sticky="ew") 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 elif mode == "excel_existing": input_file = self.excel_existing_selector.get() if not input_file: messagebox.showerror("错误", "请选择现有 Excel 文件") return if not os.path.exists(input_file): messagebox.showerror("错误", f"文件不存在：{input_file}") return elif mode == "excel_full": production_id_file = self.excel_full_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 # 确保输出目录存在 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.log_text.clear() 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, args=(mode, input_file, production_id_file, output_file), daemon=True, ) validation_thread.start() def _validation_worker( self, mode: str, input_file: str, production_id_file: str, output_file: str ): """校验工作线程""" try: # 导入校验器 from utils.material_status_validator import MaterialStatusValidator # 创建校验器实例（需要 ERP 凭据，因为可能需要登录系统） 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 = validator.validate_from_database( full_table=True, output_file=output_file ) elif mode == "database_filtered": result = validator.validate_from_database( production_id_file=production_id_file, full_table=False, output_file=output_file ) elif mode == "excel_existing": result = validator.validate_from_existing_excel( excel_file=input_file, output_file=output_file ) elif mode == "excel_full": result = validator.validate( production_id_file=production_id_file, merged_excel_file=None, output_file=output_file, ) else: raise ValueError(f"未知的校验模式: {mode}") # 获取捕获的输出并显示到日志 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: self._update_log(f"校验完成，结果已保存到：{output_file}", "SUCCESS") # 加载结果显示 self._load_results(output_file) else: self._update_log("校验失败", "ERROR") except Exception as e: self._update_log(f"校验过程中发生错误：{str(e)}", "ERROR") finally: # 更新 UI 状态 self.after(0, self._validation_complete) def _validation_complete(self): """校验完成后的 UI 更新""" self.validating = False self.start_button.config(state=tk.NORMAL) def _load_results(self, file_path: str): """加载校验结果到表格""" try: df = pd.read_excel(file_path) # 在主线程中更新表格 def update_table(): for _, row in df.iterrows(): self.tree.insert( "", tk.END, values=( row.get("材料名称", ""), row.get("匹配的MaterialName", ""), row.get("负责人", ""), row.get("匹配状态", ""), ), ) if len(df) > 0: self.export_button.config(state=tk.NORMAL) self._update_log(f"已加载 {len(df)} 条结果", "INFO") self.after(0, update_table) except Exception as e: self._update_log(f"加载结果失败：{str(e)}", "ERROR") 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"] data.append(values) if not data: messagebox.showwarning("警告", "没有数据可导出") return # 创建 DataFrame 并保存 df = pd.DataFrame( data, columns=["材料名称", "匹配的MaterialName", "负责人", "匹配状态"] ) 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)}")