581 lines
21 KiB
Python
581 lines
21 KiB
Python
"""
|
||
模型拖拽服务
|
||
|
||
集中管理：
|
||
1. 系统外部文件拖入窗口（Windows WM_DROPFILES）
|
||
2. 资源管理器内部拖拽（移动资源、拖入场景、拖入场景树）
|
||
3. 模型导入后的命名冲突处理
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import ctypes
|
||
import os
|
||
from pathlib import Path
|
||
from typing import List, Optional, Tuple
|
||
|
||
from imgui_bundle import imgui
|
||
|
||
try:
|
||
from ctypes import wintypes
|
||
except Exception: # pragma: no cover - non-Windows fallback
|
||
wintypes = None
|
||
|
||
|
||
class NativeFileDropMonitor:
|
||
"""Windows 文件拖拽桥接，收集外部文件拖入事件。"""
|
||
|
||
WM_DROPFILES = 0x0233
|
||
WM_NCDESTROY = 0x0082
|
||
GWL_WNDPROC = -4
|
||
|
||
def __init__(self, world):
|
||
self.world = world
|
||
self.enabled = False
|
||
self._hwnd = None
|
||
self._window_proc = None
|
||
self._old_wndproc = None
|
||
self._set_wndproc = None
|
||
self._user32 = None
|
||
self._shell32 = None
|
||
self._pending_events = []
|
||
self._hook_wndproc_ptr = None
|
||
|
||
def start(self):
|
||
"""启动系统文件拖拽捕获。"""
|
||
if self.enabled:
|
||
return
|
||
|
||
if os.environ.get("EG_DISABLE_NATIVE_DROP") == "1":
|
||
print("DragDropMonitor: 已通过 EG_DISABLE_NATIVE_DROP 禁用原生拖拽")
|
||
return
|
||
|
||
if os.name != "nt" or wintypes is None:
|
||
print("DragDropMonitor: 非 Windows 平台，未启用系统拖拽桥接")
|
||
return
|
||
|
||
try:
|
||
win_handle = self.world.win.getWindowHandle() if self.world and self.world.win else None
|
||
if not win_handle:
|
||
return
|
||
|
||
self._hwnd = int(win_handle.getIntHandle())
|
||
if not self._hwnd:
|
||
return
|
||
|
||
self._user32 = ctypes.windll.user32
|
||
self._shell32 = ctypes.windll.shell32
|
||
|
||
long_ptr_t = ctypes.c_ssize_t
|
||
wparam_t = wintypes.WPARAM
|
||
lparam_t = wintypes.LPARAM
|
||
wndproc_t = ctypes.WINFUNCTYPE(
|
||
long_ptr_t, wintypes.HWND, wintypes.UINT, wparam_t, lparam_t
|
||
)
|
||
|
||
if ctypes.sizeof(ctypes.c_void_p) == 8:
|
||
self._set_wndproc = self._user32.SetWindowLongPtrW
|
||
get_wndproc = self._user32.GetWindowLongPtrW
|
||
else:
|
||
self._set_wndproc = self._user32.SetWindowLongW
|
||
get_wndproc = self._user32.GetWindowLongW
|
||
|
||
self._set_wndproc.argtypes = [wintypes.HWND, ctypes.c_int, long_ptr_t]
|
||
self._set_wndproc.restype = long_ptr_t
|
||
get_wndproc.argtypes = [wintypes.HWND, ctypes.c_int]
|
||
get_wndproc.restype = long_ptr_t
|
||
|
||
self._user32.CallWindowProcW.argtypes = [
|
||
ctypes.c_void_p, wintypes.HWND, wintypes.UINT, wparam_t, lparam_t
|
||
]
|
||
self._user32.CallWindowProcW.restype = long_ptr_t
|
||
self._user32.IsWindow.argtypes = [wintypes.HWND]
|
||
self._user32.IsWindow.restype = wintypes.BOOL
|
||
|
||
self._shell32.DragQueryFileW.argtypes = [wintypes.HANDLE, wintypes.UINT, wintypes.LPWSTR, wintypes.UINT]
|
||
self._shell32.DragQueryFileW.restype = wintypes.UINT
|
||
self._shell32.DragQueryPoint.argtypes = [wintypes.HANDLE, ctypes.POINTER(wintypes.POINT)]
|
||
self._shell32.DragQueryPoint.restype = wintypes.BOOL
|
||
self._shell32.DragFinish.argtypes = [wintypes.HANDLE]
|
||
self._shell32.DragFinish.restype = None
|
||
self._shell32.DragAcceptFiles.argtypes = [wintypes.HWND, wintypes.BOOL]
|
||
self._shell32.DragAcceptFiles.restype = None
|
||
|
||
self._old_wndproc = get_wndproc(wintypes.HWND(self._hwnd), self.GWL_WNDPROC)
|
||
if not self._old_wndproc:
|
||
raise RuntimeError("GetWindowLongPtrW returned null WndProc")
|
||
|
||
def _window_proc(hwnd, msg, wparam, lparam):
|
||
try:
|
||
if msg == self.WM_DROPFILES:
|
||
files, drop_pos = self._extract_drop_files(wparam)
|
||
if files:
|
||
self._pending_events.append((files, drop_pos))
|
||
return 0
|
||
if msg == self.WM_NCDESTROY:
|
||
result = self._user32.CallWindowProcW(
|
||
ctypes.c_void_p(self._old_wndproc), hwnd, msg, wparam, lparam
|
||
)
|
||
self.enabled = False
|
||
self._hwnd = None
|
||
self._old_wndproc = None
|
||
self._hook_wndproc_ptr = None
|
||
return result
|
||
except Exception as cb_err:
|
||
print(f"[DragDropMonitor] 回调异常: {cb_err}")
|
||
|
||
if not self._old_wndproc:
|
||
return 0
|
||
return self._user32.CallWindowProcW(
|
||
ctypes.c_void_p(self._old_wndproc), hwnd, msg, wparam, lparam
|
||
)
|
||
|
||
self._window_proc = wndproc_t(_window_proc)
|
||
self._hook_wndproc_ptr = ctypes.cast(self._window_proc, ctypes.c_void_p).value
|
||
if not self._hook_wndproc_ptr:
|
||
raise RuntimeError("无法获取 WndProc 回调指针")
|
||
|
||
self._set_wndproc(
|
||
wintypes.HWND(self._hwnd),
|
||
self.GWL_WNDPROC,
|
||
self._hook_wndproc_ptr,
|
||
)
|
||
installed_wndproc = get_wndproc(wintypes.HWND(self._hwnd), self.GWL_WNDPROC)
|
||
if int(installed_wndproc) != int(self._hook_wndproc_ptr):
|
||
raise RuntimeError("WndProc 安装校验失败")
|
||
|
||
self._shell32.DragAcceptFiles(wintypes.HWND(self._hwnd), True)
|
||
self.enabled = True
|
||
print("拖拽监控已启用")
|
||
except Exception as e:
|
||
print(f"拖拽监控启动失败: {e}")
|
||
self.enabled = False
|
||
|
||
def stop(self):
|
||
"""恢复窗口过程并停止拖拽捕获。"""
|
||
if not self.enabled or os.name != "nt" or wintypes is None:
|
||
return
|
||
|
||
try:
|
||
if self._shell32 and self._hwnd:
|
||
self._shell32.DragAcceptFiles(wintypes.HWND(self._hwnd), False)
|
||
if (
|
||
self._set_wndproc
|
||
and self._hwnd
|
||
and self._old_wndproc
|
||
and self._user32
|
||
and self._user32.IsWindow(wintypes.HWND(self._hwnd))
|
||
):
|
||
self._set_wndproc(
|
||
wintypes.HWND(self._hwnd),
|
||
self.GWL_WNDPROC,
|
||
self._old_wndproc,
|
||
)
|
||
except Exception:
|
||
pass
|
||
finally:
|
||
self.enabled = False
|
||
self._window_proc = None
|
||
self._hook_wndproc_ptr = None
|
||
|
||
def _extract_drop_files(self, hdrop):
|
||
"""从 Win32 HDROP 中提取文件路径和投放点。"""
|
||
files = []
|
||
drop_pos = None
|
||
if not self._shell32 or wintypes is None:
|
||
return files, drop_pos
|
||
|
||
hdrop_handle = wintypes.HANDLE(int(hdrop))
|
||
if not hdrop_handle:
|
||
return files, drop_pos
|
||
|
||
try:
|
||
file_count = self._shell32.DragQueryFileW(hdrop_handle, 0xFFFFFFFF, None, 0)
|
||
for i in range(file_count):
|
||
path_len = self._shell32.DragQueryFileW(hdrop_handle, i, None, 0)
|
||
buffer = ctypes.create_unicode_buffer(path_len + 1)
|
||
self._shell32.DragQueryFileW(hdrop_handle, i, buffer, path_len + 1)
|
||
files.append(buffer.value)
|
||
|
||
point = wintypes.POINT()
|
||
if self._shell32.DragQueryPoint(hdrop_handle, ctypes.byref(point)):
|
||
drop_pos = (int(point.x), int(point.y))
|
||
except Exception as e:
|
||
print(f"[DragDropMonitor] 解析拖拽数据失败: {e}")
|
||
finally:
|
||
try:
|
||
self._shell32.DragFinish(hdrop_handle)
|
||
except Exception:
|
||
pass
|
||
|
||
return files, drop_pos
|
||
|
||
def add_file_from_external(self, file_path):
|
||
"""手动压入外部文件（保留给其他集成调用）。"""
|
||
if file_path:
|
||
self._pending_events.append(([str(file_path)], None))
|
||
|
||
def pop_pending_events(self):
|
||
"""弹出并清空待处理的外部拖拽事件。"""
|
||
events = list(self._pending_events)
|
||
self._pending_events.clear()
|
||
return events
|
||
|
||
|
||
class ModelDragDropService:
|
||
"""模型拖拽业务入口。"""
|
||
|
||
SUPPORTED_MODEL_EXTS = {".gltf", ".glb", ".fbx", ".bam", ".egg", ".obj"}
|
||
|
||
def __init__(self, app):
|
||
self.app = app
|
||
|
||
def setup_drag_drop_support(self):
|
||
"""初始化系统外部拖拽支持。"""
|
||
try:
|
||
monitor = NativeFileDropMonitor(self.app)
|
||
monitor.start()
|
||
self.app.drag_drop_monitor = monitor
|
||
except Exception as e:
|
||
print(f"拖拽监控启动失败: {e}")
|
||
|
||
def is_point_in_resource_manager(self, drop_pos):
|
||
"""判断投放坐标是否命中资源管理器窗口。"""
|
||
if not drop_pos:
|
||
return False
|
||
rect = getattr(self.app, "_resource_manager_window_rect", None)
|
||
if not rect:
|
||
return False
|
||
|
||
x, y = drop_pos
|
||
rx, ry, rw, rh = rect
|
||
return (rx <= x <= rx + rw) and (ry <= y <= ry + rh)
|
||
|
||
def resolve_resource_drop_target_dir(self, drop_pos):
|
||
"""根据投放坐标解析资源管理器中的目标目录。"""
|
||
rm = getattr(self.app, "resource_manager", None)
|
||
default_dir = rm.current_path if rm else None
|
||
targets = getattr(self.app, "_resource_drop_targets", None) or []
|
||
if not drop_pos or not targets:
|
||
return default_dir
|
||
|
||
x, y = drop_pos
|
||
hits = []
|
||
for tx, ty, tw, th, path in targets:
|
||
if tx <= x <= tx + tw and ty <= y <= ty + th:
|
||
hits.append((tw * th, path))
|
||
|
||
if not hits:
|
||
return default_dir
|
||
|
||
hits.sort(key=lambda item: item[0])
|
||
target_path = Path(hits[0][1])
|
||
if target_path.exists() and target_path.is_dir():
|
||
return target_path
|
||
return default_dir
|
||
|
||
def process_external_drop_events(self):
|
||
"""处理外部拖入（系统文件拖拽）事件。"""
|
||
monitor = getattr(self.app, "drag_drop_monitor", None)
|
||
if not monitor:
|
||
return
|
||
|
||
events = monitor.pop_pending_events()
|
||
if not events:
|
||
return
|
||
|
||
for file_paths, drop_pos in events:
|
||
self.handle_external_drop(file_paths, drop_pos)
|
||
|
||
def handle_external_drop(self, file_paths, drop_pos=None):
|
||
"""把外部拖入文件分发到资源管理器或场景。"""
|
||
if not file_paths:
|
||
return
|
||
|
||
paths = [Path(p) for p in file_paths if p]
|
||
if not paths:
|
||
return
|
||
|
||
if self.is_point_in_resource_manager(drop_pos):
|
||
target_dir = self.resolve_resource_drop_target_dir(drop_pos)
|
||
imported, errors = self.app.resource_manager.import_external_files(paths, target_dir=target_dir)
|
||
if imported:
|
||
target_label = self.app.resource_manager.get_relative_path(target_dir) if target_dir else "资源管理器"
|
||
self.app.add_success_message(f"已导入 {len(imported)} 个外部资源到 {target_label}")
|
||
if errors:
|
||
self.app.add_warning_message(f"有 {len(errors)} 个资源导入失败，请查看控制台日志")
|
||
for err in errors:
|
||
print(f"[资源导入] {err}")
|
||
return
|
||
|
||
imported_models = []
|
||
for path in paths:
|
||
if not path.exists() or not self._is_supported_model(path):
|
||
continue
|
||
try:
|
||
model_node = self.app._import_model_with_menu_logic(
|
||
str(path),
|
||
select_model=False,
|
||
set_origin=True,
|
||
show_info_message=False,
|
||
show_success_message=False,
|
||
)
|
||
if not model_node:
|
||
continue
|
||
imported_models.append(model_node)
|
||
except Exception as e:
|
||
print(f"[外部拖拽] 导入模型失败 {path}: {e}")
|
||
|
||
if imported_models:
|
||
if getattr(self.app, "selection", None):
|
||
try:
|
||
self.app.selection.updateSelection(imported_models[-1])
|
||
except Exception as e:
|
||
print(f"[外部拖拽] 更新选择失败: {e}")
|
||
self.app.add_success_message(f"外部拖拽导入场景成功: {len(imported_models)} 个模型")
|
||
else:
|
||
self.app.add_info_message("将文件拖放到“资源管理器”窗口可导入到项目资源")
|
||
|
||
def draw_drag_drop_interface(self):
|
||
"""每帧处理资源管理器内部拖拽的释放事件。"""
|
||
rm = getattr(self.app, "resource_manager", None)
|
||
if not rm:
|
||
return
|
||
|
||
if rm.is_dragging():
|
||
self.app.is_dragging = True
|
||
self.app.dragged_files = rm.get_dragged_files()
|
||
|
||
if self.app.is_dragging and self.app.dragged_files and imgui.is_mouse_released(0):
|
||
mouse_pos = imgui.get_mouse_pos()
|
||
self.handle_internal_drag_drop_completion((float(mouse_pos.x), float(mouse_pos.y)))
|
||
|
||
def handle_internal_drag_drop_completion(self, mouse_pos: Optional[Tuple[float, float]] = None):
|
||
"""处理资源管理器内部拖拽落点。"""
|
||
rm = self.app.resource_manager
|
||
dragged_files = list(self.app.dragged_files) if self.app.dragged_files else rm.get_dragged_files()
|
||
if not dragged_files:
|
||
self._reset_drag_state()
|
||
return
|
||
|
||
drop_kind, target_dir, target_parent = self._resolve_drop_target(mouse_pos)
|
||
|
||
if drop_kind == "resource_manager":
|
||
moved, errors = rm.move_files_to_directory(dragged_files, target_dir)
|
||
if moved:
|
||
label = rm.get_relative_path(target_dir) if target_dir else "资源管理器"
|
||
self.app.add_success_message(f"已移动 {len(moved)} 个资源到 {label}")
|
||
if errors:
|
||
self.app.add_warning_message(f"有 {len(errors)} 个资源移动失败，请查看控制台")
|
||
for err in errors:
|
||
print(f"[资源移动] {err}")
|
||
self._reset_drag_state()
|
||
return
|
||
|
||
if drop_kind not in ("scene_view", "scene_tree"):
|
||
self._reset_drag_state()
|
||
return
|
||
|
||
imported_models = []
|
||
for file_path in dragged_files:
|
||
path = Path(file_path)
|
||
if not self._is_supported_model(path):
|
||
continue
|
||
try:
|
||
model_node = self.app._import_model_with_menu_logic(
|
||
str(path),
|
||
select_model=False,
|
||
set_origin=True,
|
||
show_info_message=False,
|
||
show_success_message=False,
|
||
)
|
||
if not model_node:
|
||
continue
|
||
|
||
if drop_kind == "scene_tree" and self._is_valid_drop_parent(target_parent):
|
||
try:
|
||
model_node.wrtReparentTo(target_parent)
|
||
except Exception:
|
||
model_node.reparentTo(target_parent)
|
||
|
||
imported_models.append(model_node)
|
||
except Exception as e:
|
||
print(f"[拖拽导入] 导入失败 {path}: {e}")
|
||
|
||
if imported_models and getattr(self.app, "selection", None):
|
||
try:
|
||
self.app.selection.updateSelection(imported_models[-1])
|
||
except Exception:
|
||
pass
|
||
|
||
if imported_models:
|
||
self.app.add_success_message(f"拖拽导入成功: {len(imported_models)} 个模型")
|
||
|
||
self._reset_drag_state()
|
||
|
||
def add_dragged_file(self, file_path):
|
||
"""添加拖拽的文件。"""
|
||
if file_path not in self.app.dragged_files:
|
||
self.app.dragged_files.append(file_path)
|
||
self.app.is_dragging = True
|
||
self.app.show_drag_overlay = True
|
||
print(f"检测到拖拽文件: {file_path}")
|
||
|
||
def clear_dragged_files(self):
|
||
"""清空拖拽文件列表。"""
|
||
self.app.dragged_files.clear()
|
||
self.app.is_dragging = False
|
||
self.app.show_drag_overlay = False
|
||
|
||
def process_dragged_files(self):
|
||
"""处理拖拽的文件。"""
|
||
if not self.app.dragged_files:
|
||
return
|
||
|
||
imported_count = 0
|
||
for file_path in self.app.dragged_files:
|
||
if self.import_model_from_path(file_path):
|
||
imported_count += 1
|
||
|
||
if imported_count > 0:
|
||
self.app.add_success_message(f"成功导入 {imported_count} 个模型文件")
|
||
else:
|
||
self.app.add_error_message("没有成功导入任何文件")
|
||
|
||
self.clear_dragged_files()
|
||
|
||
def import_model_from_path(self, file_path):
|
||
"""从路径导入模型的内部方法。"""
|
||
try:
|
||
path = Path(file_path)
|
||
if not path.exists():
|
||
self.app.add_error_message(f"文件不存在: {path}")
|
||
return False
|
||
|
||
if not self._is_supported_model(path):
|
||
self.app.add_error_message(f"不支持的文件格式: {path.suffix.lower()}")
|
||
return False
|
||
|
||
model_node = self.app._import_model_with_menu_logic(str(path))
|
||
if not model_node:
|
||
return False
|
||
|
||
return True
|
||
except Exception as e:
|
||
self.app.add_error_message(f"导入模型时发生错误: {e}")
|
||
return False
|
||
|
||
@staticmethod
|
||
def draw_drag_overlay():
|
||
"""兼容接口：Unity 风格不再绘制拖拽遮罩。"""
|
||
return
|
||
|
||
@staticmethod
|
||
def draw_drag_status():
|
||
"""兼容接口：Unity 风格不再绘制拖拽状态窗。"""
|
||
return
|
||
|
||
def _postprocess_imported_model(self, model_node, set_origin=False):
|
||
if getattr(self.app, "scene_manager", None):
|
||
try:
|
||
self.app.scene_manager.processMaterials(model_node)
|
||
except Exception:
|
||
pass
|
||
if set_origin:
|
||
try:
|
||
model_node.setPos(0, 0, 0)
|
||
except Exception:
|
||
pass
|
||
self._ensure_unique_name(model_node)
|
||
|
||
def _resolve_drop_target(self, mouse_pos: Optional[Tuple[float, float]]):
|
||
point = mouse_pos
|
||
if point is None:
|
||
m = imgui.get_mouse_pos()
|
||
point = (float(m.x), float(m.y))
|
||
|
||
if self._point_in_rect(point, getattr(self.app, "_resource_manager_window_rect", None)):
|
||
target_dir = self.resolve_resource_drop_target_dir(point)
|
||
return "resource_manager", target_dir, None
|
||
|
||
if self._point_in_rect(point, getattr(self.app, "_scene_tree_window_rect", None)):
|
||
return "scene_tree", None, getattr(self.app, "_drag_scene_tree_hover_node", None)
|
||
|
||
for rect_name in (
|
||
"_property_panel_window_rect",
|
||
"_script_panel_window_rect",
|
||
"_console_window_rect",
|
||
"_toolbar_window_rect",
|
||
):
|
||
if self._point_in_rect(point, getattr(self.app, rect_name, None)):
|
||
return "ui_panel", None, None
|
||
|
||
if imgui.get_io().want_capture_mouse:
|
||
return "ui_panel", None, None
|
||
|
||
return "scene_view", None, None
|
||
|
||
@staticmethod
|
||
def _point_in_rect(point, rect):
|
||
if not point or not rect:
|
||
return False
|
||
x, y = point
|
||
rx, ry, rw, rh = rect
|
||
return rx <= x <= rx + rw and ry <= y <= ry + rh
|
||
|
||
@classmethod
|
||
def _is_supported_model(cls, path: Path):
|
||
return path.suffix.lower() in cls.SUPPORTED_MODEL_EXTS
|
||
|
||
@staticmethod
|
||
def _is_valid_drop_parent(node):
|
||
if not node:
|
||
return False
|
||
try:
|
||
return not node.isEmpty()
|
||
except Exception:
|
||
return False
|
||
|
||
@staticmethod
|
||
def _ensure_unique_name(model_node):
|
||
if not model_node or model_node.isEmpty():
|
||
return
|
||
parent = model_node.getParent()
|
||
if not parent or parent.isEmpty():
|
||
return
|
||
|
||
current_name = model_node.getName() or "model"
|
||
stem, suffix = os.path.splitext(current_name)
|
||
if not stem:
|
||
stem = current_name
|
||
|
||
existing_names = set()
|
||
try:
|
||
siblings = parent.getChildren()
|
||
for i in range(siblings.getNumPaths()):
|
||
sibling = siblings.getPath(i)
|
||
if sibling == model_node:
|
||
continue
|
||
existing_names.add(sibling.getName())
|
||
except Exception:
|
||
return
|
||
|
||
if current_name not in existing_names:
|
||
return
|
||
|
||
index = 1
|
||
while True:
|
||
candidate = f"{stem}_{index}{suffix}" if suffix else f"{stem}_{index}"
|
||
if candidate not in existing_names:
|
||
model_node.setName(candidate)
|
||
return
|
||
index += 1
|
||
|
||
def _reset_drag_state(self):
|
||
self.app.is_dragging = False
|
||
self.app.dragged_files.clear()
|
||
self.app.show_drag_overlay = False
|
||
if getattr(self.app, "resource_manager", None):
|
||
self.app.resource_manager.clear_drag()
|
||
self.app._drag_scene_tree_hover_node = None
|