From 2096a51f55bc0949ef7581f8c09c3855795f1345 Mon Sep 17 00:00:00 2001 From: Misaka_Company Date: Fri, 6 Feb 2026 13:08:31 +0800 Subject: [PATCH] feat: add database-driven material validation with multi-mode support MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This commit enhances the material validation functionality to support database-driven workflows alongside the existing Excel-based approach. ## New Features ### DAO Layer - Add ProductionContractDataDAO for querying production contract data - Add MaterialsToBeDeletedDAO with full CRUD operations - Enhance DiscreteMaterialPlanDAO with query_all(), query_by_source_numbers(), and get_unique_material_names() methods ### Validation Modes Support for 4 validation modes in MaterialValidationTab: 1. Database Full - Query all materials from DiscreteMaterialPlanData 2. Database Filtered - Query by ProductionID.txt file 3. Excel Existing - Validate from existing Excel file 4. Excel Full - Complete workflow with ERP extraction ### Configuration - Add ValidationConfig dataclass with data_source, batch_size, match_mode, enable_crud_operations, and default_manager fields - Update ConfigLoader to support validation configuration - Add validation settings section in SettingsTab GUI ### Query Chain Implementation of full query chain: ProductionID.txt (总排号) → productionContractData (生产订单号) → DiscreteMaterialPlanData (SourceNumber) → MaterialName → MaterialsToBeDeleted comparison ## Backward Compatibility All existing Excel-based validation methods remain unchanged, ensuring no breaking changes for existing workflows. Co-Authored-By: Claude Sonnet 4.5 --- config/defaults.py | 9 + config/loader.py | 12 +- config/schema.py | 52 +++++ db/discrete_material_plan_dao.py | 101 ++++++++- db/materials_to_be_deleted_dao.py | 345 +++++++++++++++++++++++++++++ db/production_contract_data_dao.py | 99 +++++++++ gui/material_validation_tab.py | 186 +++++++++++----- gui/settings_tab.py | 78 ++++++- utils/material_status_validator.py | 151 +++++++++++++ 9 files changed, 976 insertions(+), 57 deletions(-) create mode 100644 db/materials_to_be_deleted_dao.py create mode 100644 db/production_contract_data_dao.py diff --git a/config/defaults.py b/config/defaults.py index bd00773..a30fe54 100644 --- a/config/defaults.py +++ b/config/defaults.py @@ -10,6 +10,7 @@ from config.schema import ( DatabaseConfig, PathConfig, ExtractionConfig, + ValidationConfig, AppConfig, ) @@ -45,6 +46,14 @@ DEFAULT_APP_CONFIG = AppConfig( merge_batches=True, enable_db_persistence=False, # Disabled by default ), + validation=ValidationConfig( + data_source="database_full", + use_database=True, + batch_size=2000, + enable_crud_operations=False, + default_manager="", + match_mode="substring", + ), ) diff --git a/config/loader.py b/config/loader.py index ef76f49..4b27cb0 100644 --- a/config/loader.py +++ b/config/loader.py @@ -107,6 +107,7 @@ class ConfigLoader: database_dict = settings.get("database", {}) paths_dict = settings.get("paths", {}) extraction_dict = settings.get("extraction", {}) + validation_dict = settings.get("validation", {}) return AppConfig( erp=ERPConfig( @@ -142,9 +143,18 @@ class ConfigLoader: verbose=extraction_dict.get("verbose", True), auto_convert=extraction_dict.get("auto_convert", True), merge_batches=extraction_dict.get("merge_batches", True), + enable_db_persistence=extraction_dict.get("enable_db_persistence", False), + ), + validation=ValidationConfig( + data_source=validation_dict.get("data_source", "database_full"), + use_database=validation_dict.get("use_database", True), + batch_size=validation_dict.get("batch_size", 2000), + enable_crud_operations=validation_dict.get("enable_crud_operations", False), + default_manager=validation_dict.get("default_manager", ""), + match_mode=validation_dict.get("match_mode", "substring"), ), ) # 为了兼容旧代码，导入必要的类型 -from config.schema import ERPConfig, DatabaseConfig, PathConfig, ExtractionConfig +from config.schema import ERPConfig, DatabaseConfig, PathConfig, ExtractionConfig, ValidationConfig diff --git a/config/schema.py b/config/schema.py index aaeb2d4..2c73fe4 100644 --- a/config/schema.py +++ b/config/schema.py @@ -97,6 +97,48 @@ class ExtractionConfig: return errors +@dataclass +class ValidationConfig: + """物料校验配置""" + + data_source: str = "database_full" + use_database: bool = True + batch_size: int = 2000 + enable_crud_operations: bool = False + default_manager: str = "" + match_mode: str = "substring" + + def validate(self) -> list[str]: + """验证配置，返回错误列表""" + errors = [] + + valid_sources = [ + "database_full", + "database_filtered", + "excel_existing", + "excel_full" + ] + if self.data_source not in valid_sources: + errors.append( + f"无效的数据源: {self.data_source}。" + f"有效选项: {', '.join(valid_sources)}" + ) + + if self.batch_size <= 0: + errors.append("批次大小必须大于 0") + if self.batch_size > 2000: + errors.append("批次大小不应超过 2000（SQL Server 参数限制）") + + valid_match_modes = ["substring", "exact"] + if self.match_mode not in valid_match_modes: + errors.append( + f"无效的匹配模式: {self.match_mode}。" + f"有效选项: {', '.join(valid_match_modes)}" + ) + + return errors + + @dataclass class AppConfig: """应用总配置""" @@ -105,6 +147,7 @@ class AppConfig: database: DatabaseConfig paths: PathConfig extraction: ExtractionConfig + validation: ValidationConfig def validate(self) -> list[str]: """验证所有配置，返回错误列表""" @@ -113,6 +156,7 @@ class AppConfig: errors.extend(self.database.validate()) errors.extend(self.paths.validate()) errors.extend(self.extraction.validate()) + errors.extend(self.validation.validate()) return errors def to_dict(self) -> dict: @@ -147,4 +191,12 @@ class AppConfig: "merge_batches": self.extraction.merge_batches, "enable_db_persistence": self.extraction.enable_db_persistence, }, + "validation": { + "data_source": self.validation.data_source, + "use_database": self.validation.use_database, + "batch_size": self.validation.batch_size, + "enable_crud_operations": self.validation.enable_crud_operations, + "default_manager": self.validation.default_manager, + "match_mode": self.validation.match_mode, + }, } diff --git a/db/discrete_material_plan_dao.py b/db/discrete_material_plan_dao.py index e95bf18..1fecfc0 100644 --- a/db/discrete_material_plan_dao.py +++ b/db/discrete_material_plan_dao.py @@ -26,12 +26,12 @@ class DiscreteMaterialPlanDAO: def __exit__(self, exc_type, exc_val, exc_tb): """Exit context manager and close database connection""" if self.db: - self.db.close() + self.db.disconnect() def close(self): """Close database connection""" if self.db: - self.db.close() + self.db.disconnect() def save_dataframe_with_replace(self, df: pd.DataFrame) -> Dict[str, int]: """ @@ -311,3 +311,100 @@ class DiscreteMaterialPlanDAO: """ result = db.execute_query(sql) return result[0] if result else {} + + # ==================== ENHANCED QUERY METHODS ==================== + + def query_all(self) -> List[Dict]: + """ + Query all records from DiscreteMaterialPlanData table. + + Returns: + List of dictionaries representing all records + """ + with get_connection() as db: + sql = "SELECT * FROM DiscreteMaterialPlanData" + return db.execute_query(sql) + + def query_by_source_numbers(self, source_numbers: List[str]) -> List[Dict]: + """ + Query records by SourceNumber list (生产订单号). + + Args: + source_numbers: List of SourceNumber values to query + + Returns: + List of dictionaries representing records + """ + if not source_numbers: + return [] + + # SQL Server parameter limit requires batching + batch_size = 2000 + all_results = [] + + for i in range(0, len(source_numbers), batch_size): + batch = source_numbers[i:i + batch_size] + placeholders = ','.join(['?' for _ in batch]) + sql = f"SELECT * FROM DiscreteMaterialPlanData WHERE SourceNumber IN ({placeholders})" + with get_connection() as db: + results = db.execute_query(sql, tuple(batch)) + all_results.extend(results) + + return all_results + + def get_unique_material_names(self, source_numbers: List[str] = None) -> List[str]: + """ + Get unique material names, optionally filtered by SourceNumber. + + Args: + source_numbers: Optional list of SourceNumber values to filter by + + Returns: + List of unique material names + """ + if source_numbers is None or not source_numbers: + # No filter - get all unique material names + sql = "SELECT DISTINCT MaterialName FROM DiscreteMaterialPlanData WHERE MaterialName IS NOT NULL" + with get_connection() as db: + results = db.execute_query(sql) + return [r['MaterialName'] for r in results if r.get('MaterialName')] + else: + # Filter by SourceNumber list + batch_size = 2000 + all_material_names = set() + + for i in range(0, len(source_numbers), batch_size): + batch = source_numbers[i:i + batch_size] + placeholders = ','.join(['?' for _ in batch]) + sql = f""" + SELECT DISTINCT MaterialName + FROM DiscreteMaterialPlanData + WHERE SourceNumber IN ({placeholders}) + AND MaterialName IS NOT NULL + """ + with get_connection() as db: + results = db.execute_query(sql, tuple(batch)) + batch_materials = [r['MaterialName'] for r in results if r.get('MaterialName')] + all_material_names.update(batch_materials) + + return list(all_material_names) + + def get_material_names_by_总排号(self, 总排号_list: List[str]) -> List[str]: + """ + Get unique material names by 总排号 list. + This method combines query from production contract data and discrete material plan. + + Args: + 总排号_list: List of 总排号 values + + Returns: + List of unique material names + """ + from db.production_contract_data_dao import ProductionContractDataDAO + + # First get SourceNumbers from production contract data + contract_dao = ProductionContractDataDAO() + source_numbers = contract_dao.get_source_numbers_by_总排号(总排号_list) + + # Then get material names filtered by these SourceNumbers + return self.get_unique_material_names(source_numbers) diff --git a/db/materials_to_be_deleted_dao.py b/db/materials_to_be_deleted_dao.py new file mode 100644 index 0000000..a3768c9 --- /dev/null +++ b/db/materials_to_be_deleted_dao.py @@ -0,0 +1,345 @@ +""" +Data Access Object for MaterialsToBeDeleted table. + +This module provides CRUD operations for the MaterialsToBeDeleted table, +which tracks materials that need to be deleted by their managers. +""" + +from typing import List, Dict, Any, Tuple, Optional +from db.connection import get_connection + + +class MaterialsToBeDeletedDAO: + """Data Access Object for MaterialsToBeDeleted table CRUD operations""" + + def __init__(self): + self.db = None + + def __enter__(self): + """Enter context manager and establish database connection""" + self.db = get_connection() + self.db.connect() + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + """Exit context manager and close database connection""" + if self.db: + self.db.disconnect() + + def close(self): + """Close database connection""" + if self.db: + self.db.disconnect() + + # ==================== CREATE ==================== + + def insert_material( + self, material_name: str, manager_name: str + ) -> bool: + """ + Insert a single material record. + + Args: + material_name: Material name + manager_name: Manager name + + Returns: + True if successful, False otherwise + """ + sql = """ + INSERT INTO [dbo].[MaterialsToBeDeleted] ([MaterialName], [ManagerName]) + VALUES (?, ?) + """ + try: + with get_connection() as db: + db.execute_update(sql, (material_name, manager_name)) + return True + except Exception as e: + print(f"Error inserting material: {e}") + return False + + def insert_batch(self, materials: List[Tuple[str, str]]) -> int: + """ + Insert multiple material records in batch. + + Args: + materials: List of tuples (material_name, manager_name) + + Returns: + Number of records inserted + """ + if not materials: + return 0 + + sql = """ + INSERT INTO [dbo].[MaterialsToBeDeleted] ([MaterialName], [ManagerName]) + VALUES (?, ?) + """ + + inserted_count = 0 + try: + with get_connection() as db: + for material_name, manager_name in materials: + db.execute_update(sql, (material_name, manager_name)) + inserted_count += 1 + return inserted_count + except Exception as e: + print(f"Error inserting batch materials: {e}") + return inserted_count + + # ==================== READ ==================== + + def get_all_materials(self) -> List[Dict[str, Any]]: + """ + Get all material records. + + Returns: + List of all materials with MaterialName and ManagerName + """ + sql = """ + SELECT [MaterialName], [ManagerName] + FROM [dbo].[MaterialsToBeDeleted] + WHERE [MaterialName] IS NOT NULL + ORDER BY [ManagerName], [MaterialName] + """ + with get_connection() as db: + return db.execute_query(sql) + + def get_materials_by_manager(self, manager_name: str) -> List[Dict[str, Any]]: + """ + Get all materials for a specific manager. + + Args: + manager_name: Manager name + + Returns: + List of materials for the specified manager + """ + sql = """ + SELECT [MaterialName], [ManagerName] + FROM [dbo].[MaterialsToBeDeleted] + WHERE [ManagerName] = ? AND [MaterialName] IS NOT NULL + ORDER BY [MaterialName] + """ + with get_connection() as db: + return db.execute_query(sql, (manager_name,)) + + def get_managers(self) -> List[str]: + """ + Get list of unique manager names. + + Returns: + List of unique manager names + """ + sql = """ + SELECT DISTINCT [ManagerName] + FROM [dbo].[MaterialsToBeDeleted] + WHERE [ManagerName] IS NOT NULL + ORDER BY [ManagerName] + """ + with get_connection() as db: + results = db.execute_query(sql) + return [r['ManagerName'] for r in results if r.get('ManagerName')] + + def get_material_names_by_manager(self, manager_name: str) -> List[str]: + """ + Get material names for a specific manager. + + Args: + manager_name: Manager name + + Returns: + List of material names for the specified manager + """ + results = self.get_materials_by_manager(manager_name) + return [r['MaterialName'] for r in results if r.get('MaterialName')] + + # ==================== UPDATE ==================== + + def update_manager( + self, + material_name: str, + old_manager: str, + new_manager: str + ) -> bool: + """ + Update manager for a specific material. + + Args: + material_name: Material name + old_manager: Current manager name + new_manager: New manager name + + Returns: + True if successful, False otherwise + """ + sql = """ + UPDATE [dbo].[MaterialsToBeDeleted] + SET [ManagerName] = ? + WHERE [MaterialName] = ? AND [ManagerName] = ? + """ + try: + with get_connection() as db: + affected = db.execute_update(sql, (new_manager, material_name, old_manager)) + return affected > 0 + except Exception as e: + print(f"Error updating manager: {e}") + return False + + # ==================== DELETE ==================== + + def delete_material( + self, + material_name: str, + manager_name: str + ) -> bool: + """ + Delete a specific material record. + + Args: + material_name: Material name + manager_name: Manager name + + Returns: + True if successful, False otherwise + """ + sql = """ + DELETE FROM [dbo].[MaterialsToBeDeleted] + WHERE [MaterialName] = ? AND [ManagerName] = ? + """ + try: + with get_connection() as db: + affected = db.execute_update(sql, (material_name, manager_name)) + return affected > 0 + except Exception as e: + print(f"Error deleting material: {e}") + return False + + def delete_by_manager(self, manager_name: str) -> int: + """ + Delete all materials for a specific manager. + + Args: + manager_name: Manager name + + Returns: + Number of records deleted + """ + sql = """ + DELETE FROM [dbo].[MaterialsToBeDeleted] + WHERE [ManagerName] = ? + """ + try: + with get_connection() as db: + return db.execute_update(sql, (manager_name,)) + except Exception as e: + print(f"Error deleting by manager: {e}") + return 0 + + def delete_all_materials(self) -> int: + """ + Delete all material records. + + Returns: + Number of records deleted + """ + sql = "DELETE FROM [dbo].[MaterialsToBeDeleted]" + try: + with get_connection() as db: + return db.execute_update(sql) + except Exception as e: + print(f"Error deleting all materials: {e}") + return 0 + + # ==================== UTILITIES ==================== + + def material_exists(self, material_name: str) -> bool: + """ + Check if a material exists in the table. + + Args: + material_name: Material name + + Returns: + True if material exists, False otherwise + """ + sql = """ + SELECT COUNT(*) as count + FROM [dbo].[MaterialsToBeDeleted] + WHERE [MaterialName] = ? + """ + with get_connection() as db: + result = db.execute_query(sql, (material_name,)) + return result[0]['count'] > 0 if result else False + + def count_by_manager(self, manager_name: str) -> int: + """ + Count materials for a specific manager. + + Args: + manager_name: Manager name + + Returns: + Number of materials for the manager + """ + sql = """ + SELECT COUNT(*) as count + FROM [dbo].[MaterialsToBeDeleted] + WHERE [ManagerName] = ? + """ + with get_connection() as db: + result = db.execute_query(sql, (manager_name,)) + return result[0]['count'] if result else 0 + + def get_statistics(self) -> Dict[str, Any]: + """ + Get comprehensive statistics about the data. + + Returns: + Dictionary with statistics including total materials, + unique managers, and materials per manager + """ + sql = """ + SELECT + COUNT(*) as total_materials, + COUNT(DISTINCT ManagerName) as unique_managers + FROM [dbo].[MaterialsToBeDeleted] + WHERE [MaterialName] IS NOT NULL + """ + with get_connection() as db: + result = db.execute_query(sql) + stats = result[0] if result else {} + + # Get materials per manager + manager_sql = """ + SELECT [ManagerName], COUNT(*) as count + FROM [dbo].[MaterialsToBeDeleted] + WHERE [ManagerName] IS NOT NULL + GROUP BY [ManagerName] + ORDER BY count DESC + """ + manager_results = db.execute_query(manager_sql) + stats['materials_per_manager'] = [ + {r['ManagerName']: r['count']} for r in manager_results + ] + + return stats + + def search_materials(self, keyword: str) -> List[Dict[str, Any]]: + """ + Search materials by keyword (partial match). + + Args: + keyword: Keyword to search for + + Returns: + List of matching materials + """ + sql = """ + SELECT [MaterialName], [ManagerName] + FROM [dbo].[MaterialsToBeDeleted] + WHERE [MaterialName] LIKE ? + ORDER BY [ManagerName], [MaterialName] + """ + with get_connection() as db: + return db.execute_query(sql, (f'%{keyword}%',)) diff --git a/db/production_contract_data_dao.py b/db/production_contract_data_dao.py new file mode 100644 index 0000000..3c4dc7d --- /dev/null +++ b/db/production_contract_data_dao.py @@ -0,0 +1,99 @@ +""" +Data Access Object for production contract data. + +This module provides query operations for accessing production contract data +from the [productionContractData].[26年压力表合同数据] table. +""" + +from typing import List, Dict, Any +from db.connection import get_connection + + +class ProductionContractDataDAO: + """Data Access Object for production contract data queries""" + + def __init__(self): + self.db = None + + def __enter__(self): + """Enter context manager and establish database connection""" + self.db = get_connection() + self.db.connect() + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + """Exit context manager and close database connection""" + if self.db: + self.db.disconnect() + + def close(self): + """Close database connection""" + if self.db: + self.db.disconnect() + + def query_by_总排号(self, 总排号_list: List[str]) -> List[Dict[str, Any]]: + """ + Query production contract data by 总排号 list. + + Args: + 总排号_list: List of 总排号 values to query + + Returns: + List of dictionaries containing 总排号, 生产订单号, 序号, 订单号, 客户名称, 产品型号 + """ + if not 总排号_list: + return [] + + # SQL Server parameter limit requires batching + batch_size = 2000 + all_results = [] + + for i in range(0, len(总排号_list), batch_size): + batch = 总排号_list[i:i + batch_size] + placeholders = ','.join(['?' for _ in batch]) + sql = f""" + SELECT [总排号], [生产订单号], [序号], [订单号], [客户名称], [产品型号] + FROM [productionContractData].[26年压力表合同数据] + WHERE [总排号] IN ({placeholders}) + ORDER BY [序号] + """ + + with get_connection() as db: + results = db.execute_query(sql, tuple(batch)) + all_results.extend(results) + + return all_results + + def get_source_numbers_by_总排号(self, 总排号_list: List[str]) -> List[str]: + """ + Extract unique 生产订单号 values by 总排号 list. + + Args: + 总排号_list: List of 总排号 values to query + + Returns: + List of unique 生产订单号 values (SourceNumber) + """ + results = self.query_by_总排号(总排号_list) + # Extract unique 生产订单号 values, excluding None/null values + source_numbers = list(set( + [r['生产订单号'] for r in results if r.get('生产订单号')] + )) + return source_numbers + + def get_生产订单号_map(self, 总排号_list: List[str]) -> Dict[str, str]: + """ + Get mapping between 总排号 and 生产订单号. + + Args: + 总排号_list: List of 总排号 values to query + + Returns: + Dictionary mapping 总排号 -> 生产订单号 + """ + results = self.query_by_总排号(总排号_list) + return { + r['总排号']: r['生产订单号'] + for r in results + if r.get('总排号') and r.get('生产订单号') + } diff --git a/gui/material_validation_tab.py b/gui/material_validation_tab.py index 1af7f34..4a9b5ee 100644 --- a/gui/material_validation_tab.py +++ b/gui/material_validation_tab.py @@ -4,12 +4,17 @@ 物料校验标签页 校验物料状态并匹配待删除物料。 +支持 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 +from tkinter import ttk, messagebox, filedialog from pathlib import Path from io import StringIO from contextlib import redirect_stdout @@ -64,58 +69,95 @@ class MaterialValidationTab(ttk.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="existing") + self.source_mode = tk.StringVar(value="database_full") + + # 数据库模式 ttk.Radiobutton( source_group, - text="使用现有 Excel 文件", + text="数据库 - 全表校验", variable=self.source_mode, - value="existing", + value="database_full", command=self._on_source_mode_change, ).grid(row=0, column=0, sticky="w", padx=5) ttk.Radiobutton( source_group, - text="完整工作流 (提取 + 校验)", + text="数据库 - ProductionID 过滤", variable=self.source_mode, - value="full", + 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) - # 现有 Excel 文件 - self.existing_excel_frame = ttk.Frame(file_group) - self.existing_excel_frame.grid(row=0, column=0, columnspan=2, sticky="ew") + # 数据库全表模式 - 无需输入文件 + self.db_full_frame = ttk.Frame(file_group) + ttk.Label( + self.db_full_frame, + text="数据库全表模式：将查询 DiscreteMaterialPlanData 表中的所有材料", + foreground="gray" + ).pack(anchor="w") - self.existing_excel_selector = FileSelector( - self.existing_excel_frame, - label_text="现有 Excel 文件:", - file_type="file", - file_types=[("Excel 文件", "*.xlsx"), ("所有文件", "*.*")], - initial_dir=self.config.get("paths.data_dir", "data/"), - ) - self.existing_excel_selector.pack(fill=tk.X) - - # ProductionID 文件 (完整工作流模式) - self.production_id_frame = ttk.Frame(file_group) - # 初始隐藏 - - self.production_id_selector = FileSelector( - self.production_id_frame, + # 数据库过滤模式 - 需要 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.production_id_selector.pack(fill=tk.X) + 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) @@ -196,12 +238,23 @@ class MaterialValidationTab(ttk.Frame): def _on_source_mode_change(self): """数据源模式切换""" - if self.source_mode.get() == "existing": - self.existing_excel_frame.grid(row=0, column=0, columnspan=2, sticky="ew") - self.production_id_frame.grid_remove() - else: - self.existing_excel_frame.grid_remove() - self.production_id_frame.grid(row=0, column=0, columnspan=2, sticky="ew") + 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): """开始校验""" @@ -211,25 +264,38 @@ class MaterialValidationTab(ttk.Frame): messagebox.showerror("错误", "请指定输出文件路径") return - # 获取输入文件 - if self.source_mode.get() == "existing": - input_file = self.existing_excel_selector.get() - if not input_file: - messagebox.showerror("错误", "请选择现有 Excel 文件") - return - if not os.path.exists(input_file): - messagebox.showerror("错误", f"文件不存在：{input_file}") - return - production_id_file = None - else: - production_id_file = self.production_id_selector.get() + 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 - input_file = None # 确保输出目录存在 output_dir = os.path.dirname(output_file) @@ -240,7 +306,7 @@ class MaterialValidationTab(ttk.Frame): self.validating = True self.start_button.config(state=tk.DISABLED) self.log_text.clear() - self.log_text.info("开始物料校验...") + self.log_text.info(f"开始物料校验（模式: {mode}）...") # 清空结果表格 for item in self.tree.get_children(): @@ -249,13 +315,13 @@ class MaterialValidationTab(ttk.Frame): # 在后台线程中执行校验 validation_thread = threading.Thread( target=self._validation_worker, - args=(input_file, production_id_file, output_file), + args=(mode, input_file, production_id_file, output_file), daemon=True, ) validation_thread.start() def _validation_worker( - self, input_file: str, production_id_file: str, output_file: str + self, mode: str, input_file: str, production_id_file: str, output_file: str ): """校验工作线程""" try: @@ -273,18 +339,32 @@ class MaterialValidationTab(ttk.Frame): # 捕获 stdout 输出 captured_output = StringIO() - # 执行校验并捕获输出 + # 根据模式执行校验 with redirect_stdout(captured_output): - if self.source_mode.get() == "existing": - result = validator.validate_from_existing_excel( - excel_file=input_file, output_file=output_file + if mode == "database_full": + result = validator.validate_from_database( + full_table=True, + output_file=output_file ) - else: + 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, # 将在内部生成 + merged_excel_file=None, output_file=output_file, ) + else: + raise ValueError(f"未知的校验模式: {mode}") # 获取捕获的输出并显示到日志 output_text = captured_output.getvalue() diff --git a/gui/settings_tab.py b/gui/settings_tab.py index e0b12ad..3055148 100644 --- a/gui/settings_tab.py +++ b/gui/settings_tab.py @@ -57,9 +57,12 @@ class SettingsTab(ttk.Frame): # 处理配置组 self._create_extraction_group(scrollable_frame) + # 校验配置组 + self._create_validation_group(scrollable_frame) + # 按钮区域 button_frame = ttk.Frame(scrollable_frame) - button_frame.grid(row=5, column=0, columnspan=2, pady=20, sticky="ew") + button_frame.grid(row=6, column=0, columnspan=2, pady=20, sticky="ew") ttk.Button( button_frame, text="测试 ERP 连接", command=self.test_erp_connection @@ -224,6 +227,63 @@ class SettingsTab(ttk.Frame): group, text="保存到数据库 (同时写入 SQL Server)", variable=self.enable_db_persistence_var ).grid(row=4, column=0, columnspan=2, sticky="w", pady=5) + def _create_validation_group(self, parent): + """创建物料校验配置组""" + group = ttk.LabelFrame(parent, text="物料校验设置", padding=10) + group.grid(row=4, column=0, columnspan=2, pady=10, padx=10, sticky="ew") + + # 数据源选择 + ttk.Label(group, text="默认数据源:").grid(row=0, column=0, sticky="w", pady=5) + self.validation_data_source_var = tk.StringVar() + data_source_combo = ttk.Combobox( + group, + textvariable=self.validation_data_source_var, + values=["database_full", "database_filtered", "excel_existing", "excel_full"], + state="readonly", + width=30, + ) + data_source_combo.grid(row=0, column=1, sticky="w", pady=5) + + # 使用数据库 + self.validation_use_database_var = tk.BooleanVar() + ttk.Checkbutton( + group, text="使用数据库作为数据源", variable=self.validation_use_database_var + ).grid(row=1, column=0, columnspan=2, sticky="w", pady=5) + + # 批次大小 + ttk.Label(group, text="数据库批次大小:").grid(row=2, column=0, sticky="w", pady=5) + self.validation_batch_size_var = tk.IntVar(value=2000) + ttk.Spinbox( + group, from_=100, to=2000, textvariable=self.validation_batch_size_var, width=10 + ).grid(row=2, column=1, sticky="w", pady=5) + + # 匹配模式 + ttk.Label(group, text="匹配模式:").grid(row=3, column=0, sticky="w", pady=5) + self.validation_match_mode_var = tk.StringVar() + match_mode_combo = ttk.Combobox( + group, + textvariable=self.validation_match_mode_var, + values=["substring", "exact"], + state="readonly", + width=30, + ) + match_mode_combo.grid(row=3, column=1, sticky="w", pady=5) + + # CRUD 操作 + self.validation_enable_crud_var = tk.BooleanVar() + ttk.Checkbutton( + group, text="启用 CRUD 操作（管理待删除物料）", variable=self.validation_enable_crud_var + ).grid(row=4, column=0, columnspan=2, sticky="w", pady=5) + + # 默认负责人 + ttk.Label(group, text="默认负责人:").grid(row=5, column=0, sticky="w", pady=5) + self.validation_default_manager_var = tk.StringVar() + ttk.Entry(group, textvariable=self.validation_default_manager_var, width=30).grid( + row=5, column=1, sticky="w", pady=5 + ) + + group.columnconfigure(1, weight=1) + def load_settings(self): """从配置加载设置到界面""" # ERP 设置 @@ -255,6 +315,14 @@ class SettingsTab(ttk.Frame): self.merge_batches_var.set(self.config.get("extraction.merge_batches", True)) self.enable_db_persistence_var.set(self.config.get("extraction.enable_db_persistence", False)) + # 校验设置 + self.validation_data_source_var.set(self.config.get("validation.data_source", "database_full")) + self.validation_use_database_var.set(self.config.get("validation.use_database", True)) + self.validation_batch_size_var.set(self.config.get("validation.batch_size", 2000)) + self.validation_match_mode_var.set(self.config.get("validation.match_mode", "substring")) + self.validation_enable_crud_var.set(self.config.get("validation.enable_crud_operations", False)) + self.validation_default_manager_var.set(self.config.get("validation.default_manager", "")) + def save_settings(self): """保存界面设置到配置""" # ERP 设置 @@ -284,6 +352,14 @@ class SettingsTab(ttk.Frame): self.config.set("extraction.merge_batches", self.merge_batches_var.get()) self.config.set("extraction.enable_db_persistence", self.enable_db_persistence_var.get()) + # 校验设置 + self.config.set("validation.data_source", self.validation_data_source_var.get()) + self.config.set("validation.use_database", self.validation_use_database_var.get()) + self.config.set("validation.batch_size", self.validation_batch_size_var.get()) + self.config.set("validation.match_mode", self.validation_match_mode_var.get()) + self.config.set("validation.enable_crud_operations", self.validation_enable_crud_var.get()) + self.config.set("validation.default_manager", self.validation_default_manager_var.get()) + # 保存到文件 if self.config.save(): messagebox.showinfo("成功", "设置已保存") diff --git a/utils/material_status_validator.py b/utils/material_status_validator.py index 8e9dbbc..cdb2749 100644 --- a/utils/material_status_validator.py +++ b/utils/material_status_validator.py @@ -1,6 +1,10 @@ """ 物料状态校验工具 校验订单中的物料状态，匹配待删除物料 + +支持两种数据源： +1. Excel 文件（原有方式） +2. 数据库驱动（新增方式） """ import os @@ -8,6 +12,9 @@ import pandas as pd from typing import List, Dict, Any from utils.离散备料计划维护数据提取 import DiscreteMaterialPlanExtractor from db.materials_to_delete import get_all_materials_to_delete +from db.production_contract_data_dao import ProductionContractDataDAO +from db.discrete_material_plan_dao import DiscreteMaterialPlanDAO +from db.materials_to_be_deleted_dao import MaterialsToBeDeletedDAO class MaterialStatusValidator: @@ -195,3 +202,147 @@ class MaterialStatusValidator: self._print(f" 未匹配: {len(results) - matched_count}") return output_file + + # ==================== DATABASE-DRIVEN VALIDATION ==================== + + def _read_production_ids(self, production_id_file: str) -> List[str]: + """ + 读取 ProductionID.txt 文件 + + Args: + production_id_file: ProductionID.txt 文件路径 + + Returns: + List[str]: 总排号列表 + """ + with open(production_id_file, 'r', encoding='utf-8') as f: + production_ids = [line.strip() for line in f if line.strip()] + return production_ids + + def _get_source_numbers_from_production_ids( + self, production_ids: List[str] + ) -> List[str]: + """ + 通过 ProductionID 查询获取 SourceNumber 列表 + + 查询链路: + ProductionID (总排号) -> productionContractData.26年压力表合同数据.生产订单号 + + Args: + production_ids: 总排号列表 + + Returns: + List[str]: 生产订单号列表 + """ + self._print(f"[INFO] 正在查询 {len(production_ids)} 个总排号对应的生产订单号...") + + contract_dao = ProductionContractDataDAO() + source_numbers = contract_dao.get_source_numbers_by_总排号(production_ids) + + self._print(f"[INFO] 找到 {len(source_numbers)} 个唯一的生产订单号") + return source_numbers + + def _get_material_names_from_db( + self, source_numbers: List[str] = None + ) -> List[str]: + """ + 从数据库获取材料名称，可选按 SourceNumber 过滤 + + Args: + source_numbers: 可选的 SourceNumber 列表进行过滤 + + Returns: + List[str]: 唯一材料名称列表 + """ + if source_numbers is None or not source_numbers: + self._print("[INFO] 查询所有材料的名称...") + else: + self._print(f"[INFO] 查询 {len(source_numbers)} 个生产订单对应的材料名称...") + + dao = DiscreteMaterialPlanDAO() + material_names = dao.get_unique_material_names(source_numbers) + + self._print(f"[INFO] 找到 {len(material_names)} 个唯一材料名称") + return material_names + + def validate_from_database( + self, + production_id_file: str = None, + full_table: bool = False, + output_file: str = None + ) -> str: + """ + 使用数据库作为数据源执行校验 + + 支持两种模式： + 1. 全表校验 (full_table=True): 查询整个 DiscreteMaterialPlanData 表 + 2. ProductionID 过滤校验 (production_id_file 指定): 基于 ProductionID.txt 文件过滤 + + 查询链路（模式2）: + ProductionID.txt (总排号) + -> productionContractData.26年压力表合同数据.生产订单号 (SourceNumber) + -> DiscreteMaterialPlanData.SourceNumber + -> DiscreteMaterialPlanData.MaterialName + -> 对比 MaterialsToBeDeleted.MaterialName + + Args: + production_id_file: ProductionID.txt 路径（模式2） + full_table: 是否全表校验（模式1） + output_file: 输出文件路径 + + Returns: + 输出文件路径 + """ + # 设置默认输出路径 + if output_file is None: + output_file = "D:/python/playwrite/data/物料状态校验结果.xlsx" + + self._print("=" * 60) + self._print("使用数据库数据源执行校验...") + + # 根据模式获取材料名称 + if full_table: + self._print("\n模式: 全表校验") + self._print("[INFO] 查询 DiscreteMaterialPlanData 表中的所有材料...") + material_names = self._get_material_names_from_db(None) + elif production_id_file: + self._print("\n模式: ProductionID 过滤校验") + self._print(f"[INFO] 读取 ProductionID 文件: {production_id_file}") + + # 1. 读取 ProductionID.txt + production_ids = self._read_production_ids(production_id_file) + self._print(f"[INFO] 读取到 {len(production_ids)} 个总排号") + + # 2. 查询获取 SourceNumbers + source_numbers = self._get_source_numbers_from_production_ids(production_ids) + + # 3. 获取材料名称 + material_names = self._get_material_names_from_db(source_numbers) + else: + raise ValueError( + "必须指定 full_table=True 或提供 production_id_file 参数" + ) + + # 从数据库获取待删除物料 + self._print("\n从数据库获取待删除物料...") + db_materials = get_all_materials_to_delete() + self._print(f"获取到 {len(db_materials)} 条待删除物料记录") + + # 匹配物料 + self._print("\n匹配物料...") + results = self.match_materials(material_names, db_materials) + + # 输出结果 + self._print("\n输出结果...") + result_df = pd.DataFrame(results) + result_df.to_excel(output_file, index=False) + self._print(f"结果已保存: {output_file}") + + # 打印统计信息 + matched_count = sum(1 for r in results if r["匹配状态"] == "匹配成功") + self._print(f"\n统计信息:") + self._print(f" 总材料数: {len(results)}") + self._print(f" 匹配成功: {matched_count}") + self._print(f" 未匹配: {len(results) - matched_count}") + + return output_file