From a1677678906f7897468b91926c178e224ced0f03 Mon Sep 17 00:00:00 2001 From: Rowland <975945824@qq.com> Date: Tue, 27 Jan 2026 10:04:02 +0800 Subject: [PATCH] =?UTF-8?q?ui=E6=9B=BF=E6=8D=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- IFLOW.md | 121 ++++++--- core/drag_drop/__init__.py | 10 - core/drag_drop/base_detector.py | 120 --------- core/drag_drop/fallback_detector.py | 83 ------ core/drag_drop/linux_detector.py | 395 ---------------------------- core/drag_drop/macos_detector.py | 105 -------- core/drag_drop/platform_detector.py | 66 ----- core/drag_drop/simple_detector.py | 233 ---------------- core/drag_drop/windows_detector.py | 221 ---------------- core/drag_drop/x11_drag_receiver.py | 311 ---------------------- 10 files changed, 88 insertions(+), 1577 deletions(-) delete mode 100644 core/drag_drop/__init__.py delete mode 100644 core/drag_drop/base_detector.py delete mode 100644 core/drag_drop/fallback_detector.py delete mode 100644 core/drag_drop/linux_detector.py delete mode 100644 core/drag_drop/macos_detector.py delete mode 100644 core/drag_drop/platform_detector.py delete mode 100644 core/drag_drop/simple_detector.py delete mode 100644 core/drag_drop/windows_detector.py delete mode 100644 core/drag_drop/x11_drag_receiver.py diff --git a/IFLOW.md b/IFLOW.md index 300862c1..6061528d 100644 --- a/IFLOW.md +++ b/IFLOW.md @@ -6,18 +6,22 @@ 核心架构围绕 `MyWorld` 类构建，该类继承自 `CoreWorld`，并集成了各种管理器模块，如选择系统、工具管理器、脚本管理器、GUI 管理器、场景管理器、项目管理器、地形管理器、碰撞管理器、视频管理器、资源管理器和 VR 管理器。 +项目目前处于 "imgui" 分支，已全面转向基于 ImGui 的现代化用户界面，提供了完整的 dockable 界面系统，包括菜单栏、工具栏、场景树、资源管理器、属性面板和控制台等组件。 + ## 核心技术栈 - **核心引擎**: Panda3D 1.10.15 - **图形渲染**: 可选择普通渲染或 RenderPipeline 高级渲染管线 - **VR 支持**: OpenVR/SteamVR 2.2.0 -- **用户界面**: PyQt5 5.15.9 + ImGui 集成 +- **用户界面**: PyQt5 5.15.9 + ImGui 集成 (imgui_bundle) - **3D 模型格式**: 支持 glTF, FBX (需转换), BAM, EGG, OBJ 等 - **脚本语言**: Python (内嵌脚本系统) - **物理/碰撞**: Panda3D 内置碰撞系统 - **视频支持**: MovieTexture (MP4, AVI, MOV, MKV, WebM) - **插件架构**: 动态插件加载系统 - **资源管理**: 文件浏览、拖拽、右键菜单集成 +- **GUI 框架**: p3dimgui + imgui_bundle (支持 docking 功能) +- **拖拽系统**: 跨平台拖拽文件监控系统 (core/drag_drop/) ## 项目结构 @@ -27,6 +31,7 @@ │ ├── world.py # CoreWorld 基础世界类 │ ├── vr/ # VR 子模块 (完整模块化结构) │ │ ├── __init__.py +│ │ ├── README.md │ │ ├── config/ # VR 配置管理 │ │ ├── interaction/ # VR 交互系统 (动作、抓取、摇杆、传送) │ │ ├── performance/ # VR 性能监控 @@ -34,7 +39,6 @@ │ │ ├── testing/ # VR 测试模式 │ │ ├── tracking/ # VR 设备跟踪 │ │ └── visualization/ # VR 可视化 (控制器模型) -│ ├── drag_drop/ # 拖拽文件监控系统 │ ├── vr_manager.py # VR 管理器主文件 │ ├── selection.py # 选择系统 │ ├── tool_manager.py # 工具管理器 @@ -49,45 +53,77 @@ │ ├── resource_manager.py # 资源管理器 (ImGui版本) │ ├── CustomMouseController.py # 自定义鼠标控制器 │ ├── maintenance_gui.py # 维护GUI系统 -│ └── assembly_interaction.py # 装配交互系统 -├── demo/ # 示例和测试文件 (包含大量测试和演示) +│ ├── assembly_interaction.py # 装配交互系统 +│ ├── imgui_style_manager.py # ImGui样式管理器 +│ └── [工具模型文件] # .fbx 格式的工具手柄模型 +│ ├── RotationHandleFull.fbx +│ ├── RotationHandleQuarter.fbx +│ ├── TranslateArrowHandle.fbx +│ └── UniformScaleHandle.fbx ├── gui/ # GUI 相关模块 +│ ├── __init__.py +│ └── gui_manager.py # GUI管理器 ├── project/ # 项目管理模块 +│ ├── __init__.py +│ └── project_manager.py ├── scene/ # 场景管理模块 +│ ├── __init__.py +│ ├── scene_manager.py +│ └── util.py ├── scripts/ # 脚本文件目录 -├── ui/ # UI 组件和主窗口 -│ ├── widgets.py # 自定义UI组件 -│ ├── property_panel.py # 属性面板 -│ ├── interface_manager.py # 界面管理器 -│ ├── main_window.py # 主窗口 -│ ├── icon_manager.py # 图标管理器 -│ └── maintenance_system.py # 维护系统UI +│ └── [各种脚本文件] +├── ui/ # UI 组件 +│ ├── __init__.py +│ └── icon_manager.py # 图标管理器 ├── plugins/ # 插件系统 +│ ├── __init__.py +│ ├── plugin_manager.py +│ ├── plugin_interface_spec.py │ ├── core/ # 核心插件 │ ├── third_party/ # 第三方插件 -│ ├── user/ # 用户插件 -│ ├── plugin_manager.py # 插件管理器 -│ └── plugin_interface_spec.py # 插件接口规范 +│ └── user/ # 用户插件 ├── QPanda3D/ # Panda3D 与 PyQt 集成库 ├── QMeta3D/ # Meta3D 集成库 ├── RenderPipelineFile/ # RenderPipeline 高级渲染管线 ├── Resources/ # 资源文件 (模型、纹理等) +│ ├── a/ # 资源子目录 +│ ├── animations/ +│ ├── c/ +│ ├── icons/ +│ ├── materials/ +│ ├── models/ +│ └── textures/ ├── config/ # 配置文件 │ └── vr_settings.json # VR 配置文件 ├── requirements/ # 依赖项文件 │ ├── requirements.txt # 主要依赖项 │ ├── conda-requirements.txt # Conda环境依赖 -│ └── clean-requirements.txt # 精简依赖项 +│ ├── clean-requirements.txt # 精简依赖项 +│ └── DEPLOYMENT_README.md # 部署说明 ├── vr_actions/ # VR动作配置 +│ ├── actions.json +│ ├── bindings_index.json +│ ├── bindings_oculus.json +│ └── bindings_vive.json ├── icons/ # 图标资源 +│ └── file_types/ # 文件类型图标 ├── templates/ # 模板文件 +│ └── main_template.py ├── tools/ # 工具脚本 +│ └── open_source_rate.py ├── tex/ # 纹理资源 -├── main.py # 程序入口点 -├── demo.py # ImGui演示程序 -├── gui_preview_window.py # GUI预览窗口 +│ ├── empty_basecolor.png +│ ├── empty_normal.png +│ ├── empty_roughness.png +│ └── empty_specular.png +├── new/ # 新项目目录 +│ └── project.json +├── test_project/ # 测试项目 +│ └── project.json +├── main.py # 程序入口点 (集成ImGui界面) ├── Start_Run.py # 启动脚本 -└── imgui.ini # ImGui配置文件 +├── imgui.ini # ImGui配置文件 +└── IFLOW.md # 项目文档 ``` ## 核心功能模块 @@ -134,15 +170,25 @@ - `VRTestMode`: 提供不同的测试显示模式和功能开关。 - `VRPerformanceMonitor`: 性能监控和报告。 -### 3. GUI 系统 (core/gui_manager.py, gui/) +### 3. GUI 系统 (core/gui_manager.py, gui/, core/imgui_style_manager.py) - **GUIManager**: 管理 2D 和 3D GUI 元素的创建、编辑、删除。 +- **ImGui 集成**: 完整的 ImGui 界面系统，支持 docking 功能。 - **功能**: - 创建按钮、标签、输入框、2D/3D 图像、视频屏幕等。 - GUI 编辑模式，支持拖拽创建和属性编辑。 - 与场景树和属性面板集成。 - - 独立的 GUI 预览窗口 (`gui_preview_window.py`)。 - - ImGui 集成支持现代化的界面设计。 + - **现代化界面**: 基于 imgui_bundle 的完整 docking 界面系统，包括： + - 菜单栏 (文件、编辑、创建、视图、工具、VR、帮助) + - 工具栏 (选择、移动、旋转、缩放工具) + - 场景树 (层级结构显示和管理) + - 资源管理器 (文件浏览、拖拽导入) + - 属性面板 (对象属性编辑) + - 控制台 (信息输出) + - 脚本管理面板 + - **ImGui 样式管理**: `ImGuiStyleManager` 提供统一的界面风格管理。 + - **中文字体支持**: 自动加载中文字体，适合中文环境开发。 + - **快捷键支持**: 完整的键盘快捷键系统。 ### 4. 场景与模型管理 (scene/scene_manager.py) @@ -259,37 +305,43 @@ ### 入口点 -- `main.py`: 主程序入口，创建 `MyWorld` 实例并启动 PyQt5 主窗口。 -- `demo.py`: ImGui演示程序，展示现代化界面功能。 +- `main.py`: 主程序入口，创建 `MyWorld` 实例并启动基于 ImGui 的现代化界面。 - `Start_Run.py`: 启动脚本。 -- `gui_preview_window.py`: GUI预览窗口，用于独立测试GUI元素。 ### 运行方式 1. 确保已安装所有依赖项（详见 requirements/requirements.txt）。 -2. 运行 `python main.py` 启动主应用程序。 -3. 运行 `python demo.py` 启动ImGui演示程序。 -4. 如果连接了 VR 设备并安装了 SteamVR，可以在应用内启用 VR 模式。 -5. VR配置可通过 `config/vr_settings.json` 进行调整。 +2. 运行 `python main.py` 启动主应用程序，将显示完整的 ImGui 界面。 +3. 如果连接了 VR 设备并安装了 SteamVR，可以通过菜单 "VR" -> "进入VR模式" 启用 VR 模式。 +4. VR配置可通过 `config/vr_settings.json` 进行调整，或在界面中通过 "VR" -> "VR设置" 进行配置。 ### 依赖项 项目提供完整的依赖项文件在 `requirements/` 目录： -- **requirements.txt**: 主要依赖项，包含： +- **requirements.txt**: 完整依赖项，包含： - `Panda3D==1.10.15` (核心3D引擎) - `PyQt5==5.15.9` (GUI框架) + - `PySide6>=6.8.1` (Qt6支持) - `openvr==2.2.0` (VR支持) - `p3dimgui` (ImGui集成) - `QPanda3D==0.2.10` (Panda3D与PyQt5集成) + - `Pillow>=9.0.1` (图像处理) + - `python-dotenv>=1.0.1` (环境变量管理) + - `pyassimp>=5.2.5` (3D模型导入) - 以及其他辅助库 -- **conda-requirements.txt**: Conda环境依赖项 -- **clean-requirements.txt**: 精简依赖项 +- **clean-requirements.txt**: 精简依赖项，仅包含核心库 +- **conda-requirements.txt**: Conda环境依赖项 (基础Python环境) +- **DEPLOYMENT_README.md**: 部署说明文档 安装命令： ```bash +# 安装完整依赖 pip install -r requirements/requirements.txt + +# 或安装精简依赖 +pip install -r requirements/clean-requirements.txt ``` ## 开发约定 @@ -306,4 +358,7 @@ pip install -r requirements/requirements.txt - **国际化支持**: 内置中文字体支持，适合中文环境开发。 - **事件驱动**: 完善的事件处理系统，支持键盘快捷键（如F键聚焦、P键巡检）。 - **命令模式**: 使用命令系统实现撤销/重做功能。 -- **演示系统**: demo/ 目录包含大量测试和演示文件，便于功能验证和学习。 +- **拖拽系统**: 跨平台拖拽文件监控系统，支持从外部文件管理器拖拽模型到场景中。 +- **ImGui Docking**: 完整的 docking 界面系统，支持窗口自由布局和停靠。 +- **分支管理**: 当前开发在 "imgui" 分支，专注于基于 ImGui 的现代化界面开发。 +- **工具集成**: 内置多种工具手柄模型，支持专业的3D变换操作。 diff --git a/core/drag_drop/__init__.py b/core/drag_drop/__init__.py deleted file mode 100644 index 74ea592b..00000000 --- a/core/drag_drop/__init__.py +++ /dev/null @@ -1,10 +0,0 @@ -""" -跨平台拖拽检测模块 - -提供统一的拖拽检测接口，支持Windows、Linux和macOS平台。 -""" - -from .base_detector import BaseDragDetector -from .platform_detector import PlatformDragDetector - -__all__ = ['BaseDragDetector', 'PlatformDragDetector'] \ No newline at end of file diff --git a/core/drag_drop/base_detector.py b/core/drag_drop/base_detector.py deleted file mode 100644 index 07d638b3..00000000 --- a/core/drag_drop/base_detector.py +++ /dev/null @@ -1,120 +0,0 @@ -""" -拖拽检测器基类 - -定义所有平台拖拽检测器的通用接口。 -""" - -import os -import platform -from abc import ABC, abstractmethod -from typing import List, Optional, Callable -import threading -import queue -import time - - -class BaseDragDetector(ABC): - """拖拽检测器基类""" - - def __init__(self, supported_formats: List[str] = None): - """ - 初始化拖拽检测器 - - Args: - supported_formats: 支持的文件格式列表，如 ['.gltf', '.glb', '.fbx'] - """ - self.supported_formats = supported_formats or ['.gltf', '.glb', '.fbx', '.bam', '.egg', '.obj'] - self.is_running = False - self.monitor_thread = None - self.file_queue = queue.Queue() - self.drop_callback: Optional[Callable[[List[str]], None]] = None - - def set_drop_callback(self, callback: Callable[[List[str]], None]): - """ - 设置文件拖拽回调函数 - - Args: - callback: 接收文件路径列表的回调函数 - """ - self.drop_callback = callback - - def start_monitoring(self): - """开始监控拖拽事件""" - if not self.is_running: - self.is_running = True - self.monitor_thread = threading.Thread(target=self._monitor_loop, daemon=True) - self.monitor_thread.start() - - def stop_monitoring(self): - """停止监控拖拽事件""" - self.is_running = False - if self.monitor_thread: - self.monitor_thread.join(timeout=1.0) - - def get_dropped_files(self) -> List[str]: - """获取拖拽的文件列表""" - files = [] - while not self.file_queue.empty(): - try: - files.append(self.file_queue.get_nowait()) - except queue.Empty: - break - return files - - def add_dropped_file(self, file_path: str): - """ - 添加拖拽的文件路径 - - Args: - file_path: 拖拽的文件路径 - """ - if self._is_supported_format(file_path): - self.file_queue.put(file_path) - if self.drop_callback: - self.drop_callback([file_path]) - - def _is_supported_format(self, file_path: str) -> bool: - """检查文件格式是否支持""" - _, ext = os.path.splitext(file_path.lower()) - return ext in self.supported_formats - - @abstractmethod - def _monitor_loop(self): - """监控循环，由子类实现""" - pass - - @abstractmethod - def is_supported(self) -> bool: - """检查当前平台是否支持此检测器""" - pass - - -class DragDetectorFactory: - """拖拽检测器工厂类""" - - @staticmethod - def create_detector(supported_formats: List[str] = None) -> BaseDragDetector: - """ - 根据当前平台创建合适的拖拽检测器 - - Args: - supported_formats: 支持的文件格式列表 - - Returns: - 适合当前平台的拖拽检测器实例 - """ - system = platform.system().lower() - - if system == 'windows': - from .windows_detector import WindowsDragDetector - return WindowsDragDetector(supported_formats) - elif system == 'linux': - from .linux_detector import LinuxDragDetector - return LinuxDragDetector(supported_formats) - elif system == 'darwin': # macOS - from .macos_detector import MacOSDragDetector - return MacOSDragDetector(supported_formats) - else: - # 降级到基础检测器 - from .fallback_detector import FallbackDragDetector - return FallbackDragDetector(supported_formats) \ No newline at end of file diff --git a/core/drag_drop/fallback_detector.py b/core/drag_drop/fallback_detector.py deleted file mode 100644 index 980b9015..00000000 --- a/core/drag_drop/fallback_detector.py +++ /dev/null @@ -1,83 +0,0 @@ -""" -降级拖拽检测器 - -当平台特定检测器不可用时的降级方案。 -""" - -import os -import time -import threading -from typing import List -from .base_detector import BaseDragDetector - - -class FallbackDragDetector(BaseDragDetector): - """降级拖拽检测器""" - - def __init__(self, supported_formats: List[str] = None): - """ - 初始化降级检测器 - - Args: - supported_formats: 支持的文件格式列表 - """ - super().__init__(supported_formats) - self._watch_directory = os.path.expanduser("~/Desktop") # 监控桌面 - self._known_files = set() - self._scan_interval = 0.5 # 扫描间隔（秒） - - # 初始化已知文件列表 - self._scan_known_files() - - def _scan_known_files(self): - """扫描已知文件""" - self._known_files.clear() - if os.path.exists(self._watch_directory): - for filename in os.listdir(self._watch_directory): - if self._is_supported_format(filename): - filepath = os.path.join(self._watch_directory, filename) - self._known_files.add(filepath) - - def _monitor_loop(self): - """监控循环 - 通过文件系统变化检测""" - while self.is_running: - try: - if os.path.exists(self._watch_directory): - current_files = set() - for filename in os.listdir(self._watch_directory): - if self._is_supported_format(filename): - filepath = os.path.join(self._watch_directory, filename) - current_files.add(filepath) - - # 检测新文件 - new_files = current_files - self._known_files - for filepath in new_files: - # 检查文件是否最近创建（1秒内） - try: - file_time = os.path.getmtime(filepath) - if time.time() - file_time < 1.0: - self.add_dropped_file(filepath) - except OSError: - pass - - self._known_files = current_files - - except Exception as e: - print(f"降级检测器错误: {e}") - - time.sleep(self._scan_interval) - - def is_supported(self) -> bool: - """降级检测器总是支持""" - return True - - def set_watch_directory(self, directory: str): - """ - 设置监控目录 - - Args: - directory: 要监控的目录路径 - """ - if os.path.exists(directory): - self._watch_directory = directory - self._scan_known_files() \ No newline at end of file diff --git a/core/drag_drop/linux_detector.py b/core/drag_drop/linux_detector.py deleted file mode 100644 index cb0162a7..00000000 --- a/core/drag_drop/linux_detector.py +++ /dev/null @@ -1,395 +0,0 @@ -""" -Linux平台拖拽检测器 - -支持X11和Wayland窗口系统的拖拽检测。 -""" - -import os -import sys -import time -import threading -import subprocess -from typing import List, Optional -from .base_detector import BaseDragDetector - - -class LinuxDragDetector(BaseDragDetector): - """Linux平台拖拽检测器""" - - def __init__(self, supported_formats: List[str] = None): - """ - 初始化Linux拖拽检测器 - - Args: - supported_formats: 支持的文件格式列表 - """ - super().__init__(supported_formats) - self._window_id = None - self._display = None - self._is_x11 = self._check_x11() - self._is_wayland = self._check_wayland() - self._xdg_active = self._check_xdg() - - def _check_x11(self) -> bool: - """检查是否使用X11""" - try: - # 检查DISPLAY环境变量 - if os.environ.get('DISPLAY'): - # 尝试连接X11显示 - result = subprocess.run(['xset', 'q'], - capture_output=True, - timeout=2) - return result.returncode == 0 - except (subprocess.TimeoutExpired, FileNotFoundError): - pass - return False - - def _check_wayland(self) -> bool: - """检查是否使用Wayland""" - return os.environ.get('WAYLAND_DISPLAY') is not None - - def _check_xdg(self) -> bool: - """检查XDG桌面环境是否可用""" - try: - # 检查是否安装了xdg-user-dir - result = subprocess.run(['which', 'xdg-user-dir'], - capture_output=True) - return result.returncode == 0 - except FileNotFoundError: - return False - - def _get_window_id(self) -> Optional[str]: - """获取当前窗口ID""" - try: - # 使用xdotool获取活动窗口 - result = subprocess.run(['xdotool', 'getactivewindow'], - capture_output=True, - text=True, - timeout=2) - if result.returncode == 0: - return result.stdout.strip() - except (subprocess.TimeoutExpired, FileNotFoundError): - pass - - # 备用方法：使用wmctrl - try: - result = subprocess.run(['wmctrl', '-l'], - capture_output=True, - text=True, - timeout=2) - if result.returncode == 0: - lines = result.stdout.strip().split('\n') - if lines: - # 获取第一个窗口的ID - return lines[0].split()[0] - except (subprocess.TimeoutExpired, FileNotFoundError): - pass - - return None - - def _monitor_x11_drag(self): - """监控X11拖拽事件""" - try: - # 使用xev监控窗口事件 - window_id = self._get_window_id() - if not window_id: - print("无法获取窗口ID，切换到文件监控") - self._monitor_file_drag() - return - - cmd = ['xev', '-id', window_id, '-event', 'pointer'] - process = subprocess.Popen(cmd, - stdout=subprocess.PIPE, - stderr=subprocess.DEVNULL, - text=True) - - print(f"开始监控X11窗口 {window_id} 的拖拽事件") - - while self.is_running: - line = process.stdout.readline() - if not line: - break - - # 检测拖拽相关事件 - if any(keyword in line for keyword in ['DND', 'Xdnd', 'ClientMessage', 'SelectionNotify']): - # 尝试获取拖拽的文件 - self._extract_x11_drag_files() - - except Exception as e: - print(f"X11拖拽监控错误: {e}") - print("切换到文件监控模式") - self._monitor_file_drag() - finally: - if 'process' in locals(): - process.terminate() - - def _extract_x11_drag_files(self): - """从X11拖拽事件中提取文件路径""" - try: - # 使用xclip获取剪贴板内容（可能包含拖拽的文件） - result = subprocess.run(['xclip', '-selection', 'primary', '-o'], - capture_output=True, - text=True, - timeout=1) - if result.returncode == 0 and result.stdout.strip(): - self._process_clipboard_content(result.stdout.strip()) - - except (subprocess.TimeoutExpired, FileNotFoundError): - # 尝试使用xsel作为备选 - try: - result = subprocess.run(['xsel', '--primary', '--output'], - capture_output=True, - text=True, - timeout=1) - if result.returncode == 0 and result.stdout.strip(): - self._process_clipboard_content(result.stdout.strip()) - except (subprocess.TimeoutExpired, FileNotFoundError): - pass - - def _process_clipboard_content(self, content): - """处理剪贴板内容，提取文件路径""" - # 检查是否包含文件路径 - lines = content.split('\n') - for line in lines: - line = line.strip() - if line.startswith('file://'): - # URI格式 - file_path = line[7:] # 移除 'file://' 前缀 - file_path = file_path.replace('%20', ' ') # 解码空格 - if self._is_supported_format(file_path): - self.add_dropped_file(file_path) - elif os.path.isabs(line) and self._is_supported_format(line): - # 绝对路径 - self.add_dropped_file(line) - - def _monitor_wayland_drag(self): - """监控Wayland拖拽事件""" - try: - # 尝试使用wlrctl监控拖拽事件 - if self._check_command_exists('wlrctl'): - self._monitor_wlrctl_drag() - else: - # 尝试使用wl-paste监控剪贴板 - if self._check_command_exists('wl-paste'): - self._monitor_wayland_clipboard() - else: - print("Wayland拖拽监控工具不可用，切换到文件监控") - self._monitor_file_drag() - except Exception as e: - print(f"Wayland拖拽监控错误: {e}") - print("切换到文件监控模式") - self._monitor_file_drag() - - def _check_command_exists(self, command): - """检查命令是否存在""" - try: - result = subprocess.run(['which', command], - capture_output=True, - timeout=1) - return result.returncode == 0 - except (subprocess.TimeoutExpired, FileNotFoundError): - return False - - def _monitor_wlrctl_drag(self): - """使用wlrctl监控拖拽事件""" - print("使用wlrctl监控Wayland拖拽事件") - try: - # wlrctl可能支持拖拽监控 - process = subprocess.Popen(['wlrctl', 'drag'], - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True) - - while self.is_running: - line = process.stdout.readline() - if not line: - break - - # 解析wlrctl输出 - if 'file://' in line or os.path.isabs(line.strip()): - self._process_drag_line(line.strip()) - - except Exception as e: - print(f"wlrctl监控失败: {e}") - raise - finally: - if 'process' in locals(): - process.terminate() - - def _monitor_wayland_clipboard(self): - """监控Wayland剪贴板变化""" - print("使用wl-paste监控Wayland剪贴板") - last_clipboard = "" - - while self.is_running: - try: - # 获取当前剪贴板内容 - result = subprocess.run(['wl-paste', '--type', 'text/uri-list'], - capture_output=True, - text=True, - timeout=1) - - if result.returncode == 0: - current_clipboard = result.stdout.strip() - if current_clipboard and current_clipboard != last_clipboard: - last_clipboard = current_clipboard - self._process_clipboard_content(current_clipboard) - - except subprocess.TimeoutExpired: - pass - except Exception as e: - print(f"Wayland剪贴板监控错误: {e}") - - time.sleep(0.5) - - def _process_drag_line(self, line): - """处理拖拽行内容""" - if line.startswith('file://'): - file_path = line[7:] - file_path = file_path.replace('%20', ' ') - if self._is_supported_format(file_path): - self.add_dropped_file(file_path) - elif os.path.isabs(line) and self._is_supported_format(line): - self.add_dropped_file(line) - - def _get_user_directories(self): - """获取用户常用目录""" - directories = [] - - # 获取桌面目录 - desktop = os.path.expanduser("~/Desktop") - if os.path.exists(desktop): - directories.append(desktop) - - # 获取下载目录 - downloads = os.path.expanduser("~/Downloads") - if os.path.exists(downloads): - directories.append(downloads) - - # 获取主目录 - home = os.path.expanduser("~") - if os.path.exists(home): - directories.append(home) - - # 获取临时目录 - temp_dirs = ['/tmp', '/var/tmp'] - for temp_dir in temp_dirs: - if os.path.exists(temp_dir): - directories.append(temp_dir) - - return directories - - def _monitor_file_drag(self): - """通过文件系统变化监控拖拽""" - watch_dirs = self._get_user_directories() - known_files = {} - recent_files = {} # 用于跟踪最近添加的文件 - - # 初始化已知文件 - for watch_dir in watch_dirs: - if os.path.exists(watch_dir): - for filename in os.listdir(watch_dir): - filepath = os.path.join(watch_dir, filename) - if self._is_supported_format(filepath): - try: - stat = os.stat(filepath) - known_files[filepath] = stat.st_mtime - except OSError: - pass - - print(f"开始监控目录: {watch_dirs}") - print(f"已知文件数量: {len(known_files)}") - - while self.is_running: - try: - current_time = time.time() - - for watch_dir in watch_dirs: - if not os.path.exists(watch_dir): - continue - - try: - # 获取当前目录中的文件 - current_files = {} - for filename in os.listdir(watch_dir): - filepath = os.path.join(watch_dir, filename) - if self._is_supported_format(filepath): - try: - stat = os.stat(filepath) - current_files[filepath] = stat.st_mtime - - # 检查是否是新文件（最近5秒内创建） - if filepath not in known_files: - time_diff = current_time - stat.st_mtime - if time_diff < 5.0: - # 检查文件大小是否稳定（避免检测到正在下载的文件） - if self._is_file_stable(filepath): - print(f"检测到新文件: {filepath}") - recent_files[filepath] = current_time - self.add_dropped_file(filepath) - # 检查已知文件是否被更新 - elif filepath in known_files: - if stat.st_mtime > known_files[filepath]: - time_diff = current_time - stat.st_mtime - if time_diff < 5.0 and filepath not in recent_files: - print(f"检测到文件更新: {filepath}") - recent_files[filepath] = current_time - self.add_dropped_file(filepath) - - except OSError: - continue - - # 更新已知文件列表 - known_files.update(current_files) - - except PermissionError: - print(f"无权限访问目录: {watch_dir}") - continue - except Exception as e: - print(f"监控目录 {watch_dir} 时出错: {e}") - - # 清理过期的最近文件记录（超过30秒） - expired_files = [path for path, timestamp in recent_files.items() - if current_time - timestamp > 30] - for path in expired_files: - del recent_files[path] - - except Exception as e: - print(f"文件拖拽监控错误: {e}") - - time.sleep(0.5) - - def _is_file_stable(self, filepath, check_interval=0.5, max_checks=3): - """检查文件大小是否稳定（避免检测到正在下载的文件）""" - try: - initial_size = os.path.getsize(filepath) - for i in range(max_checks): - time.sleep(check_interval) - current_size = os.path.getsize(filepath) - if current_size != initial_size: - return False - return True - except OSError: - return False - - def _monitor_loop(self): - """主监控循环""" - if self._is_x11: - try: - self._monitor_x11_drag() - except Exception as e: - print(f"X11监控失败，切换到文件监控: {e}") - self._monitor_file_drag() - elif self._is_wayland: - try: - self._monitor_wayland_drag() - except Exception as e: - print(f"Wayland监控失败，切换到文件监控: {e}") - self._monitor_file_drag() - else: - # 降级到文件监控 - self._monitor_file_drag() - - def is_supported(self) -> bool: - """检查Linux平台是否支持""" - return self._is_x11 or self._is_wayland or self._xdg_active \ No newline at end of file diff --git a/core/drag_drop/macos_detector.py b/core/drag_drop/macos_detector.py deleted file mode 100644 index b8f3bdb0..00000000 --- a/core/drag_drop/macos_detector.py +++ /dev/null @@ -1,105 +0,0 @@ -""" -macOS平台拖拽检测器 - -使用macOS Cocoa API实现系统级拖拽检测。 -""" - -import os -import sys -import time -import threading -from typing import List, Optional -from .base_detector import BaseDragDetector - - -class MacOSDragDetector(BaseDragDetector): - """macOS平台拖拽检测器""" - - def __init__(self, supported_formats: List[str] = None): - """ - 初始化macOS拖拽检测器 - - Args: - supported_formats: 支持的文件格式列表 - """ - super().__init__(supported_formats) - self._app_kit_available = self._check_app_kit() - self._cocoa_available = self._check_cocoa() - - def _check_app_kit(self) -> bool: - """检查AppKit是否可用""" - try: - import AppKit - return True - except ImportError: - return False - - def _check_cocoa(self) -> bool: - """检查Cocoa是否可用""" - try: - import Cocoa - return True - except ImportError: - return False - - def _setup_cocoa_drag_drop(self) -> bool: - """设置Cocoa拖拽接收""" - try: - if not self._app_kit_available: - return False - - import AppKit - - # 创建拖拽接收器 - # 这里需要实现macOS的拖拽API调用 - # 暂时返回False，需要进一步实现 - return False - except Exception as e: - print(f"设置Cocoa拖拽失败: {e}") - return False - - def _monitor_file_drag(self): - """通过文件系统变化监控拖拽""" - # 监控用户桌面和下载目录 - desktop_path = os.path.expanduser("~/Desktop") - downloads_path = os.path.expanduser("~/Downloads") - - watch_dirs = [desktop_path, downloads_path] - - while self.is_running: - try: - current_time = time.time() - for watch_dir in watch_dirs: - if not os.path.exists(watch_dir): - continue - - for filename in os.listdir(watch_dir): - filepath = os.path.join(watch_dir, filename) - try: - file_time = os.path.getmtime(filepath) - # 如果文件是最近1秒内创建的 - if current_time - file_time < 1.0: - if self._is_supported_format(filepath): - self.add_dropped_file(filepath) - except OSError: - continue - - except Exception as e: - print(f"文件拖拽监控错误: {e}") - - time.sleep(0.5) - - def _monitor_loop(self): - """主监控循环""" - if self._app_kit_available and self._cocoa_available: - # 尝试设置真正的拖拽接收 - if not self._setup_cocoa_drag_drop(): - # 降级到文件系统监控 - self._monitor_file_drag() - else: - # 降级到文件系统监控 - self._monitor_file_drag() - - def is_supported(self) -> bool: - """检查macOS平台是否支持""" - return sys.platform.startswith('darwin') \ No newline at end of file diff --git a/core/drag_drop/platform_detector.py b/core/drag_drop/platform_detector.py deleted file mode 100644 index abb5a1a0..00000000 --- a/core/drag_drop/platform_detector.py +++ /dev/null @@ -1,66 +0,0 @@ -""" -平台拖拽检测器 - -自动选择合适的平台特定拖拽检测器。 -""" - -from .base_detector import BaseDragDetector, DragDetectorFactory -from typing import List - - -class PlatformDragDetector: - """平台拖拽检测器包装类""" - - def __init__(self, supported_formats: List[str] = None): - """ - 初始化平台拖拽检测器 - - Args: - supported_formats: 支持的文件格式列表 - """ - self.supported_formats = supported_formats or ['.gltf', '.glb', '.fbx', '.bam', '.egg', '.obj'] - self.detector = DragDetectorFactory.create_detector(self.supported_formats) - self._is_monitoring = False - - def set_drop_callback(self, callback): - """设置拖拽回调函数""" - if self.detector: - self.detector.set_drop_callback(callback) - - def start_monitoring(self): - """开始监控""" - if self.detector and not self._is_monitoring: - self.detector.start_monitoring() - self._is_monitoring = True - - def stop_monitoring(self): - """停止监控""" - if self.detector and self._is_monitoring: - self.detector.stop_monitoring() - self._is_monitoring = False - - def get_dropped_files(self): - """获取拖拽的文件""" - if self.detector: - return self.detector.get_dropped_files() - return [] - - def is_supported(self) -> bool: - """检查当前平台是否支持""" - return self.detector.is_supported() if self.detector else False - - def add_dropped_file(self, file_path: str): - """添加拖拽的文件路径""" - if self.detector: - self.detector.add_dropped_file(file_path) - - def get_platform_info(self) -> dict: - """获取平台信息""" - import platform - return { - 'system': platform.system(), - 'release': platform.release(), - 'version': platform.version(), - 'detector_type': type(self.detector).__name__, - 'supported': self.is_supported() - } \ No newline at end of file diff --git a/core/drag_drop/simple_detector.py b/core/drag_drop/simple_detector.py deleted file mode 100644 index cc7edc15..00000000 --- a/core/drag_drop/simple_detector.py +++ /dev/null @@ -1,233 +0,0 @@ -#!/usr/bin/env python3 -""" -简化的拖拽检测器 - -使用文件系统监控和剪贴板检测实现拖拽功能 -""" - -import os -import sys -import time -import subprocess -import threading -from typing import List, Optional, Callable -from collections import deque -import tempfile - -class SimpleDragDetector: - """简化的拖拽检测器""" - - def __init__(self, supported_formats: List[str]): - self.supported_formats = supported_formats - self.is_running = False - self.drop_callback = None - self.dropped_files = deque() - self.monitor_thread = None - self.last_clipboard = "" - - def set_drop_callback(self, callback: Callable[[List[str]], None]): - """设置拖拽回调函数""" - self.drop_callback = callback - - def start_monitoring(self): - """开始监控""" - if self.is_running: - return - - self.is_running = True - self.monitor_thread = threading.Thread(target=self._monitor_loop) - self.monitor_thread.daemon = True - self.monitor_thread.start() - print("✓ 简化拖拽监控已启动") - - def stop_monitoring(self): - """停止监控""" - self.is_running = False - if self.monitor_thread: - self.monitor_thread.join(timeout=1) - - def get_dropped_files(self) -> List[str]: - """获取拖拽的文件列表""" - files = list(self.dropped_files) - self.dropped_files.clear() - return files - - def add_dropped_file(self, file_path: str): - """添加拖拽的文件""" - if os.path.exists(file_path): - file_ext = os.path.splitext(file_path)[1].lower() - if file_ext in self.supported_formats: - self.dropped_files.append(file_path) - print(f"检测到拖拽文件: {file_path}") - - def _monitor_loop(self): - """监控循环""" - watch_dirs = self._get_watch_directories() - known_files = set() - - # 初始化已知文件 - for watch_dir in watch_dirs: - self._scan_directory(watch_dir, known_files) - - print(f"监控目录: {watch_dirs}") - print(f"已知文件数量: {len(known_files)}") - - while self.is_running: - try: - current_time = time.time() - - # 1. 检查剪贴板变化 - self._check_clipboard() - - # 2. 检查文件系统变化 - for watch_dir in watch_dirs: - if not os.path.exists(watch_dir): - continue - - try: - current_files = set() - for filename in os.listdir(watch_dir): - filepath = os.path.join(watch_dir, filename) - if self._is_supported_format(filepath): - try: - stat = os.stat(filepath) - current_files.add(filepath) - - # 检查新文件 - if filepath not in known_files: - time_diff = current_time - stat.st_mtime - if time_diff < 5.0: # 5秒内创建的文件 - if self._is_file_stable(filepath): - print(f"检测到新文件: {filepath}") - self.add_dropped_file(filepath) - if self.drop_callback: - self.drop_callback([filepath]) - - except OSError: - continue - - known_files.update(current_files) - - except PermissionError: - continue - except Exception as e: - print(f"监控目录 {watch_dir} 时出错: {e}") - - except Exception as e: - print(f"监控循环错误: {e}") - - time.sleep(0.3) # 更频繁的检查 - - def _check_clipboard(self): - """检查剪贴板变化""" - try: - # 检查剪贴板 - result = subprocess.run(['xclip', '-selection', 'clipboard', '-o'], - capture_output=True, text=True, timeout=0.5) - if result.returncode == 0 and result.stdout.strip(): - current_clipboard = result.stdout.strip() - if current_clipboard != self.last_clipboard: - self.last_clipboard = current_clipboard - files = self._parse_clipboard_content(current_clipboard) - if files: - print(f"剪贴板检测到文件: {files}") - for file_path in files: - self.add_dropped_file(file_path) - if self.drop_callback: - self.drop_callback(files) - - except Exception: - # 尝试使用xsel - try: - result = subprocess.run(['xsel', '--clipboard', '--output'], - capture_output=True, text=True, timeout=0.5) - if result.returncode == 0 and result.stdout.strip(): - current_clipboard = result.stdout.strip() - if current_clipboard != self.last_clipboard: - self.last_clipboard = current_clipboard - files = self._parse_clipboard_content(current_clipboard) - if files: - for file_path in files: - self.add_dropped_file(file_path) - if self.drop_callback: - self.drop_callback(files) - except Exception: - pass - - def _parse_clipboard_content(self, content: str) -> List[str]: - """解析剪贴板内容""" - files = [] - for line in content.split('\n'): - line = line.strip() - if line.startswith('file://'): - file_path = line[7:] - file_path = file_path.replace('%20', ' ') - if os.path.exists(file_path): - files.append(file_path) - elif os.path.isabs(line) and os.path.exists(line): - files.append(line) - return files - - def _get_watch_directories(self) -> List[str]: - """获取监控目录""" - dirs = [ - os.path.expanduser('~/Desktop'), - os.path.expanduser('~/Downloads'), - tempfile.gettempdir() - ] - - # 添加一些常见的拖拽目标目录 - additional_dirs = [ - os.path.expanduser('~/Documents'), - os.path.expanduser('~'), - '/tmp' - ] - - for dir_path in additional_dirs: - if os.path.exists(dir_path) and os.path.isdir(dir_path): - dirs.append(dir_path) - - return [d for d in dirs if os.path.exists(d)] - - def _scan_directory(self, directory: str, known_files: set): - """扫描目录""" - try: - for filename in os.listdir(directory): - filepath = os.path.join(directory, filename) - if self._is_supported_format(filepath): - try: - known_files.add(filepath) - except OSError: - pass - except Exception: - pass - - def _is_supported_format(self, file_path: str) -> bool: - """检查文件格式是否支持""" - if not os.path.isfile(file_path): - return False - file_ext = os.path.splitext(file_path)[1].lower() - return file_ext in self.supported_formats - - def _is_file_stable(self, file_path: str, wait_time: float = 0.5) -> bool: - """检查文件是否稳定""" - try: - initial_size = os.path.getsize(file_path) - time.sleep(wait_time) - current_size = os.path.getsize(file_path) - return initial_size == current_size - except: - return False - - def is_supported(self) -> bool: - """检查是否支持""" - return True # 这个版本总是支持的 - - def get_platform_info(self) -> dict: - """获取平台信息""" - return { - 'system': 'Linux', - 'detector_type': 'SimpleDragDetector', - 'supported': True, - 'method': 'File system + clipboard monitoring' - } diff --git a/core/drag_drop/windows_detector.py b/core/drag_drop/windows_detector.py deleted file mode 100644 index 12945781..00000000 --- a/core/drag_drop/windows_detector.py +++ /dev/null @@ -1,221 +0,0 @@ -""" -Windows平台拖拽检测器 - -使用Windows API实现系统级拖拽检测。 -""" - -import os -import sys -import time -import threading -from typing import List, Optional -from .base_detector import BaseDragDetector - - -class WindowsDragDetector(BaseDragDetector): - """Windows平台拖拽检测器""" - - def __init__(self, supported_formats: List[str] = None): - """ - 初始化Windows拖拽检测器 - - Args: - supported_formats: 支持的文件格式列表 - """ - super().__init__(supported_formats) - self._hwnd = None - self._old_win_proc = None - self._is_com_initialized = False - self._ole_initialized = False - - def _initialize_com(self) -> bool: - """初始化COM""" - try: - import ctypes - from ctypes import wintypes - - # 初始化COM - ctypes.windll.ole32.CoInitializeEx(None, 0) # COINIT_APARTMENTTHREADED - self._is_com_initialized = True - return True - except Exception as e: - print(f"COM初始化失败: {e}") - return False - - def _initialize_ole(self) -> bool: - """初始化OLE""" - try: - import ctypes - from ctypes import wintypes - - # 初始化OLE - ctypes.windll.ole32.OleInitialize(None) - self._ole_initialized = True - return True - except Exception as e: - print(f"OLE初始化失败: {e}") - return False - - def _get_window_handle(self) -> Optional[int]: - """获取Panda3D窗口句柄""" - try: - # 这里需要获取Panda3D窗口的句柄 - # 可能需要通过Panda3D的API获取 - # 暂时返回None，需要进一步实现 - return None - except Exception as e: - print(f"获取窗口句柄失败: {e}") - return None - - def _setup_drag_drop(self) -> bool: - """设置拖拽接收""" - try: - import ctypes - from ctypes import wintypes - - hwnd = self._get_window_handle() - if not hwnd: - return False - - # 注册拖拽接收窗口 - # 这里需要实现Windows拖拽API的调用 - # 暂时返回False，需要进一步实现 - return False - except Exception as e: - print(f"设置拖拽接收失败: {e}") - return False - - def _monitor_clipboard_drag(self): - """通过剪贴板监控拖拽""" - try: - import ctypes - from ctypes import wintypes - - user32 = ctypes.windll.user32 - kernel32 = ctypes.windll.kernel32 - - # 监控剪贴板变化 - while self.is_running: - # 检查剪贴板是否包含文件 - if user32.OpenClipboard(None): - try: - # 检查是否是文件格式 - format_id = 15 # CF_HDROP - if user32.IsClipboardFormatAvailable(format_id): - # 获取拖拽文件列表 - h_drop = user32.GetClipboardData(format_id) - if h_drop: - files = self._parse_drop_files(h_drop) - for file_path in files: - self.add_dropped_file(file_path) - finally: - user32.CloseClipboard() - - time.sleep(0.1) - - except Exception as e: - print(f"剪贴板监控错误: {e}") - - def _parse_drop_files(self, h_drop) -> List[str]: - """解析拖拽文件列表""" - try: - import ctypes - from ctypes import wintypes - - kernel32 = ctypes.windll.kernel32 - shell32 = ctypes.windll.shell32 - - # 获取文件数量 - file_count = shell32.DragQueryFileW(h_drop, 0xFFFFFFFF, None, 0) - files = [] - - # 获取每个文件路径 - for i in range(file_count): - # 获取文件路径长度 - length = shell32.DragQueryFileW(h_drop, i, None, 0) - if length > 0: - # 分配缓冲区 - buffer = ctypes.create_unicode_buffer(length + 1) - # 获取文件路径 - shell32.DragQueryFileW(h_drop, i, buffer, length + 1) - files.append(buffer.value) - - return files - except Exception as e: - print(f"解析拖拽文件失败: {e}") - return [] - - def _monitor_file_drag(self): - """通过文件系统变化监控拖拽""" - # 监控用户桌面和下载目录 - import winreg - - try: - # 获取桌面路径 - key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, - r"Software\Microsoft\Windows\CurrentVersion\Explorer\Shell Folders") - desktop_path = winreg.QueryValueEx(key, "Desktop")[0] - downloads_path = winreg.QueryValueEx(key, "{374DE290-123F-4565-9164-39C4925E467B}")[0] # 下载文件夹 - winreg.CloseKey(key) - except Exception: - # 降级到默认路径 - desktop_path = os.path.expanduser("~/Desktop") - downloads_path = os.path.expanduser("~/Downloads") - - watch_dirs = [desktop_path, downloads_path] - - while self.is_running: - try: - current_time = time.time() - for watch_dir in watch_dirs: - if not os.path.exists(watch_dir): - continue - - for filename in os.listdir(watch_dir): - filepath = os.path.join(watch_dir, filename) - try: - file_time = os.path.getmtime(filepath) - # 如果文件是最近1秒内创建的 - if current_time - file_time < 1.0: - if self._is_supported_format(filepath): - self.add_dropped_file(filepath) - except OSError: - continue - - except Exception as e: - print(f"文件拖拽监控错误: {e}") - - time.sleep(0.5) - - def _monitor_loop(self): - """主监控循环""" - # 尝试初始化COM和OLE - com_ready = self._initialize_com() - ole_ready = self._initialize_ole() - - if com_ready and ole_ready: - # 尝试设置真正的拖拽接收 - if self._setup_drag_drop(): - # 使用Windows API拖拽检测 - pass - else: - # 降级到剪贴板监控 - self._monitor_clipboard_drag() - else: - # 降级到文件系统监控 - self._monitor_file_drag() - - # 清理COM和OLE - try: - if self._ole_initialized: - import ctypes - ctypes.windll.ole32.OleUninitialize() - if self._is_com_initialized: - import ctypes - ctypes.windll.ole32.CoUninitialize() - except Exception: - pass - - def is_supported(self) -> bool: - """检查Windows平台是否支持""" - return sys.platform.startswith('win') \ No newline at end of file diff --git a/core/drag_drop/x11_drag_receiver.py b/core/drag_drop/x11_drag_receiver.py deleted file mode 100644 index 1031e3dc..00000000 --- a/core/drag_drop/x11_drag_receiver.py +++ /dev/null @@ -1,311 +0,0 @@ -#!/usr/bin/env python3 -""" -X11窗口拖拽事件接收器 - -使用X11的Xdnd协议实现真正的窗口拖拽功能 -""" - -import os -import sys -import time -import subprocess -import threading -from typing import List, Optional, Callable -from collections import deque - -class X11DragReceiver: - """X11窗口拖拽事件接收器""" - - def __init__(self, supported_formats: List[str]): - self.supported_formats = supported_formats - self.window_id = None - self.is_running = False - self.drop_callback = None - self.dropped_files = deque() - self.monitor_thread = None - - # X11相关 - self.x11_display = os.environ.get('DISPLAY', ':0') - self.xdnd_aware = True # 标记窗口支持拖拽 - - def set_window(self, window_id: int): - """设置要监控的窗口ID""" - self.window_id = window_id - print(f"设置拖拽监控窗口: {window_id}") - - def set_drop_callback(self, callback: Callable[[List[str]], None]): - """设置拖拽回调函数""" - self.drop_callback = callback - - def start_monitoring(self): - """开始监控拖拽事件""" - if self.is_running: - return - - self.is_running = True - self.monitor_thread = threading.Thread(target=self._monitor_drag_events) - self.monitor_thread.daemon = True - self.monitor_thread.start() - print("X11拖拽监控已启动") - - def stop_monitoring(self): - """停止监控""" - self.is_running = False - if self.monitor_thread: - self.monitor_thread.join(timeout=1) - - def get_dropped_files(self) -> List[str]: - """获取拖拽的文件列表""" - files = list(self.dropped_files) - self.dropped_files.clear() - return files - - def add_dropped_file(self, file_path: str): - """添加拖拽的文件""" - if os.path.exists(file_path): - file_ext = os.path.splitext(file_path)[1].lower() - if file_ext in self.supported_formats: - self.dropped_files.append(file_path) - print(f"检测到拖拽文件: {file_path}") - - def _monitor_drag_events(self): - """监控拖拽事件""" - if not self.window_id: - print("窗口ID未设置，无法监控拖拽事件") - return - - try: - # 首先注册窗口为拖拽目标 - self._register_drag_target() - - # 使用xev监控窗口事件 - cmd = ['xev', '-id', str(self.window_id), '-event', 'dnd'] - process = subprocess.Popen(cmd, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True) - - print(f"开始监控窗口 {self.window_id} 的拖拽事件") - - while self.is_running: - line = process.stdout.readline() - if not line: - break - - # 解析拖拽事件 - if 'XdndEnter' in line: - self._handle_drag_enter(line) - elif 'XdndPosition' in line: - self._handle_drag_position(line) - elif 'XdndDrop' in line: - self._handle_drag_drop(line) - elif 'XdndLeave' in line: - self._handle_drag_leave(line) - - except Exception as e: - print(f"拖拽事件监控错误: {e}") - # 降级到文件监控 - self._fallback_to_file_monitor() - finally: - if 'process' in locals(): - process.terminate() - - def _register_drag_target(self): - """注册窗口为拖拽目标""" - try: - # 使用xprop设置窗口属性，标记支持拖拽 - cmd = ['xprop', '-id', str(self.window_id), '-f', '_NET_WM_WINDOW_TYPE', '32a', - '-set', '_NET_WM_WINDOW_TYPE', '_NET_WM_WINDOW_TYPE_NORMAL'] - subprocess.run(cmd, capture_output=True, timeout=2) - - # 设置XdndAware属性 - cmd = ['xprop', '-id', str(self.window_id), '-f', 'XdndAware', '32c', - '-set', 'XdndAware', '5'] - subprocess.run(cmd, capture_output=True, timeout=2) - - print(f"窗口 {self.window_id} 已注册为拖拽目标") - - except Exception as e: - print(f"注册拖拽目标失败: {e}") - - def _handle_drag_enter(self, event_line: str): - """处理拖拽进入事件""" - print("检测到拖拽进入") - - def _handle_drag_position(self, event_line: str): - """处理拖拽位置事件""" - # 可以在这里更新拖拽位置显示 - pass - - def _handle_drag_drop(self, event_line: str): - """处理拖拽释放事件""" - print("检测到拖拽释放") - - # 尝试获取拖拽的文件 - files = self._extract_dropped_files() - if files: - for file_path in files: - self.add_dropped_file(file_path) - - # 调用回调函数 - if self.drop_callback: - self.drop_callback(list(self.dropped_files)) - - def _handle_drag_leave(self, event_line: str): - """处理拖拽离开事件""" - print("拖拽离开窗口") - - def _extract_dropped_files(self) -> List[str]: - """提取拖拽的文件路径""" - files = [] - - try: - # 方法1：尝试从剪贴板获取 - result = subprocess.run(['xclip', '-selection', 'clipboard', '-o'], - capture_output=True, text=True, timeout=1) - if result.returncode == 0 and result.stdout.strip(): - files.extend(self._parse_uri_list(result.stdout.strip())) - - # 方法2：尝试从主选择获取 - result = subprocess.run(['xclip', '-selection', 'primary', '-o'], - capture_output=True, text=True, timeout=1) - if result.returncode == 0 and result.stdout.strip(): - files.extend(self._parse_uri_list(result.stdout.strip())) - - # 方法3：使用xsel作为备选 - if not files: - result = subprocess.run(['xsel', '--clipboard', '--output'], - capture_output=True, text=True, timeout=1) - if result.returncode == 0 and result.stdout.strip(): - files.extend(self._parse_uri_list(result.stdout.strip())) - - except Exception as e: - print(f"提取拖拽文件失败: {e}") - - return files - - def _parse_uri_list(self, uri_list: str) -> List[str]: - """解析URI列表，提取文件路径""" - files = [] - - for line in uri_list.split('\n'): - line = line.strip() - if not line or line.startswith('#'): - continue - - if line.startswith('file://'): - # URI格式 - file_path = line[7:] # 移除 'file://' 前缀 - file_path = file_path.replace('%20', ' ') # 解码空格 - file_path = file_path.strip() - - # 移除可能的回车符 - if file_path.endswith('\r'): - file_path = file_path[:-1] - - if os.path.exists(file_path): - files.append(file_path) - elif os.path.isabs(line): - # 绝对路径 - if os.path.exists(line): - files.append(line) - - return files - - def _fallback_to_file_monitor(self): - """降级到文件监控""" - print("降级到文件监控模式") - - # 监控桌面和下载目录 - watch_dirs = [ - os.path.expanduser('~/Desktop'), - os.path.expanduser('~/Downloads'), - '/tmp' - ] - - known_files = set() - - # 初始化已知文件 - for watch_dir in watch_dirs: - if os.path.exists(watch_dir): - for filename in os.listdir(watch_dir): - filepath = os.path.join(watch_dir, filename) - if self._is_supported_format(filepath): - try: - known_files.add(filepath) - except OSError: - pass - - while self.is_running: - try: - current_time = time.time() - - for watch_dir in watch_dirs: - if not os.path.exists(watch_dir): - continue - - try: - for filename in os.listdir(watch_dir): - filepath = os.path.join(watch_dir, filename) - if self._is_supported_format(filepath): - try: - file_time = os.path.getmtime(filepath) - - # 检查是否是新文件（最近5秒内创建） - if filepath not in known_files and current_time - file_time < 5.0: - print(f"检测到新文件: {filepath}") - self.add_dropped_file(filepath) - - # 调用回调函数 - if self.drop_callback: - self.drop_callback([filepath]) - - known_files.add(filepath) - - except OSError: - continue - - except PermissionError: - continue - except Exception as e: - print(f"监控目录 {watch_dir} 时出错: {e}") - - except Exception as e: - print(f"文件监控错误: {e}") - - time.sleep(0.5) - - def _is_supported_format(self, file_path: str) -> bool: - """检查文件格式是否支持""" - if not os.path.isfile(file_path): - return False - - file_ext = os.path.splitext(file_path)[1].lower() - return file_ext in self.supported_formats - - def is_supported(self) -> bool: - """检查是否支持拖拽功能""" - # 检查X11环境 - if not os.environ.get('DISPLAY'): - return False - - # 检查必要工具 - required_tools = ['xev', 'xprop', 'xclip'] - for tool in required_tools: - try: - subprocess.run(['which', tool], capture_output=True, check=True) - except subprocess.CalledProcessError: - print(f"缺少必要工具: {tool}") - return False - - return True - - def get_platform_info(self) -> dict: - """获取平台信息""" - return { - 'system': 'Linux', - 'detector_type': 'X11DragReceiver', - 'supported': self.is_supported(), - 'display': self.x11_display, - 'window_id': self.window_id - } \ No newline at end of file