7819570ef3
- 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>
140 lines
4.0 KiB
Python
140 lines
4.0 KiB
Python
"""
|
||
生产订单号查询组件
|
||
从 ProductionID.txt 读取总排号或生产订单号，智能识别并处理
|
||
"""
|
||
|
||
import re
|
||
from db.connection import get_connection
|
||
from config.schema import DatabaseType
|
||
from config.loader import ConfigLoader
|
||
|
||
|
||
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(file_path):
|
||
"""
|
||
读取输入文件，获取输入项列表
|
||
|
||
Args:
|
||
file_path: 输入文件路径
|
||
|
||
Returns:
|
||
输入项列表（可能是总排号或生产订单号）
|
||
"""
|
||
with open(file_path, "r", encoding="utf-8") as f:
|
||
# 去除空白行和空格
|
||
items = [line.strip() for line in f if line.strip()]
|
||
return items
|
||
|
||
|
||
def _query_order_numbers_from_db(production_ids, db_type):
|
||
"""
|
||
根据总排号列表从数据库查询生产订单号（内部函数）
|
||
|
||
Args:
|
||
production_ids: 总排号列表
|
||
db_type: 数据库类型
|
||
|
||
Returns:
|
||
生产订单号列表
|
||
"""
|
||
if not production_ids:
|
||
return []
|
||
|
||
# SQL Server 限制每个查询最多 2100 个参数
|
||
BATCH_SIZE = 2000
|
||
all_results = []
|
||
|
||
# 分批查询
|
||
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()
|
||
placeholders = ",".join([placeholder for _ in batch])
|
||
|
||
# 根据数据库类型选择表名和列名格式
|
||
if db_type == DatabaseType.MYSQL:
|
||
table_name = "productionContractData_26年压力表合同数据"
|
||
query = f"""
|
||
SELECT 生产订单号
|
||
FROM {table_name}
|
||
WHERE 总排号 IN ({placeholders})
|
||
"""
|
||
else:
|
||
table_name = "[productionContractData].[26年压力表合同数据]"
|
||
query = f"""
|
||
SELECT [生产订单号]
|
||
FROM {table_name}
|
||
WHERE [总排号] IN ({placeholders})
|
||
"""
|
||
|
||
results = conn.execute_query(query, tuple(batch))
|
||
# 提取生产订单号并去除空值
|
||
batch_numbers = [row["生产订单号"] for row in results if row.get("生产订单号")]
|
||
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
|