From 0a62877cdec8a222e032ffb9b59647e4f9ba00f9 Mon Sep 17 00:00:00 2001 From: sorrow404null Date: Thu, 17 Sep 2026 16:40:05 +0800 Subject: [PATCH] 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. --- tests/test_global_hotkey_platforms.py | 279 +++++++++++++++++++++ tests/test_screen_capture_platforms.py | 293 +++++++++++++++++++++++ ui/views/system_tools/desktop_session.py | 76 ++++++ ui/views/system_tools/portal_capture.py | 163 +++++++++++++ ui/views/system_tools/screen_capture.py | 7 + ui/views/system_tools/x11_hotkey.py | 196 +++++++++++++++ 6 files changed, 1014 insertions(+) create mode 100644 tests/test_global_hotkey_platforms.py create mode 100644 tests/test_screen_capture_platforms.py create mode 100644 ui/views/system_tools/desktop_session.py create mode 100644 ui/views/system_tools/portal_capture.py create mode 100644 ui/views/system_tools/x11_hotkey.py diff --git a/tests/test_global_hotkey_platforms.py b/tests/test_global_hotkey_platforms.py new file mode 100644 index 0000000..e7646c0 --- /dev/null +++ b/tests/test_global_hotkey_platforms.py @@ -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) diff --git a/tests/test_screen_capture_platforms.py b/tests/test_screen_capture_platforms.py new file mode 100644 index 0000000..12d8c8a --- /dev/null +++ b/tests/test_screen_capture_platforms.py @@ -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) diff --git a/ui/views/system_tools/desktop_session.py b/ui/views/system_tools/desktop_session.py new file mode 100644 index 0000000..18557b3 --- /dev/null +++ b/ui/views/system_tools/desktop_session.py @@ -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/无显示）→ " + "截图功能不可用；聊天与其他功能不受影响") diff --git a/ui/views/system_tools/portal_capture.py b/ui/views/system_tools/portal_capture.py new file mode 100644 index 0000000..6bc1793 --- /dev/null +++ b/ui/views/system_tools/portal_capture.py @@ -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) diff --git a/ui/views/system_tools/screen_capture.py b/ui/views/system_tools/screen_capture.py index a4fafd0..8d8ca07 100644 --- a/ui/views/system_tools/screen_capture.py +++ b/ui/views/system_tools/screen_capture.py @@ -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() diff --git a/ui/views/system_tools/x11_hotkey.py b/ui/views/system_tools/x11_hotkey.py new file mode 100644 index 0000000..dcbe7d3 --- /dev/null +++ b/ui/views/system_tools/x11_hotkey.py @@ -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