init repo

This commit is contained in:
2026-09-17 16:30:02 +08:00
commit 0ba53cc7e3
163 changed files with 33634 additions and 0 deletions
+7
View File
@@ -0,0 +1,7 @@
# -*- coding: utf-8 -*-
"""
系统级工具集合 (system_tools)
- screen_capture: 屏幕截图覆盖层
- global_hotkey: 全局热键监听线程
- file_reader: 文本/代码文件读取(编码探测 + 二进制探测)
"""
+76
View File
@@ -0,0 +1,76 @@
# -*- coding: utf-8 -*-
"""P1-04:桌面会话探测 + 截图/全局热键能力路由(纯 stdlib,可在任何平台导入)。
契约(REPAIR_BACKLOG P1-04):
- Windows 保持现有行为(Win32 RegisterHotKey + Qt grabWindow 覆盖层)。
- Linux X11:原生全局快捷键(XGrabKey,见 x11_hotkey.py+ 原生屏幕捕获(Qt grabWindowX11 可用)。
- Linux Wayland:截图走 xdg-desktop-portal(用户授权,不绕过 compositor,见 portal_capture.py);
全局热键依赖 compositor 桌面协议(ext-global-shortcut 等),本版本无免依赖实现 →
明确告知不可用,仅保留应用内 Alt+S 快捷键,主程序其余功能不受影响。
- offscreen/无显示(unknown):能力不可用要有明确日志,主程序仍可聊天。
"""
import os
import sys
def session_kind() -> str:
"""返回 "win32" / "x11" / "wayland" / "unknown"
判定顺序(Linux,综合 Qt 平台名与 XDG_SESSION_TYPE,见 PLATFORM_PLAN):
1. QT_QPA_PLATFORM 以 offscreen 开头 → unknown(自动化离屏,无桌面能力)
2. QT_QPA_PLATFORM 以 wayland 开头,或 WAYLAND_DISPLAY 已设置,
或 XDG_SESSION_TYPE=wayland → wayland
3. DISPLAY 已设置,或 XDG_SESSION_TYPE=x11 → x11
4. 其余(headless/无显示)→ unknown
"""
if sys.platform == "win32":
return "win32"
if sys.platform != "linux":
# 其他 POSIX(macOS 等)不在本任务支持矩阵内
return "unknown"
qt_plat = os.environ.get("QT_QPA_PLATFORM", "").strip()
if qt_plat.startswith("offscreen"):
return "unknown"
xdg_type = os.environ.get("XDG_SESSION_TYPE", "").strip().lower()
if (qt_plat.startswith("wayland")
or os.environ.get("WAYLAND_DISPLAY", "").strip()
or xdg_type == "wayland"):
return "wayland"
if os.environ.get("DISPLAY", "").strip() or xdg_type == "x11":
return "x11"
return "unknown"
def hotkey_plan(kind: str):
"""全局热键能力路由 → (thread_factory | None, message)。
thread_factory() 返回 QThread(带 triggered 信号);None 表示无全局热键能力
(调用方应保留应用内 QShortcut 兜底)。message 需要打印以明确当前能力。
"""
if kind == "win32":
from ui.views.system_tools.global_hotkey import GlobalHotkeyThread
return GlobalHotkeyThread, "[GlobalHotkey] Windows:系统级全局热键 Alt+SRegisterHotKey"
if kind == "x11":
from ui.views.system_tools.x11_hotkey import X11HotkeyThread
return X11HotkeyThread, "[GlobalHotkey] X11:原生全局热键 Alt+SXGrabKey"
if kind == "wayland":
return None, ("[GlobalHotkey] Wayland:全局热键需要 compositor 桌面协议"
"ext-global-shortcut 等),本版本未启用 → 仅提供应用内 Alt+S 快捷键"
"(窗口获焦时生效)与截图按钮;其余功能不受影响")
return None, ("[GlobalHotkey] 当前环境无显示服务(offscreen/无 DISPLAY)→ 全局热键不可用;"
"应用内 Alt+S 快捷键同样不可用,可用截图按钮以外的全部功能")
def capture_plan(kind: str):
"""截图能力路由 → (mode, message)。
mode: "overlay"Qt grabWindow 覆盖层,win32/x11
"portal"xdg-desktop-portal 交互截图,wayland
"unavailable"(unknown:明确告知,主程序其余功能不受影响)
"""
if kind in ("win32", "x11"):
return "overlay", None
if kind == "wayland":
return "portal", None
return "unavailable", ("[Screenshot] 当前环境无法获取屏幕画面(offscreen/无显示)→ "
"截图功能不可用;聊天与其他功能不受影响")
+71
View File
@@ -0,0 +1,71 @@
# -*- coding: utf-8 -*-
"""
文本/代码文件读取工具 (file_reader)
编码探测 + 二进制探测 + 大小守卫,供附件系统(拖拽/粘贴/上传文本与代码文件)复用。
用法:read_text_file(path) -> (content, encoding, size_kb, lines),失败抛 ValueError。
说明:纯函数模块,不依赖 Qt,便于单元测试。
"""
import os
# 明确拒绝的二进制/文档格式(Word/Excel/PPT/PDF/压缩包/可执行/媒体等)
BINARY_EXTS = frozenset({
# Office 文档(明确排除 Word/Excel
'.doc', '.docx', '.xls', '.xlsx', '.ppt', '.pptx',
'.odt', '.ods', '.odp', '.rtf',
# 电子书(.pdf 已交由 tools/builtin_tools/pdf_reader.py 专门解析,不在此拦截)
'.epub', '.mobi',
# 压缩包 / 磁盘镜像
'.zip', '.rar', '.7z', '.tar', '.gz', '.bz2', '.xz', '.iso', '.dmg',
# 可执行 / 编译产物
'.exe', '.dll', '.so', '.dylib', '.msi', '.bin', '.apk', '.jar',
'.class', '.pyc', '.pyd', '.o', '.a',
# 音视频
'.mp3', '.wav', '.flac', '.m4a', '.aac', '.ogg',
'.mp4', '.avi', '.mov', '.mkv', '.webm', '.wmv',
# 字体
'.ttf', '.otf', '.woff', '.woff2', '.eot',
# 数据库
'.db', '.sqlite', '.sqlite3', '.mdb',
# 设计稿
'.psd', '.ai', '.sketch', '.fig',
})
# 单文件大小上限(字节):超过则拒绝,避免内存与上下文爆炸
MAX_ATTACH_FILE_BYTES = 2 * 1024 * 1024 # 2 MB
# 编码降级链:UTF-8(含BOM) → GB18030(⊇GBK/GB2312) → latin-1(永不失败)
_ENCODINGS = ('utf-8-sig', 'gb18030', 'latin-1')
def read_text_file(path, max_bytes=MAX_ATTACH_FILE_BYTES):
"""读取一个文本/代码文件。
返回 (content, encoding, size_kb, lines)。
文件过大或判定为二进制时抛出 ValueError(附带可读原因)。
"""
size = os.path.getsize(path)
if size > max_bytes:
raise ValueError(
f"文件过大({size / 1024 / 1024:.1f} MB > {max_bytes / 1024 / 1024:.0f} MB"
)
with open(path, 'rb') as f:
raw = f.read()
# 二进制探测:前 8KB 含 NUL 字节即判定为二进制(git 同款启发式)
if b'\x00' in raw[:8192]:
raise ValueError("二进制文件,无法作为文本读取")
text, used_enc = None, 'utf-8'
for enc in _ENCODINGS:
try:
text = raw.decode(enc)
used_enc = enc
break
except UnicodeDecodeError:
continue
if text is None: # 理论上 latin-1 兜底永不失败
raise ValueError("无法识别文件编码")
lines = text.count('\n') + 1
return text, used_enc, round(size / 1024, 2), lines
+105
View File
@@ -0,0 +1,105 @@
# -*- coding: utf-8 -*-
"""
全局热键监听器 (Windows)
独立线程 + Win32 RegisterHotKey 消息循环,实现主窗口失焦也能触发的系统级全局快捷键。
用法:
hotkey = GlobalHotkeyThread() # 默认 Alt+S
hotkey.triggered.connect(on_triggered) # 跨线程信号,槽函数在主线程执行
hotkey.start() # 启动线程
hotkey.stop() # 注销热键并退出线程
"""
import ctypes
import sys
import threading
from ctypes import wintypes
from PyQt6 import QtCore
# Win32 常量
WM_HOTKEY = 0x0312
WM_QUIT = 0x0012
MOD_ALT = 0x0001
MOD_CONTROL = 0x0002
MOD_SHIFT = 0x0004
MOD_NOREPEAT = 0x4000 # 按住不重复触发 (Windows 7+)
VK_S = 0x53
_is_windows = sys.platform == "win32"
if _is_windows:
_user32 = ctypes.windll.user32
_kernel32 = ctypes.windll.kernel32
# 显式声明函数签名,避免 ctypes 默认 int 截断指针/句柄
_user32.RegisterHotKey.argtypes = [wintypes.HWND, ctypes.c_int, wintypes.UINT, wintypes.UINT]
_user32.RegisterHotKey.restype = wintypes.BOOL
_user32.UnregisterHotKey.argtypes = [wintypes.HWND, ctypes.c_int]
_user32.UnregisterHotKey.restype = wintypes.BOOL
_user32.GetMessageW.argtypes = [
ctypes.POINTER(wintypes.MSG), wintypes.HWND, wintypes.UINT, wintypes.UINT
]
_user32.GetMessageW.restype = wintypes.BOOL # >0 正常 / 0 WM_QUIT / -1 出错
_user32.PostThreadMessageW.argtypes = [
wintypes.DWORD, wintypes.UINT, wintypes.WPARAM, wintypes.LPARAM
]
_user32.PostThreadMessageW.restype = wintypes.BOOL
_kernel32.GetCurrentThreadId.restype = wintypes.DWORD
class GlobalHotkeyThread(QtCore.QThread):
"""在独立线程注册系统级全局热键并运行 Win32 消息循环。
收到 WM_HOTKEY 后通过 Qt 信号 triggered 通知主线程,
信号槽机制保证槽函数在主线程执行,操作 Qt 控件安全。
"""
triggered = QtCore.pyqtSignal()
def __init__(self, hotkey_id=9001, mod=MOD_ALT, vk=VK_S, parent=None):
super().__init__(parent)
self._hotkey_id = hotkey_id
self._mod = mod
self._vk = vk
self._thread_id = 0
self._ready = threading.Event() # run() 记录线程 ID 后置位
self._registered = False
def run(self):
"""线程入口:注册热键 -> 消息循环 -> 退出时注销"""
if not _is_windows:
self._ready.set()
return
self._thread_id = _kernel32.GetCurrentThreadId()
ok = _user32.RegisterHotKey(
None, self._hotkey_id, self._mod | MOD_NOREPEAT, self._vk
)
self._registered = bool(ok)
self._ready.set()
if not ok:
print(
f"[GlobalHotkey] 注册热键失败 (id={self._hotkey_id}, mod={self._mod:#x}, "
f"vk={self._vk:#x}),可能已被其他程序占用"
)
return
# Win32 消息循环:hwnd=None 取本线程所有消息
msg = wintypes.MSG()
while _user32.GetMessageW(ctypes.byref(msg), None, 0, 0) > 0:
if msg.message == WM_HOTKEY and msg.wParam == self._hotkey_id:
self.triggered.emit()
# 收到 WM_QUIT 退出循环 -> 注销热键
_user32.UnregisterHotKey(None, self._hotkey_id)
self._registered = False
def stop(self):
"""请求线程退出:向线程消息队列投递 WM_QUIT,然后等待结束"""
if not _is_windows:
return
# 等 run() 至少记录好线程 ID(注册成功或失败都行)
self._ready.wait(timeout=2.0)
if self._thread_id:
_user32.PostThreadMessageW(self._thread_id, WM_QUIT, 0, 0)
self.wait(2000)
+163
View File
@@ -0,0 +1,163 @@
# -*- coding: utf-8 -*-
"""
xdg-desktop-portal 截图(P1-04Linux Wayland 专用窄适配器)
Wayland compositor 不允许应用直接抓取屏幕画面(安全模型),截图必须走
xdg-desktop-portal 的 org.freedesktop.portal.Screenshot
1. gdbus 调 Screenshot(handle, parent_window=/, {}) → 返回 request 对象路径;
2. gdbus monitor 监听该 request 对象:
FilePicked(file_uri, ...) → 用户已授权并保存成功 → 返回文件路径;
Finished(无 FilePicked → 用户取消/授权被拒 → 返回 denied;
3. 全程有显式时间预算;超时 → timeout。
实现只用系统自带 CLI(gdbus,GLib 的一部分)+ 标准库,不新增 Python 依赖、
不绕过 compositor、不伪造成功。portal/桌面环境不支持时返回明确的不可用原因。
"""
import json
import os
import shutil
import subprocess
import sys
import time
import urllib.parse
from PyQt6 import QtCore
PORTAL_DEST = "org.freedesktop.portal.Desktop"
PORTAL_PATH = "/org/freedesktop/portal/desktop"
SCREENSHOT_IFACE = "org.freedesktop.portal.Screenshot"
REQUEST_IFACE = "org.freedesktop.portal.Request"
def detect_portal():
"""探测 portal 截图能力 → (ok: bool, reason: str | None)。"""
if sys.platform != "linux":
return False, "portal 截图仅用于 Linux"
if not (os.environ.get("XDG_RUNTIME_DIR", "").strip()
or os.environ.get("DBUS_SESSION_BUS_ADDRESS", "").strip()):
return False, ("未检测到 D-Bus 会话(XDG_RUNTIME_DIR/DBUS_SESSION_BUS_ADDRESS 均缺失)"
" → portal 截图不可用")
if not shutil.which("gdbus"):
return False, ("未找到 gdbus CLIGLib 组件)→ portal 截图不可用;"
"聊天与其他功能不受影响")
return True, None
def _gdbus_screenshot_request(handle: str, timeout_s: float):
"""调 Screenshot 方法,返回 (request_path, stderr);失败抛 RuntimeError。"""
cmd = ["gdbus", "call", "--session", "--dest", PORTAL_DEST,
"--object-path", PORTAL_PATH,
f"--method={SCREENSHOT_IFACE}.Screenshot",
handle, "/", "{}"]
try:
p = subprocess.run(cmd, capture_output=True, text=True,
encoding="utf-8", errors="replace",
timeout=timeout_s, check=False)
except subprocess.TimeoutExpired:
raise RuntimeError(f"portal Screenshot 调用超时({timeout_s:.0f}s 预算)")
if p.returncode != 0:
err = (p.stderr or p.stdout or "").strip().splitlines()
raise RuntimeError("portal 不可用: " + (err[-1] if err else "未知错误"))
out = (p.stdout or "").strip()
# gdbus 输出形如:('/org/freedesktop/portal/desktop/request/1000/haocode/42', <>)
start = out.find("('")
end = out.find("',") if start != -1 else -1
if start == -1 or end == -1:
raise RuntimeError(f"无法解析 Screenshot 返回: {out[:120]!r}")
return out[start + 2:end], (p.stderr or "").strip()
def _gdbus_monitor_request(request_path: str, budget_s: float):
"""监听 request 对象直到 FilePicked / Finished / 预算耗尽。
返回 (status, path_or_reason)status ∈ {"ok", "denied", "timeout"}。
"""
cmd = ["gdbus", "monitor", "--session", "--object-path", request_path]
p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
text=True, encoding="utf-8", errors="replace")
deadline = time.monotonic() + budget_s
file_path = None
finished = False
try:
while time.monotonic() < deadline:
line = p.stdout.readline()
if not line:
if p.poll() is not None:
break
continue
line = line.strip()
if not line.startswith("{"):
continue
try:
ev = json.loads(line)
except Exception:
continue
member = ev.get("member", "")
body = ev.get("body", [])
if member == "FilePicked" and body:
uri = str(body[0])
# 剥前缀 + unquote(比 urlparse 稳健:兼容 file:///tmp/x 与 file://D%3A%5Cx
file_path = urllib.parse.unquote(uri[len("file://"):]) \
if uri.startswith("file://") else uri
elif member == "Finished" and ev.get("interface", "") == REQUEST_IFACE:
finished = True
break
if file_path:
return "ok", file_path
if finished and time.monotonic() <= deadline:
return "denied", "用户在 portal 授权窗口取消/拒绝"
return "timeout", f"等待 portal 授权超时({budget_s:.0f}s 预算)"
finally:
try:
p.kill()
except Exception:
pass
try:
p.wait(timeout=3)
except Exception:
pass
def portal_screenshot_sync(handle: str = "haocode-shot",
request_timeout_s: float = 10.0,
wait_budget_s: float = 120.0):
"""同步执行 portal 截图(调用方须保证不在 GUI 线程阻塞,或用 PortalScreenshotWorker)。
返回 (ok: bool, path_or_reason: str)。
"""
ok, reason = detect_portal()
if not ok:
return False, reason
try:
request_path, _err = _gdbus_screenshot_request(handle, request_timeout_s)
except RuntimeError as e:
return False, str(e)
status, result = _gdbus_monitor_request(request_path, wait_budget_s)
if status == "ok":
return True, result
if status == "denied":
return False, result
return False, result
class PortalScreenshotWorker(QtCore.QThread):
"""GUI 线程安全的 portal 截图:done(ok, path_or_reason) 信号回主线程。"""
done = QtCore.pyqtSignal(bool, str)
def __init__(self, handle: str = "haocode-shot",
request_timeout_s: float = 10.0,
wait_budget_s: float = 120.0, parent=None):
super().__init__(parent)
self._handle = handle
self._request_timeout_s = request_timeout_s
self._wait_budget_s = wait_budget_s
def run(self):
try:
ok, result = portal_screenshot_sync(self._handle,
self._request_timeout_s,
self._wait_budget_s)
except Exception as e:
ok, result = False, f"portal 截图异常({type(e).__name__}: {e}"
self.done.emit(ok, result)
+200
View File
@@ -0,0 +1,200 @@
# -*- coding: utf-8 -*-
"""
屏幕截图覆盖层 (ScreenCaptureOverlay)
全屏半透明遮罩 + 鼠标拖拽选区 + 确认/取消工具条 + 截图完成发射 QImage 信号
用法:调用 start() 启动截图,监听 screenshot_captured 信号获取结果
"""
from PyQt6 import QtCore, QtGui, QtWidgets
import os
class ScreenCaptureOverlay(QtWidgets.QWidget):
"""全屏截图覆盖层:半透明遮罩 + 鼠标拖拽选区 + 确认/取消按钮"""
screenshot_captured = QtCore.pyqtSignal(QtGui.QImage)
# 按钮尺寸
BTN_W = 36
BTN_H = 30
BTN_GAP = 4
def __init__(self, parent=None):
super().__init__(parent)
self.setWindowFlags(
QtCore.Qt.WindowType.FramelessWindowHint
| QtCore.Qt.WindowType.WindowStaysOnTopHint
| QtCore.Qt.WindowType.Tool
)
self.setCursor(QtCore.Qt.CursorShape.CrossCursor)
self._full_pixmap = None
self._start_pos = QtCore.QPoint()
self._end_pos = QtCore.QPoint()
self._is_drawing = False
self._has_selection = False
self._current_rect = QtCore.QRect()
# SVG 图标路径(项目根目录下 svg/ 文件夹)
root_dir = os.path.abspath(
os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "..", "..")
)
# 确认 / 取消按钮(使用 SVG 图标,灰白简约风背景)
self._btn_confirm = QtWidgets.QPushButton(self)
self._btn_cancel = QtWidgets.QPushButton(self)
self._btn_confirm.setIcon(QtGui.QIcon(os.path.join(root_dir, "svg", "check.svg")))
self._btn_cancel.setIcon(QtGui.QIcon(os.path.join(root_dir, "svg", "cross.svg")))
btn_style = (
"QPushButton { background-color: #f9f9f9; border: 1px solid #ddd; border-radius: 4px; }"
"QPushButton:hover { background-color: #eee; border-color: #ccc; }"
)
for btn in (self._btn_confirm, self._btn_cancel):
btn.setFixedSize(self.BTN_W, self.BTN_H)
btn.setIconSize(QtCore.QSize(20, 20))
btn.setCursor(QtCore.Qt.CursorShape.PointingHandCursor)
btn.setStyleSheet(btn_style)
btn.hide()
self._btn_confirm.clicked.connect(self._on_confirm)
self._btn_cancel.clicked.connect(self._on_cancel)
def start(self):
"""开始截图:抓取屏幕全图并显示覆盖层"""
screen = QtWidgets.QApplication.primaryScreen()
if not screen:
print("[Screenshot] 无主屏幕 → 截图不可用", flush=True)
return
self._full_pixmap = screen.grabWindow(0)
# P1-04X11 某些 compositor/环境下 grabWindow 可能拿到空图(Wayland 已改走 portal
if self._full_pixmap is None or self._full_pixmap.isNull() or self._full_pixmap.width() == 0:
print("[Screenshot] 屏幕抓取返回空画面(当前 compositor/环境限制)→ 本次截图取消;"
"聊天与其他功能不受影响", flush=True)
self._full_pixmap = None
return
self.setGeometry(screen.geometry())
self.show()
self.activateWindow()
self.raise_()
def paintEvent(self, event):
if not self._full_pixmap:
return
painter = QtGui.QPainter(self)
# 1. 绘制屏幕截图作为背景
painter.drawPixmap(0, 0, self._full_pixmap)
# 2. 半透明遮罩
painter.fillRect(self.rect(), QtGui.QColor(0, 0, 0, 100))
# 绘制中 或 已有选区 时,绘制选区高亮
if self._is_drawing or self._has_selection:
if self._is_drawing:
rect = QtCore.QRect(self._start_pos, self._end_pos).normalized()
else:
rect = self._current_rect
if rect.width() > 0 and rect.height() > 0:
# 3. 选区内重绘原图(去掉遮罩,形成高亮效果)
painter.drawPixmap(rect, self._full_pixmap, rect)
# 4. 蓝色边框
pen = QtGui.QPen(QtGui.QColor(0, 120, 215), 2)
painter.setPen(pen)
painter.setBrush(QtCore.Qt.BrushStyle.NoBrush)
painter.drawRect(rect)
# 5. 尺寸标注
size_text = f"{rect.width()} x {rect.height()}"
font = painter.font()
font.setPointSize(9)
painter.setFont(font)
fm = painter.fontMetrics()
text_w = fm.horizontalAdvance(size_text)
text_h = fm.height()
text_x = rect.x()
text_y = rect.y() - text_h - 2
if text_y < 0:
text_y = rect.bottom() + 2
painter.fillRect(text_x, text_y, text_w + 10, text_h, QtGui.QColor(0, 120, 215))
painter.setPen(QtGui.QColor(255, 255, 255))
painter.drawText(text_x + 5, text_y + fm.ascent(), size_text)
painter.end()
def mousePressEvent(self, event):
if event.button() == QtCore.Qt.MouseButton.LeftButton:
# 开始新选区,隐藏按钮
self._hide_buttons()
self._has_selection = False
self._start_pos = event.position().toPoint()
self._end_pos = self._start_pos
self._is_drawing = True
self.update()
elif event.button() == QtCore.Qt.MouseButton.RightButton:
self.close()
def mouseMoveEvent(self, event):
if self._is_drawing:
self._end_pos = event.position().toPoint()
self.update()
def mouseReleaseEvent(self, event):
if event.button() == QtCore.Qt.MouseButton.LeftButton and self._is_drawing:
self._is_drawing = False
rect = QtCore.QRect(self._start_pos, self._end_pos).normalized()
if rect.width() > 5 and rect.height() > 5:
# 保留选区,显示确认/取消按钮
self._current_rect = rect
self._has_selection = True
self._position_buttons(rect)
self._btn_confirm.show()
self._btn_cancel.show()
self.update()
else:
self.close()
def keyPressEvent(self, event):
if event.key() == QtCore.Qt.Key.Key_Escape:
self.close()
elif event.key() in (QtCore.Qt.Key.Key_Return, QtCore.Qt.Key.Key_Enter):
if self._has_selection:
self._on_confirm()
def _position_buttons(self, rect):
"""将确认/取消按钮定位到选区右下角"""
total_w = self.BTN_W * 2 + self.BTN_GAP
# 默认放在选区右下角外侧
x = rect.right() - total_w
y = rect.bottom() + 4
# 边界检测:超出屏幕底部时翻到选区内侧
if y + self.BTN_H > self.height():
y = rect.bottom() - self.BTN_H - 4
if x < 0:
x = 0
self._btn_confirm.move(x, y)
self._btn_cancel.move(x + self.BTN_W + self.BTN_GAP, y)
def _hide_buttons(self):
self._btn_confirm.hide()
self._btn_cancel.hide()
def _on_confirm(self):
"""确认截图:裁剪并发射信号"""
if self._full_pixmap and self._current_rect.width() > 5 and self._current_rect.height() > 5:
dpr = self._full_pixmap.devicePixelRatio()
phys_rect = QtCore.QRect(
int(self._current_rect.x() * dpr),
int(self._current_rect.y() * dpr),
int(self._current_rect.width() * dpr),
int(self._current_rect.height() * dpr),
)
captured = self._full_pixmap.toImage().copy(phys_rect)
self.screenshot_captured.emit(captured)
self.close()
def _on_cancel(self):
"""取消截图:直接关闭"""
self.close()
def closeEvent(self, event):
self._is_drawing = False
self._has_selection = False
self._hide_buttons()
self._full_pixmap = None
super().closeEvent(event)
+196
View File
@@ -0,0 +1,196 @@
# -*- coding: utf-8 -*-
"""
X11 原生全局热键(P1-04Linux X11 专用窄适配器)
用 ctypes 调 libX11XOpenDisplay → XGrabKey(root, keycode, Mod1Mask) →
select(display_fd) + XNextEvent 消息循环,命中后发射 triggered 信号。
与 GlobalHotkeyThreadWindows)同一公开面:triggered / start() / stop()。
约束:
- 只支持现有截图快捷键(Alt+S),不增加任意键监听/记录/重映射;
- 注册失败/无显示/无 libX11 → 打印明确"不可用"日志后安静退出,绝不影响主程序;
- 不引入任何新的 Python 依赖(libX11 是 X11 桌面必然存在的系统库)。
"""
import ctypes
import ctypes.util
import select
import threading
from PyQt6 import QtCore
# ---- X11 常量 ----
KeyPress = 2
Mod1Mask = 0x0001 # Alt
KeyPressMask = 0x00002 # XSelectInput event mask
AnyModifier = 0xFFFFFF
DEFAULT_MOD = 1 # 与 global_hotkey.MOD_ALT 同值
DEFAULT_VK = 0x53 # 与 global_hotkey.VK_S 同值('S'
#: Win32 VK → X11 keysym 的窄映射(只覆盖现有快捷键;扩展需同步本表)
_VK_TO_KEYSYM = {
0x53: 0x73, # S → 's'
}
_XEVENT_FIELDS = [
("type", ctypes.c_int),
("serial", ctypes.c_ulong),
("send_event", ctypes.c_int),
("display", ctypes.c_void_p),
("window", ctypes.c_uint),
("root", ctypes.c_uint),
("subwindow", ctypes.c_uint),
("time", ctypes.c_ulong),
("x", ctypes.c_int),
("y", ctypes.c_int),
("root_x", ctypes.c_int),
("root_y", ctypes.c_int),
("state", ctypes.c_uint),
("keycode", ctypes.c_uint),
("same_screen", ctypes.c_int),
]
class XEvent(ctypes.Structure):
"""XEvent union 的按键事件视图(字段布局与 xproto.h XKeyEvent 一致,64 位下 keycode 位于偏移 76"""
_fields_ = _XEVENT_FIELDS
def _load_x11():
"""加载 libX11 并声明最小 API 原型。失败抛 OSError(调用方转成"不可用"日志)。"""
name = ctypes.util.find_library("X11") or "libX11.so.6"
x11 = ctypes.CDLL(name)
x11.XOpenDisplay.argtypes = [ctypes.c_char_p]
x11.XOpenDisplay.restype = ctypes.c_void_p
x11.XCloseDisplay.argtypes = [ctypes.c_void_p]
x11.XConnectionNumber.argtypes = [ctypes.c_void_p]
x11.XConnectionNumber.restype = ctypes.c_int
x11.XDefaultRootWindow.argtypes = [ctypes.c_void_p]
x11.XDefaultRootWindow.restype = ctypes.c_uint
x11.XKeysymToKeycode.argtypes = [ctypes.c_void_p, ctypes.c_ulong]
x11.XKeysymToKeycode.restype = ctypes.c_int
x11.XSelectInput.argtypes = [ctypes.c_void_p, ctypes.c_uint, ctypes.c_long]
x11.XGrabKey.argtypes = [ctypes.c_void_p, ctypes.c_uint, ctypes.c_uint, ctypes.c_uint, ctypes.c_int]
x11.XGrabKey.restype = ctypes.c_int
x11.XUngrabKey.argtypes = [ctypes.c_void_p, ctypes.c_uint, ctypes.c_uint, ctypes.c_uint]
x11.XPending.argtypes = [ctypes.c_void_p]
x11.XPending.restype = ctypes.c_int
x11.XNextEvent.argtypes = [ctypes.c_void_p, ctypes.POINTER(XEvent)]
return x11
def _open_x11():
"""可注入入口(测试用替身替换)。返回 (x11, display)display 为 0/None 表示无显示。"""
x11 = _load_x11()
disp = x11.XOpenDisplay(None)
return x11, disp
class X11HotkeyThread(QtCore.QThread):
"""X11 全局热键线程(公开面与 GlobalHotkeyThread 一致:triggered/start/stop)。"""
triggered = QtCore.pyqtSignal()
def __init__(self, hotkey_id=9001, mod=DEFAULT_MOD, vk=DEFAULT_VK, parent=None):
super().__init__(parent)
self._hotkey_id = hotkey_id
self._mod = mod
self._vk = vk
self._ready = threading.Event()
self._stop_flag = False
self._ready_flag = False
self._registered = False
self._keycode = 0
self._x11 = None
self._display = None
self._fd = None
# ------------------------------------------------------------------
def _setup(self):
"""打开显示、注册热键。成功返回 True;失败打印明确日志并返回 False。"""
try:
keysym = _VK_TO_KEYSYM.get(self._vk)
if keysym is None or self._mod != DEFAULT_MOD:
print(f"[X11Hotkey] 暂不支持的快捷键组合 (mod={self._mod:#x}, vk={self._vk:#x})"
f" → 全局热键不可用(本适配器只覆盖现有 Alt+S)", flush=True)
return False
x11, disp = _open_x11()
if not disp:
print("[X11Hotkey] XOpenDisplay 失败(无 X11 显示服务)→ 全局热键不可用;"
"应用内快捷键/截图按钮以外的功能不受影响", flush=True)
return False
self._x11 = x11
self._display = disp
self._fd = x11.XConnectionNumber(disp)
root = x11.XDefaultRootWindow(disp)
keycode = x11.XKeysymToKeycode(disp, keysym)
if not keycode:
print("[X11Hotkey] 键码解析失败 → 全局热键不可用", flush=True)
self._cleanup()
return False
self._keycode = keycode
x11.XSelectInput(disp, root, ctypes.c_long(KeyPressMask))
if x11.XGrabKey(disp, keycode, Mod1Mask, root, 1) == 0:
# BadAccess:键已被其他程序抢占
print(f"[X11Hotkey] XGrabKey 注册失败(Alt+S 可能已被其他程序占用)"
f" → 全局热键不可用;应用内快捷键/截图按钮以外的功能不受影响", flush=True)
self._cleanup()
return False
self._registered = True
return True
except Exception as e:
print(f"[X11Hotkey] 初始化失败({type(e).__name__}: {e})→ 全局热键不可用", flush=True)
self._cleanup()
return False
def _cleanup(self):
if self._display and self._x11:
try:
self._x11.XUngrabKey(self._display, self._keycode, AnyModifier,
self._x11.XDefaultRootWindow(self._display))
except Exception:
pass
try:
self._x11.XCloseDisplay(self._display)
except Exception:
pass
self._display = None
self._x11 = None
self._fd = None
self._registered = False
# ------------------------------------------------------------------
def run(self):
if not self._setup():
self._ready.set()
return
self._ready.set()
try:
while not self._stop_flag:
if self._fd is None:
break
try:
r, _, _ = select.select([self._fd], [], [], 0.2)
except (OSError, ValueError):
break
if not r:
continue
if not self._x11.XPending(self._display):
continue
ev = XEvent()
self._x11.XNextEvent(self._display, ctypes.byref(ev))
if ev.type == KeyPress and ev.keycode == self._keycode:
self.triggered.emit()
except BaseException as e:
# 绝不允许异常逃逸出 QThread.run()PyQt6 会 abort 整个应用)
print(f"[X11Hotkey] 事件循环异常({type(e).__name__}: {e})→ 全局热键不可用;其余功能不受影响",
flush=True)
finally:
self._cleanup()
def wait_ready(self, timeout_s: float = 2.0) -> bool:
"""等待注册流程结束(成功或失败都算就绪;测试/诊断用)"""
return self._ready.wait(timeout_s)
def stop(self):
"""请求退出(select 0.2s 超时轮询 stop 标志 → 线程安全退出,无需向 X 连接发消息)"""
self._stop_flag = True