diff --git a/db/materials_to_delete.py b/db/materials_to_delete.py index 3f933f2..76bd622 100644 --- a/db/materials_to_delete.py +++ b/db/materials_to_delete.py @@ -2,6 +2,7 @@ 待删除物料查询组件 从数据库查询指定负责人需要删除的物料名称 """ +from typing import List, Dict, Any from db.connection import get_connection @@ -26,3 +27,22 @@ def get_materials_to_delete(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 diff --git a/materialDelete.py b/materialDelete.py deleted file mode 100644 index e69de29..0000000 diff --git a/utils/material_status_validator.py b/utils/material_status_validator.py new file mode 100644 index 0000000..a85ba2e --- /dev/null +++ b/utils/material_status_validator.py @@ -0,0 +1,186 @@ +""" +物料状态校验工具 +校验订单中的物料状态，匹配待删除物料 +""" +import os +import pandas as pd +from typing import List, Dict, Any +from utils.离散备料计划维护数据提取 import DiscreteMaterialPlanExtractor +from db.materials_to_delete import get_all_materials_to_delete + + +class MaterialStatusValidator: + """物料状态校验器""" + + def __init__(self, username, password, headless=False, verbose=True): + """ + 初始化校验器 + + Args: + username: ERP系统用户名 + password: ERP系统密码 + headless: 是否无头模式运行浏览器 + verbose: 是否打印详细日志 + """ + self.username = username + self.password = password + self.headless = headless + self.verbose = verbose + + def _print(self, *args, **kwargs): + """打印日志（如果 verbose=True）""" + if self.verbose: + print(*args, **kwargs) + + def extract_material_names(self, excel_file: str) -> List[str]: + """ + 从Excel文件的Q列提取材料名称并去重 + + Args: + excel_file: Excel文件路径 + + Returns: + List[str]: 去重后的材料名称列表 + """ + df = pd.read_excel(excel_file) + # Q列索引为16（Python从0开始） + material_names = df.iloc[:, 16].dropna().unique().tolist() + return material_names + + def match_materials(self, material_names: List[str], + db_materials: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + """ + 匹配材料名称 + + Args: + material_names: Excel中的材料名称列表 + db_materials: 数据库中的物料记录列表 + + Returns: + List[Dict]: 匹配结果 + """ + results = [] + for material_name in material_names: + matched = None + for db_record in db_materials: + # 如果数据库的MaterialName出现在Excel的材料名称中 + if db_record['MaterialName'] in material_name: + matched = db_record + break + + results.append({ + '材料名称': material_name, + '匹配的MaterialName': matched['MaterialName'] if matched else None, + '负责人': matched['ManagerName'] if matched else None, + '匹配状态': '匹配成功' if matched else '未匹配' + }) + + return results + + def validate(self, production_id_file: str, + merged_excel_file: str = None, + output_file: str = None) -> str: + """ + 执行完整的校验流程 + + Args: + production_id_file: ProductionID.txt文件路径 + merged_excel_file: 合并后的Excel文件路径（可选） + output_file: 输出文件路径（可选） + + Returns: + 输出文件路径 + """ + # 设置默认文件路径 + if merged_excel_file is None: + merged_excel_file = "D:/python/playwrite/data/离散备料计划维护_合并.xlsx" + if output_file is None: + output_file = "D:/python/playwrite/data/物料状态校验结果.xlsx" + + # 1. 调用数据提取工具 + self._print("=" * 60) + self._print("步骤1: 提取备料计划数据...") + extractor = DiscreteMaterialPlanExtractor( + username=self.username, + password=self.password, + headless=self.headless, + verbose=self.verbose + ) + extractor.extract(production_id_file, output_file=merged_excel_file) + self._print(f"数据提取完成: {merged_excel_file}") + + # 2. 提取材料名称 + self._print("\n步骤2: 提取材料名称...") + material_names = self.extract_material_names(merged_excel_file) + self._print(f"提取到 {len(material_names)} 个唯一材料名称") + + # 3. 从数据库获取待删除物料 + self._print("\n步骤3: 从数据库获取待删除物料...") + db_materials = get_all_materials_to_delete() + self._print(f"获取到 {len(db_materials)} 条待删除物料记录") + + # 4. 匹配物料 + self._print("\n步骤4: 匹配物料...") + results = self.match_materials(material_names, db_materials) + + # 5. 输出结果 + self._print("\n步骤5: 输出结果...") + result_df = pd.DataFrame(results) + result_df.to_excel(output_file, index=False) + self._print(f"结果已保存: {output_file}") + + # 打印统计信息 + matched_count = sum(1 for r in results if r['匹配状态'] == '匹配成功') + self._print(f"\n统计信息:") + self._print(f" 总材料数: {len(results)}") + self._print(f" 匹配成功: {matched_count}") + self._print(f" 未匹配: {len(results) - matched_count}") + + return output_file + + def validate_from_existing_excel(self, excel_file: str, output_file: str = None) -> str: + """ + 从已存在的Excel文件执行校验（不需要重新提取数据） + + Args: + excel_file: 已存在的Excel文件路径 + output_file: 输出文件路径（可选） + + Returns: + 输出文件路径 + """ + # 设置默认输出路径 + if output_file is None: + output_file = "D:/python/playwrite/data/物料状态校验结果.xlsx" + + self._print("=" * 60) + self._print("从已存在的Excel文件执行校验...") + + # 1. 提取材料名称 + self._print("\n步骤1: 提取材料名称...") + material_names = self.extract_material_names(excel_file) + self._print(f"提取到 {len(material_names)} 个唯一材料名称") + + # 2. 从数据库获取待删除物料 + self._print("\n步骤2: 从数据库获取待删除物料...") + db_materials = get_all_materials_to_delete() + self._print(f"获取到 {len(db_materials)} 条待删除物料记录") + + # 3. 匹配物料 + self._print("\n步骤3: 匹配物料...") + results = self.match_materials(material_names, db_materials) + + # 4. 输出结果 + self._print("\n步骤4: 输出结果...") + result_df = pd.DataFrame(results) + result_df.to_excel(output_file, index=False) + self._print(f"结果已保存: {output_file}") + + # 打印统计信息 + matched_count = sum(1 for r in results if r['匹配状态'] == '匹配成功') + self._print(f"\n统计信息:") + self._print(f" 总材料数: {len(results)}") + self._print(f" 匹配成功: {matched_count}") + self._print(f" 未匹配: {len(results) - matched_count}") + + return output_file diff --git a/validate_material_status.py b/validate_material_status.py new file mode 100644 index 0000000..da5a1dd --- /dev/null +++ b/validate_material_status.py @@ -0,0 +1,48 @@ +""" +物料状态校验脚本 +校验订单中的物料状态，匹配待删除物料 +""" +import os +import sys + +# 添加项目根目录到 sys.path +project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if project_root not in sys.path: + sys.path.insert(0, project_root) + +from utils.material_status_validator import MaterialStatusValidator + + +def main(): + """主函数""" + # 创建校验器实例 + validator = MaterialStatusValidator( + username="BLDpengqiangqiang", + password="Cqbld123456.", + headless=True, + verbose=True + ) + + # 设置文件路径 + production_id_file = "D:/python/playwrite/ProductionID.txt" + output_file = "D:/python/playwrite/data/物料状态校验结果.xlsx" + + # 执行校验 + print("=" * 60) + print("物料状态校验工具") + print("=" * 60) + + # 方式1: 从头开始（提取数据 + 校验） + validator.validate(production_id_file, output_file=output_file) + + # 方式2: 如果已经有Excel文件，可以直接校验 + # excel_file = "D:/python/playwrite/data/离散备料计划维护_合并.xlsx" + # validator.validate_from_existing_excel(excel_file, output_file=output_file) + + print("\n" + "=" * 60) + print("校验完成！") + print("=" * 60) + + +if __name__ == "__main__": + main()