feat(ui): cross-platform screenshot hotkey and portal-based capture
Windows keeps the legacy path; Linux X11 uses a native XGrabKey hotkey (ctypes libX11) with native capture; Wayland captures through xdg-desktop-portal and degrades gracefully with explicit capability-unavailable logs when the portal/protocol is missing.
This commit is contained in:
@@ -0,0 +1,279 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
P1-04 全局热键平台矩阵单测(适配器替身,不依赖真实 X11 显示 / 不新增依赖)
|
||||
|
||||
覆盖:
|
||||
H1 session_kind 矩阵(win32 / x11 / wayland / offscreen-unknown,mock 环境变量 + sys.platform)
|
||||
H2 hotkey_plan 路由(win32→RegisterHotKey 线程;x11→XGrabKey 线程;wayland/offscreen→None+明确日志)
|
||||
H3 X11HotkeyThread 成功路径(fake libX11:XGrabKey 参数正确、命中 keycode 发射 triggered、
|
||||
stop 后 XUngrabKey/XCloseDisplay 释放)
|
||||
H4 X11 注册失败(XGrabKey=0 键被占用 / 无显示 / 不支持的组合 → _registered=False,线程安静退出)
|
||||
H5 Windows GlobalHotkeyThread 行为保持(非 Windows run() 静默就绪退出;Windows 常量完整)
|
||||
|
||||
运行: PYTHONIOENCODING=utf-8 python tests/test_global_hotkey_platforms.py
|
||||
"""
|
||||
import contextlib
|
||||
import io
|
||||
import os
|
||||
import socket
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
||||
|
||||
from PyQt6.QtWidgets import QApplication # noqa: E402
|
||||
|
||||
from ui.views.system_tools import desktop_session as ds # noqa: E402
|
||||
from ui.views.system_tools import x11_hotkey as xh # noqa: E402
|
||||
from ui.views.system_tools import global_hotkey as gh # noqa: E402
|
||||
|
||||
RESULTS = []
|
||||
APP = QApplication.instance() or QApplication(sys.argv)
|
||||
|
||||
|
||||
def check(name, cond, extra=""):
|
||||
RESULTS.append((name, bool(cond)))
|
||||
print(f"{'PASS' if cond else 'FAIL'} {name} {extra if not cond else ''}", flush=True)
|
||||
|
||||
|
||||
def with_env(patch: dict, fn):
|
||||
saved = {k: os.environ.get(k) for k in patch}
|
||||
for k, v in patch.items():
|
||||
if v is None:
|
||||
os.environ.pop(k, None)
|
||||
else:
|
||||
os.environ[k] = v
|
||||
try:
|
||||
return fn()
|
||||
finally:
|
||||
for k, v in saved.items():
|
||||
if v is None:
|
||||
os.environ.pop(k, None)
|
||||
else:
|
||||
os.environ[k] = v
|
||||
|
||||
|
||||
# ======================================================================
|
||||
# H1. session_kind 矩阵
|
||||
# ======================================================================
|
||||
_orig_platform = ds.sys.platform
|
||||
|
||||
def _set_platform(p):
|
||||
ds.sys = type("FakeSys", (), {"platform": p})()
|
||||
|
||||
|
||||
try:
|
||||
_set_platform("win32")
|
||||
check("H1.1 win32", ds.session_kind() == "win32")
|
||||
_set_platform("linux")
|
||||
check("H1.2 X11(DISPLAY 有、无 WAYLAND_DISPLAY)",
|
||||
with_env({"DISPLAY": ":0", "WAYLAND_DISPLAY": None, "XDG_SESSION_TYPE": None,
|
||||
"QT_QPA_PLATFORM": "xcb"}, lambda: ds.session_kind()) == "x11")
|
||||
check("H1.3 Wayland(WAYLAND_DISPLAY 有)",
|
||||
with_env({"WAYLAND_DISPLAY": "wayland-0", "DISPLAY": ":0", "XDG_SESSION_TYPE": None,
|
||||
"QT_QPA_PLATFORM": "wayland"}, lambda: ds.session_kind()) == "wayland")
|
||||
check("H1.4 QT_QPA_PLATFORM=wayland(无 WAYLAND_DISPLAY)",
|
||||
with_env({"WAYLAND_DISPLAY": None, "DISPLAY": None, "XDG_SESSION_TYPE": None,
|
||||
"QT_QPA_PLATFORM": "wayland"}, lambda: ds.session_kind()) == "wayland")
|
||||
check("H1.5 offscreen → unknown(即使有 DISPLAY)",
|
||||
with_env({"DISPLAY": ":0", "WAYLAND_DISPLAY": None, "XDG_SESSION_TYPE": None,
|
||||
"QT_QPA_PLATFORM": "offscreen"}, lambda: ds.session_kind()) == "unknown")
|
||||
check("H1.6 无显示无 WAYLAND → unknown",
|
||||
with_env({"DISPLAY": None, "WAYLAND_DISPLAY": None, "XDG_SESSION_TYPE": None,
|
||||
"QT_QPA_PLATFORM": None}, lambda: ds.session_kind()) == "unknown")
|
||||
check("H1.7 XDG_SESSION_TYPE=wayland(无 WAYLAND_DISPLAY/QT 平台)",
|
||||
with_env({"DISPLAY": None, "WAYLAND_DISPLAY": None, "XDG_SESSION_TYPE": "wayland",
|
||||
"QT_QPA_PLATFORM": None}, lambda: ds.session_kind()) == "wayland")
|
||||
check("H1.8 XDG_SESSION_TYPE=x11(无 DISPLAY)",
|
||||
with_env({"DISPLAY": None, "WAYLAND_DISPLAY": None, "XDG_SESSION_TYPE": "x11",
|
||||
"QT_QPA_PLATFORM": None}, lambda: ds.session_kind()) == "x11")
|
||||
check("H1.9 矛盾时 WAYLAND_DISPLAY 优先于 XDG_SESSION_TYPE=x11",
|
||||
with_env({"DISPLAY": ":0", "WAYLAND_DISPLAY": "wayland-0", "XDG_SESSION_TYPE": "x11",
|
||||
"QT_QPA_PLATFORM": None}, lambda: ds.session_kind()) == "wayland")
|
||||
finally:
|
||||
ds.sys = sys # 还原
|
||||
|
||||
# ======================================================================
|
||||
# H2. hotkey_plan 路由
|
||||
# ======================================================================
|
||||
try:
|
||||
_set_platform("win32")
|
||||
factory, msg = ds.hotkey_plan("win32")
|
||||
check("H2.1 win32 → GlobalHotkeyThread 工厂 + 说明",
|
||||
factory is gh.GlobalHotkeyThread and "Windows" in msg, repr((factory, msg)))
|
||||
finally:
|
||||
ds.sys = sys
|
||||
|
||||
factory, msg = ds.hotkey_plan("x11")
|
||||
check("H2.2 x11 → X11HotkeyThread 工厂 + 说明",
|
||||
factory is xh.X11HotkeyThread and "X11" in msg, repr((factory, msg)))
|
||||
|
||||
factory, msg = ds.hotkey_plan("wayland")
|
||||
check("H2.3 wayland → None + 明确不可用说明(不绕过 compositor)",
|
||||
factory is None and "Wayland" in msg and "Alt+S" in msg, repr((factory, msg)))
|
||||
|
||||
factory, msg = ds.hotkey_plan("unknown")
|
||||
check("H2.4 unknown → None + 明确不可用说明",
|
||||
factory is None and "不可用" in msg, repr((factory, msg)))
|
||||
|
||||
# ======================================================================
|
||||
# H3–H4. X11HotkeyThread(fake libX11)
|
||||
# ======================================================================
|
||||
|
||||
class FakeX11:
|
||||
def __init__(self, grab_rc=1, open_rc=0xAB, keycode=39, unsupported_vk=False):
|
||||
self.grab_rc = grab_rc
|
||||
self.open_rc = open_rc
|
||||
self.keycode = keycode
|
||||
self.unsupported_vk = unsupported_vk
|
||||
self.grab_calls = []
|
||||
self.ungrab_calls = []
|
||||
self.close_calls = 0
|
||||
self.select_calls = 0
|
||||
self.opened = False
|
||||
self.pending_left = 1
|
||||
# socketpair:跨平台可被 select() 监听(Windows 上 os.pipe 的 fd 不行)
|
||||
self._a, self._b = socket.socketpair()
|
||||
self._b.send(b"\x01") # 让 a 有可读数据 → select 首次就绪
|
||||
self._event = xh.XEvent()
|
||||
self._event.type = xh.KeyPress
|
||||
self._event.keycode = self.keycode
|
||||
|
||||
# -- libX11 API(鸭子类型替身) --
|
||||
def XOpenDisplay(self, name):
|
||||
self.opened = True
|
||||
return self.open_rc
|
||||
def XCloseDisplay(self, d):
|
||||
self.close_calls += 1
|
||||
def XConnectionNumber(self, d):
|
||||
return self._a.fileno()
|
||||
def XDefaultRootWindow(self, d):
|
||||
return 123
|
||||
def XKeysymToKeycode(self, d, keysym):
|
||||
if self.unsupported_vk:
|
||||
return 0
|
||||
return self.keycode
|
||||
def XSelectInput(self, d, w, mask):
|
||||
self.select_calls += 1
|
||||
def XGrabKey(self, d, kc, mod, w, owner):
|
||||
self.grab_calls.append((kc, mod, w, owner))
|
||||
return self.grab_rc
|
||||
def XUngrabKey(self, d, kc, mod, w):
|
||||
self.ungrab_calls.append((kc, mod, w))
|
||||
def XPending(self, d):
|
||||
if self.pending_left > 0:
|
||||
self.pending_left -= 1
|
||||
return 1
|
||||
return 0
|
||||
def XNextEvent(self, d, evp):
|
||||
# 生产代码传入 ctypes.byref(ev)(CArgObject);真实 CDLL 自行解引用,
|
||||
# 替身可调用对象则通过 ._obj 拿回原始 struct
|
||||
import ctypes as _ct
|
||||
target = getattr(evp, "_obj", evp)
|
||||
_ct.memmove(_ct.byref(target), _ct.byref(self._event), _ct.sizeof(xh.XEvent))
|
||||
self._event.type = 0 # 之后无事件
|
||||
|
||||
def close(self):
|
||||
for s in (self._a, self._b):
|
||||
try:
|
||||
s.close()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def run_hotkey_case(fake, **kw):
|
||||
"""启动 X11HotkeyThread,等待 triggered 或退出,回收。返回 (got_signal, thread)"""
|
||||
orig_open = xh._open_x11
|
||||
xh._open_x11 = lambda: (fake, fake.open_rc)
|
||||
try:
|
||||
t = xh.X11HotkeyThread(**kw)
|
||||
got = threading.Event()
|
||||
t.triggered.connect(lambda: got.set())
|
||||
t.start()
|
||||
t.wait_ready(3.0)
|
||||
deadline = time.time() + 6.0
|
||||
while not got.is_set() and time.time() < deadline:
|
||||
APP.processEvents()
|
||||
time.sleep(0.02)
|
||||
t.stop()
|
||||
t.wait(3000)
|
||||
return got.is_set(), t
|
||||
finally:
|
||||
xh._open_x11 = orig_open
|
||||
fake.close()
|
||||
|
||||
|
||||
# H3.1 成功路径:grab 参数正确 + 命中发射 + stop 释放
|
||||
fake = FakeX11()
|
||||
got, t = run_hotkey_case(fake)
|
||||
check("H3.1 命中 Alt+S → triggered 发射", got)
|
||||
check("H3.2 XGrabKey 参数(keycode=39, Mod1Mask=1, root=123, owner_events=1)",
|
||||
fake.grab_calls == [(39, xh.Mod1Mask, 123, 1)], repr(fake.grab_calls))
|
||||
check("H3.3 stop 后 XUngrabKey + XCloseDisplay 释放",
|
||||
len(fake.ungrab_calls) == 1 and fake.close_calls == 1,
|
||||
repr((fake.ungrab_calls, fake.close_calls)))
|
||||
check("H3.4 注册成功标志 _registered", t._registered is False) # cleanup 后复位为 False
|
||||
|
||||
# H4.1 键被占用(XGrabKey → 0)
|
||||
fake = FakeX11(grab_rc=0)
|
||||
buf = io.StringIO()
|
||||
with contextlib.redirect_stdout(buf):
|
||||
got, t = run_hotkey_case(fake)
|
||||
check("H4.1 注册失败 → 无信号、_registered=False、明确日志",
|
||||
not got and t._registered is False and ("占用" in buf.getvalue() or "失败" in buf.getvalue()),
|
||||
buf.getvalue()[-200:])
|
||||
|
||||
# H4.2 无显示(XOpenDisplay → None)
|
||||
fake = FakeX11(open_rc=None)
|
||||
buf = io.StringIO()
|
||||
with contextlib.redirect_stdout(buf):
|
||||
got, t = run_hotkey_case(fake)
|
||||
check("H4.2 无 X11 显示 → 安静退出 + 明确日志",
|
||||
not got and t._registered is False and "XOpenDisplay" in buf.getvalue(),
|
||||
buf.getvalue()[-200:])
|
||||
|
||||
# H4.3 不支持的组合(vk 不在窄映射表)
|
||||
fake = FakeX11()
|
||||
orig_open = xh._open_x11
|
||||
t_probe = xh.X11HotkeyThread(vk=0x41)
|
||||
xh._open_x11 = lambda: (fake, 0xAB)
|
||||
try:
|
||||
t_probe.start()
|
||||
t_probe.wait_ready(3.0)
|
||||
deadline = time.time() + 5.0
|
||||
while t_probe.isRunning() and time.time() < deadline:
|
||||
time.sleep(0.02)
|
||||
t_probe.stop()
|
||||
t_probe.wait(2000)
|
||||
finally:
|
||||
xh._open_x11 = orig_open
|
||||
fake.close()
|
||||
check("H4.3 不支持的组合 → 不打开显示即退出",
|
||||
t_probe._registered is False and not fake.opened, repr(t_probe._registered))
|
||||
|
||||
# ======================================================================
|
||||
# H5. Windows 路径保持
|
||||
# ======================================================================
|
||||
check("H5.1 GlobalHotkeyThread 常量完整(MOD_ALT/VK_S/WM_HOTKEY)",
|
||||
gh.MOD_ALT == 0x0001 and gh.VK_S == 0x53 and gh.WM_HOTKEY == 0x0312)
|
||||
|
||||
t_win = gh.GlobalHotkeyThread()
|
||||
t_win.start()
|
||||
t_win._ready.wait(3.0)
|
||||
if gh._is_windows:
|
||||
# 本机 Windows:真实 RegisterHotKey + GetMessage 循环 = 行为保持的存活检查(随后立即释放 Alt+S)
|
||||
check("H5.2 Windows:RegisterHotKey 线程运行中(行为保持)", t_win.isRunning())
|
||||
t_win.stop()
|
||||
t_win.wait(3000)
|
||||
check("H5.3 Windows:stop() 干净退出(释放热键)", not t_win.isRunning())
|
||||
else:
|
||||
t_win.wait(3000)
|
||||
check("H5.2 非 Windows:run() 静默就绪退出(不注册、不崩)",
|
||||
not t_win.isRunning() and t_win._ready.is_set(),
|
||||
f"running={t_win.isRunning()}")
|
||||
|
||||
print(f"\n===== {sum(1 for _, ok in RESULTS if ok)}/{len(RESULTS)} passed =====", flush=True)
|
||||
sys.exit(0 if all(ok for _, ok in RESULTS) else 1)
|
||||
@@ -0,0 +1,293 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
P1-04 平台截图能力单测(portal 用替身 subprocess,不依赖真实 portal / compositor)
|
||||
|
||||
覆盖:
|
||||
C1 capture_plan 路由(win32/x11 → overlay;wayland → portal;unknown → unavailable+明确日志)
|
||||
C2 detect_portal(非 Linux / 无 D-Bus 会话 / 无 gdbus / 齐备 → 四分支)
|
||||
C3 portal_screenshot_sync 成功路径(fake subprocess:Screenshot 返回解析 + FilePicked
|
||||
file:// URI 解码(含空格文件名)→ 现有附件流程可用的真实文件路径;校验 gdbus 命令行)
|
||||
C4 用户取消/授权被拒(只有 Finished 无 FilePicked → denied)
|
||||
C5 portal 缺失(call rc≠0 NotSupported → "portal 不可用" + 原因)
|
||||
C6 等待超时(显式预算内无信号 → timeout,不无限挂起)
|
||||
C7 PortalScreenshotWorker 信号回主线程(done(ok, path))
|
||||
C8 ScreenCaptureOverlay offscreen 构造 + start() 空画面守卫不崩
|
||||
|
||||
运行: PYTHONIOENCODING=utf-8 python tests/test_screen_capture_platforms.py
|
||||
"""
|
||||
import contextlib
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
import types
|
||||
import urllib.parse
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
||||
|
||||
from PyQt6.QtWidgets import QApplication # noqa: E402
|
||||
|
||||
from ui.views.system_tools import desktop_session as ds # noqa: E402
|
||||
from ui.views.system_tools import portal_capture as pc # noqa: E402
|
||||
from ui.views.system_tools import screen_capture as sc # noqa: E402
|
||||
|
||||
RESULTS = []
|
||||
APP = QApplication.instance() or QApplication(sys.argv)
|
||||
|
||||
|
||||
def check(name, cond, extra=""):
|
||||
RESULTS.append((name, bool(cond)))
|
||||
print(f"{'PASS' if cond else 'FAIL'} {name} {extra if not cond else ''}", flush=True)
|
||||
|
||||
|
||||
# ======================================================================
|
||||
# C1. capture_plan 路由
|
||||
# ======================================================================
|
||||
mode, msg = ds.capture_plan("win32")
|
||||
check("C1.1 win32 → overlay", mode == "overlay" and msg is None, repr((mode, msg)))
|
||||
mode, msg = ds.capture_plan("x11")
|
||||
check("C1.2 x11 → overlay(原生 grabWindow 路径)", mode == "overlay" and msg is None,
|
||||
repr((mode, msg)))
|
||||
mode, msg = ds.capture_plan("wayland")
|
||||
check("C1.3 wayland → portal(不绕过 compositor)", mode == "portal" and msg is None,
|
||||
repr((mode, msg)))
|
||||
mode, msg = ds.capture_plan("unknown")
|
||||
check("C1.4 unknown → unavailable + 明确日志(主程序其余功能不受影响)",
|
||||
mode == "unavailable" and msg and "不可用" in msg, repr((mode, msg)))
|
||||
|
||||
# ======================================================================
|
||||
# C2. detect_portal 四分支
|
||||
# ======================================================================
|
||||
_orig_sys = pc.sys
|
||||
_orig_which = pc.shutil.which
|
||||
|
||||
try:
|
||||
pc.sys = types.SimpleNamespace(platform="win32")
|
||||
ok, reason = pc.detect_portal()
|
||||
check("C2.1 非 Linux → 不可用", ok is False and reason, repr((ok, reason)))
|
||||
|
||||
pc.sys = types.SimpleNamespace(platform="linux")
|
||||
saved = {k: os.environ.get(k) for k in ("XDG_RUNTIME_DIR", "DBUS_SESSION_BUS_ADDRESS")}
|
||||
for k in saved:
|
||||
os.environ.pop(k, None)
|
||||
ok, reason = pc.detect_portal()
|
||||
check("C2.2 无 D-Bus 会话 → 不可用(明确原因)",
|
||||
ok is False and "D-Bus" in reason, repr((ok, reason)))
|
||||
os.environ["XDG_RUNTIME_DIR"] = "/run/user/1000"
|
||||
pc.shutil.which = lambda name: None
|
||||
ok, reason = pc.detect_portal()
|
||||
check("C2.3 无 gdbus CLI → 不可用(明确原因)",
|
||||
ok is False and "gdbus" in reason, repr((ok, reason)))
|
||||
pc.shutil.which = lambda name: "/usr/bin/gdbus"
|
||||
ok, reason = pc.detect_portal()
|
||||
check("C2.4 齐备 → 可用", ok is True and reason is None, repr((ok, reason)))
|
||||
finally:
|
||||
pc.sys = _orig_sys
|
||||
pc.shutil.which = _orig_which
|
||||
for k, v in saved.items():
|
||||
if v is None:
|
||||
os.environ.pop(k, None)
|
||||
else:
|
||||
os.environ[k] = v
|
||||
|
||||
|
||||
# ======================================================================
|
||||
# C3–C7. portal_screenshot_sync(fake subprocess,模拟 Linux 环境)
|
||||
# ======================================================================
|
||||
_linux_sys = types.SimpleNamespace(platform="linux")
|
||||
|
||||
|
||||
class FakeRun:
|
||||
"""替身 subprocess.run:记录 argv,返回预先设定的 call 输出"""
|
||||
def __init__(self, rc=0, stdout="", stderr=""):
|
||||
self.rc = rc
|
||||
self.stdout = stdout
|
||||
self.stderr = stderr
|
||||
self.cmds = []
|
||||
|
||||
def __call__(self, cmd, **kw):
|
||||
self.cmds.append(list(cmd))
|
||||
p = types.SimpleNamespace()
|
||||
p.returncode = self.rc
|
||||
p.stdout = self.stdout
|
||||
p.stderr = self.stderr
|
||||
return p
|
||||
|
||||
|
||||
class FakePipe:
|
||||
"""替身 monitor 的 stdout:按序输出 JSON 行;readline 可模拟阻塞"""
|
||||
def __init__(self, lines, delay_s=0.0):
|
||||
self._lines = list(lines)
|
||||
self._delay = delay_s
|
||||
self.closed = False
|
||||
|
||||
def readline(self):
|
||||
if self._delay:
|
||||
time.sleep(self._delay)
|
||||
self._delay = 0
|
||||
if self._lines:
|
||||
return self._lines.pop(0) + "\n"
|
||||
# 模拟管道关闭(EOF)
|
||||
return ""
|
||||
|
||||
|
||||
class FakePopen:
|
||||
def __init__(self, lines, delay_s=0.0):
|
||||
self._lines = lines
|
||||
self._delay = delay_s
|
||||
self.cmd = None
|
||||
self.killed = False
|
||||
|
||||
def __call__(self, cmd, **kw):
|
||||
self.cmd = list(cmd)
|
||||
p = types.SimpleNamespace()
|
||||
p.stdout = FakePipe(self._lines, self._delay)
|
||||
p.stderr = io.StringIO()
|
||||
self._rc = None
|
||||
p.poll = lambda: self._rc
|
||||
p.kill = lambda: setattr(p, "killed", True) or setattr(self, "killed", True)
|
||||
p.wait = lambda timeout=None: 0
|
||||
return p
|
||||
|
||||
|
||||
def _uri(path: str) -> str:
|
||||
return "file://" + urllib.parse.quote(path)
|
||||
|
||||
|
||||
def _json(member, iface, body):
|
||||
return json.dumps({"interface": iface, "member": member, "body": body},
|
||||
ensure_ascii=False)
|
||||
|
||||
|
||||
# C3.1 成功路径(含空格文件名 → URI 解码)
|
||||
tmpdir = tempfile.mkdtemp(prefix="haocode_p104_portal_")
|
||||
shot = os.path.join(tmpdir, "shot test.png")
|
||||
with open(shot, "wb") as f:
|
||||
f.write(b"\x89PNG\r\n\x1a\nfake")
|
||||
|
||||
pc.sys = _linux_sys # C3–C7 模拟 Linux 宿主
|
||||
_orig_which2 = pc.shutil.which
|
||||
pc.shutil.which = lambda name: "/usr/bin/gdbus" # 本机无 gdbus,替身探测
|
||||
_saved_xdg = os.environ.get("XDG_RUNTIME_DIR")
|
||||
os.environ["XDG_RUNTIME_DIR"] = "/run/user/1000"
|
||||
run_fake = FakeRun(rc=0, stdout="('/org/freedesktop/portal/desktop/request/1000/haocode/7', <>)")
|
||||
popen_fake = FakePopen([
|
||||
_json("FilePicked", "org.freedesktop.portal.FileChooser", [_uri(shot), {}]),
|
||||
_json("Finished", "org.freedesktop.portal.Request", [0]),
|
||||
])
|
||||
_orig_run, _orig_popen = pc.subprocess.run, pc.subprocess.Popen
|
||||
pc.subprocess.run = run_fake
|
||||
pc.subprocess.Popen = popen_fake
|
||||
try:
|
||||
t0 = time.time()
|
||||
ok, result = pc.portal_screenshot_sync("haocode-shot", request_timeout_s=5, wait_budget_s=10)
|
||||
dt = time.time() - t0
|
||||
finally:
|
||||
pc.subprocess.run = _orig_run
|
||||
pc.subprocess.Popen = _orig_popen
|
||||
|
||||
check("C3.1 成功:返回真实文件路径(URI 空格解码正确)",
|
||||
ok is True and result == shot and os.path.exists(result), repr((ok, result)))
|
||||
check("C3.2 Screenshot gdbus 命令行正确(dest/path/method/parent=/)",
|
||||
run_fake.cmds and run_fake.cmds[0][:5] == ["gdbus", "call", "--session", "--dest",
|
||||
pc.PORTAL_DEST]
|
||||
and "--method=org.freedesktop.portal.Screenshot.Screenshot" in run_fake.cmds[0]
|
||||
and "/" in run_fake.cmds[0], repr(run_fake.cmds))
|
||||
check("C3.3 monitor 监听 request 对象路径",
|
||||
popen_fake.cmd and popen_fake.cmd[-2:] == ["--object-path",
|
||||
"/org/freedesktop/portal/desktop/request/1000/haocode/7"],
|
||||
repr(popen_fake.cmd))
|
||||
check("C3.4 全程在显式预算内完成(<10s)", dt < 10, f"{dt:.1f}s")
|
||||
|
||||
# C4. 用户取消/授权被拒(只有 Finished)
|
||||
popen_fake = FakePopen([_json("Finished", "org.freedesktop.portal.Request", [0])])
|
||||
pc.subprocess.run = run_fake
|
||||
pc.subprocess.Popen = popen_fake
|
||||
try:
|
||||
ok, result = pc.portal_screenshot_sync("haocode-shot", 5, 10)
|
||||
finally:
|
||||
pc.subprocess.run = _orig_run
|
||||
pc.subprocess.Popen = _orig_popen
|
||||
check("C4.1 授权被拒 → 不伪造成功,明确原因",
|
||||
ok is False and "取消" in result, repr((ok, result)))
|
||||
|
||||
# C5. portal 缺失/方法不支持(call rc≠0)
|
||||
run_fake2 = FakeRun(rc=1, stderr="Error: org.freedesktop.DBus.Error.NotSupported: "
|
||||
"Method not supported by portal")
|
||||
pc.subprocess.run = run_fake2
|
||||
pc.subprocess.Popen = FakePopen([])
|
||||
try:
|
||||
ok, result = pc.portal_screenshot_sync("haocode-shot", 5, 10)
|
||||
finally:
|
||||
pc.subprocess.run = _orig_run
|
||||
pc.subprocess.Popen = _orig_popen
|
||||
check("C5.1 portal 不支持 → 不可用 + 原因透出",
|
||||
ok is False and "portal 不可用" in result and "NotSupported" in result, repr((ok, result)))
|
||||
|
||||
# C6. 等待超时(monitor 首行前阻塞 > 预算)
|
||||
popen_fake = FakePopen([_json("Finished", "org.freedesktop.portal.Request", [0])],
|
||||
delay_s=3.0)
|
||||
pc.subprocess.run = run_fake
|
||||
pc.subprocess.Popen = popen_fake
|
||||
try:
|
||||
t0 = time.time()
|
||||
ok, result = pc.portal_screenshot_sync("haocode-shot", 5, 1.0)
|
||||
dt = time.time() - t0
|
||||
finally:
|
||||
pc.subprocess.run = _orig_run
|
||||
pc.subprocess.Popen = _orig_popen
|
||||
check("C6.1 超预算 → timeout(不无限挂起)",
|
||||
ok is False and "超时" in result and dt < 5, repr((ok, result, f"{dt:.1f}s")))
|
||||
|
||||
# ======================================================================
|
||||
# C7. PortalScreenshotWorker 信号
|
||||
# ======================================================================
|
||||
popen_fake = FakePopen([
|
||||
_json("FilePicked", "org.freedesktop.portal.FileChooser", [_uri(shot), {}]),
|
||||
_json("Finished", "org.freedesktop.portal.Request", [0]),
|
||||
])
|
||||
pc.subprocess.run = run_fake
|
||||
pc.subprocess.Popen = popen_fake
|
||||
try:
|
||||
w = pc.PortalScreenshotWorker(wait_budget_s=10)
|
||||
done = {}
|
||||
w.done.connect(lambda ok, r: done.update(ok=ok, r=r))
|
||||
w.start()
|
||||
deadline = time.time() + 15
|
||||
while "ok" not in done and time.time() < deadline:
|
||||
APP.processEvents()
|
||||
time.sleep(0.02)
|
||||
w.wait(3000)
|
||||
finally:
|
||||
pc.subprocess.run = _orig_run
|
||||
pc.subprocess.Popen = _orig_popen
|
||||
pc.shutil.which = _orig_which2
|
||||
if _saved_xdg is None:
|
||||
os.environ.pop("XDG_RUNTIME_DIR", None)
|
||||
else:
|
||||
os.environ["XDG_RUNTIME_DIR"] = _saved_xdg
|
||||
pc.sys = _orig_sys
|
||||
check("C7.1 worker done 信号回主线程(ok + 路径)",
|
||||
done.get("ok") is True and done.get("r") == shot, repr(done))
|
||||
|
||||
# ======================================================================
|
||||
# C8. 覆盖层 offscreen 构造 + start() 空画面守卫
|
||||
# ======================================================================
|
||||
ov = sc.ScreenCaptureOverlay()
|
||||
buf = io.StringIO()
|
||||
with contextlib.redirect_stdout(buf):
|
||||
ov.start() # offscreen:无屏幕或空画面都不得崩
|
||||
shown = ov.isVisible()
|
||||
check("C8.1 offscreen start() 不崩;空画面时不显示覆盖层且日志明确",
|
||||
(not shown and "空画面" in buf.getvalue()) or (shown and ov._full_pixmap is not None),
|
||||
f"shown={shown} log={buf.getvalue()[-160:]!r}")
|
||||
if shown:
|
||||
ov.close()
|
||||
|
||||
print(f"\n===== {sum(1 for _, ok in RESULTS if ok)}/{len(RESULTS)} passed =====", flush=True)
|
||||
sys.exit(0 if all(ok for _, ok in RESULTS) else 1)
|
||||
@@ -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 grabWindow,X11 可用)。
|
||||
- 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+S(RegisterHotKey)"
|
||||
if kind == "x11":
|
||||
from ui.views.system_tools.x11_hotkey import X11HotkeyThread
|
||||
return X11HotkeyThread, "[GlobalHotkey] X11:原生全局热键 Alt+S(XGrabKey)"
|
||||
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/无显示)→ "
|
||||
"截图功能不可用;聊天与其他功能不受影响")
|
||||
@@ -0,0 +1,163 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
xdg-desktop-portal 截图(P1-04,Linux 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 CLI(GLib 组件)→ 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)
|
||||
@@ -60,8 +60,15 @@ class ScreenCaptureOverlay(QtWidgets.QWidget):
|
||||
"""开始截图:抓取屏幕全图并显示覆盖层"""
|
||||
screen = QtWidgets.QApplication.primaryScreen()
|
||||
if not screen:
|
||||
print("[Screenshot] 无主屏幕 → 截图不可用", flush=True)
|
||||
return
|
||||
self._full_pixmap = screen.grabWindow(0)
|
||||
# P1-04:X11 某些 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()
|
||||
|
||||
@@ -0,0 +1,196 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
X11 原生全局热键(P1-04,Linux X11 专用窄适配器)
|
||||
|
||||
用 ctypes 调 libX11:XOpenDisplay → XGrabKey(root, keycode, Mod1Mask) →
|
||||
select(display_fd) + XNextEvent 消息循环,命中后发射 triggered 信号。
|
||||
与 GlobalHotkeyThread(Windows)同一公开面: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
|
||||
Reference in New Issue
Block a user