admin/playwrite
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: support both productionID and order number input formats

Browse Source 
- 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) <noreply@anthropic.com>
This commit is contained in:
Misaka
2026-02-25 22:29:51 +08:00
parent 0a2c3c55c6
commit 7819570ef3
2 changed files with 183 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

144
utils/material_status_validator.py
View File

@@ -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 参数")