From 7819570ef3b1cac25b04b3248845f75799cef45e Mon Sep 17 00:00:00 2001 From: Misaka Date: Wed, 25 Feb 2026 22:29:51 +0800 Subject: [PATCH] feat: support both productionID and order number input formats - Add identify_input_type() function to detect input format: - production_id: 2-digit + 1-letter + serial (e.g., 25A1, 25A12345) - order_number: SC + 14 digits (e.g., SC70202602120120) - Refactor MaterialStatusValidator to intelligently handle both input types in validate_from_database() and validate_from_database_enhanced() - Update log messages to reflect the new dual-input capability Co-Authored-By: Claude (glm-5) --- db/production_order_query.py | 89 +++++++++++++++--- utils/material_status_validator.py | 144 +++++++++++++++++++++-------- 2 files changed, 183 insertions(+), 50 deletions(-) diff --git a/db/production_order_query.py b/db/production_order_query.py index 168d4bd..98355ea 100644 --- a/db/production_order_query.py +++ b/db/production_order_query.py @@ -1,35 +1,62 @@ """ 生产订单号查询组件 -从 ProductionID.txt 读取总排号，查询数据库获取生产订单号 +从 ProductionID.txt 读取总排号或生产订单号，智能识别并处理 """ +import re from db.connection import get_connection from config.schema import DatabaseType from config.loader import ConfigLoader -def read_production_ids(file_path): +def identify_input_type(input_str: str) -> str: """ - 读取 ProductionID.txt 文件，获取总排号列表 + 识别输入字符串的类型 Args: - file_path: ProductionID.txt 文件路径 + 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(file_path): + """ + 读取输入文件，获取输入项列表 + + Args: + file_path: 输入文件路径 + + Returns: + 输入项列表（可能是总排号或生产订单号） """ with open(file_path, "r", encoding="utf-8") as f: # 去除空白行和空格 - production_ids = [line.strip() for line in f if line.strip()] - return production_ids + items = [line.strip() for line in f if line.strip()] + return items -def query_production_order_numbers(production_ids): +def _query_order_numbers_from_db(production_ids, db_type): """ - 根据总排号列表，从数据库查询生产订单号 + 根据总排号列表从数据库查询生产订单号（内部函数） Args: production_ids: 总排号列表 + db_type: 数据库类型 Returns: 生产订单号列表 @@ -37,10 +64,6 @@ def query_production_order_numbers(production_ids): if not production_ids: return [] - # 获取当前数据库类型 - app_config = ConfigLoader.load() - db_type = app_config.database.db_type - # SQL Server 限制每个查询最多 2100 个参数 BATCH_SIZE = 2000 all_results = [] @@ -48,7 +71,7 @@ def query_production_order_numbers(production_ids): # 分批查询 for i in range(0, len(production_ids), BATCH_SIZE): batch = production_ids[i : i + BATCH_SIZE] - + with get_connection() as conn: # 获取正确的占位符 placeholder = conn.get_placeholder() @@ -76,3 +99,41 @@ def query_production_order_numbers(production_ids): all_results.extend(batch_numbers) return all_results + + +def query_production_order_numbers(inputs): + """ + 根据输入列表，智能处理并返回生产订单号列表 + + 对于 productionID（总排号）：查询数据库获取生产订单号 + 对于生产订单号：直接使用 + + Args: + inputs: 输入项列表（可能是总排号或生产订单号） + + Returns: + 生产订单号列表 + """ + if not inputs: + return [] + + production_ids = [] # 需要查询数据库的 + order_numbers = [] # 直接使用的 + + for item in inputs: + input_type = identify_input_type(item) + if input_type == "order_number": + order_numbers.append(item) + elif input_type == "production_id": + production_ids.append(item) + + # 获取当前数据库类型 + app_config = ConfigLoader.load() + db_type = app_config.database.db_type + + # 查询数据库获取总排号对应的生产订单号 + if production_ids: + db_order_numbers = _query_order_numbers_from_db(production_ids, db_type) + order_numbers.extend(db_order_numbers) + + return order_numbers diff --git a/utils/material_status_validator.py b/utils/material_status_validator.py index 87d707f..43dc165 100644 --- a/utils/material_status_validator.py +++ b/utils/material_status_validator.py @@ -5,9 +5,14 @@ 支持两种数据源： 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 @@ -221,6 +226,31 @@ class MaterialStatusValidator: # ==================== 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 文件 @@ -229,34 +259,57 @@ class MaterialStatusValidator: production_id_file: ProductionID.txt 文件路径 Returns: - List[str]: 总排号列表 + 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 + items = [line.strip() for line in f if line.strip()] + return items - def _get_source_numbers_from_production_ids( - self, production_ids: List[str] + def _get_source_numbers_from_inputs( + self, inputs: List[str] ) -> List[str]: """ - 通过 ProductionID 查询获取 SourceNumber 列表 + 根据输入列表智能获取 SourceNumber（生产订单号）列表 - 查询链路: - ProductionID (总排号) -> productionContractData.26年压力表合同数据.生产订单号 + 对于 productionID（总排号）：查询数据库获取生产订单号 + 对于生产订单号：直接使用 Args: - production_ids: 总排号列表 + inputs: 输入项列表（可能是总排号或生产订单号） Returns: List[str]: 生产订单号列表 """ - self._print(f"[INFO] 正在查询 {len(production_ids)} 个总排号对应的生产订单号...") + production_ids = [] # 需要查询数据库的 + order_numbers = [] # 直接使用的 - contract_dao = ProductionContractDataDAO() - source_numbers = contract_dao.get_source_numbers_by_总排号(production_ids) + 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) - self._print(f"[INFO] 找到 {len(source_numbers)} 个唯一的生产订单号") - return source_numbers + # 统计输入类型 + 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 @@ -292,17 +345,21 @@ class MaterialStatusValidator: 支持两种模式： 1. 全表校验 (full_table=True): 查询整个 DiscreteMaterialPlanData 表 - 2. ProductionID 过滤校验 (production_id_file 指定): 基于 ProductionID.txt 文件过滤 + 2. 输入过滤校验 (production_id_file 指定): 基于输入文件过滤 + - 支持总排号格式 (如 25A1, 25A12345) + - 支持生产订单号格式 (如 SC00000000000001) + - 支持混合输入 查询链路（模式2）: - ProductionID.txt (总排号) - -> productionContractData.26年压力表合同数据.生产订单号 (SourceNumber) + 输入文件 (总排号或生产订单号) + -> 总排号需查询: productionContractData.26年压力表合同数据.生产订单号 (SourceNumber) + -> 生产订单号直接使用 -> DiscreteMaterialPlanData.SourceNumber -> DiscreteMaterialPlanData.MaterialName -> 对比 MaterialsTypeToBeDeleted.MaterialName Args: - production_id_file: ProductionID.txt 路径（模式2） + production_id_file: 输入文件路径（模式2） full_table: 是否全表校验（模式1） output_file: 输出文件路径 @@ -322,15 +379,15 @@ class MaterialStatusValidator: 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}") + self._print("\n模式: 输入过滤校验") + self._print(f"[INFO] 读取输入文件: {production_id_file}") - # 1. 读取 ProductionID.txt - production_ids = self._read_production_ids(production_id_file) - self._print(f"[INFO] 读取到 {len(production_ids)} 个总排号") + # 1. 读取输入文件 + inputs = self._read_production_ids(production_id_file) + self._print(f"[INFO] 读取到 {len(inputs)} 个输入项") - # 2. 查询获取 SourceNumbers - source_numbers = self._get_source_numbers_from_production_ids(production_ids) + # 2. 智能识别并获取 SourceNumbers + source_numbers = self._get_source_numbers_from_inputs(inputs) # 3. 获取材料名称 material_names = self._get_material_names_from_db(source_numbers) @@ -428,10 +485,17 @@ class MaterialStatusValidator: """ Enhanced database validation with complete record information. + 支持两种模式： + 1. 全表校验 (full_table=True): 查询整个 DiscreteMaterialPlanData 表 + 2. 输入过滤校验 (production_id_file 指定): 基于输入文件过滤 + - 支持总排号格式 (如 25A1, 25A12345) + - 支持生产订单号格式 (如 SC00000000000001) + - 支持混合输入 + Args: - production_id_file: ProductionID.txt path (for filtered mode) - full_table: Whether to query full table (for full table mode) - output_file: Output Excel file path + production_id_file: 输入文件路径（模式2） + full_table: 是否全表校验（模式1） + output_file: 输出 Excel 文件路径 Returns: Tuple of (output_file_path, List[MaterialValidationResult]) @@ -459,15 +523,23 @@ class MaterialStatusValidator: if dedup_count > 0: self._print(f"[INFO] 基于 MaterialCode 去重：移除了 {dedup_count} 条重复记录") elif production_id_file: - self._print("\n模式: ProductionID 过滤校验") - self._print(f"[INFO] 读取 ProductionID 文件: {production_id_file}") + self._print("\n模式: 输入过滤校验") + self._print(f"[INFO] 读取输入文件: {production_id_file}") - # 1. Read ProductionID.txt - production_ids = self._read_production_ids(production_id_file) - self._print(f"[INFO] 读取到 {len(production_ids)} 个总排号") + # 1. Read input file + inputs = self._read_production_ids(production_id_file) + self._print(f"[INFO] 读取到 {len(inputs)} 个输入项") - # 2. Query SourceNumbers - source_numbers = self._get_source_numbers_from_production_ids(production_ids) + # 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 去重）...") @@ -491,7 +563,7 @@ class MaterialStatusValidator: self._print("[ERROR] 1. 这些生产订单的物料数据还没有提取到数据库") self._print("[ERROR] 2. 请先运行【正式备料计划数据提取】工具") self._print("[ERROR] 3. 提取时勾选【持久化到数据库】选项") - self._print("[ERROR] 4. 将这 10 个 ProductionID 的物料数据保存到 DiscreteMaterialPlanData 表") + self._print(f"[ERROR] 4. 将这些输入项的物料数据保存到 DiscreteMaterialPlanData 表") else: raise ValueError("必须指定 full_table=True 或提供 production_id_file 参数")