playwrite/utils/material_status_validator.py

"""
物料状态校验工具
校验订单中的物料状态，匹配待删除物料
"""
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