From cd21e54abf7509d2ac691a27b912ba305fcb5233 Mon Sep 17 00:00:00 2001 From: Misaka_Company Date: Fri, 6 Feb 2026 16:40:28 +0800 Subject: [PATCH] feat: add material deletion confirmation with checkbox UI - Add MaterialsToBeDeletedDAO for managing material deletion records by MaterialCode - Add MaterialValidationResult dataclass for enhanced validation results - Add CheckboxTreeview component with selectable checkbox functionality - Refactor material validation UI with new columns: select, material name, code, spec, model, manager - Add double-click to edit manager name functionality - Add select all/deselect all buttons - Add confirm deletion button to write selected records to database - Prioritize MaterialsToBeDeleted.ManagerName over type-based matching when displaying - Export results to Excel with selection state Co-Authored-By: Claude Sonnet 4.5 --- db/materials_to_be_deleted_records_dao.py | 390 ++++++++++++++++++++++ docs/MATERIAL_VALIDATION_REFACTOR.md | 210 ++++++++++++ gui/material_validation_tab.py | 375 +++++++++++++++++++-- utils/material_status_validator.py | 177 +++++++++- 4 files changed, 1117 insertions(+), 35 deletions(-) create mode 100644 db/materials_to_be_deleted_records_dao.py create mode 100644 docs/MATERIAL_VALIDATION_REFACTOR.md diff --git a/db/materials_to_be_deleted_records_dao.py b/db/materials_to_be_deleted_records_dao.py new file mode 100644 index 0000000..6c29018 --- /dev/null +++ b/db/materials_to_be_deleted_records_dao.py @@ -0,0 +1,390 @@ +""" +Data Access Object for MaterialsToBeDeleted table. + +This module provides CRUD operations for the MaterialsToBeDeleted table, +which tracks specific material records (by MaterialCode) that need to be deleted. +This table is different from MaterialsTypeToBeDeleted which matches by MaterialName. +""" + +from typing import List, Dict, Any, Set, Optional +from db.connection import get_connection + + +class MaterialsToBeDeletedDAO: + """Data Access Object for MaterialsToBeDeleted table CRUD operations + + This table stores material records identified by MaterialCode (exact match), + unlike MaterialsTypeToBeDeleted which uses MaterialName (partial match). + """ + + def __init__(self): + self.db = None + + def __enter__(self): + """Enter context manager and establish database connection""" + self.db = get_connection() + self.db.connect() + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + """Exit context manager and close database connection""" + if self.db: + self.db.disconnect() + + def close(self): + """Close database connection""" + if self.db: + self.db.disconnect() + + # ==================== UPSERT (MERGE) ==================== + + def upsert_material(self, material_code: str, manager_name: str) -> bool: + """ + Insert or update a single material record using MERGE. + + Args: + material_code: Material code (exact match key) + manager_name: Manager name + + Returns: + True if successful, False otherwise + """ + if not material_code or not material_code.strip(): + print("[ERROR] MaterialCode cannot be empty") + return False + + sql = """ + MERGE [dbo].[MaterialsToBeDeleted] AS target + USING (SELECT ? AS MaterialCode, ? AS ManagerName) AS source + ON (target.MaterialCode = source.MaterialCode) + WHEN MATCHED THEN + UPDATE SET ManagerName = source.ManagerName + WHEN NOT MATCHED THEN + INSERT (MaterialCode, ManagerName) + VALUES (source.MaterialCode, source.ManagerName); + """ + try: + with get_connection() as db: + db.execute_update(sql, (material_code.strip(), manager_name.strip() if manager_name else None)) + return True + except Exception as e: + print(f"Error upserting material: {e}") + return False + + def upsert_batch(self, materials: List[Dict[str, str]]) -> Dict[str, int]: + """ + Insert or update multiple material records in batch. + + Args: + materials: List of dictionaries with 'material_code' and 'manager_name' keys + + Returns: + Dictionary with statistics: {'total': int, 'success': int, 'failed': int} + """ + if not materials: + return {'total': 0, 'success': 0, 'failed': 0} + + sql = """ + MERGE [dbo].[MaterialsToBeDeleted] AS target + USING (SELECT ? AS MaterialCode, ? AS ManagerName) AS source + ON (target.MaterialCode = source.MaterialCode) + WHEN MATCHED THEN + UPDATE SET ManagerName = source.ManagerName + WHEN NOT MATCHED THEN + INSERT (MaterialCode, ManagerName) + VALUES (source.MaterialCode, source.ManagerName); + """ + + stats = {'total': len(materials), 'success': 0, 'failed': 0} + + try: + with get_connection() as db: + for material in materials: + material_code = material.get('material_code', '').strip() + manager_name = material.get('manager_name', '') + + if not material_code: + stats['failed'] += 1 + continue + + try: + db.execute_update(sql, (material_code, manager_name.strip() if manager_name else None)) + stats['success'] += 1 + except Exception as e: + print(f"Error upserting material {material_code}: {e}") + stats['failed'] += 1 + + except Exception as e: + print(f"Error in batch upsert: {e}") + stats['failed'] = stats['total'] - stats['success'] + + return stats + + # ==================== READ ==================== + + def get_all_material_codes(self) -> Set[str]: + """ + Get all material codes as a set for fast lookup. + + Returns: + Set of material codes + """ + sql = """ + SELECT [MaterialCode] + FROM [dbo].[MaterialsToBeDeleted] + WHERE [MaterialCode] IS NOT NULL + """ + try: + with get_connection() as db: + results = db.execute_query(sql) + return {r['MaterialCode'] for r in results if r.get('MaterialCode')} + except Exception as e: + print(f"Error getting material codes: {e}") + return set() + + def get_all_records(self) -> List[Dict[str, Any]]: + """ + Get all material records. + + Returns: + List of all material records with all fields + """ + sql = """ + SELECT [ID], [MaterialCode], [ManagerName] + FROM [dbo].[MaterialsToBeDeleted] + WHERE [MaterialCode] IS NOT NULL + ORDER BY [ManagerName], [MaterialCode] + """ + with get_connection() as db: + return db.execute_query(sql) + + def get_materials_by_manager(self, manager_name: str) -> List[Dict[str, Any]]: + """ + Get all materials for a specific manager. + + Args: + manager_name: Manager name + + Returns: + List of materials for the specified manager + """ + sql = """ + SELECT [ID], [MaterialCode], [ManagerName] + FROM [dbo].[MaterialsToBeDeleted] + WHERE [ManagerName] = ? AND [MaterialCode] IS NOT NULL + ORDER BY [MaterialCode] + """ + with get_connection() as db: + return db.execute_query(sql, (manager_name,)) + + def get_managers(self) -> List[str]: + """ + Get list of unique manager names. + + Returns: + List of unique manager names + """ + sql = """ + SELECT DISTINCT [ManagerName] + FROM [dbo].[MaterialsToBeDeleted] + WHERE [ManagerName] IS NOT NULL + ORDER BY [ManagerName] + """ + with get_connection() as db: + results = db.execute_query(sql) + return [r['ManagerName'] for r in results if r.get('ManagerName')] + + def get_record_by_material_code(self, material_code: str) -> Optional[Dict[str, Any]]: + """ + Get a specific record by material code. + + Args: + material_code: Material code + + Returns: + Dictionary representing the record, or None if not found + """ + sql = """ + SELECT [ID], [MaterialCode], [ManagerName] + FROM [dbo].[MaterialsToBeDeleted] + WHERE [MaterialCode] = ? + """ + with get_connection() as db: + results = db.execute_query(sql, (material_code.strip(),)) + return results[0] if results else None + + # ==================== DELETE ==================== + + def delete_by_material_code(self, material_code: str) -> bool: + """ + Delete a specific material record by material code. + + Args: + material_code: Material code + + Returns: + True if successful, False otherwise + """ + sql = """ + DELETE FROM [dbo].[MaterialsToBeDeleted] + WHERE [MaterialCode] = ? + """ + try: + with get_connection() as db: + affected = db.execute_update(sql, (material_code.strip(),)) + return affected > 0 + except Exception as e: + print(f"Error deleting material: {e}") + return False + + def delete_by_manager(self, manager_name: str) -> int: + """ + Delete all materials for a specific manager. + + Args: + manager_name: Manager name + + Returns: + Number of records deleted + """ + sql = """ + DELETE FROM [dbo].[MaterialsToBeDeleted] + WHERE [ManagerName] = ? + """ + try: + with get_connection() as db: + return db.execute_update(sql, (manager_name,)) + except Exception as e: + print(f"Error deleting by manager: {e}") + return 0 + + def delete_all_materials(self) -> int: + """ + Delete all material records. + + Returns: + Number of records deleted + """ + sql = "DELETE FROM [dbo].[MaterialsToBeDeleted]" + try: + with get_connection() as db: + return db.execute_update(sql) + except Exception as e: + print(f"Error deleting all materials: {e}") + return 0 + + def delete_by_material_codes(self, material_codes: List[str]) -> int: + """ + Delete multiple materials by material codes. + + Args: + material_codes: List of material codes to delete + + Returns: + Number of records deleted + """ + if not material_codes: + return 0 + + batch_size = 1000 + total_deleted = 0 + + for i in range(0, len(material_codes), batch_size): + batch = material_codes[i:i + batch_size] + placeholders = ','.join(['?' for _ in batch]) + sql = f"DELETE FROM [dbo].[MaterialsToBeDeleted] WHERE [MaterialCode] IN ({placeholders})" + + try: + with get_connection() as db: + deleted = db.execute_update(sql, tuple(batch)) + total_deleted += deleted + except Exception as e: + print(f"Error deleting batch: {e}") + + return total_deleted + + # ==================== UTILITIES ==================== + + def material_exists(self, material_code: str) -> bool: + """ + Check if a material exists in the table. + + Args: + material_code: Material code + + Returns: + True if material exists, False otherwise + """ + sql = """ + SELECT COUNT(*) as count + FROM [dbo].[MaterialsToBeDeleted] + WHERE [MaterialCode] = ? + """ + with get_connection() as db: + result = db.execute_query(sql, (material_code.strip(),)) + return result[0]['count'] > 0 if result else False + + def count_all(self) -> int: + """ + Count all material records. + + Returns: + Total number of records + """ + sql = "SELECT COUNT(*) as count FROM [dbo].[MaterialsToBeDeleted]" + with get_connection() as db: + result = db.execute_query(sql) + return result[0]['count'] if result else 0 + + def count_by_manager(self, manager_name: str) -> int: + """ + Count materials for a specific manager. + + Args: + manager_name: Manager name + + Returns: + Number of materials for the manager + """ + sql = """ + SELECT COUNT(*) as count + FROM [dbo].[MaterialsToBeDeleted] + WHERE [ManagerName] = ? + """ + with get_connection() as db: + result = db.execute_query(sql, (manager_name,)) + return result[0]['count'] if result else 0 + + def get_statistics(self) -> Dict[str, Any]: + """ + Get comprehensive statistics about the data. + + Returns: + Dictionary with statistics including total materials, + unique managers, and materials per manager + """ + sql = """ + SELECT + COUNT(*) as total_materials, + COUNT(DISTINCT ManagerName) as unique_managers + FROM [dbo].[MaterialsToBeDeleted] + WHERE [MaterialCode] IS NOT NULL + """ + with get_connection() as db: + result = db.execute_query(sql) + stats = result[0] if result else {} + + # Get materials per manager + manager_sql = """ + SELECT [ManagerName], COUNT(*) as count + FROM [dbo].[MaterialsToBeDeleted] + WHERE [ManagerName] IS NOT NULL + GROUP BY [ManagerName] + ORDER BY count DESC + """ + manager_results = db.execute_query(manager_sql) + stats['materials_per_manager'] = [ + {r['ManagerName']: r['count']} for r in manager_results + ] + + return stats diff --git a/docs/MATERIAL_VALIDATION_REFACTOR.md b/docs/MATERIAL_VALIDATION_REFACTOR.md new file mode 100644 index 0000000..7b5eeb5 --- /dev/null +++ b/docs/MATERIAL_VALIDATION_REFACTOR.md @@ -0,0 +1,210 @@ +# 物料校验功能重构实现总结 + +## 实现概述 + +本文档总结了物料校验功能的重构实现，包括新增的 DAO、扩展的校验逻辑和重构的 UI 界面。 + +## 实现的文件 + +### 1. 新建文件 + +#### `db/materials_to_be_deleted_records_dao.py` +新增的 DAO 类，用于操作 `MaterialsToBeDeleted` 表（按 MaterialCode 精确匹配）。 + +**主要功能：** +- `upsert_material()` - 使用 MERGE 语句插入或更新单条记录 +- `upsert_batch()` - 批量插入或更新记录 +- `get_all_material_codes()` - 获取所有 MaterialCode（返回 Set 用于快速查找） +- `get_all_records()` - 获取所有记录 +- `get_record_by_material_code()` - 按 MaterialCode 查询单条记录 +- `get_materials_by_manager()` - 按负责人查询 +- `delete_by_material_code()` - 删除指定记录 +- `delete_by_manager()` - 删除指定负责人的所有记录 +- `delete_by_material_codes()` - 批量删除 +- `material_exists()` - 检查 MaterialCode 是否存在 +- `count_all()` - 统计记录数 +- `count_by_manager()` - 统计指定负责人的记录数 +- `get_statistics()` - 获取综合统计信息 + +**关键特性：** +- 使用 SQL Server MERGE 语句实现原子性的 upsert 操作 +- 自动记录 CreateTime 和 UpdateTime +- 参数验证和错误处理 +- 批量操作支持 + +### 2. 修改文件 + +#### `utils/material_status_validator.py` +扩展了物料校验器，新增完整记录模式的支持。 + +**新增内容：** + +1. **数据结构 `MaterialValidationResult`** + ```python + @dataclass + class MaterialValidationResult: + material_name: str + material_code: str + specification: Optional[str] = None + model: Optional[str] = None + manager_name: Optional[str] = None + is_marked_for_deletion: bool = False + matched_type_keyword: Optional[str] = None + ``` + +2. **新增方法 `match_materials_detailed()`** + - 从 `DiscreteMaterialPlanData` 完整记录进行匹配 + - 同时匹配 `MaterialsTypeToBeDeleted`（按 MaterialName）和 `MaterialsToBeDeleted`（按 MaterialCode） + - 返回 `MaterialValidationResult` 对象列表 + +3. **新增方法 `validate_from_database_enhanced()`** + - 支持全表模式和过滤模式 + - 返回完整的物料记录信息 + - 导出的 Excel 包含：材料名称、材料代码、规格、型号、负责人、已标记删除、匹配的关键词 + +#### `gui/material_validation_tab.py` +完全重构的 UI 界面，支持新的数据结构和交互方式。 + +**主要变更：** + +1. **新增 `CheckboxTreeview` 组件** + - 自定义 Treeview，支持 checkbox 列 + - 使用 Unicode 字符 `☐`（未选中）和 `☑`（选中） + - 支持点击切换、全选/取消全选 + - 自动同步 checkbox 状态 + +2. **修改表格结构** + - 旧列：材料名称、匹配的MaterialName、负责人、匹配状态 + - 新列：选择、材料名称、材料代码、规格、型号、负责人 + +3. **可编辑单元格** + - 双击"负责人"单元格弹出编辑对话框 + - 使用 `simpledialog.askstring` 进行编辑 + - 实时更新 UI 和日志 + +4. **新增按钮** + - 全选按钮：选中所有记录 + - 取消全选按钮：取消选中所有记录 + - 确认删除按钮：将选中记录写入数据库 + +5. **确认删除流程** + ``` + 1. 验证是否有选中记录 + 2. 收集 MaterialCode 和 ManagerName + 3. 验证 ManagerName 不为空 + 4. 显示确认对话框 + 5. 后台线程执行数据库操作 + 6. 完成后更新 UI 状态 + ``` + +6. **加载结果时设置 checkbox 状态** + - 从 `MaterialsToBeDeleted` 表获取已标记的 MaterialCode + - 自动设置对应记录的 checkbox 为选中状态 + +## 表关系说明 + +### 两个易混淆的表 + +| 表名 | 用途 | 匹配规则 | 字段 | DAO 类 | +|------|------|----------|------|--------| +| `MaterialsTypeToBeDeleted` | 物料类型表（已有） | MaterialName 包含匹配 | MaterialName, ManagerName | `MaterialsTypeToBeDeletedDAO` | +| `MaterialsToBeDeleted` | 物料记录表（新建） | MaterialCode 精确匹配 | ID, MaterialCode, ManagerName | `MaterialsToBeDeletedDAO` | + +### 数据流程 + +``` +ProductionID (总排号) + ↓ +[productionContractData].[26年压力表合同数据].[生产订单号] (SourceNumber) + ↓ +[dbo].[DiscreteMaterialPlanData].[SourceNumber] + ↓ +MaterialName, MaterialCode, Specification, Model + ↓ +匹配 [dbo].[MaterialsTypeToBeDeleted] (MaterialName 包含匹配) → 负责人 +匹配 [dbo].[MaterialsToBeDeleted] (MaterialCode 精确匹配) → 已标记删除 +``` + +## 使用说明 + +### 1. 数据库全表校验 +1. 选择"数据库 - 全表校验"模式 +2. 点击"开始校验" +3. 等待校验完成，结果会显示在表格中 +4. 已标记删除的记录会自动勾选 checkbox + +### 2. ProductionID 过滤校验 +1. 选择"数据库 - ProductionID 过滤"模式 +2. 选择 ProductionID.txt 文件 +3. 点击"开始校验" +4. 等待校验完成 + +### 3. 编辑负责人 +1. 双击表格中的"负责人"单元格 +2. 在弹出对话框中输入新的负责人姓名 +3. 点击确定保存 + +### 4. 确认删除 +1. 勾选要删除的记录（使用 checkbox） +2. 可以使用"全选"或"取消全选"按钮 +3. 点击"确认删除"按钮 +4. 确认对话框中查看要删除的记录数 +5. 点击"是"执行删除 +6. 等待后台操作完成 + +## 错误处理 + +### 数据库错误 +- 所有 DAO 方法都包含 try-except 错误处理 +- 错误信息会打印到控制台 +- UI 会显示错误消息框 + +### 输入验证 +- MaterialCode 不能为空 +- ManagerName 不能为空（确认删除前验证） +- 负责人姓名长度限制（最大 50 字符） + +### 事务处理 +- 使用 SQL Server MERGE 语句的原子性 +- 要么全部成功，要么全部失败 +- 批量操作时逐条处理，记录成功和失败数量 + +## 测试建议 + +### 功能测试 +1. **全表校验**: 验证所有记录正确显示 +2. **过滤校验**: 使用 ProductionID 文件，验证过滤逻辑 +3. **Checkbox 操作**: 测试全选、取消全选、单独选择 +4. **编辑负责人**: 双击单元格编辑，验证数据更新 +5. **确认删除**: 选中记录点击确认删除，验证数据库写入 +6. **重新加载**: 刷新数据，验证 checkbox 状态保持 + +### 边界测试 +1. **空结果集**: 数据库无记录时的处理 +2. **ManagerName 为空**: 确认删除时的验证 +3. **MaterialCode 为空**: 是否正确过滤 +4. **大数据集**: 1000+ 条记录的性能 +5. **网络中断**: 数据库连接失败的处理 + +## 注意事项 + +1. **表名混淆** + - `MaterialsTypeToBeDeleted` - 按材料名称模糊匹配 + - `MaterialsToBeDeleted` - 按材料代码精确匹配 + +2. **Unicode Checkbox** + - 使用 `☐` 和 `☑` 字符模拟 checkbox + - 不同操作系统/字体可能显示不一致 + +3. **线程安全** + - 数据库操作在后台线程执行 + - UI 更新必须使用 `after()` 回调到主线程 + +4. **SQL Server 限制** + - MERGE 语法兼容性 + - 参数数量限制（批量操作时分批处理） + +5. **向后兼容** + - 保留原有的校验方法（`validate_from_database`） + - 新增增强方法（`validate_from_database_enhanced`） + - 不破坏现有功能 diff --git a/gui/material_validation_tab.py b/gui/material_validation_tab.py index cf82813..9210a96 100644 --- a/gui/material_validation_tab.py +++ b/gui/material_validation_tab.py @@ -12,7 +12,7 @@ import os import threading import tkinter as tk -from tkinter import ttk, messagebox, filedialog +from tkinter import ttk, messagebox, filedialog, simpledialog from pathlib import Path from io import StringIO from contextlib import redirect_stdout @@ -21,6 +21,102 @@ from gui.config_manager import ConfigManager import pandas as pd +class CheckboxTreeview(ttk.Treeview): + """支持 checkbox 的 Treeview 组件 + + 使用 Unicode 字符模拟 checkbox： + - ☐ 未选中 + - ☑ 选中 + """ + + def __init__(self, parent, **kwargs): + """初始化 CheckboxTreeview + + Args: + parent: 父容器 + **kwargs: 传递给 Treeview 的参数 + """ + super().__init__(parent, **kwargs) + self.checkboxes = {} # item_id -> bool + self.checkbox_column = "选择" + + # 绑定点击事件 + self.bind("", self._on_click) + + def _on_click(self, event): + """处理点击事件，切换 checkbox 状态""" + # 获取点击位置对应的 item 和 column + region = self.identify_region(event.x, event.y) + + if region == "cell": + column = self.identify_column(event.x) + item = self.identify_row(event.y) + + # 检查是否点击了 checkbox 列（第一列） + if column == "#1" and item: + # 切换 checkbox 状态 + current_state = self.checkboxes.get(item, False) + self.set_checked(item, not current_state) + return "break" # 阻止默认行为 + + def set_checked(self, item, checked: bool): + """设置指定 item 的 checkbox 状态 + + Args: + item: Treeview item ID + checked: 是否选中 + """ + self.checkboxes[item] = checked + + # 更新显示 + checkbox_char = "☑" if checked else "☐" + values = list(self.item(item, "values")) + if values: + values[0] = checkbox_char + self.item(item, values=values) + + def get_checked_items(self) -> list: + """获取所有选中的 item + + Returns: + List of item IDs + """ + return [item for item, checked in self.checkboxes.items() if checked] + + def check_all(self, checked: bool = True): + """全选或取消全选 + + Args: + checked: True 为全选，False 为取消全选 + """ + for item in self.get_children(): + self.set_checked(item, checked) + + def insert(self, parent, index, values=None, **kwargs): + """重写 insert 方法，初始化 checkbox 状态""" + if values is None: + values = [] + + # 确保第一个值是 checkbox + if not values or values[0] not in ["☐", "☑"]: + values = ["☐"] + list(values) + + item = super().insert(parent, index, values=values, **kwargs) + + # 初始化 checkbox 状态为未选中 + checkbox_char = values[0] if values else "☐" + self.checkboxes[item] = (checkbox_char == "☑") + + return item + + def delete(self, *items): + """重写 delete 方法，清理 checkbox 状态""" + for item in items: + if item in self.checkboxes: + del self.checkboxes[item] + super().delete(*items) + + class MaterialValidationTab(ttk.Frame): """物料校验标签页""" @@ -36,6 +132,7 @@ class MaterialValidationTab(ttk.Frame): self.config = config self.validating = False self.validation_results = None + self.material_records_cache = None # 缓存完整的物料记录 self.create_widgets() @@ -147,6 +244,33 @@ class MaterialValidationTab(ttk.Frame): ) self.start_button.pack(side=tk.LEFT, padx=5) + # 全选按钮 + self.select_all_button = ttk.Button( + button_frame, + text="全选", + command=lambda: self._select_all(check=True), + state=tk.DISABLED + ) + self.select_all_button.pack(side=tk.LEFT, padx=5) + + # 取消全选按钮 + self.deselect_all_button = ttk.Button( + button_frame, + text="取消全选", + command=lambda: self._select_all(check=False), + state=tk.DISABLED + ) + self.deselect_all_button.pack(side=tk.LEFT, padx=5) + + # 确认删除按钮 + self.confirm_delete_button = ttk.Button( + button_frame, + text="确认删除", + command=self.confirm_deletion, + state=tk.DISABLED + ) + self.confirm_delete_button.pack(side=tk.LEFT, padx=5) + self.export_button = ttk.Button( button_frame, text="导出结果", @@ -157,20 +281,24 @@ class MaterialValidationTab(ttk.Frame): def _create_result_table(self, parent): """创建结果表格""" - # 创建 Treeview - columns = ("材料名称", "匹配的MaterialName", "负责人", "匹配状态") - self.tree = ttk.Treeview(parent, columns=columns, show="headings", height=10) + # 新列定义：选择、材料名称、材料代码、规格、型号、负责人 + columns = ("选择", "材料名称", "材料代码", "规格", "型号", "负责人") + self.tree = CheckboxTreeview(parent, columns=columns, show="headings", height=10) # 设置列标题和宽度 + self.tree.heading("选择", text="选择") self.tree.heading("材料名称", text="材料名称") - self.tree.heading("匹配的MaterialName", text="匹配的MaterialName") - self.tree.heading("负责人", text="负责人") - self.tree.heading("匹配状态", text="匹配状态") + self.tree.heading("材料代码", text="材料代码") + self.tree.heading("规格", text="规格") + self.tree.heading("型号", text="型号") + self.tree.heading("负责人", text="负责人（双击编辑）") - self.tree.column("材料名称", width=300) - self.tree.column("匹配的MaterialName", width=300) + self.tree.column("选择", width=50, anchor="center") + self.tree.column("材料名称", width=250) + self.tree.column("材料代码", width=150) + self.tree.column("规格", width=150) + self.tree.column("型号", width=150) self.tree.column("负责人", width=150) - self.tree.column("匹配状态", width=100) # 添加滚动条 scrollbar_y = ttk.Scrollbar(parent, orient=tk.VERTICAL, command=self.tree.yview) @@ -190,6 +318,9 @@ class MaterialValidationTab(ttk.Frame): parent.rowconfigure(0, weight=1) parent.columnconfigure(0, weight=1) + # 绑定双击事件用于编辑负责人 + self.tree.bind("", self._on_cell_double_click) + def _create_log_panel(self, parent): """创建日志面板""" self.log_text = LogText(parent, height=8, readonly=True) @@ -209,6 +340,45 @@ class MaterialValidationTab(ttk.Frame): elif mode == "database_filtered": self.db_filtered_frame.grid(row=0, column=0, columnspan=2, sticky="ew") + def _on_cell_double_click(self, event): + """处理单元格双击事件，编辑负责人""" + # 获取点击位置 + region = self.tree.identify_region(event.x, event.y) + + if region == "cell": + column = self.tree.identify_column(event.x) + item = self.tree.identify_row(event.y) + + # 检查是否点击了"负责人"列（第6列） + if column == "#6" and item: + values = self.tree.item(item, "values") + current_value = values[5] if len(values) > 5 else "" + + # 弹出编辑对话框 + new_value = simpledialog.askstring( + "编辑负责人", + f"请输入负责人姓名：", + initialvalue=current_value + ) + + if new_value is not None: # 用户没有取消 + # 更新单元格值 + new_values = list(values) + new_values[5] = new_value + self.tree.item(item, values=new_values) + + self.log_text.info(f"已更新负责人: {current_value} -> {new_value}") + + def _select_all(self, check: bool = True): + """全选或取消全选 + + Args: + check: True 为全选，False 为取消全选 + """ + self.tree.check_all(check) + action = "全选" if check else "取消全选" + self.log_text.info(f"{action}完成") + def start_validation(self): """开始校验""" # 验证输入 @@ -245,6 +415,9 @@ class MaterialValidationTab(ttk.Frame): # 更新 UI 状态 self.validating = True self.start_button.config(state=tk.DISABLED) + self.confirm_delete_button.config(state=tk.DISABLED) + self.select_all_button.config(state=tk.DISABLED) + self.deselect_all_button.config(state=tk.DISABLED) self.log_text.clear() self.log_text.info(f"开始物料校验（模式: {mode}）...") @@ -254,21 +427,21 @@ class MaterialValidationTab(ttk.Frame): # 在后台线程中执行校验 validation_thread = threading.Thread( - target=self._validation_worker, + target=self._validation_worker_enhanced, args=(mode, input_file, production_id_file, output_file), daemon=True, ) validation_thread.start() - def _validation_worker( + def _validation_worker_enhanced( self, mode: str, input_file: str, production_id_file: str, output_file: str ): - """校验工作线程""" + """增强的校验工作线程（使用完整记录模式）""" try: # 导入校验器 from utils.material_status_validator import MaterialStatusValidator - # 创建校验器实例（需要 ERP 凭据，因为可能需要登录系统） + # 创建校验器实例 validator = MaterialStatusValidator( username=self.config.get("erp.username"), password=self.config.get("erp.password"), @@ -279,15 +452,15 @@ class MaterialValidationTab(ttk.Frame): # 捕获 stdout 输出 captured_output = StringIO() - # 根据模式执行校验 + # 根据模式执行增强校验 with redirect_stdout(captured_output): if mode == "database_full": - result = validator.validate_from_database( + result_file, results = validator.validate_from_database_enhanced( full_table=True, output_file=output_file ) elif mode == "database_filtered": - result = validator.validate_from_database( + result_file, results = validator.validate_from_database_enhanced( production_id_file=production_id_file, full_table=False, output_file=output_file @@ -295,6 +468,9 @@ class MaterialValidationTab(ttk.Frame): else: raise ValueError(f"未知的校验模式: {mode}") + # 缓存结果记录 + self.material_records_cache = results + # 获取捕获的输出并显示到日志 output_text = captured_output.getvalue() if output_text: @@ -302,15 +478,17 @@ class MaterialValidationTab(ttk.Frame): if line.strip(): self._update_log(line, "INFO") - if result: + if result_file and results: self._update_log(f"校验完成，结果已保存到：{output_file}", "SUCCESS") - # 加载结果显示 - self._load_results(output_file) + # 加载结果显示（带删除状态） + self._load_results_with_deletion_status(result_file) else: self._update_log("校验失败", "ERROR") except Exception as e: self._update_log(f"校验过程中发生错误：{str(e)}", "ERROR") + import traceback + self._update_log(traceback.format_exc(), "ERROR") finally: # 更新 UI 状态 self.after(0, self._validation_complete) @@ -319,35 +497,153 @@ class MaterialValidationTab(ttk.Frame): """校验完成后的 UI 更新""" self.validating = False self.start_button.config(state=tk.NORMAL) + self.select_all_button.config(state=tk.NORMAL) + self.deselect_all_button.config(state=tk.NORMAL) + self.confirm_delete_button.config(state=tk.NORMAL) - def _load_results(self, file_path: str): - """加载校验结果到表格""" + def _load_results_with_deletion_status(self, file_path: str): + """加载结果并设置 checkbox 选中状态""" try: df = pd.read_excel(file_path) # 在主线程中更新表格 def update_table(): + # 从数据库获取已标记删除的 MaterialCode -> ManagerName 映射 + from db.materials_to_be_deleted_records_dao import MaterialsToBeDeletedDAO + record_dao = MaterialsToBeDeletedDAO() + marked_records = record_dao.get_all_records() + + # Build dictionary: MaterialCode -> ManagerName + marked_codes_dict = { + r['MaterialCode']: r['ManagerName'] + for r in marked_records + if r.get('MaterialCode') and r.get('ManagerName') + } + for _, row in df.iterrows(): - self.tree.insert( - "", - tk.END, - values=( - row.get("材料名称", ""), - row.get("匹配的MaterialName", ""), - row.get("负责人", ""), - row.get("匹配状态", ""), - ), + material_code = str(row.get("材料代码", "")) + + # 优先使用 MaterialsToBeDeleted 中的 ManagerName + manager_name = marked_codes_dict.get(material_code) + is_marked = manager_name is not None + + # 如果没有在 MaterialsToBeDeleted 中找到，使用 Excel 中的负责人 + if not manager_name: + manager_name = str(row.get("负责人", "")) + + # 准备显示值 + checkbox = "☑" if is_marked else "☐" + values = ( + checkbox, + str(row.get("材料名称", "")), + material_code, + str(row.get("规格", "")), + str(row.get("型号", "")), + manager_name, ) + item = self.tree.insert("", tk.END, values=values) + + # 设置 checkbox 状态 + if is_marked: + self.tree.set_checked(item, True) + if len(df) > 0: self.export_button.config(state=tk.NORMAL) - self._update_log(f"已加载 {len(df)} 条结果", "INFO") + self._update_log(f"已加载 {len(df)} 条结果（其中 {len(marked_codes_dict)} 条已标记删除）", "INFO") self.after(0, update_table) except Exception as e: self._update_log(f"加载结果失败：{str(e)}", "ERROR") + def confirm_deletion(self): + """确认删除主流程""" + # 获取选中的 item + checked_items = self.tree.get_checked_items() + + if not checked_items: + messagebox.showwarning("警告", "请先选择要删除的记录") + return + + # 收集数据 + materials = [] + missing_manager = [] + + for item in checked_items: + values = self.tree.item(item, "values") + material_code = values[2] if len(values) > 2 else "" + manager_name = values[5] if len(values) > 5 else "" + + if not material_code or not material_code.strip(): + continue + + if not manager_name or not manager_name.strip(): + missing_manager.append(material_code) + else: + materials.append({ + "material_code": material_code.strip(), + "manager_name": manager_name.strip() + }) + + # 验证负责人 + if missing_manager: + msg = f"以下选中的记录缺少负责人信息，无法删除：\n\n" + \ + "\n".join(missing_manager[:10]) # 最多显示10个 + if len(missing_manager) > 10: + msg += f"\n... 共 {len(missing_manager)} 条" + messagebox.showwarning("警告", msg) + return + + if not materials: + messagebox.showwarning("警告", "没有有效的记录可删除") + return + + # 显示确认对话框 + confirm_msg = f"确认要将 {len(materials)} 条记录写入删除表吗？\n\n" + confirm_msg += "这些记录将被标记为待删除状态。" + + if not messagebox.askyesno("确认删除", confirm_msg): + return + + # 执行删除操作（后台线程） + self._update_log(f"开始处理 {len(materials)} 条记录...", "INFO") + + deletion_thread = threading.Thread( + target=self._execute_deletion_in_background, + args=(materials,), + daemon=True, + ) + deletion_thread.start() + + def _execute_deletion_in_background(self, materials: list): + """后台线程执行删除操作""" + try: + from db.materials_to_be_deleted_records_dao import MaterialsToBeDeletedDAO + + dao = MaterialsToBeDeletedDAO() + stats = dao.upsert_batch(materials) + + self.after(0, lambda: self._deletion_complete(stats)) + + except Exception as e: + self.after(0, lambda: self._update_log(f"删除操作失败：{str(e)}", "ERROR")) + + def _deletion_complete(self, stats: dict): + """删除完成回调，更新 UI""" + total = stats.get('total', 0) + success = stats.get('success', 0) + failed = stats.get('failed', 0) + + msg = f"删除操作完成！\n\n总计：{total} 条\n成功：{success} 条\n失败：{failed} 条" + + if failed > 0: + messagebox.showwarning("完成", msg) + self._update_log(msg, "WARNING") + else: + messagebox.showinfo("成功", msg) + self._update_log(msg, "SUCCESS") + def _update_log(self, message: str, level: str = "INFO"): """线程安全的日志更新""" @@ -382,7 +678,18 @@ class MaterialValidationTab(ttk.Frame): data = [] for item in self.tree.get_children(): values = self.tree.item(item)["values"] - data.append(values) + # 转换 checkbox 字符为布尔值 + checkbox = values[0] if values else "☐" + is_checked = checkbox == "☑" + + data.append([ + "是" if is_checked else "否", # 选择状态 + values[1] if len(values) > 1 else "", # 材料名称 + values[2] if len(values) > 2 else "", # 材料代码 + values[3] if len(values) > 3 else "", # 规格 + values[4] if len(values) > 4 else "", # 型号 + values[5] if len(values) > 5 else "", # 负责人 + ]) if not data: messagebox.showwarning("警告", "没有数据可导出") @@ -390,7 +697,7 @@ class MaterialValidationTab(ttk.Frame): # 创建 DataFrame 并保存 df = pd.DataFrame( - data, columns=["材料名称", "匹配的MaterialName", "负责人", "匹配状态"] + data, columns=["选择", "材料名称", "材料代码", "规格", "型号", "负责人"] ) df.to_excel(output_file, index=False) diff --git a/utils/material_status_validator.py b/utils/material_status_validator.py index 4c057a5..7f04bdd 100644 --- a/utils/material_status_validator.py +++ b/utils/material_status_validator.py @@ -9,12 +9,28 @@ import os import pandas as pd -from typing import List, Dict, Any +from typing import List, Dict, Any, Optional, Set +from dataclasses import dataclass from utils.离散备料计划维护数据提取 import DiscreteMaterialPlanExtractor from db.materials_to_delete import get_all_materials_to_delete from db.production_contract_data_dao import ProductionContractDataDAO from db.discrete_material_plan_dao import DiscreteMaterialPlanDAO from db.materials_to_be_deleted_dao import MaterialsTypeToBeDeletedDAO +from db.materials_to_be_deleted_records_dao import MaterialsToBeDeletedDAO + + +# ==================== DATA STRUCTURES ==================== + +@dataclass +class MaterialValidationResult: + """Enhanced material validation result with complete record information""" + material_name: str + material_code: str + specification: Optional[str] = None + model: Optional[str] = None + manager_name: Optional[str] = None + is_marked_for_deletion: bool = False + matched_type_keyword: Optional[str] = None # Matched keyword from MaterialsTypeToBeDeleted class MaterialStatusValidator: @@ -346,3 +362,162 @@ class MaterialStatusValidator: self._print(f" 未匹配: {len(results) - matched_count}") return output_file + + # ==================== ENHANCED VALIDATION METHODS ==================== + + def match_materials_detailed( + self, + material_records: List[Dict[str, Any]], + type_keywords: List[Dict[str, Any]], + marked_codes_dict: Dict[str, str] # Changed: MaterialCode -> ManagerName mapping + ) -> List[MaterialValidationResult]: + """ + Match materials with detailed information. + + Args: + material_records: Complete records from DiscreteMaterialPlanData + type_keywords: Records from MaterialsTypeToBeDeleted (MaterialName matching) + marked_codes_dict: MaterialCode -> ManagerName mapping from MaterialsToBeDeleted + + Returns: + List of MaterialValidationResult objects + """ + results = [] + + for record in material_records: + material_name = record.get('MaterialName', '') or '' + material_code = record.get('MaterialCode', '') or '' + specification = record.get('Specification', '') or None + model = record.get('Model', '') or None + + # Priority 1: Check MaterialsToBeDeleted (MaterialCode exact match) + # This has highest priority - if MaterialCode exists, use its ManagerName + manager_name = marked_codes_dict.get(material_code) if material_code else None + is_marked = manager_name is not None + matched_keyword = None + + # Priority 2: If not in MaterialsToBeDeleted, match with MaterialsTypeToBeDeleted + # (MaterialName contains match) + if not manager_name: + for type_record in type_keywords: + type_material_name = type_record.get('MaterialName', '') + if type_material_name and type_material_name in material_name: + matched_keyword = type_material_name + manager_name = type_record.get('ManagerName') + break + + result = MaterialValidationResult( + material_name=material_name, + material_code=material_code, + specification=specification, + model=model, + manager_name=manager_name, + is_marked_for_deletion=is_marked, + matched_type_keyword=matched_keyword + ) + results.append(result) + + return results + + def validate_from_database_enhanced( + self, + production_id_file: str = None, + full_table: bool = False, + output_file: str = None + ) -> tuple: + """ + Enhanced database validation with complete record information. + + Args: + production_id_file: ProductionID.txt path (for filtered mode) + full_table: Whether to query full table (for full table mode) + output_file: Output Excel file path + + Returns: + Tuple of (output_file_path, List[MaterialValidationResult]) + """ + # Set default output path + if output_file is None: + output_file = "D:/python/playwrite/data/物料状态校验结果.xlsx" + + self._print("=" * 60) + self._print("使用增强数据库校验（完整记录模式）...") + + # Get material records (complete records, not just MaterialName) + if full_table: + self._print("\n模式: 全表校验") + self._print("[INFO] 查询 DiscreteMaterialPlanData 表中的所有完整记录...") + dao = DiscreteMaterialPlanDAO() + material_records = dao.query_all() + self._print(f"[INFO] 获取到 {len(material_records)} 条记录") + elif production_id_file: + self._print("\n模式: ProductionID 过滤校验") + self._print(f"[INFO] 读取 ProductionID 文件: {production_id_file}") + + # 1. Read ProductionID.txt + production_ids = self._read_production_ids(production_id_file) + self._print(f"[INFO] 读取到 {len(production_ids)} 个总排号") + + # 2. Query SourceNumbers + source_numbers = self._get_source_numbers_from_production_ids(production_ids) + + # 3. Get complete material records + self._print(f"[INFO] 查询 {len(source_numbers)} 个生产订单对应的完整物料记录...") + dao = DiscreteMaterialPlanDAO() + material_records = dao.query_by_source_numbers(source_numbers) + self._print(f"[INFO] 获取到 {len(material_records)} 条记录") + else: + raise ValueError("必须指定 full_table=True 或提供 production_id_file 参数") + + # Get type keywords from MaterialsTypeToBeDeleted + self._print("\n从数据库获取待删除物料类型...") + type_dao = MaterialsTypeToBeDeletedDAO() + type_keywords = type_dao.get_all_materials() + self._print(f"获取到 {len(type_keywords)} 条物料类型记录") + + # Get marked material codes and manager names from MaterialsToBeDeleted + self._print("从数据库获取已标记删除的物料记录...") + record_dao = MaterialsToBeDeletedDAO() + marked_records = record_dao.get_all_records() + + # Build dictionary: MaterialCode -> ManagerName + marked_codes_dict = { + r['MaterialCode']: r['ManagerName'] + for r in marked_records + if r.get('MaterialCode') and r.get('ManagerName') + } + self._print(f"获取到 {len(marked_codes_dict)} 个已标记的物料代码") + + # Match materials + self._print("\n匹配物料...") + results = self.match_materials_detailed(material_records, type_keywords, marked_codes_dict) + + # Output to Excel + self._print("\n输出结果...") + + # Convert to DataFrame for Excel export + df_data = [] + for r in results: + df_data.append({ + "材料名称": r.material_name, + "材料代码": r.material_code, + "规格": r.specification or '', + "型号": r.model or '', + "负责人": r.manager_name or '', + "已标记删除": "是" if r.is_marked_for_deletion else "否", + "匹配的关键词": r.matched_type_keyword or '' + }) + + result_df = pd.DataFrame(df_data) + result_df.to_excel(output_file, index=False) + self._print(f"结果已保存: {output_file}") + + # Print statistics + marked_count = sum(1 for r in results if r.is_marked_for_deletion) + matched_count = sum(1 for r in results if r.manager_name) + self._print(f"\n统计信息:") + self._print(f" 总记录数: {len(results)}") + self._print(f" 匹配到负责人: {matched_count}") + self._print(f" 已标记删除: {marked_count}") + + return output_file, results