c6b97443e5
Add MaterialStatusValidator tool to check material status and match materials to be deleted: - Add get_all_materials_to_delete() function to fetch all materials from database - Add utils/material_status_validator.py with MaterialStatusValidator class - Add validate_material_status.py script to run the validation - Delete obsolete materialDelete.py file Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
49 lines
1.3 KiB
Python
49 lines
1.3 KiB
Python
"""
|
|
待删除物料查询组件
|
|
从数据库查询指定负责人需要删除的物料名称
|
|
"""
|
|
from typing import List, Dict, Any
|
|
from db.connection import get_connection
|
|
|
|
|
|
def get_materials_to_delete(manager_name):
|
|
"""
|
|
根据负责人名称查询待删除物料名称列表
|
|
|
|
Args:
|
|
manager_name: 负责人姓名
|
|
|
|
Returns:
|
|
物料名称列表（关键字）
|
|
"""
|
|
query = """
|
|
SELECT [MaterialName]
|
|
FROM [dbo].[MaterialsToBeDeleted]
|
|
WHERE [ManagerName] = ?
|
|
"""
|
|
|
|
with get_connection() as conn:
|
|
results = conn.execute_query(query, (manager_name,))
|
|
# 提取物料名称并去除空值
|
|
material_names = [row['MaterialName'] for row in results if row['MaterialName']]
|
|
return material_names
|
|
|
|
|
|
def get_all_materials_to_delete() -> List[Dict[str, Any]]:
|
|
"""
|
|
获取所有待删除物料记录
|
|
|
|
Returns:
|
|
List[Dict[str, Any]]: 包含MaterialName和ManagerName的记录列表
|
|
"""
|
|
query = """
|
|
SELECT [MaterialName], [ManagerName]
|
|
FROM [dbo].[MaterialsToBeDeleted]
|
|
WHERE [MaterialName] IS NOT NULL
|
|
ORDER BY [ManagerName], [MaterialName]
|
|
"""
|
|
|
|
with get_connection() as conn:
|
|
results = conn.execute_query(query)
|
|
return results
|