395 lines
16 KiB
Python
395 lines
16 KiB
Python
"""
|
||
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 |