2fb44a2a9b
- Add MaterialTypeManagementDialog for managing MaterialsTypeToBeDeleted table - Implement checkbox-based filtering by manager name with dynamic manager list - Support CRUD operations: create, read, update, delete with local caching - Add "类型管理" button in material validation tab to open the dialog - Include select all/deselect all functionality for manager filters - Track pending changes and show unsaved changes warning Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
818 lines
30 KiB
Python
818 lines
30 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
物料校验标签页
|
||
|
||
校验物料状态并匹配待删除物料。
|
||
支持 2 种校验模式：
|
||
1. database_full - 数据库全表校验
|
||
2. database_filtered - 数据库 ProductionID 过滤校验
|
||
"""
|
||
|
||
import os
|
||
import threading
|
||
import tkinter as tk
|
||
from tkinter import ttk, messagebox, filedialog, simpledialog
|
||
from pathlib import Path
|
||
from io import StringIO
|
||
from contextlib import redirect_stdout
|
||
from gui.widgets import FileSelector, LogText
|
||
from gui.config_manager import ConfigManager
|
||
from gui.material_type_management_dialog import MaterialTypeManagementDialog
|
||
import pandas as pd
|
||
|
||
|
||
class CheckboxTreeview(ttk.Treeview):
|
||
"""支持 checkbox 的 Treeview 组件
|
||
|
||
使用 Unicode 字符模拟 checkbox：
|
||
- ☐ 未选中
|
||
- ☑ 选中
|
||
"""
|
||
|
||
def __init__(self, parent, on_checkbox_change=None, **kwargs):
|
||
"""初始化 CheckboxTreeview
|
||
|
||
Args:
|
||
parent: 父容器
|
||
on_checkbox_change: checkbox 状态改变时的回调函数
|
||
**kwargs: 传递给 Treeview 的参数
|
||
"""
|
||
super().__init__(parent, **kwargs)
|
||
self.checkboxes = {} # item_id -> bool
|
||
self.checkbox_column = "选择"
|
||
self.on_checkbox_change = on_checkbox_change # checkbox 状态改变回调
|
||
|
||
# 绑定点击事件
|
||
self.bind("<Button-1>", 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)
|
||
new_state = not current_state
|
||
self.set_checked(item, new_state)
|
||
|
||
# 通知父组件 checkbox 状态已改变
|
||
if self.on_checkbox_change:
|
||
self.on_checkbox_change(item, new_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):
|
||
"""物料校验标签页"""
|
||
|
||
def __init__(self, parent, config: ConfigManager):
|
||
"""
|
||
初始化物料校验标签页
|
||
|
||
Args:
|
||
parent: 父容器
|
||
config: 配置管理器
|
||
"""
|
||
super().__init__(parent)
|
||
self.config = config
|
||
self.validating = False
|
||
self.validation_results = None
|
||
self.material_records_cache = None # 缓存完整的物料记录
|
||
|
||
self.create_widgets()
|
||
|
||
# 稍后显示就绪消息
|
||
try:
|
||
self.log_text.info("物料校验标签页已就绪")
|
||
except:
|
||
pass # 如果窗口还未完全就绪，忽略错误
|
||
|
||
def create_widgets(self):
|
||
"""创建界面组件"""
|
||
# 主容器
|
||
main_container = ttk.Frame(self)
|
||
main_container.pack(fill=tk.BOTH, expand=True, padx=10, pady=10)
|
||
|
||
# 上部：控制面板
|
||
control_frame = ttk.Frame(main_container)
|
||
control_frame.pack(fill=tk.X, pady=(0, 10))
|
||
|
||
# 中部：结果表格
|
||
result_frame = ttk.LabelFrame(main_container, text="校验结果", padding=5)
|
||
result_frame.pack(fill=tk.BOTH, expand=True, pady=(0, 10))
|
||
|
||
# 下部：日志输出
|
||
log_frame = ttk.LabelFrame(main_container, text="日志", padding=5)
|
||
log_frame.pack(fill=tk.BOTH, expand=True)
|
||
|
||
self._create_control_panel(control_frame)
|
||
self._create_result_table(result_frame)
|
||
self._create_log_panel(log_frame)
|
||
|
||
# 初始化文件选择器显示状态
|
||
self._on_source_mode_change()
|
||
|
||
def _create_control_panel(self, parent):
|
||
"""创建控制面板"""
|
||
# 数据来源选择
|
||
source_group = ttk.LabelFrame(parent, text="数据来源", padding=10)
|
||
source_group.pack(fill=tk.X, pady=5)
|
||
|
||
self.source_mode = tk.StringVar(value="database_full")
|
||
|
||
# 数据库模式
|
||
ttk.Radiobutton(
|
||
source_group,
|
||
text="数据库 - 全表校验",
|
||
variable=self.source_mode,
|
||
value="database_full",
|
||
command=self._on_source_mode_change,
|
||
).grid(row=0, column=0, sticky="w", padx=5)
|
||
|
||
ttk.Radiobutton(
|
||
source_group,
|
||
text="数据库 - ProductionID 过滤",
|
||
variable=self.source_mode,
|
||
value="database_filtered",
|
||
command=self._on_source_mode_change,
|
||
).grid(row=0, column=1, sticky="w", padx=5)
|
||
|
||
# 文件选择
|
||
file_group = ttk.LabelFrame(parent, text="文件选择", padding=10)
|
||
file_group.pack(fill=tk.X, pady=5)
|
||
|
||
# 数据库全表模式 - 无需输入文件
|
||
self.db_full_frame = ttk.Frame(file_group)
|
||
ttk.Label(
|
||
self.db_full_frame,
|
||
text="数据库全表模式：将查询 DiscreteMaterialPlanData 表中的所有材料",
|
||
foreground="gray"
|
||
).pack(anchor="w")
|
||
|
||
# 数据库过滤模式 - 需要 ProductionID 文件
|
||
self.db_filtered_frame = ttk.Frame(file_group)
|
||
self.db_filtered_production_id_selector = FileSelector(
|
||
self.db_filtered_frame,
|
||
label_text="ProductionID 文件:",
|
||
file_type="file",
|
||
file_types=[("文本文件", "*.txt"), ("所有文件", "*.*")],
|
||
initial_dir="D:/python/playwrite/",
|
||
)
|
||
self.db_filtered_production_id_selector.pack(fill=tk.X)
|
||
|
||
# 输出文件
|
||
output_frame = ttk.Frame(file_group)
|
||
output_frame.grid(row=1, column=0, columnspan=2, sticky="ew", pady=(10, 0))
|
||
|
||
self.output_file_selector = FileSelector(
|
||
output_frame,
|
||
label_text="输出文件:",
|
||
file_type="file",
|
||
file_types=[("Excel 文件", "*.xlsx"), ("所有文件", "*.*")],
|
||
initial_dir=self.config.get("paths.data_dir", "data/"),
|
||
)
|
||
self.output_file_selector.pack(fill=tk.X)
|
||
|
||
# 设置默认输出
|
||
default_output = os.path.join(
|
||
self.config.get("paths.data_dir", "data/"),
|
||
self.config.get("paths.validation_output", "物料状态校验结果.xlsx"),
|
||
)
|
||
self.output_file_selector.set(default_output)
|
||
|
||
# 控制按钮
|
||
button_frame = ttk.Frame(parent)
|
||
button_frame.pack(fill=tk.X, pady=10)
|
||
|
||
self.start_button = ttk.Button(
|
||
button_frame, text="开始校验", command=self.start_validation
|
||
)
|
||
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="导出结果",
|
||
command=self.export_results,
|
||
state=tk.DISABLED,
|
||
)
|
||
self.export_button.pack(side=tk.LEFT, padx=5)
|
||
|
||
# 类型管理按钮
|
||
self.type_management_button = ttk.Button(
|
||
button_frame,
|
||
text="类型管理",
|
||
command=self.open_type_management
|
||
)
|
||
self.type_management_button.pack(side=tk.LEFT, padx=5)
|
||
|
||
def _create_result_table(self, parent):
|
||
"""创建结果表格"""
|
||
# 新列定义：选择、材料名称、材料代码、规格、型号、负责人
|
||
columns = ("选择", "材料名称", "材料代码", "规格", "型号", "负责人")
|
||
self.tree = CheckboxTreeview(
|
||
parent,
|
||
columns=columns,
|
||
show="headings",
|
||
height=10,
|
||
on_checkbox_change=self._sync_checkbox_by_material_code
|
||
)
|
||
|
||
# 设置列标题和宽度
|
||
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=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)
|
||
|
||
# 添加滚动条
|
||
scrollbar_y = ttk.Scrollbar(parent, orient=tk.VERTICAL, command=self.tree.yview)
|
||
scrollbar_x = ttk.Scrollbar(
|
||
parent, orient=tk.HORIZONTAL, command=self.tree.xview
|
||
)
|
||
|
||
self.tree.configure(
|
||
yscrollcommand=scrollbar_y.set, xscrollcommand=scrollbar_x.set
|
||
)
|
||
|
||
# 布局
|
||
self.tree.grid(row=0, column=0, sticky="nsew")
|
||
scrollbar_y.grid(row=0, column=1, sticky="ns")
|
||
scrollbar_x.grid(row=1, column=0, sticky="ew")
|
||
|
||
parent.rowconfigure(0, weight=1)
|
||
parent.columnconfigure(0, weight=1)
|
||
|
||
# 绑定双击事件用于编辑负责人
|
||
self.tree.bind("<Double-1>", self._on_cell_double_click)
|
||
|
||
def _create_log_panel(self, parent):
|
||
"""创建日志面板"""
|
||
self.log_text = LogText(parent, height=8, readonly=True)
|
||
self.log_text.pack(fill=tk.BOTH, expand=True)
|
||
|
||
def _on_source_mode_change(self):
|
||
"""数据源模式切换"""
|
||
mode = self.source_mode.get()
|
||
|
||
# 隐藏所有文件选择框架
|
||
self.db_full_frame.grid_remove()
|
||
self.db_filtered_frame.grid_remove()
|
||
|
||
# 根据模式显示对应的文件选择器
|
||
if mode == "database_full":
|
||
self.db_full_frame.grid(row=0, column=0, columnspan=2, sticky="ew")
|
||
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 _sync_checkbox_by_material_code(self, changed_item: str, new_state: bool):
|
||
"""同步相同材料代码的所有记录的选择状态
|
||
|
||
当用户点击某个记录的 checkbox 时，自动将所有具有相同材料代码的记录
|
||
的 checkbox 状态同步更新。
|
||
|
||
Args:
|
||
changed_item: 被点击的 item ID
|
||
new_state: 新的 checkbox 状态
|
||
"""
|
||
# 获取被点击行的材料代码
|
||
values = self.tree.item(changed_item, "values")
|
||
if not values or len(values) <= 2:
|
||
return
|
||
material_code = values[2] # 材料代码在第3列（索引2）
|
||
|
||
# 同步所有具有相同材料代码的记录
|
||
synced_count = 0
|
||
for item in self.tree.get_children():
|
||
item_values = self.tree.item(item, "values")
|
||
if item_values and len(item_values) > 2:
|
||
if item_values[2] == material_code and item != changed_item:
|
||
# 只更新状态不同的行，避免重复更新
|
||
current_state = self.tree.checkboxes.get(item, False)
|
||
if current_state != new_state:
|
||
self.tree.set_checked(item, new_state)
|
||
synced_count += 1
|
||
|
||
if synced_count > 0:
|
||
action = "选中" if new_state else "取消选中"
|
||
self.log_text.info(f"已同步 {synced_count} 条相同材料代码的记录{action}")
|
||
|
||
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):
|
||
"""开始校验"""
|
||
# 验证输入
|
||
output_file = self.output_file_selector.get()
|
||
if not output_file:
|
||
messagebox.showerror("错误", "请指定输出文件路径")
|
||
return
|
||
|
||
mode = self.source_mode.get()
|
||
input_file = None
|
||
production_id_file = None
|
||
|
||
# 根据模式验证输入文件
|
||
if mode == "database_full":
|
||
# 无需输入文件
|
||
pass
|
||
elif mode == "database_filtered":
|
||
production_id_file = self.db_filtered_production_id_selector.get()
|
||
if not production_id_file:
|
||
messagebox.showerror("错误", "请选择 ProductionID 文件")
|
||
return
|
||
if not os.path.exists(production_id_file):
|
||
messagebox.showerror("错误", f"文件不存在：{production_id_file}")
|
||
return
|
||
else:
|
||
messagebox.showerror("错误", f"未知的校验模式: {mode}")
|
||
return
|
||
|
||
# 确保输出目录存在
|
||
output_dir = os.path.dirname(output_file)
|
||
if output_dir and not os.path.exists(output_dir):
|
||
os.makedirs(output_dir, exist_ok=True)
|
||
|
||
# 更新 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}）...")
|
||
|
||
# 清空结果表格
|
||
for item in self.tree.get_children():
|
||
self.tree.delete(item)
|
||
|
||
# 在后台线程中执行校验
|
||
validation_thread = threading.Thread(
|
||
target=self._validation_worker_enhanced,
|
||
args=(mode, input_file, production_id_file, output_file),
|
||
daemon=True,
|
||
)
|
||
validation_thread.start()
|
||
|
||
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
|
||
|
||
# 创建校验器实例
|
||
validator = MaterialStatusValidator(
|
||
username=self.config.get("erp.username"),
|
||
password=self.config.get("erp.password"),
|
||
headless=self.config.get("erp.headless", True),
|
||
verbose=True,
|
||
)
|
||
|
||
# 捕获 stdout 输出
|
||
captured_output = StringIO()
|
||
|
||
# 根据模式执行增强校验
|
||
with redirect_stdout(captured_output):
|
||
if mode == "database_full":
|
||
result_file, results = validator.validate_from_database_enhanced(
|
||
full_table=True,
|
||
output_file=output_file
|
||
)
|
||
elif mode == "database_filtered":
|
||
result_file, results = validator.validate_from_database_enhanced(
|
||
production_id_file=production_id_file,
|
||
full_table=False,
|
||
output_file=output_file
|
||
)
|
||
else:
|
||
raise ValueError(f"未知的校验模式: {mode}")
|
||
|
||
# 缓存结果记录
|
||
self.material_records_cache = results
|
||
|
||
# 获取捕获的输出并显示到日志
|
||
output_text = captured_output.getvalue()
|
||
if output_text:
|
||
for line in output_text.split("\n"):
|
||
if line.strip():
|
||
self._update_log(line, "INFO")
|
||
|
||
if result_file and results:
|
||
self._update_log(f"校验完成，结果已保存到：{output_file}", "SUCCESS")
|
||
# 加载结果显示（带删除状态）
|
||
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)
|
||
|
||
def _validation_complete(self):
|
||
"""校验完成后的 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_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():
|
||
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)} 条结果（其中 {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):
|
||
"""确认删除主流程
|
||
|
||
处理所有行：
|
||
- 已勾选的行：写入/更新到数据库（负责人不能为空）
|
||
- 未勾选的行：从数据库中删除
|
||
"""
|
||
if not self.tree.get_children():
|
||
messagebox.showwarning("警告", "没有可处理的数据")
|
||
return
|
||
|
||
# 收集所有数据
|
||
to_upsert = [] # 需要写入/更新的记录
|
||
to_delete = [] # 需要删除的记录
|
||
missing_manager = [] # 缺少负责人的记录
|
||
|
||
for item in self.tree.get_children():
|
||
values = self.tree.item(item, "values")
|
||
is_checked = self.tree.checkboxes.get(item, False)
|
||
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
|
||
|
||
material_code = material_code.strip()
|
||
|
||
if is_checked:
|
||
# 已勾选：需要写入/更新
|
||
if not manager_name or not manager_name.strip():
|
||
missing_manager.append(material_code)
|
||
else:
|
||
to_upsert.append({
|
||
"material_code": material_code,
|
||
"manager_name": manager_name.strip()
|
||
})
|
||
else:
|
||
# 未勾选：需要删除
|
||
to_delete.append(material_code)
|
||
|
||
# 验证：已勾选的记录必须有负责人
|
||
if missing_manager:
|
||
msg = f"以下已勾选的记录缺少负责人信息，无法保存：\n\n" + \
|
||
"\n".join(missing_manager[:10])
|
||
if len(missing_manager) > 10:
|
||
msg += f"\n... 共 {len(missing_manager)} 条"
|
||
messagebox.showwarning("警告", msg)
|
||
return
|
||
|
||
# 构建确认消息
|
||
confirm_parts = []
|
||
if to_upsert:
|
||
confirm_parts.append(f"写入/更新 {len(to_upsert)} 条记录")
|
||
if to_delete:
|
||
confirm_parts.append(f"删除 {len(to_delete)} 条记录")
|
||
|
||
if not to_upsert and not to_delete:
|
||
messagebox.showinfo("提示", "没有需要处理的记录")
|
||
return
|
||
|
||
confirm_msg = "确认以下操作吗？\n\n" + "\n".join(confirm_parts)
|
||
if not messagebox.askyesno("确认操作", confirm_msg):
|
||
return
|
||
|
||
# 在后台线程中执行
|
||
self._update_log(f"开始处理：写入/更新 {len(to_upsert)} 条，删除 {len(to_delete)} 条...", "INFO")
|
||
|
||
deletion_thread = threading.Thread(
|
||
target=self._execute_sync_in_background,
|
||
args=(to_upsert, to_delete),
|
||
daemon=True,
|
||
)
|
||
deletion_thread.start()
|
||
|
||
def _execute_sync_in_background(self, to_upsert: list, to_delete: list):
|
||
"""后台线程执行同步操作
|
||
|
||
Args:
|
||
to_upsert: 需要写入/更新的记录列表
|
||
to_delete: 需要删除的材料代码列表
|
||
"""
|
||
try:
|
||
from db.materials_to_be_deleted_records_dao import MaterialsToBeDeletedDAO
|
||
|
||
dao = MaterialsToBeDeletedDAO()
|
||
|
||
# 执行写入/更新操作
|
||
upsert_stats = {'total': 0, 'success': 0, 'failed': 0}
|
||
if to_upsert:
|
||
upsert_stats = dao.upsert_batch(to_upsert)
|
||
|
||
# 执行删除操作
|
||
delete_count = 0
|
||
if to_delete:
|
||
delete_count = dao.delete_by_material_codes(to_delete)
|
||
|
||
# 组合统计信息
|
||
combined_stats = {
|
||
'upsert_success': upsert_stats['success'],
|
||
'upsert_failed': upsert_stats['failed'],
|
||
'deleted': delete_count
|
||
}
|
||
|
||
self.after(0, lambda: self._sync_complete(combined_stats))
|
||
|
||
except Exception as e:
|
||
self.after(0, lambda: self._update_log(f"同步操作失败：{str(e)}", "ERROR"))
|
||
|
||
def _sync_complete(self, stats: dict):
|
||
"""同步完成回调，更新 UI
|
||
|
||
Args:
|
||
stats: 包含 upsert_success, upsert_failed, deleted 的字典
|
||
"""
|
||
upsert_success = stats.get('upsert_success', 0)
|
||
upsert_failed = stats.get('upsert_failed', 0)
|
||
deleted = stats.get('deleted', 0)
|
||
|
||
msg_parts = []
|
||
if upsert_success > 0:
|
||
msg_parts.append(f"写入/更新成功：{upsert_success} 条")
|
||
if upsert_failed > 0:
|
||
msg_parts.append(f"写入/更新失败：{upsert_failed} 条")
|
||
if deleted > 0:
|
||
msg_parts.append(f"删除成功：{deleted} 条")
|
||
|
||
msg = "同步操作完成！\n\n" + "\n".join(msg_parts)
|
||
|
||
if upsert_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"):
|
||
"""线程安全的日志更新"""
|
||
|
||
def update():
|
||
if level == "INFO":
|
||
self.log_text.info(message)
|
||
elif level == "SUCCESS":
|
||
self.log_text.success(message)
|
||
elif level == "WARNING":
|
||
self.log_text.warning(message)
|
||
elif level == "ERROR":
|
||
self.log_text.error(message)
|
||
|
||
self.after(0, update)
|
||
|
||
def export_results(self):
|
||
"""导出结果到 Excel"""
|
||
output_file = self.output_file_selector.get()
|
||
|
||
if not output_file:
|
||
output_file = filedialog.asksaveasfilename(
|
||
title="保存结果",
|
||
defaultextension=".xlsx",
|
||
filetypes=[("Excel 文件", "*.xlsx"), ("所有文件", "*.*")],
|
||
)
|
||
|
||
if not output_file:
|
||
return
|
||
|
||
try:
|
||
# 收集表格数据
|
||
data = []
|
||
for item in self.tree.get_children():
|
||
values = self.tree.item(item)["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("警告", "没有数据可导出")
|
||
return
|
||
|
||
# 创建 DataFrame 并保存
|
||
df = pd.DataFrame(
|
||
data, columns=["选择", "材料名称", "材料代码", "规格", "型号", "负责人"]
|
||
)
|
||
df.to_excel(output_file, index=False)
|
||
|
||
messagebox.showinfo("成功", f"结果已导出到：{output_file}")
|
||
self.log_text.success(f"结果已导出到：{output_file}")
|
||
|
||
except Exception as e:
|
||
messagebox.showerror("错误", f"导出失败：{str(e)}")
|
||
self.log_text.error(f"导出失败：{str(e)}")
|
||
|
||
def open_type_management(self):
|
||
"""打开类型管理窗口"""
|
||
dialog = MaterialTypeManagementDialog(self, title="物料类型管理")
|