#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ SUW Core - Deletion Manager Module 拆分自: suw_impl.py (Line 4274-4800, 6970-7100) 用途: Blender删除管理、对象清理、数据结构维护 版本: 1.0.0 作者: SUWood Team """ from .memory_manager import memory_manager, execute_in_main_thread import time import logging import threading from typing import Dict, Any, List, Optional # 设置日志 logger = logging.getLogger(__name__) # 检查Blender可用性 try: import bpy BLENDER_AVAILABLE = True except ImportError: BLENDER_AVAILABLE = False # 导入依赖模块 # ==================== 删除管理器类 ==================== class DeletionManager: """删除管理器 - 负责所有删除相关操作""" def __init__(self, suw_impl=None): """ 初始化删除管理器 Args: suw_impl: SUWImpl实例的引用（可选） """ self.suw_impl = suw_impl # 删除统计 self.deletion_stats = { "units_deleted": 0, "zones_deleted": 0, "parts_deleted": 0, "hardwares_deleted": 0, "objects_deleted": 0, "deletion_errors": 0 } logger.info("✅ 删除管理器初始化完成") # ==================== 原始命令方法 ==================== def c09(self, data: Dict[str, Any]): """del_entity - 删除实体 - 完全对应c03/c04创建逻辑的析构函数""" try: if not BLENDER_AVAILABLE: logger.warning("Blender 不可用，跳过删除操作") return None def delete_entities(): try: # 确保在主线程中执行 if threading.current_thread() != threading.main_thread(): logger.warning("删除操作转移到主线程执行") return 0 # 清除所有选择 if self.suw_impl and hasattr(self.suw_impl, 'sel_clear'): self.suw_impl.sel_clear() uid = data.get("uid") typ = data.get("typ") # uid/zid/cp/work/hw/pull/wall oid = data.get("oid", 0) logger.info(f"🗑️ 开始删除实体: uid={uid}, typ={typ}, oid={oid}") # 【构造/析构对称性】根据类型执行对应的删除逻辑 if typ == "uid": # 删除整个单元 - 对应c03/c04的完整创建 self._del_unit_complete(uid) elif typ == "zid": # 删除区域 - 对应c03的zone创建 self._del_zone_complete(uid, oid) elif typ == "cp": # 删除部件 - 对应c04的part创建 self._del_part_complete(uid, oid) elif typ == "wall": # 删除墙体实体 self._del_wall_entity_safe(data, uid, oid) elif typ == "hw": # 删除硬件 self._del_hardware_complete(uid, oid) else: # 其他类型的删除 self._del_other_entity_safe(data, uid, typ, oid) # 清理标签和维度标注 self._clear_labels_safe() # 强制更新视图 if BLENDER_AVAILABLE: bpy.context.view_layer.update() logger.info(f"✅ 删除实体完成: uid={uid}, typ={typ}, oid={oid}") return True except Exception as e: logger.error(f"删除实体失败: {e}") self.deletion_stats["deletion_errors"] += 1 import traceback logger.error(traceback.format_exc()) return False # 在主线程中执行删除操作 if self.suw_impl and hasattr(self.suw_impl, 'execute_in_main_thread'): result = self.suw_impl.execute_in_main_thread(delete_entities) # 确保返回有效值 return result if result is not None else True else: # 直接执行 result = delete_entities() return result if result is not None else True except Exception as e: logger.error(f"❌ 删除实体失败: {e}") self.deletion_stats["deletion_errors"] += 1 return None def c03(self, data: Dict[str, Any]): """删除层次结构 - 超级安全版本""" try: if not BLENDER_AVAILABLE: logger.warning("Blender 不可用，跳过层次删除") return None logger.info("🗑️ 执行c03命令: 删除层次结构") # 使用保守的删除策略 def delete_hierarchy_ultra_safe(root_obj): """超级保守的层次删除""" try: if not root_obj or not self._is_object_valid(root_obj): return 0 deleted_count = 0 children_to_delete = [] # 收集所有子对象 def collect_children(obj, collection): if obj and self._is_object_valid(obj): collection.append(obj) for child in obj.children: collect_children(child, collection) collect_children(root_obj, children_to_delete) # 从叶子节点开始删除 for obj in reversed(children_to_delete): if self._delete_object_safe(obj): deleted_count += 1 return deleted_count except Exception as e: logger.error(f"层次删除失败: {e}") return 0 # 获取目标对象并执行删除 uid = data.get("uid", "") target_name = data.get("target", f"Unit_{uid}") target_obj = bpy.data.objects.get(target_name) if target_obj: deleted_count = delete_hierarchy_ultra_safe(target_obj) logger.info(f"✅ 层次删除完成: 删除了 {deleted_count} 个对象") return deleted_count else: logger.warning(f"目标对象不存在: {target_name}") return 0 except Exception as e: logger.error(f"c03命令执行失败: {e}") self.deletion_stats["deletion_errors"] += 1 return None # ==================== 核心删除方法 ==================== def _del_unit_complete(self, uid: str): """完整删除单元 - 对应c03/c04的完整创建逻辑""" try: logger.info(f"🗑️ 开始完整删除单元: {uid}") if not self.suw_impl: logger.error("缺少SUWImpl引用，无法删除单元") return # 1. 删除所有区域 (对应c03创建的zones) if hasattr(self.suw_impl, 'zones') and uid in self.suw_impl.zones: zones_to_delete = list(self.suw_impl.zones[uid].keys()) for zid in zones_to_delete: self._del_zone_complete(uid, zid) # 清空zones字典 del self.suw_impl.zones[uid] logger.info(f"✅ 清理了单元 {uid} 的所有区域数据") # 2. 删除所有部件 (对应c04创建的parts) if hasattr(self.suw_impl, 'parts') and uid in self.suw_impl.parts: parts_to_delete = list(self.suw_impl.parts[uid].keys()) for cp in parts_to_delete: self._del_part_complete(uid, cp) # 清空parts字典 del self.suw_impl.parts[uid] logger.info(f"✅ 清理了单元 {uid} 的所有部件数据") # 3. 删除所有硬件 (对应c08创建的hardwares) if hasattr(self.suw_impl, 'hardwares') and uid in self.suw_impl.hardwares: hardwares_to_delete = list(self.suw_impl.hardwares[uid].keys()) for hw_id in hardwares_to_delete: self._del_hardware_complete(uid, hw_id) # 清空hardwares字典 del self.suw_impl.hardwares[uid] logger.info(f"✅ 清理了单元 {uid} 的所有硬件数据") # 4. 删除所有加工 (对应c05创建的machinings) if hasattr(self.suw_impl, 'machinings') and uid in self.suw_impl.machinings: del self.suw_impl.machinings[uid] logger.info(f"✅ 清理了单元 {uid} 的所有加工数据") # 5. 删除所有尺寸标注 (对应c07创建的dimensions) if hasattr(self.suw_impl, 'dimensions') and uid in self.suw_impl.dimensions: del self.suw_impl.dimensions[uid] logger.info(f"✅ 清理了单元 {uid} 的所有尺寸标注数据") # 6. 清理单元级别的数据 self._cleanup_unit_data(uid) # 7. 清理c15缓存 if hasattr(self.suw_impl, '_clear_c15_cache'): self.suw_impl._clear_c15_cache(uid) # 更新统计 self.deletion_stats["units_deleted"] += 1 logger.info(f"🎉 单元 {uid} 完整删除完成") except Exception as e: logger.error(f"完整删除单元失败 {uid}: {e}") self.deletion_stats["deletion_errors"] += 1 def _del_zone_complete(self, uid: str, zid: int): """完整删除区域 - 对应c03的zone创建逻辑""" try: logger.info(f"🗑️ 开始删除区域: uid={uid}, zid={zid}") # 1. 找到Zone对象 zone_name = f"Zone_{zid}" zone_obj = bpy.data.objects.get(zone_name) if zone_obj: # 2. 递归删除所有子对象 (对应create_face_safe创建的子面) children_to_delete = list(zone_obj.children) for child in children_to_delete: logger.info(f"删除Zone子对象: {child.name}") self._delete_object_safe(child) # 3. 删除Zone对象本身 logger.info(f"删除Zone对象: {zone_name}") self._delete_object_safe(zone_obj) else: logger.warning(f"Zone对象不存在: {zone_name}") # 4. 从数据结构中移除 (对应c03中的存储逻辑) if (self.suw_impl and hasattr(self.suw_impl, 'zones') and uid in self.suw_impl.zones and zid in self.suw_impl.zones[uid]): del self.suw_impl.zones[uid][zid] logger.info(f"✅ 从zones数据结构中移除: uid={uid}, zid={zid}") # 更新统计 self.deletion_stats["zones_deleted"] += 1 logger.info(f"✅ 区域删除完成: uid={uid}, zid={zid}") except Exception as e: logger.error(f"删除区域失败 uid={uid}, zid={zid}: {e}") self.deletion_stats["deletion_errors"] += 1 def _del_part_complete(self, uid: str, cp: int): """完整删除部件 - 与c04完全对称的删除逻辑""" try: logger.info(f"🗑️ 开始删除部件: uid={uid}, cp={cp}") if not self.suw_impl: logger.error("缺少SUWImpl引用，无法删除部件") return # 【数据结构优先策略】先从数据结构获取信息 parts = self.suw_impl.get_parts({'uid': uid}) part_exists_in_data = cp in parts if part_exists_in_data: part = parts[cp] logger.info(f"📊 数据结构中找到部件: {part.name if part else 'None'}") # 获取创建记录（如果存在） created_objects = part.get( "sw_created_objects", {}) if part else {} # 1. 清理材质引用（对应c04的材质设置） for material_name in created_objects.get("materials", []): material = bpy.data.materials.get(material_name) if material: logger.info(f"🗑️ 清理材质引用: {material_name}") # 不删除材质本身，只清理引用 # 2. 删除板材（对应c04的板材创建） for board_name in created_objects.get("boards", []): board = bpy.data.objects.get(board_name) if board: logger.info(f"🗑️ 删除记录的板材: {board_name}") self._delete_object_safe(board) else: logger.info(f"📊 数据结构中未找到部件: uid={uid}, cp={cp}") part = None # 3. 查找Blender中的Part对象 part_name = f"Part_{cp}" part_obj = bpy.data.objects.get(part_name) deleted_objects_count = 0 if part_obj: logger.info(f"🎯 在Blender中找到部件对象: {part_name}") # 3. 递归删除所有子对象 (对应各种板材创建方法) children_to_delete = list(part_obj.children) logger.info(f"📦 找到 {len(children_to_delete)} 个子对象需要删除") for child in children_to_delete: try: # 在删除前记录名称，避免删除后访问 child_name = child.name if hasattr( child, 'name') else 'unknown' logger.info(f"🗑️ 删除Part子对象: {child_name}") # 检查是否是板材 try: if child.get("sw_face_type") == "board": logger.info(f"📋 删除板材对象: {child_name}") except (ReferenceError, AttributeError): # 对象可能已经被删除 pass success = self._delete_object_safe(child) if success: deleted_objects_count += 1 logger.info(f"✅ 子对象删除成功: {child_name}") else: logger.warning(f"⚠️ 子对象删除失败: {child_name}") except (ReferenceError, AttributeError) as e: logger.warning(f"⚠️ 子对象已被删除，跳过: {e}") # 对象已被删除，计为成功 deleted_objects_count += 1 # 4. 删除Part对象本身 try: logger.info(f"🗑️ 删除Part对象: {part_name}") success = self._delete_object_safe(part_obj) if success: deleted_objects_count += 1 logger.info(f"✅ Part对象删除成功: {part_name}") else: logger.warning(f"⚠️ Part对象删除失败: {part_name}") except (ReferenceError, AttributeError) as e: logger.warning(f"⚠️ Part对象已被删除，跳过: {e}") # 对象已被删除，计为成功 deleted_objects_count += 1 else: logger.warning(f"❌ Part对象不存在: {part_name}") # 5. 全面搜索并删除所有可能的相关板材对象 board_patterns = [ f"Board_Part_{cp}", # 标准板材 f"Board_Part_{cp}_default", # 默认板材 f"Board_Surface_Part_{cp}", # 表面板材 ] # 搜索带时间戳的板材 if BLENDER_AVAILABLE: all_objects = list(bpy.data.objects) for obj in all_objects: # 检查是否是该Part的板材（包含时间戳的情况） if obj.name.startswith(f"Board_Part_{cp}_") and obj.name != f"Board_Part_{cp}_default": logger.info(f"🔍 发现时间戳板材: {obj.name}") board_patterns.append(obj.name) orphaned_boards_deleted = 0 for pattern in board_patterns: board_obj = bpy.data.objects.get(pattern) if board_obj: try: logger.info(f"🗑️ 删除孤立板材: {pattern}") success = self._delete_object_safe(board_obj) if success: orphaned_boards_deleted += 1 logger.info(f"✅ 孤立板材删除成功: {pattern}") else: logger.warning(f"⚠️ 孤立板材删除失败: {pattern}") except (ReferenceError, AttributeError) as e: logger.warning(f"⚠️ 孤立板材已被删除，跳过: {pattern}, {e}") # 对象已被删除，计为成功 orphaned_boards_deleted += 1 # 6. 搜索所有可能的相关对象（基于属性） if BLENDER_AVAILABLE: attribute_based_objects = [] for obj in bpy.data.objects: # 检查对象属性 if (obj.get("sw_uid") == uid and obj.get("sw_cp") == cp) or \ (obj.get("sw_face_type") == "board" and f"Part_{cp}" in obj.name): attribute_based_objects.append(obj) if attribute_based_objects: logger.info( f"🔍 通过属性找到 {len(attribute_based_objects)} 个相关对象") for obj in attribute_based_objects: try: obj_name = obj.name if hasattr( obj, 'name') else 'unknown' if obj_name not in [o.name for o in [part_obj] + (list(part_obj.children) if part_obj else [])]: logger.info(f"🗑️ 删除属性相关对象: {obj_name}") success = self._delete_object_safe(obj) if success: orphaned_boards_deleted += 1 logger.info(f"✅ 属性相关对象删除成功: {obj_name}") except (ReferenceError, AttributeError) as e: logger.warning(f"⚠️ 属性相关对象已被删除，跳过: {e}") # 对象已被删除，计为成功 orphaned_boards_deleted += 1 # 7. 最后清理数据结构（对应c04的数据结构存储） if part_exists_in_data: del parts[cp] logger.info(f"✅ 从parts数据结构中移除: uid={uid}, cp={cp}") # 检查是否清空了整个uid的parts if not parts: logger.info(f"📊 uid={uid} 的所有部件已清空") else: logger.info(f"📊 数据结构中没有需要清理的部件引用: uid={uid}, cp={cp}") # 8. 显示所有剩余的Part对象用于调试 if BLENDER_AVAILABLE: remaining_parts = [ obj for obj in bpy.data.objects if obj.name.startswith("Part_")] if remaining_parts: logger.info( f"🔍 场景中剩余的Part对象: {[obj.name for obj in remaining_parts]}") else: logger.info("🔍 场景中没有剩余的Part对象") total_deleted = deleted_objects_count + orphaned_boards_deleted self.deletion_stats["parts_deleted"] += 1 self.deletion_stats["objects_deleted"] += total_deleted logger.info( f"🎉 部件删除完成: uid={uid}, cp={cp}, 共删除 {total_deleted} 个对象") except Exception as e: logger.error(f"❌ 删除部件失败 uid={uid}, cp={cp}: {e}") self.deletion_stats["deletion_errors"] += 1 import traceback logger.error(traceback.format_exc()) def _del_hardware_complete(self, uid: str, hw_id: int): """完整删除硬件 - 对应c08的hardware创建逻辑""" try: logger.info(f"🗑️ 开始删除硬件: uid={uid}, hw_id={hw_id}") if not self.suw_impl: logger.error("缺少SUWImpl引用，无法删除硬件") return # 从数据结构中查找并删除硬件对象 if (hasattr(self.suw_impl, 'hardwares') and uid in self.suw_impl.hardwares and hw_id in self.suw_impl.hardwares[uid]): hw_obj = self.suw_impl.hardwares[uid][hw_id] if hw_obj and self._is_object_valid(hw_obj): logger.info(f"删除硬件对象: {hw_obj.name}") self._delete_object_safe(hw_obj) # 从数据结构中移除 del self.suw_impl.hardwares[uid][hw_id] logger.info(f"✅ 从hardwares数据结构中移除: uid={uid}, hw_id={hw_id}") else: logger.warning(f"硬件不存在: uid={uid}, hw_id={hw_id}") # 更新统计 self.deletion_stats["hardwares_deleted"] += 1 logger.info(f"✅ 硬件删除完成: uid={uid}, hw_id={hw_id}") except Exception as e: logger.error(f"删除硬件失败 uid={uid}, hw_id={hw_id}: {e}") self.deletion_stats["deletion_errors"] += 1 def _del_other_entity_safe(self, data: Dict[str, Any], uid: str, typ: str, oid: int): """删除其他类型实体 - 兼容旧逻辑""" try: logger.info(f"🗑️ 删除其他实体: uid={uid}, typ={typ}, oid={oid}") # 获取相应的实体集合 if typ == "work": # 工作实体，可能需要特殊处理 logger.info(f"删除工作实体: uid={uid}, oid={oid}") # 这里可以添加具体的工作实体删除逻辑 elif typ == "pull": # 拉手实体，可能需要特殊处理 logger.info(f"删除拉手实体: uid={uid}, oid={oid}") # 这里可以添加具体的拉手实体删除逻辑 else: logger.warning(f"未知实体类型: {typ}") logger.info(f"✅ 其他实体删除完成: uid={uid}, typ={typ}, oid={oid}") except Exception as e: logger.error(f"删除其他实体失败 uid={uid}, typ={typ}, oid={oid}: {e}") self.deletion_stats["deletion_errors"] += 1 def _del_wall_entity_safe(self, data: Dict[str, Any], uid: str, oid: int): """安全删除墙体实体""" try: logger.info(f"删除墙体实体: uid={uid}, oid={oid}") if not BLENDER_AVAILABLE: return # 查找并删除墙体对象 objects_to_delete = [] for obj in list(bpy.data.objects): try: if not self._is_object_valid(obj): continue # 检查是否是墙体对象 obj_uid = obj.get("sw_uid") obj_oid = obj.get("sw_oid") obj_type = obj.get("sw_typ") if obj_uid == uid and obj_oid == oid and obj_type == "wall": objects_to_delete.append(obj) logger.debug(f"标记删除墙体对象: {obj.name}") except Exception as e: logger.warning(f"检查墙体对象失败: {e}") continue # 删除找到的墙体对象 deleted_count = 0 for obj in objects_to_delete: try: if self._delete_object_safe(obj): deleted_count += 1 except Exception as e: logger.error( f"删除墙体对象失败 {obj.name if hasattr(obj, 'name') else 'unknown'}: {e}") self.deletion_stats["objects_deleted"] += deleted_count logger.info( f"墙体删除完成: {deleted_count}/{len(objects_to_delete)} 个对象") except Exception as e: logger.error(f"删除墙体实体失败: {e}") self.deletion_stats["deletion_errors"] += 1 # ==================== 辅助方法 ==================== def _is_object_valid(self, obj) -> bool: """检查对象是否仍然有效""" try: if not BLENDER_AVAILABLE or not obj: return False # 尝试访问对象的基本属性 _ = obj.name _ = obj.type # 检查对象是否仍在数据中 return obj.name in bpy.data.objects except (ReferenceError, AttributeError): # 对象已被删除或无效 return False except Exception: # 其他错误，假设对象无效 return False def _delete_object_safe(self, obj) -> bool: """安全删除对象 - 极简化版本，避免网格删除冲突""" try: # 确保在主线程中执行 if threading.current_thread() != threading.main_thread(): logger.warning("对象删除操作必须在主线程中执行") return False if not self._is_object_valid(obj): logger.debug(f"对象已无效，跳过删除") return True obj_name = obj.name obj_type = obj.type # 递归删除子对象 children = list(obj.children) for child in children: if self._is_object_valid(child): self._delete_object_safe(child) # 从所有集合中移除对象 for collection in obj.users_collection: collection.objects.unlink(obj) # 删除对象 bpy.data.objects.remove(obj, do_unlink=True) # 注册到内存管理器 memory_manager.cleanup_orphaned_data() logger.debug(f"✅ 对象删除成功: {obj_name} ({obj_type})") return True except Exception as e: logger.error( f"❌ 删除对象失败 {obj_name if 'obj_name' in locals() else 'unknown'}: {e}") return False def _cleanup_unit_data(self, uid: str): """清理单元数据""" try: logger.info(f"🧹 清理单元数据: {uid}") if not self.suw_impl: return # 清理单元变换数据 if hasattr(self.suw_impl, 'unit_trans') and uid in self.suw_impl.unit_trans: del self.suw_impl.unit_trans[uid] logger.info(f"✅ 清理单元变换数据: {uid}") # 清理单元参数数据 if hasattr(self.suw_impl, 'unit_param') and uid in self.suw_impl.unit_param: del self.suw_impl.unit_param[uid] logger.info(f"✅ 清理单元参数数据: {uid}") # 强制垃圾回收 import gc gc.collect() except Exception as e: logger.error(f"清理单元数据失败: {e}") def _clear_labels_safe(self): """安全清理标签""" try: if not BLENDER_AVAILABLE: return # 查找并删除标签对象 labels_to_delete = [] for obj in bpy.data.objects: if obj.get("sw_typ") == "label" or "Label" in obj.name: labels_to_delete.append(obj) for label in labels_to_delete: self._delete_object_safe(label) if labels_to_delete: logger.info(f"✅ 清理了 {len(labels_to_delete)} 个标签对象") except Exception as e: logger.error(f"清理标签失败: {e}") # ==================== 统计和管理方法 ==================== def get_deletion_stats(self) -> Dict[str, Any]: """获取删除统计信息""" try: return { "deletion_stats": self.deletion_stats.copy(), "total_deletions": ( self.deletion_stats["units_deleted"] + self.deletion_stats["zones_deleted"] + self.deletion_stats["parts_deleted"] + self.deletion_stats["hardwares_deleted"] ), "success_rate": ( 1.0 - (self.deletion_stats["deletion_errors"] / max(1, sum(self.deletion_stats.values()))) ) * 100 } except Exception as e: logger.error(f"获取删除统计失败: {e}") return {"error": str(e)} def reset_deletion_stats(self): """重置删除统计""" self.deletion_stats = { "units_deleted": 0, "zones_deleted": 0, "parts_deleted": 0, "hardwares_deleted": 0, "objects_deleted": 0, "deletion_errors": 0 } logger.info("删除统计已重置") def cleanup(self): """清理删除管理器""" try: # 重置统计 self.reset_deletion_stats() logger.info("✅ 删除管理器清理完成") except Exception as e: logger.error(f"清理删除管理器失败: {e}") # ==================== 全局删除管理器实例 ==================== # 全局实例 deletion_manager = None def init_deletion_manager(suw_impl): """初始化全局删除管理器实例""" global deletion_manager deletion_manager = DeletionManager(suw_impl) return deletion_manager def get_deletion_manager(): """获取全局删除管理器实例""" return deletion_manager