cff5e78c1d
- Change from fuzzy material name matching to exact material code matching - Migrate from MaterialsTypeToBeDeleted to MaterialsToBeDeleted table - Add should_delete_material() for on-demand database queries - Remove in-memory material list loading to reduce memory footprint - Update documentation to reflect new matching logic - Remove main scripts from git tracking (contain user-specific config) Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
75 lines
2.0 KiB
Python
75 lines
2.0 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 [MaterialCode]
|
||
FROM [dbo].[MaterialsToBeDeleted]
|
||
WHERE [ManagerName] = ?
|
||
"""
|
||
|
||
with get_connection() as conn:
|
||
results = conn.execute_query(query, (manager_name,))
|
||
# 提取物料编码并去除空值
|
||
material_codes = [row["MaterialCode"] for row in results if row["MaterialCode"]]
|
||
return material_codes
|
||
|
||
|
||
def get_all_materials_to_delete() -> List[Dict[str, Any]]:
|
||
"""
|
||
获取所有待删除物料记录
|
||
|
||
Returns:
|
||
List[Dict[str, Any]]: 包含MaterialCode和ManagerName的记录列表
|
||
"""
|
||
query = """
|
||
SELECT [MaterialCode], [ManagerName]
|
||
FROM [dbo].[MaterialsToBeDeleted]
|
||
WHERE [MaterialCode] IS NOT NULL
|
||
ORDER BY [ManagerName], [MaterialCode]
|
||
"""
|
||
|
||
with get_connection() as conn:
|
||
results = conn.execute_query(query)
|
||
return results
|
||
|
||
|
||
def should_delete_material(manager_name: str, material_code: str) -> bool:
|
||
"""
|
||
检查指定物料编码是否需要删除
|
||
|
||
直接在数据库层面查询，避免将大量数据加载到内存
|
||
|
||
Args:
|
||
manager_name: 负责人姓名
|
||
material_code: 物料编码
|
||
|
||
Returns:
|
||
bool: 如果物料需要删除返回 True，否则返回 False
|
||
"""
|
||
query = """
|
||
SELECT COUNT(*) as count
|
||
FROM [dbo].[MaterialsToBeDeleted]
|
||
WHERE [ManagerName] = ?
|
||
AND [MaterialCode] = ?
|
||
"""
|
||
|
||
with get_connection() as conn:
|
||
results = conn.execute_query(query, (manager_name, material_code))
|
||
return results[0]["count"] > 0 if results else False
|