suwoodblender/blenderpython/suw_core/deletion_manager - 副本.py

750 lines
31 KiB
Python
Raw Permalink Normal View History

2025-07-18 17:09:39 +08:00
#!/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