""" 物料状态校验工具 校验订单中的物料状态，匹配待删除物料 支持两种数据源： 1. Excel 文件（原有方式） 2. 数据库驱动（新增方式） 支持两种输入格式： 1. productionID（总排号）: 2位数字 + 1位字母 + 流水号 (如 25A1, 25A12345) 2. 生产订单号: SC + 14位数字 (如 SC00000000000001) """ import os import re import pandas as pd from typing import List, Dict, Any, Optional, Set from dataclasses import dataclass from utils.discrete_material_plan_extractor 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 MaterialsTypeToBeDeletedDAO from db.materials_to_be_deleted_records_dao import MaterialsToBeDeletedDAO # ==================== DATA STRUCTURES ==================== @dataclass class MaterialValidationResult: """Enhanced material validation result with complete record information""" material_name: str material_code: str specification: Optional[str] = None model: Optional[str] = None manager_name: Optional[str] = None is_marked_for_deletion: bool = False matched_type_keyword: Optional[str] = ( None # Matched keyword from MaterialsTypeToBeDeleted ) class MaterialStatusValidator: """物料状态校验器""" def __init__(self, username, password, headless=False, verbose=True): """ 初始化校验器 Args: username: ERP系统用户名 password: ERP系统密码 headless: 是否无头模式运行浏览器 verbose: 是否打印详细日志 """ self.username = username self.password = password self.headless = headless self.verbose = verbose def _print(self, *args, **kwargs): """打印日志（如果 verbose=True）""" if self.verbose: print(*args, **kwargs) def extract_material_names(self, excel_file: str) -> List[str]: """ 从Excel文件的Q列提取材料名称并去重 Args: excel_file: Excel文件路径 Returns: List[str]: 去重后的材料名称列表 """ df = pd.read_excel(excel_file) # R列索引为17（Python从0开始） material_names = df.iloc[:, 17].dropna().unique().tolist() # 确保所有元素都是字符串类型 material_names = [str(name) for name in material_names] return material_names def match_materials( self, material_names: List[str], db_materials: List[Dict[str, Any]] ) -> List[Dict[str, Any]]: """ 匹配材料名称 Args: material_names: Excel中的材料名称列表 db_materials: 数据库中的物料记录列表 Returns: List[Dict]: 匹配结果 """ results = [] for material_name in material_names: matched = None for db_record in db_materials: # 如果数据库的MaterialName出现在Excel的材料名称中 if db_record["MaterialName"] in material_name: matched = db_record break results.append( { "材料名称": material_name, "匹配的MaterialName": matched["MaterialName"] if matched else None, "负责人": matched["ManagerName"] if matched else None, "匹配状态": "匹配成功" if matched else "未匹配", } ) return results def validate( self, production_id_file: str, merged_excel_file: str = None, output_file: str = None, ) -> str: """ 执行完整的校验流程 Args: production_id_file: ProductionID.txt文件路径 merged_excel_file: 合并后的Excel文件路径（可选） output_file: 输出文件路径（可选） Returns: 输出文件路径 """ # 设置默认文件路径 if merged_excel_file is None: merged_excel_file = "D:/python/playwrite/data/离散备料计划维护_合并.xlsx" if output_file is None: output_file = "D:/python/playwrite/data/物料状态校验结果.xlsx" # 1. 调用数据提取工具 self._print("=" * 60) self._print("步骤1: 提取备料计划数据...") extractor = DiscreteMaterialPlanExtractor( username=self.username, password=self.password, headless=self.headless, verbose=self.verbose, ) extractor.extract(production_id_file, output_file=merged_excel_file) self._print(f"数据提取完成: {merged_excel_file}") # 2. 提取材料名称 self._print("\n步骤2: 提取材料名称...") material_names = self.extract_material_names(merged_excel_file) self._print(f"提取到 {len(material_names)} 个唯一材料名称") # 3. 从数据库获取待删除物料 self._print("\n步骤3: 从数据库获取待删除物料...") db_materials = get_all_materials_to_delete() self._print(f"获取到 {len(db_materials)} 条待删除物料记录") # 4. 匹配物料 self._print("\n步骤4: 匹配物料...") results = self.match_materials(material_names, db_materials) # 5. 输出结果 self._print("\n步骤5: 输出结果...") 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 def validate_from_existing_excel( self, excel_file: str, output_file: str = None ) -> str: """ 从已存在的Excel文件执行校验（不需要重新提取数据） Args: excel_file: 已存在的Excel文件路径 output_file: 输出文件路径（可选） Returns: 输出文件路径 """ # 设置默认输出路径 if output_file is None: output_file = "D:/python/playwrite/data/物料状态校验结果.xlsx" self._print("=" * 60) self._print("从已存在的Excel文件执行校验...") # 1. 提取材料名称 self._print("\n步骤1: 提取材料名称...") material_names = self.extract_material_names(excel_file) self._print(f"提取到 {len(material_names)} 个唯一材料名称") # 2. 从数据库获取待删除物料 self._print("\n步骤2: 从数据库获取待删除物料...") db_materials = get_all_materials_to_delete() self._print(f"获取到 {len(db_materials)} 条待删除物料记录") # 3. 匹配物料 self._print("\n步骤3: 匹配物料...") results = self.match_materials(material_names, db_materials) # 4. 输出结果 self._print("\n步骤4: 输出结果...") 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 # ==================== DATABASE-DRIVEN VALIDATION ==================== @staticmethod def _identify_input_type(input_str: str) -> str: """ 识别输入字符串的类型 Args: input_str: 输入字符串 Returns: "production_id": 总排号格式 (2位数字 + 1位字母 + 流水号) "order_number": 生产订单号格式 (SC + 14位数字) "unknown": 无法识别 """ input_str = input_str.strip() # 生产订单号: SC + 14位数字 if re.match(r"^SC\d{14}$", input_str): return "order_number" # 总排号: 2位数字 + 1位字母 + 流水号(1-6位数字) if re.match(r"^\d{2}[A-Za-z]\d{1,6}$", input_str): return "production_id" return "unknown" 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: items = [line.strip() for line in f if line.strip()] return items def _get_source_numbers_from_inputs(self, inputs: List[str]) -> List[str]: """ 根据输入列表智能获取 SourceNumber（生产订单号）列表 对于 productionID（总排号）：查询数据库获取生产订单号 对于生产订单号：直接使用 Args: inputs: 输入项列表（可能是总排号或生产订单号） Returns: List[str]: 生产订单号列表 """ production_ids = [] # 需要查询数据库的 order_numbers = [] # 直接使用的 for item in inputs: input_type = self._identify_input_type(item) if input_type == "order_number": order_numbers.append(item) elif input_type == "production_id": production_ids.append(item) # 统计输入类型 if production_ids: self._print(f"[INFO] 识别到 {len(production_ids)} 个总排号") if order_numbers: self._print(f"[INFO] 识别到 {len(order_numbers)} 个生产订单号") # 查询数据库获取总排号对应的生产订单号 if production_ids: self._print( f"[INFO] 正在查询 {len(production_ids)} 个总排号对应的生产订单号..." ) contract_dao = ProductionContractDataDAO() db_order_numbers = contract_dao.get_source_numbers_by_总排号(production_ids) self._print(f"[INFO] 从数据库获取到 {len(db_order_numbers)} 个生产订单号") order_numbers.extend(db_order_numbers) # 去重 unique_order_numbers = list(dict.fromkeys(order_numbers)) if len(unique_order_numbers) != len(order_numbers): self._print( f"[INFO] 去重后得到 {len(unique_order_numbers)} 个唯一生产订单号" ) return unique_order_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. 输入过滤校验 (production_id_file 指定): 基于输入文件过滤 - 支持总排号格式 (如 25A1, 25A12345) - 支持生产订单号格式 (如 SC00000000000001) - 支持混合输入 查询链路（模式2）: 输入文件 (总排号或生产订单号) -> 总排号需查询: productionContractData.26年压力表合同数据.生产订单号 (SourceNumber) -> 生产订单号直接使用 -> DiscreteMaterialPlanData.SourceNumber -> DiscreteMaterialPlanData.MaterialName -> 对比 MaterialsTypeToBeDeleted.MaterialName Args: production_id_file: 输入文件路径（模式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模式: 输入过滤校验") self._print(f"[INFO] 读取输入文件: {production_id_file}") # 1. 读取输入文件 inputs = self._read_production_ids(production_id_file) self._print(f"[INFO] 读取到 {len(inputs)} 个输入项") # 2. 智能识别并获取 SourceNumbers source_numbers = self._get_source_numbers_from_inputs(inputs) # 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 # ==================== ENHANCED VALIDATION METHODS ==================== def match_materials_detailed( self, material_records: List[Dict[str, Any]], type_keywords: List[Dict[str, Any]], marked_codes_dict: Dict[ str, str ], # Changed: MaterialCode -> ManagerName mapping ) -> List[MaterialValidationResult]: """ Match materials with detailed information. Args: material_records: Complete records from DiscreteMaterialPlanData type_keywords: Records from MaterialsTypeToBeDeleted (MaterialName matching) marked_codes_dict: MaterialCode -> ManagerName mapping from MaterialsToBeDeleted Returns: List of MaterialValidationResult objects """ results = [] for record in material_records: material_name = record.get("MaterialName", "") or "" material_code = record.get("MaterialCode", "") or "" specification = record.get("Specification", "") or None model = record.get("Model", "") or None # Priority 1: Check MaterialsToBeDeleted (MaterialCode exact match) # This has highest priority - if MaterialCode exists, use its ManagerName manager_name = ( marked_codes_dict.get(material_code) if material_code else None ) is_marked = manager_name is not None matched_keyword = None # Priority 2: If not in MaterialsToBeDeleted, match with MaterialsTypeToBeDeleted # (MaterialName contains match) if not manager_name: for type_record in type_keywords: type_material_name = type_record.get("MaterialName", "") if type_material_name and type_material_name in material_name: matched_keyword = type_material_name manager_name = type_record.get("ManagerName") break result = MaterialValidationResult( material_name=material_name, material_code=material_code, specification=specification, model=model, manager_name=manager_name, is_marked_for_deletion=is_marked, matched_type_keyword=matched_keyword, ) results.append(result) return results def validate_from_database_enhanced( self, production_id_file: str = None, full_table: bool = False, output_file: str = None, ) -> tuple: """ Enhanced database validation with complete record information. 支持两种模式： 1. 全表校验 (full_table=True): 查询整个 DiscreteMaterialPlanData 表 2. 输入过滤校验 (production_id_file 指定): 基于输入文件过滤 - 支持总排号格式 (如 25A1, 25A12345) - 支持生产订单号格式 (如 SC00000000000001) - 支持混合输入 Args: production_id_file: 输入文件路径（模式2） full_table: 是否全表校验（模式1） output_file: 输出 Excel 文件路径 Returns: Tuple of (output_file_path, List[MaterialValidationResult]) """ # Set default output path if output_file is None: output_file = "D:/python/playwrite/data/物料状态校验结果.xlsx" self._print("=" * 60) self._print("使用增强数据库校验（完整记录模式）...") # Get material records (complete records, not just MaterialName) if full_table: self._print("\n模式: 全表校验") self._print( "[INFO] 查询 DiscreteMaterialPlanData 表中的所有完整记录（启用 MaterialCode 去重）..." ) dao = DiscreteMaterialPlanDAO() # Get original count for deduplication statistics original_count = dao.count_all() material_records = dao.query_all_distinct_by_material_code() dedup_count = original_count - len(material_records) self._print(f"[INFO] 获取到 {len(material_records)} 条记录") if dedup_count > 0: self._print( f"[INFO] 基于 MaterialCode 去重：移除了 {dedup_count} 条重复记录" ) elif production_id_file: self._print("\n模式: 输入过滤校验") self._print(f"[INFO] 读取输入文件: {production_id_file}") # 1. Read input file inputs = self._read_production_ids(production_id_file) self._print(f"[INFO] 读取到 {len(inputs)} 个输入项") # 2. Smart identify and get SourceNumbers source_numbers = self._get_source_numbers_from_inputs(inputs) if not source_numbers: self._print("\n[ERROR] 校验失败：未找到有效的生产订单号") self._print("[ERROR] 可能原因：") self._print("[ERROR] 1. 总排号在数据库中不存在对应的生产订单号") self._print("[ERROR] 2. 输入的生产订单号格式不正确") self._print("[ERROR] 3. 请检查输入文件内容") return output_file, [] # 3. Get complete material records with deduplication self._print( f"[INFO] 查询 {len(source_numbers)} 个生产订单对应的完整物料记录（启用 MaterialCode 去重）..." ) dao = DiscreteMaterialPlanDAO() # Get original count for deduplication statistics original_records = dao.query_by_source_numbers(source_numbers) material_records = dao.query_by_source_numbers_distinct(source_numbers) dedup_count = len(original_records) - len(material_records) self._print(f"[INFO] 获取到 {len(material_records)} 条记录") if dedup_count > 0: self._print( f"[INFO] 基于 MaterialCode 去重：移除了 {dedup_count} 条重复记录" ) # 如果没有找到物料记录，给出友好提示 if not material_records: self._print("\n[ERROR] 校验失败：未找到物料记录") self._print("[ERROR] 可能原因：") self._print("[ERROR] 1. 这些生产订单的物料数据还没有提取到数据库") self._print("[ERROR] 2. 请先运行【正式备料计划数据提取】工具") self._print("[ERROR] 3. 提取时勾选【持久化到数据库】选项") self._print( f"[ERROR] 4. 将这些输入项的物料数据保存到 DiscreteMaterialPlanData 表" ) else: raise ValueError("必须指定 full_table=True 或提供 production_id_file 参数") # Get type keywords from MaterialsTypeToBeDeleted self._print("\n从数据库获取待删除物料类型...") type_dao = MaterialsTypeToBeDeletedDAO() type_keywords = type_dao.get_all_materials() self._print(f"获取到 {len(type_keywords)} 条物料类型记录") # Get marked material codes and manager names from MaterialsToBeDeleted self._print("从数据库获取已标记删除的物料记录...") record_dao = MaterialsToBeDeletedDAO() 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") } self._print(f"获取到 {len(marked_codes_dict)} 个已标记的物料代码") # Match materials self._print("\n匹配物料...") results = self.match_materials_detailed( material_records, type_keywords, marked_codes_dict ) # Output to Excel self._print("\n输出结果...") # Convert to DataFrame for Excel export df_data = [] for r in results: df_data.append( { "材料名称": r.material_name, "材料代码": r.material_code, "规格": r.specification or "", "型号": r.model or "", "负责人": r.manager_name or "", "已标记删除": "是" if r.is_marked_for_deletion else "否", "匹配的关键词": r.matched_type_keyword or "", } ) result_df = pd.DataFrame(df_data) result_df.to_excel(output_file, index=False) self._print(f"结果已保存: {output_file}") # Print statistics marked_count = sum(1 for r in results if r.is_marked_for_deletion) matched_count = sum(1 for r in results if r.manager_name) self._print(f"\n统计信息:") self._print(f" 总记录数: {len(results)}") self._print(f" 匹配到负责人: {matched_count}") self._print(f" 已标记删除: {marked_count}") return output_file, results