From 3a6ba0063ec5ae8d4ba5477acb4053bce0e99882 Mon Sep 17 00:00:00 2001 From: antecanis8 <42382878+antecanis8@users.noreply.github.com> Date: Sun, 4 Oct 2026 11:04:33 +0800 Subject: [PATCH 1/4] =?UTF-8?q?feat:=20=E6=94=AF=E6=8C=81=E5=9C=A8?= =?UTF-8?q?=E8=84=9A=E6=9C=AC=E9=93=BE=E4=B8=AD=E5=BC=80=E5=85=B3=E9=9A=90?= =?UTF-8?q?=E7=A7=81=E5=B1=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 通过电脑操作打开或关闭全屏隐私屏。截图和录屏看不到黑屏，脚本自身的画面识别不受影响。可用 Ctrl+Alt+Shift 加主键在显示和隐藏之间切换。 --- .gitignore | 4 +- .../base/controller/pc_button/__init__.py | 0 .../pc_button/pc_button_listener.py | 83 +++ .../widgets/setting_card/key_setting_card.py | 4 +- src/script_chainer/config/run_config.py | 21 + src/script_chainer/config/script_config.py | 41 ++ .../context/script_chainer_context.py | 2 + .../gui/page/editor_setting_interface.py | 18 + .../gui/page/script_setting_cards.py | 89 ++- .../gui/page/script_setting_dialogs.py | 28 + .../gui/page/script_setting_interface.py | 15 + src/script_chainer/utils/privacy_screen.py | 561 ++++++++++++++++++ src/script_chainer/win_exe/script_runner.py | 53 +- 13 files changed, 915 insertions(+), 4 deletions(-) create mode 100644 src/one_dragon/base/controller/pc_button/__init__.py create mode 100644 src/one_dragon/base/controller/pc_button/pc_button_listener.py create mode 100644 src/script_chainer/config/run_config.py create mode 100644 src/script_chainer/utils/privacy_screen.py diff --git a/.gitignore b/.gitignore index 03a6a5b..ed9db4f 100644 --- a/.gitignore +++ b/.gitignore @@ -13,4 +13,6 @@ __pycache__/ config/script_chain/ config/custom.yml config/push.yml -config/env.yml \ No newline at end of file +config/env.yml +config/run.yml +.pai/ diff --git a/src/one_dragon/base/controller/pc_button/__init__.py b/src/one_dragon/base/controller/pc_button/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/one_dragon/base/controller/pc_button/pc_button_listener.py b/src/one_dragon/base/controller/pc_button/pc_button_listener.py new file mode 100644 index 0000000..e4587c6 --- /dev/null +++ b/src/one_dragon/base/controller/pc_button/pc_button_listener.py @@ -0,0 +1,83 @@ +from collections.abc import Callable +from concurrent.futures import Future, ThreadPoolExecutor + +from pynput import keyboard, mouse + +from one_dragon.utils import thread_utils + + +class PcButtonListener: + + def __init__(self, + on_button_tap: Callable[[str], None], + listen_keyboard: bool = False, + listen_mouse: bool = False, + listen_gamepad: bool = False, + ): + self.keyboard_listener = keyboard.Listener(on_press=self._on_keyboard_press) + self.mouse_listener = mouse.Listener(on_click=self._on_mouse_click) + self.gamepad_listener = None + + self.on_button_tap: Callable[[str], None] = on_button_tap + + self.listen_keyboard: bool = listen_keyboard + self.listen_mouse: bool = listen_mouse + self.listen_gamepad: bool = listen_gamepad + + self._executor = ThreadPoolExecutor(thread_name_prefix='od_key_mouse_btn_listener', max_workers=8) + + def _on_keyboard_press(self, event): + if isinstance(event, keyboard.Key): + k = event.name + elif isinstance(event, keyboard.KeyCode): + # 处理小键盘按键和特殊按键 + if event.char is not None: + k = event.char + elif hasattr(event, 'vk') and event.vk is not None: + # 使用虚拟键码来识别小键盘按键 + k = self._get_numpad_key_name(event.vk) + elif hasattr(event, 'vk'): + k = f'vk_{event.vk}' # vk 为 None 的情况 + else: + k = 'unknown' # 没有 vk 属性 + else: + return + + # 确保按键名称不为空 + if k is None or k == '': + return + + self._call_button_tap_callback(k) + + def _get_numpad_key_name(self, vk: int) -> str: + """ + 根据虚拟键码获取小键盘按键名称 + :param vk: 虚拟键码 + :return: 按键名称 + """ + # 小键盘数字键: vk 96-105 对应 numpad_0 到 numpad_9 + if 96 <= vk <= 105: + return f'numpad_{vk - 96}' + + # 其他按键返回通用格式 + return f'vk_{vk}' + + def _on_mouse_click(self, x, y, button: mouse.Button, pressed): + if pressed == 1: + self._call_button_tap_callback('mouse_' + button.name) + + def _call_button_tap_callback(self, key: str) -> None: + if self.on_button_tap is not None: + future: Future = self._executor.submit(self.on_button_tap, key) + future.add_done_callback(thread_utils.handle_future_result) + + def start(self): + if self.listen_keyboard: + self.keyboard_listener.start() + if self.listen_mouse: + self.mouse_listener.start() + + def stop(self): + self.keyboard_listener.stop() + self.mouse_listener.stop() + self._executor.shutdown(wait=False, cancel_futures=True) diff --git a/src/one_dragon_qt/widgets/setting_card/key_setting_card.py b/src/one_dragon_qt/widgets/setting_card/key_setting_card.py index 07ba252..d2b8867 100644 --- a/src/one_dragon_qt/widgets/setting_card/key_setting_card.py +++ b/src/one_dragon_qt/widgets/setting_card/key_setting_card.py @@ -36,6 +36,7 @@ def __init__(self, icon: Union[str, QIcon, FluentIconBase], title: str, content: Optional[str]=None, icon_size: IconSize = IconSize(16, 16), margins: Margins = Margins(16, 16, 0, 16), + listen_mouse: bool = True, parent=None): SettingCardBase.__init__( @@ -56,6 +57,7 @@ def __init__(self, self.btn.clicked.connect(self._on_btn_clicked) # 初始化监听器和键盘事件工作者 + self.listen_mouse: bool = listen_mouse self.button_listener = None # 按键监听 self.key_worker = KeyEventWorker() self.key_worker.key_pressed.connect(self._on_key_signal) @@ -71,7 +73,7 @@ def _on_btn_clicked(self) -> None: """ if self.button_listener is None: self.button_listener = PcButtonListener( - self._on_key_press, listen_keyboard=True, listen_mouse=True + self._on_key_press, listen_keyboard=True, listen_mouse=self.listen_mouse ) self.btn.setText(gt('请按键')) self.button_listener.start() diff --git a/src/script_chainer/config/run_config.py b/src/script_chainer/config/run_config.py new file mode 100644 index 0000000..0b4bf39 --- /dev/null +++ b/src/script_chainer/config/run_config.py @@ -0,0 +1,21 @@ +from one_dragon.base.config.yaml_config import YamlConfig + + +class RunConfig(YamlConfig): + """千机链专有的运行行为配置，落在 `config/run.yml`。""" + + def __init__(self, is_mock: bool = False): + YamlConfig.__init__( + self, + 'run', + is_mock=is_mock, sample=False, copy_from_sample=False, + ) + + @property + def privacy_screen_close_key(self) -> str: + """隐私屏开关热键的主键（修饰键固定为 Ctrl+Alt+Shift）。""" + return self.get('privacy_screen_close_key', 'f12') + + @privacy_screen_close_key.setter + def privacy_screen_close_key(self, new_value: str) -> None: + self.update('privacy_screen_close_key', new_value) diff --git a/src/script_chainer/config/script_config.py b/src/script_chainer/config/script_config.py index fd10e4f..d0d1c86 100644 --- a/src/script_chainer/config/script_config.py +++ b/src/script_chainer/config/script_config.py @@ -31,6 +31,12 @@ class ScriptProcessName(Enum): OK_SCRIPTS = ConfigItem(label='OK', value=['pythonw.exe']) +class ComputerOperation(Enum): + + OPEN_PRIVACY_SCREEN = ConfigItem(label='打开隐私屏', value='open_privacy_screen', desc='打开全屏隐私屏，屏幕变黑且截图录屏拍不到画面') + CLOSE_PRIVACY_SCREEN = ConfigItem(label='关闭隐私屏', value='close_privacy_screen', desc='关闭已打开的隐私屏，恢复屏幕显示') + + class GameProcessName(Enum): GENSHIN_IMPACT_CN = ConfigItem(label='原神', value='YuanShen.exe') @@ -47,6 +53,7 @@ class GameProcessName(Enum): class ScriptType: EXTERNAL = 'external' PYTHON = 'python' + COMPUTER_OP = 'computer_op' class AttachDirection: @@ -141,6 +148,7 @@ class ScriptConfig: game_label: str = '' script_type: str = ScriptType.EXTERNAL script_path: str = '' + computer_operation: str = '' script_process_name: list[str] = field(default_factory=list) game_process_name: str = '' launcher_mode: bool = False @@ -180,6 +188,15 @@ def create_python_default(cls) -> 'ScriptConfig': notify_done=False, ) + @classmethod + def create_computer_op_default(cls) -> 'ScriptConfig': + """创建电脑操作类型的默认配置。""" + return cls( + script_type=ScriptType.COMPUTER_OP, + notify_start=False, + notify_done=False, + ) + def copy(self) -> 'ScriptConfig': """深拷贝（保留 idx）。""" new = ScriptConfig(**self.to_dict()) @@ -243,8 +260,20 @@ def check_done_display_name(self) -> str: else: return '' + @property + def computer_operation_display_name(self) -> str: + config = get_config_item_from_enum(ComputerOperation, self.computer_operation) + if config is not None: + return config.label + return '' + @property def invalid_message(self) -> str | None: + if self.script_type == ScriptType.COMPUTER_OP: + if get_config_item_from_enum(ComputerOperation, self.computer_operation) is None: + return '未选择电脑操作' + return None + if self.script_type == ScriptType.PYTHON: if not self.script_path: return 'Python 脚本路径为空' @@ -433,6 +462,18 @@ def add_one(self) -> ScriptConfig: self.save() return new_config + def add_computer_op(self) -> ScriptConfig: + """新增一个电脑操作配置并返回。 + + Returns: + 新创建的 ScriptConfig。 + """ + new_config = ScriptConfig.create_computer_op_default() + self.script_list.append(new_config) + self.init_idx() + self.save() + return new_config + def delete_one(self, index: int) -> None: """删除一个配置。 diff --git a/src/script_chainer/context/script_chainer_context.py b/src/script_chainer/context/script_chainer_context.py index 12267c0..7adf7af 100644 --- a/src/script_chainer/context/script_chainer_context.py +++ b/src/script_chainer/context/script_chainer_context.py @@ -10,6 +10,7 @@ from one_dragon.envs.project_config import ProjectConfig from one_dragon.utils import os_utils from one_dragon.utils.log_utils import log +from script_chainer.config.run_config import RunConfig from script_chainer.config.script_config import ScriptChainConfig from script_chainer.services.github_update_service import GithubUpdateService @@ -26,6 +27,7 @@ def __init__(self): self.github_update_service: GithubUpdateService = GithubUpdateService(self) self.push_service: PushService = PushService(self) self.notify_config: NotifyConfig = NotifyConfig(None, {}) + self.run_config: RunConfig = RunConfig() self._init_lock = threading.Lock() def init(self) -> None: diff --git a/src/script_chainer/gui/page/editor_setting_interface.py b/src/script_chainer/gui/page/editor_setting_interface.py index 66f98d2..e8b382b 100644 --- a/src/script_chainer/gui/page/editor_setting_interface.py +++ b/src/script_chainer/gui/page/editor_setting_interface.py @@ -23,6 +23,7 @@ from one_dragon_qt.widgets.setting_card.combo_box_setting_card import ( ComboBoxSettingCard, ) +from one_dragon_qt.widgets.setting_card.key_setting_card import KeySettingCard from one_dragon_qt.widgets.setting_card.multi_push_setting_card import ( MultiPushSettingCard, ) @@ -280,6 +281,7 @@ def get_content_widget(self) -> QWidget: content_widget = Column() content_widget.add_widget(self.get_appearance_group()) + content_widget.add_widget(self.get_run_group()) content_widget.add_widget(self.get_update_group()) content_widget.add_stretch(1) @@ -302,6 +304,19 @@ def get_appearance_group(self) -> SettingCardGroup: return group + def get_run_group(self) -> SettingCardGroup: + group = SettingCardGroup(gt('运行')) + + self.privacy_screen_key_input = KeySettingCard( + icon=FluentIcon.POWER_BUTTON, + title='隐私屏开关快捷键', + content='组合键固定 Ctrl+Alt+Shift，按下后在显示和隐藏之间切换；点击右侧按钮后按下主键完成录制', + listen_mouse=False, + ) + group.addSettingCard(self.privacy_screen_key_input) + + return group + def get_update_group(self) -> SettingCardGroup: group = SettingCardGroup(gt('更新')) @@ -359,6 +374,9 @@ def get_update_group(self) -> SettingCardGroup: def on_interface_shown(self) -> None: VerticalScrollInterface.on_interface_shown(self) self.theme_opt.init_with_adapter(self.ctx.custom_config.get_prop_adapter('theme')) + self.privacy_screen_key_input.init_with_adapter( + self.ctx.run_config.get_prop_adapter('privacy_screen_close_key') + ) self.proxy_type_opt.init_with_adapter(self.ctx.env_config.get_prop_adapter('proxy_type')) self.personal_proxy_input.init_with_adapter(self.ctx.env_config.get_prop_adapter('personal_proxy')) self.gh_proxy_url_opt.init_with_adapter(self.ctx.env_config.get_prop_adapter('gh_proxy_url')) diff --git a/src/script_chainer/gui/page/script_setting_cards.py b/src/script_chainer/gui/page/script_setting_cards.py index 548b2bd..63dd3ed 100644 --- a/src/script_chainer/gui/page/script_setting_cards.py +++ b/src/script_chainer/gui/page/script_setting_cards.py @@ -19,7 +19,7 @@ ScriptChainConfig, ScriptConfig, ) -from script_chainer.gui.page.script_setting_dialogs import ScriptRenameDialog +from script_chainer.gui.page.script_setting_dialogs import ComputerOpEditDialog, ScriptRenameDialog from script_chainer.gui.page.script_setting_utils import ( show_error, show_success, @@ -344,3 +344,90 @@ def _update_display(self) -> None: self._external_tag.setVisible(is_external) self.edit_btn.setIcon(FluentIcon.FOLDER.icon() if is_external else FluentIcon.EDIT.icon()) self.edit_btn.setToolTip('定位文件' if is_external else '编辑') + + +class ComputerOpSettingCard(ScriptCardMixin, DraggableListItem): + """电脑操作卡片，通过铅笔编辑选择要执行的操作。""" + + value_changed = Signal(ScriptConfig) + deleted = Signal(int) + + def __init__(self, config: ScriptConfig, chain_name: str, index: int = 0, parent=None, + enable_opacity_effect: bool = True): + self.config: ScriptConfig = config + self.chain_name: str = chain_name + self._setup_common_widgets() + + self.debug_btn = TransparentToolButton(FluentIcon.PLAY, None) + self.debug_btn.setToolTip('调试运行') + self.debug_btn.clicked.connect(self.on_debug_clicked) + + content_widget = MultiPushSettingCard( + icon=FluentIcon.COMMAND_PROMPT, + title=self._get_title(), + content=self._get_operation_display(), + parent=parent, + btn_list=[ + self.debug_btn, + self.edit_btn, + self.delete_btn, + self.enable_switch, + ] + ) + + self._setup_rename_btn(content_widget) + + DraggableListItem.__init__( + self, + data=config, + index=index, + content_widget=content_widget, + parent=parent, + enable_opacity_effect=enable_opacity_effect, + ) + + self.content_widget: MultiPushSettingCard + self._update_display() + + def _get_title(self) -> str: + if self.config.display_name: + return self.config.display_name + return '电脑操作' + + def _get_operation_display(self) -> str: + label = self.config.computer_operation_display_name + return label if label else '未设置' + + def on_debug_clicked(self) -> None: + """调试运行当前电脑操作""" + invalid_msg = self.config.invalid_message + if invalid_msg is not None: + show_warning(self.window(), '配置不合法', invalid_msg) + return + + display = self._get_title() + + try: + cmd, cwd = build_runner_command(self.chain_name, self.index) + launch_in_terminal( + command=cmd, + cwd=cwd, + title=f'调试 {display}', + ) + show_success(self.window(), '调试运行', f'已在终端启动 {display}') + except Exception as e: + show_error(self.window(), '启动失败', str(e)) + + def on_edit_clicked(self) -> None: + """弹窗选择要执行的电脑操作。""" + dialog = ComputerOpEditDialog(self.config.computer_operation, parent=self.window()) + if dialog.exec(): + self.config.computer_operation = dialog.get_operation() + self._update_display() + self.value_changed.emit(self.config) + + def _update_display(self) -> None: + """更新卡片显示内容""" + self.content_widget.setTitle(self._get_title()) + self.content_widget.setContent(self._get_operation_display()) + self.enable_switch.setChecked(self.config.enabled) diff --git a/src/script_chainer/gui/page/script_setting_dialogs.py b/src/script_chainer/gui/page/script_setting_dialogs.py index 0fd33d6..8843ac9 100644 --- a/src/script_chainer/gui/page/script_setting_dialogs.py +++ b/src/script_chainer/gui/page/script_setting_dialogs.py @@ -2,6 +2,9 @@ from PySide6.QtGui import QColor from qfluentwidgets import CaptionLabel, LineEdit, MessageBoxBase, SubtitleLabel +from one_dragon_qt.widgets.combo_box import ComboBox +from script_chainer.config.script_config import ComputerOperation + class ChainRenameDialog(MessageBoxBase): def __init__(self, current_name: str, parent=None): @@ -63,3 +66,28 @@ def __init__(self, current_name: str, parent=None): def get_new_name(self) -> str: return self.name_input.text().strip() + + +class ComputerOpEditDialog(MessageBoxBase): + """编辑电脑操作：选择要执行的操作。""" + + def __init__(self, current_operation: str, parent=None): + MessageBoxBase.__init__(self, parent) + self.yesButton.setText('确定') + self.cancelButton.setText('取消') + + self.title_label = SubtitleLabel(text="编辑电脑操作") + self.viewLayout.addWidget(self.title_label, alignment=Qt.AlignmentFlag.AlignCenter) + + self.operation_combo = ComboBox() + self.operation_combo.set_items( + [enum_item.value for enum_item in ComputerOperation], + target_value=current_operation or None, + ) + if self.operation_combo.currentIndex() < 0: + self.operation_combo.setCurrentIndex(0) + self.operation_combo.setFixedWidth(300) + self.viewLayout.addWidget(self.operation_combo) + + def get_operation(self) -> str: + return self.operation_combo.currentData() or '' diff --git a/src/script_chainer/gui/page/script_setting_interface.py b/src/script_chainer/gui/page/script_setting_interface.py index 1fa5ad1..d3ba4db 100644 --- a/src/script_chainer/gui/page/script_setting_interface.py +++ b/src/script_chainer/gui/page/script_setting_interface.py @@ -26,6 +26,7 @@ from script_chainer.context.script_chainer_context import ScriptChainerContext from script_chainer.gui.page.script_edit_interface import ScriptEditInterface from script_chainer.gui.page.script_setting_cards import ( + ComputerOpSettingCard, PythonScriptSettingCard, ScriptSettingCard, ) @@ -87,6 +88,7 @@ def get_fixed_top_widget(self) -> QWidget | None: add_script_menu.addAction(Action(FluentIcon.APPLICATION, '外部程序', triggered=self.on_add_script_clicked)) add_script_menu.addAction(Action(FluentIcon.CODE, '新建 Python 脚本', triggered=self.on_add_python_script_clicked)) add_script_menu.addAction(Action(FluentIcon.DOCUMENT, '选择已有 Python 脚本', triggered=self.on_import_python_script_clicked)) + add_script_menu.addAction(Action(FluentIcon.COMMAND_PROMPT, '电脑操作', triggered=self.on_add_computer_op_clicked)) self.add_script_btn = PrimaryDropDownPushButton(text='新增脚本') self.add_script_btn.setMenu(add_script_menu) @@ -200,6 +202,13 @@ def on_import_python_script_clicked(self) -> None: self.chosen_config.add_python_script_from_file(file_path) self.update_chain_display() + def on_add_computer_op_clicked(self) -> None: + """新增一个电脑操作""" + if self.chosen_config is None: + return + self.chosen_config.add_computer_op() + self.update_chain_display() + def on_run_chain_clicked(self) -> None: """拉起独立 runner 运行当前脚本链。""" if self.chosen_config is None or self._runner_launch_in_progress: @@ -249,6 +258,12 @@ def update_chain_display(self) -> None: card = PythonScriptSettingCard( script_config, chain_config=self.chosen_config, index=i) card.attach_changed.connect(self._update_attach_margins) + elif script_config.script_type == ScriptType.COMPUTER_OP: + card = ComputerOpSettingCard( + script_config, + chain_name=self.chosen_config.module_name, + index=i, + ) else: card = ScriptSettingCard( script_config, diff --git a/src/script_chainer/utils/privacy_screen.py b/src/script_chainer/utils/privacy_screen.py new file mode 100644 index 0000000..5607f9d --- /dev/null +++ b/src/script_chainer/utils/privacy_screen.py @@ -0,0 +1,561 @@ +"""运行脚本链期间的隐私屏。 + +在物理屏幕上盖一块 TOPMOST 全屏黑窗，并用 +``SetWindowDisplayAffinity(WDA_EXCLUDEFROMCAPTURE)`` 让这块窗口对截图 / 录屏 / +远程桌面不可见：人眼看到的是黑屏，捕获 API 拍到的是窗口下方的真实画面，因此 +脚本自身的画面识别不受影响。 + +实现是纯 Win32（ctypes），不依赖 Qt、不新增进程，可以直接跑在没有 Qt 事件循环的 +runner 进程里；窗口与消息循环挂在后台线程，不阻塞调用方。 + +任何一步失败都只记录日志、不向调用方抛异常，避免影响脚本链执行。 +窗口对鼠标穿透：带 ``WS_EX_LAYERED | WS_EX_TRANSPARENT``，点击会落到下层窗口。 +只返回 ``HTTRANSPARENT`` 不够，那只会把命中转给同一线程里的窗口，到不了游戏进程。 +可用 ``Ctrl + Alt + Shift + 主键`` 在显示和隐藏之间切换 +（主键默认 F12，可在设置界面录制，见 :class:`PrivacyScreen`）。 +隐藏只收起黑屏，窗口线程和热键仍在；脚本链结束或「关闭隐私屏」才会真正关掉。 +""" + +from __future__ import annotations + +import ctypes +import re +import threading + +from ctypes import wintypes + +from script_chainer.win_exe.runner_logging import log + +# ---------------------------------------------------------------- 常量 + +WS_POPUP = 0x80000000 +WS_VISIBLE = 0x10000000 +WS_EX_TOPMOST = 0x00000008 +WS_EX_TOOLWINDOW = 0x00000080 +WS_EX_TRANSPARENT = 0x00000020 +WS_EX_LAYERED = 0x00080000 +WS_EX_NOACTIVATE = 0x08000000 + +# 不透明度 255：人眼仍是纯黑，穿透靠的是 WS_EX_TRANSPARENT 而不是透明像素 +LWA_ALPHA = 0x00000002 +CLICK_THROUGH_ALPHA = 255 + +SW_HIDE = 0 +SW_SHOWNOACTIVATE = 4 +HWND_TOPMOST = -1 +SWP_NOSIZE = 0x0001 +SWP_NOMOVE = 0x0002 +SWP_NOACTIVATE = 0x0010 +SWP_SHOWWINDOW = 0x0040 + +WM_DESTROY = 0x0002 +WM_CLOSE = 0x0010 +WM_TIMER = 0x0113 +WM_HOTKEY = 0x0312 +WM_NCHITTEST = 0x0084 +WM_APP = 0x8000 +WM_PRIVACY_SHOW = WM_APP + 1 +HTTRANSPARENT = -1 + +WDA_EXCLUDEFROMCAPTURE = 0x00000011 + +SM_XVIRTUALSCREEN = 76 +SM_YVIRTUALSCREEN = 77 +SM_CXVIRTUALSCREEN = 78 +SM_CYVIRTUALSCREEN = 79 + +MOD_ALT = 0x0001 +MOD_CONTROL = 0x0002 +MOD_SHIFT = 0x0004 +MOD_NOREPEAT = 0x4000 + +ERROR_CLASS_ALREADY_EXISTS = 1410 + +TOP_MOST_TIMER_ID = 1 +TOP_MOST_INTERVAL_MS = 1000 +EMERGENCY_HOTKEY_ID = 0x91A7 + +DPI_AWARENESS_CONTEXT_PER_MONITOR_AWARE_V2 = -4 + +WINDOW_CLASS_NAME = 'OD_SCRIPT_CHAINER_PRIVACY_SCREEN' +WINDOW_TITLE = 'OneDragon ScriptChainer Privacy Screen' + +DEFAULT_CLOSE_KEY = 'f12' + +START_TIMEOUT_SECONDS = 5 +STOP_TIMEOUT_SECONDS = 5 + +# ---------------------------------------------------------------- Win32 + +_user32 = ctypes.WinDLL('user32', use_last_error=True) +_gdi32 = ctypes.WinDLL('gdi32', use_last_error=True) +_kernel32 = ctypes.WinDLL('kernel32', use_last_error=True) + +LRESULT = ctypes.c_longlong + +WNDPROC = ctypes.WINFUNCTYPE( + LRESULT, wintypes.HWND, wintypes.UINT, wintypes.WPARAM, wintypes.LPARAM +) + + +class WNDCLASSEXW(ctypes.Structure): + _fields_ = [ + ('cbSize', wintypes.UINT), + ('style', wintypes.UINT), + ('lpfnWndProc', WNDPROC), + ('cbClsExtra', ctypes.c_int), + ('cbWndExtra', ctypes.c_int), + ('hInstance', wintypes.HINSTANCE), + ('hIcon', wintypes.HICON), + ('hCursor', wintypes.HANDLE), + ('hbrBackground', wintypes.HBRUSH), + ('lpszMenuName', wintypes.LPCWSTR), + ('lpszClassName', wintypes.LPCWSTR), + ('hIconSm', wintypes.HICON), + ] + + +_user32.RegisterClassExW.argtypes = [ctypes.POINTER(WNDCLASSEXW)] +_user32.RegisterClassExW.restype = ctypes.c_ushort + +_user32.CreateWindowExW.argtypes = [ + wintypes.DWORD, wintypes.LPCWSTR, wintypes.LPCWSTR, wintypes.DWORD, + ctypes.c_int, ctypes.c_int, ctypes.c_int, ctypes.c_int, + wintypes.HWND, wintypes.HMENU, wintypes.HINSTANCE, wintypes.LPVOID, +] +_user32.CreateWindowExW.restype = wintypes.HWND + +_user32.DefWindowProcW.argtypes = [ + wintypes.HWND, wintypes.UINT, wintypes.WPARAM, wintypes.LPARAM +] +_user32.DefWindowProcW.restype = LRESULT + +_user32.GetMessageW.argtypes = [ + ctypes.POINTER(wintypes.MSG), wintypes.HWND, wintypes.UINT, wintypes.UINT +] +_user32.GetMessageW.restype = ctypes.c_int + +_user32.TranslateMessage.argtypes = [ctypes.POINTER(wintypes.MSG)] + +_user32.DispatchMessageW.argtypes = [ctypes.POINTER(wintypes.MSG)] +_user32.DispatchMessageW.restype = LRESULT + +_user32.DestroyWindow.argtypes = [wintypes.HWND] +_user32.PostQuitMessage.argtypes = [ctypes.c_int] + +_user32.PostMessageW.argtypes = [ + wintypes.HWND, wintypes.UINT, wintypes.WPARAM, wintypes.LPARAM +] +_user32.PostMessageW.restype = wintypes.BOOL + +_user32.SendMessageW.argtypes = [ + wintypes.HWND, wintypes.UINT, wintypes.WPARAM, wintypes.LPARAM +] +_user32.SendMessageW.restype = LRESULT + +_user32.ShowWindow.argtypes = [wintypes.HWND, ctypes.c_int] +_user32.UpdateWindow.argtypes = [wintypes.HWND] + +_user32.SetTimer.argtypes = [ + wintypes.HWND, ctypes.c_size_t, wintypes.UINT, ctypes.c_void_p +] +_user32.SetTimer.restype = ctypes.c_size_t + +_user32.SetWindowPos.argtypes = [ + wintypes.HWND, wintypes.HWND, ctypes.c_int, ctypes.c_int, + ctypes.c_int, ctypes.c_int, wintypes.UINT, +] +_user32.SetWindowPos.restype = wintypes.BOOL + +_user32.SetWindowDisplayAffinity.argtypes = [wintypes.HWND, wintypes.DWORD] +_user32.SetWindowDisplayAffinity.restype = wintypes.BOOL + +_user32.SetLayeredWindowAttributes.argtypes = [ + wintypes.HWND, wintypes.DWORD, wintypes.BYTE, wintypes.DWORD +] +_user32.SetLayeredWindowAttributes.restype = wintypes.BOOL + +_user32.GetSystemMetrics.argtypes = [ctypes.c_int] +_user32.GetSystemMetrics.restype = ctypes.c_int + +_user32.IsWindowVisible.argtypes = [wintypes.HWND] +_user32.IsWindowVisible.restype = wintypes.BOOL + +_user32.RegisterHotKey.argtypes = [ + wintypes.HWND, ctypes.c_int, wintypes.UINT, wintypes.UINT +] +_user32.RegisterHotKey.restype = wintypes.BOOL + +_user32.SetThreadDpiAwarenessContext.argtypes = [ctypes.c_void_p] +_user32.SetThreadDpiAwarenessContext.restype = ctypes.c_void_p + +_gdi32.CreateSolidBrush.argtypes = [wintypes.DWORD] +_gdi32.CreateSolidBrush.restype = wintypes.HBRUSH + +_kernel32.GetModuleHandleW.argtypes = [wintypes.LPCWSTR] +_kernel32.GetModuleHandleW.restype = wintypes.HMODULE + + +# ---------------------------------------------------------------- 热键解析 + +_KEY_TO_VK_EXTRA = { + 'space': 0x20, + 'tab': 0x09, + 'enter': 0x0D, + 'esc': 0x1B, + 'escape': 0x1B, + 'backspace': 0x08, + 'delete': 0x2E, + 'insert': 0x2D, + 'home': 0x24, + 'end': 0x23, + 'page_up': 0x21, + 'page_down': 0x22, + 'up': 0x26, + 'down': 0x28, + 'left': 0x25, + 'right': 0x27, + 'minus': 0xBD, + 'equals': 0xBB, + 'comma': 0xBC, + 'period': 0xBE, + 'slash': 0xBF, + 'backslash': 0xDC, + 'semicolon': 0xBA, + 'apostrophe': 0xDE, + 'grave': 0xC0, + 'l_bracket': 0xDB, + 'r_bracket': 0xDD, +} + + +def key_to_vk(key: str) -> int | None: + """把按键名解析成 Windows 虚拟键码，无法识别时返回 None。 + + 支持 ``a``~``z``、``0``~``9``、``f1``~``f24``、``numpad_0``~``numpad_9``、 + ``vk_N`` 数字码以及 ``_KEY_TO_VK_EXTRA`` 里的常用键名。 + """ + key_name = str(key or '').strip().lower() + if not key_name: + return None + + vk_match = re.fullmatch(r'vk_(\d+)', key_name) + if vk_match: + vk = int(vk_match.group(1)) + return vk if 0 <= vk <= 254 else None + + if len(key_name) == 1 and key_name.isalnum(): + return ord(key_name.upper()) + + if key_name.startswith('numpad_'): + suffix = key_name.replace('numpad_', '', 1) + if suffix.isdigit(): + num = int(suffix) + if 0 <= num <= 9: + return 0x60 + num + return None + + fn_match = re.fullmatch(r'f(\d{1,2})', key_name) + if fn_match: + fn_num = int(fn_match.group(1)) + if 1 <= fn_num <= 24: + return 0x70 + fn_num - 1 + + return _KEY_TO_VK_EXTRA.get(key_name) + + +# ---------------------------------------------------------------- 进程级状态 + +_class_lock = threading.Lock() +_class_registered = False +_wndproc_callback: WNDPROC | None = None +_background_brush: int | None = None +def _set_thread_dpi_awareness() -> int | None: + """把当前线程设为 PER_MONITOR_AWARE_V2，返回旧 context 供恢复使用。 + + 与进程级 ``SetProcessDpiAwarenessContext`` 不同， + ``SetThreadDpiAwarenessContext`` 只影响当前线程，不会污染进程内其他线程 + （例如自动化软件截图所在的线程）。 + + 返回值是旧的 DPI_AWARENESS_CONTEXT 句柄（非 0/NULL）， + 传给 :func:`_restore_thread_dpi_awareness` 即可恢复。 + 失败时返回 None。 + """ + ctypes.set_last_error(0) + old = _user32.SetThreadDpiAwarenessContext( + ctypes.c_void_p(DPI_AWARENESS_CONTEXT_PER_MONITOR_AWARE_V2) + ) + if old: + return int(old) + log.warning('SetThreadDpiAwarenessContext 失败 error=%s', ctypes.get_last_error()) + return None + + +def _restore_thread_dpi_awareness(old_context: int | None) -> None: + """恢复线程原来的 DPI 感知 context。""" + if old_context is None: + return + _user32.SetThreadDpiAwarenessContext(ctypes.c_void_p(old_context)) + + +def _register_window_class() -> bool: + """注册隐私屏窗口类，进程内只注册一次。""" + global _class_registered, _wndproc_callback, _background_brush + + with _class_lock: + if _class_registered: + return True + + # 窗口类长期存在，回调与背景刷必须一直持有引用 + _wndproc_callback = WNDPROC(_window_proc) + _background_brush = _gdi32.CreateSolidBrush(0x00000000) + + wc = WNDCLASSEXW() + wc.cbSize = ctypes.sizeof(WNDCLASSEXW) + wc.style = 0 + wc.lpfnWndProc = _wndproc_callback + wc.hInstance = _kernel32.GetModuleHandleW(None) + wc.hbrBackground = _background_brush + wc.lpszClassName = WINDOW_CLASS_NAME + + ctypes.set_last_error(0) + if not _user32.RegisterClassExW(ctypes.byref(wc)): + error = ctypes.get_last_error() + if error != ERROR_CLASS_ALREADY_EXISTS: + log.error('注册隐私屏窗口类失败 error=%s', error) + return False + + _class_registered = True + return True + + +def _window_proc(hwnd, msg, wparam, lparam): + """隐私屏窗口的消息处理。 + + 窗口本身无状态（一块纯黑板），消息处理不依赖实例，因此用模块级函数。 + """ + try: + if msg == WM_CLOSE: + _user32.DestroyWindow(hwnd) + return 0 + if msg == WM_DESTROY: + _user32.PostQuitMessage(0) + return 0 + if msg == WM_TIMER and wparam == TOP_MOST_TIMER_ID: + # 隐藏期间不能带 SWP_SHOWWINDOW，否则一秒内又会被拉起来。 + if _user32.IsWindowVisible(hwnd): + _user32.SetWindowPos( + hwnd, HWND_TOPMOST, 0, 0, 0, 0, + SWP_NOMOVE | SWP_NOSIZE | SWP_NOACTIVATE | SWP_SHOWWINDOW, + ) + return 0 + if msg == WM_NCHITTEST: + # 同一线程内的兜底。跨进程穿透靠 WS_EX_LAYERED | WS_EX_TRANSPARENT。 + return HTTRANSPARENT + if msg == WM_HOTKEY and wparam == EMERGENCY_HOTKEY_ID: + if _user32.IsWindowVisible(hwnd): + _conceal_privacy_window(hwnd) + else: + _reveal_privacy_window(hwnd) + return 0 + if msg == WM_PRIVACY_SHOW: + if not _user32.IsWindowVisible(hwnd): + _reveal_privacy_window(hwnd) + return 0 + except Exception: + log.error('隐私屏窗口消息处理异常 msg=%s', msg, exc_info=True) + return 0 + + return _user32.DefWindowProcW(hwnd, msg, wparam, lparam) + + +def _conceal_privacy_window(hwnd: int) -> None: + """暂时隐藏隐私屏。窗口、消息循环和热键都保留，以便再次打开。""" + _user32.ShowWindow(hwnd, SW_HIDE) + log.info('隐私屏已隐藏') + + +def _reveal_privacy_window(hwnd: int) -> None: + """重新显示已隐藏的隐私屏，并再次置顶、排除捕获。""" + _user32.ShowWindow(hwnd, SW_SHOWNOACTIVATE) + _user32.SetWindowPos( + hwnd, HWND_TOPMOST, 0, 0, 0, 0, + SWP_NOMOVE | SWP_NOSIZE | SWP_NOACTIVATE | SWP_SHOWWINDOW, + ) + ctypes.set_last_error(0) + if not _user32.SetWindowDisplayAffinity(hwnd, WDA_EXCLUDEFROMCAPTURE): + log.warning( + '重新显示时 SetWindowDisplayAffinity 失败 error=%s 隐私屏仍会显示但截屏可能拍到画面', + ctypes.get_last_error(), + ) + log.info('隐私屏已显示') + + +def _enable_click_through(hwnd: int) -> bool: + """把已带 WS_EX_LAYERED 的窗口设成不透明黑底，同时保留鼠标穿透。 + + 分层窗口在调用 ``SetLayeredWindowAttributes`` 之前不会正常显示。 + Alpha 取 255，画面仍是纯黑；穿透来自扩展样式，不来自透明像素。 + """ + ctypes.set_last_error(0) + if _user32.SetLayeredWindowAttributes(hwnd, 0, CLICK_THROUGH_ALPHA, LWA_ALPHA): + return True + log.error('SetLayeredWindowAttributes 失败 error=%s', ctypes.get_last_error()) + return False + + +def _create_privacy_window(close_key: str) -> int | None: + """创建并显示覆盖整个虚拟桌面的隐私屏窗口，失败返回 None。 + + 调用前应已通过 :func:`_set_thread_dpi_awareness` 设置线程级 DPI 感知， + 这样 ``GetSystemMetrics`` 才能返回物理像素而非逻辑像素。 + """ + if not _register_window_class(): + return None + + vx = _user32.GetSystemMetrics(SM_XVIRTUALSCREEN) + vy = _user32.GetSystemMetrics(SM_YVIRTUALSCREEN) + vw = _user32.GetSystemMetrics(SM_CXVIRTUALSCREEN) + vh = _user32.GetSystemMetrics(SM_CYVIRTUALSCREEN) + + hwnd = _user32.CreateWindowExW( + WS_EX_TOPMOST | WS_EX_TOOLWINDOW | WS_EX_LAYERED | WS_EX_TRANSPARENT | WS_EX_NOACTIVATE, + WINDOW_CLASS_NAME, WINDOW_TITLE, + WS_POPUP, + vx, vy, vw, vh, + None, None, _kernel32.GetModuleHandleW(None), None, + ) + if not hwnd: + log.error('创建隐私屏窗口失败 error=%s', ctypes.get_last_error()) + return None + + if not _enable_click_through(hwnd): + _user32.DestroyWindow(hwnd) + return None + + _user32.ShowWindow(hwnd, SW_SHOWNOACTIVATE) + _user32.UpdateWindow(hwnd) + + if _user32.SetWindowDisplayAffinity(hwnd, WDA_EXCLUDEFROMCAPTURE): + log.info('隐私屏已覆盖虚拟桌面 (%s,%s) %sx%s', vx, vy, vw, vh) + else: + log.warning( + 'SetWindowDisplayAffinity 失败 error=%s 隐私屏仍会显示但截屏可能拍到画面', + ctypes.get_last_error(), + ) + + # 修饰键固定 Ctrl+Alt+Shift，主键来自设置；解析失败时回退默认值。 + # MOD_NOREPEAT：按住不连发，避免一次按键把隐私屏来回切换。 + main_key = str(close_key or '').strip().lower() + main_vk = key_to_vk(main_key) + if main_vk is None: + log.warning('隐私屏热键主键无法识别 %r 回退默认主键 %s', close_key, DEFAULT_CLOSE_KEY) + main_key = DEFAULT_CLOSE_KEY + main_vk = key_to_vk(main_key) + + if not _user32.RegisterHotKey( + hwnd, + EMERGENCY_HOTKEY_ID, + MOD_CONTROL | MOD_ALT | MOD_SHIFT | MOD_NOREPEAT, + main_vk, + ): + log.warning( + '隐私屏开关热键 Ctrl+Alt+Shift+%s 注册失败 可能已被占用', + main_key.upper(), + ) + + _user32.SetTimer(hwnd, TOP_MOST_TIMER_ID, TOP_MOST_INTERVAL_MS, None) + return hwnd + + +class PrivacyScreen: + """全屏隐私屏。 + + ``start()`` 与 ``stop()`` 都是幂等的，且都不会抛异常；启动失败时 ``start()`` + 返回 False 并记录日志。 + + 快捷键只在显示和隐藏之间切换，不销毁窗口。``stop()`` 才会真正关闭。 + + Args: + close_key: 开关热键的主键（修饰键固定 Ctrl+Alt+Shift）， + 键名格式同 :func:`key_to_vk`；无法识别时回退默认主键 F12。 + """ + + def __init__(self, close_key: str = DEFAULT_CLOSE_KEY) -> None: + self._close_key = close_key + self._lock = threading.Lock() + self._thread: threading.Thread | None = None + self._hwnd: int | None = None + self._ready = threading.Event() + + def start(self) -> bool: + """在后台线程拉起隐私屏，返回是否成功盖住屏幕。 + + 已经在运行但被快捷键隐藏时，会重新显示。 + """ + with self._lock: + if self._thread is not None and self._thread.is_alive(): + hwnd = self._hwnd + if hwnd: + # 窗口线程正在等消息，同步发过去，返回时黑屏已经盖上。 + _user32.SendMessageW(hwnd, WM_PRIVACY_SHOW, 0, 0) + return True + + self._hwnd = None + self._ready.clear() + self._thread = threading.Thread( + target=self._run, name='privacy_screen', daemon=True + ) + self._thread.start() + self._ready.wait(START_TIMEOUT_SECONDS) + hwnd = self._hwnd + + if hwnd is None: + log.error('隐私屏启动失败') + return False + + return True + + def stop(self) -> None: + """关闭隐私屏，恢复屏幕显示。""" + with self._lock: + thread = self._thread + hwnd = self._hwnd + self._thread = None + self._hwnd = None + + if thread is None or not thread.is_alive(): + return + + if hwnd: + _user32.PostMessageW(hwnd, WM_CLOSE, 0, 0) + + thread.join(STOP_TIMEOUT_SECONDS) + if thread.is_alive(): + log.warning('隐私屏未在 %s 秒内关闭 窗口将由进程退出回收', STOP_TIMEOUT_SECONDS) + else: + log.info('隐私屏已关闭') + + def _run(self) -> None: + """窗口线程：建窗、显示，然后跑消息循环直到窗口销毁。""" + old_dpi = _set_thread_dpi_awareness() + try: + hwnd = _create_privacy_window(self._close_key) + except Exception: + log.error('创建隐私屏窗口异常', exc_info=True) + hwnd = None + finally: + # 建窗完成后立即恢复线程 DPI 感知，将对 DPI 敏感的 + # 范围缩到最小（窗口本身的属性在建窗时已锁定）。 + _restore_thread_dpi_awareness(old_dpi) + + self._hwnd = hwnd + self._ready.set() + + if hwnd is None: + return + + msg = wintypes.MSG() + while _user32.GetMessageW(ctypes.byref(msg), None, 0, 0) > 0: + _user32.TranslateMessage(ctypes.byref(msg)) + _user32.DispatchMessageW(ctypes.byref(msg)) diff --git a/src/script_chainer/win_exe/script_runner.py b/src/script_chainer/win_exe/script_runner.py index a7498ac..b744297 100644 --- a/src/script_chainer/win_exe/script_runner.py +++ b/src/script_chainer/win_exe/script_runner.py @@ -16,8 +16,11 @@ from colorama import Fore, Style, init from one_dragon.utils import cmd_utils +from one_dragon.base.config.config_item import get_config_item_from_enum +from script_chainer.config.run_config import RunConfig from script_chainer.config.script_config import ( CheckDoneMethods, + ComputerOperation, ScriptChainConfig, ScriptConfig, ScriptType, @@ -33,6 +36,7 @@ is_process_existed, ) from script_chainer.utils.console_close_utils import force_exit_on_console_close +from script_chainer.utils.privacy_screen import DEFAULT_CLOSE_KEY, PrivacyScreen from script_chainer.utils.process_name_utils import ( normalize_process_names, process_name_equals, @@ -585,17 +589,50 @@ def _run_script_in_group( log_notifier: LogNotifier | None = None, ctx: ScriptChainerContext | None = None, chain_name: str = '', + privacy_screen: PrivacyScreen | None = None, ) -> None: """运行运行组中的单个脚本。""" try: if script_config.script_type == ScriptType.PYTHON: _run_python_script(script_config, log_notifier) + elif script_config.script_type == ScriptType.COMPUTER_OP: + _run_computer_op(script_config, privacy_screen) else: _run_external_script_with_retries(script_config, log_notifier, ctx, chain_name) except Exception: log.error('脚本执行异常', exc_info=True) +def _run_computer_op( + script_config: ScriptConfig, + privacy_screen: PrivacyScreen | None, +) -> None: + """执行电脑操作类型的脚本项（如打开/关闭隐私屏）。 + + Args: + script_config: 脚本配置（script_type == 'computer_op'）。 + privacy_screen: 本次链运行的隐私屏实例，为 None 时跳过。 + """ + op_config = get_config_item_from_enum(ComputerOperation, script_config.computer_operation) + if op_config is None: + print_message(f'电脑操作未选择 跳过 {script_config.script_display_name}', 'ERROR') + return + if privacy_screen is None: + print_message(f'隐私屏不可用 跳过电脑操作 {op_config.label}', 'ERROR') + return + + print_message(f'执行电脑操作 {op_config.label}...') + if script_config.computer_operation == ComputerOperation.OPEN_PRIVACY_SCREEN.value.value: + if privacy_screen.start(): + print_message(f'电脑操作执行完成 {op_config.label}', level='PASS') + else: + print_message(f'电脑操作执行失败 {op_config.label}', 'ERROR') + return + + privacy_screen.stop() + print_message(f'电脑操作执行完成 {op_config.label}', level='PASS') + + def _run_python_script( script_config: ScriptConfig, log_notifier: LogNotifier | None = None, @@ -687,6 +724,15 @@ def _cleanup_active_pm(): _active_pm = None +def _get_privacy_screen_close_key() -> str: + """读取隐私屏开关热键的主键，读取失败时用默认值，避免影响脚本链。""" + try: + return RunConfig().privacy_screen_close_key + except Exception: + log.error('读取运行配置失败 隐私屏热键主键使用默认 %s', DEFAULT_CLOSE_KEY, exc_info=True) + return DEFAULT_CLOSE_KEY + + def run_chain(chain_name: str = '01', shutdown_delay: int = 0, debug_index: int | None = None) -> None: """运行指定的脚本链。 @@ -714,6 +760,8 @@ def run_chain(chain_name: str = '01', shutdown_delay: int = 0, debug_index: int except Exception as e: log.error(f'初始化上下文实例失败: {e}') + privacy_screen = PrivacyScreen(close_key=_get_privacy_screen_close_key()) + try: if not chain_config.is_file_exists: print_message(f'脚本链配置不存在 {chain_name}', "ERROR") @@ -757,7 +805,8 @@ def run_chain(chain_name: str = '01', shutdown_delay: int = 0, debug_index: int ) for script_config in group.scripts: - _run_script_in_group(script_config, log_notifier, ctx, chain_name) + _run_script_in_group(script_config, log_notifier, ctx, chain_name, + privacy_screen=privacy_screen) if group.host.notify_done: _push_chain_notification( @@ -776,6 +825,7 @@ def run_chain(chain_name: str = '01', shutdown_delay: int = 0, debug_index: int break print_message('已完成调试脚本' if debug_index is not None else '已完成全部脚本') + privacy_screen.stop() if shutdown_delay > 0: cmd_utils.shutdown_sys(shutdown_delay) @@ -784,6 +834,7 @@ def run_chain(chain_name: str = '01', shutdown_delay: int = 0, debug_index: int print_message('5秒后关闭本窗口') _exit_controller.wait(5) finally: + privacy_screen.stop() # 清理资源 if ctx is not None: try: From 2c9dfb94d483eb45adadf00bcd5087f2f67981f0 Mon Sep 17 00:00:00 2001 From: antecanis8 <42382878+antecanis8@users.noreply.github.com> Date: Fri, 16 Oct 2026 11:10:04 +0800 Subject: [PATCH 2/4] =?UTF-8?q?fix(privacy-screen):=20=E4=BF=AE=E5=A4=8D?= =?UTF-8?q?=E5=87=8F=E5=8F=B7=E7=83=AD=E9=94=AE=E5=AD=97=E7=AC=A6=E6=97=A0?= =?UTF-8?q?=E6=B3=95=E8=AF=86=E5=88=AB=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/script_chainer/utils/privacy_screen.py | 1 + 1 file changed, 1 insertion(+) diff --git a/src/script_chainer/utils/privacy_screen.py b/src/script_chainer/utils/privacy_screen.py index 5607f9d..14b244f 100644 --- a/src/script_chainer/utils/privacy_screen.py +++ b/src/script_chainer/utils/privacy_screen.py @@ -216,6 +216,7 @@ class WNDCLASSEXW(ctypes.Structure): 'left': 0x25, 'right': 0x27, 'minus': 0xBD, + '-': 0xBD, 'equals': 0xBB, 'comma': 0xBC, 'period': 0xBE, From 9ff77aa61314e334df0894396a67a43b7a92452e Mon Sep 17 00:00:00 2001 From: antecanis8 <42382878+antecanis8@users.noreply.github.com> Date: Fri, 16 Oct 2026 11:50:48 +0800 Subject: [PATCH 3/4] =?UTF-8?q?fix(privacy-screen):=20=E4=BF=AE=E5=A4=8D?= =?UTF-8?q?=E6=8D=95=E8=8E=B7=E6=8E=92=E9=99=A4=E5=A4=B1=E8=B4=A5=E6=97=B6?= =?UTF-8?q?=E4=BB=8D=E6=98=BE=E7=A4=BA=E9=BB=91=E5=B1=8F=E7=9A=84=E9=97=AE?= =?UTF-8?q?=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/script_chainer/utils/privacy_screen.py | 48 +++-- tests/test_privacy_screen.py | 232 +++++++++++++++++++++ 2 files changed, 258 insertions(+), 22 deletions(-) create mode 100644 tests/test_privacy_screen.py diff --git a/src/script_chainer/utils/privacy_screen.py b/src/script_chainer/utils/privacy_screen.py index 14b244f..01dfe71 100644 --- a/src/script_chainer/utils/privacy_screen.py +++ b/src/script_chainer/utils/privacy_screen.py @@ -21,7 +21,6 @@ import ctypes import re import threading - from ctypes import wintypes from script_chainer.win_exe.runner_logging import log @@ -361,8 +360,8 @@ def _window_proc(hwnd, msg, wparam, lparam): return 0 if msg == WM_PRIVACY_SHOW: if not _user32.IsWindowVisible(hwnd): - _reveal_privacy_window(hwnd) - return 0 + return int(_reveal_privacy_window(hwnd)) + return 1 except Exception: log.error('隐私屏窗口消息处理异常 msg=%s', msg, exc_info=True) return 0 @@ -376,20 +375,23 @@ def _conceal_privacy_window(hwnd: int) -> None: log.info('隐私屏已隐藏') -def _reveal_privacy_window(hwnd: int) -> None: - """重新显示已隐藏的隐私屏，并再次置顶、排除捕获。""" +def _reveal_privacy_window(hwnd: int) -> bool: + """先确认排除捕获，再重新显示并置顶；失败时保持隐藏。""" + ctypes.set_last_error(0) + if not _user32.SetWindowDisplayAffinity(hwnd, WDA_EXCLUDEFROMCAPTURE): + log.error( + '重新显示时 SetWindowDisplayAffinity 失败 error=%s 隐私屏保持隐藏', + ctypes.get_last_error(), + ) + return False + _user32.ShowWindow(hwnd, SW_SHOWNOACTIVATE) _user32.SetWindowPos( hwnd, HWND_TOPMOST, 0, 0, 0, 0, SWP_NOMOVE | SWP_NOSIZE | SWP_NOACTIVATE | SWP_SHOWWINDOW, ) - ctypes.set_last_error(0) - if not _user32.SetWindowDisplayAffinity(hwnd, WDA_EXCLUDEFROMCAPTURE): - log.warning( - '重新显示时 SetWindowDisplayAffinity 失败 error=%s 隐私屏仍会显示但截屏可能拍到画面', - ctypes.get_last_error(), - ) log.info('隐私屏已显示') + return True def _enable_click_through(hwnd: int) -> bool: @@ -434,16 +436,18 @@ def _create_privacy_window(close_key: str) -> int | None: _user32.DestroyWindow(hwnd) return None - _user32.ShowWindow(hwnd, SW_SHOWNOACTIVATE) - _user32.UpdateWindow(hwnd) - - if _user32.SetWindowDisplayAffinity(hwnd, WDA_EXCLUDEFROMCAPTURE): - log.info('隐私屏已覆盖虚拟桌面 (%s,%s) %sx%s', vx, vy, vw, vh) - else: - log.warning( - 'SetWindowDisplayAffinity 失败 error=%s 隐私屏仍会显示但截屏可能拍到画面', + ctypes.set_last_error(0) + if not _user32.SetWindowDisplayAffinity(hwnd, WDA_EXCLUDEFROMCAPTURE): + log.error( + 'SetWindowDisplayAffinity 失败 error=%s 取消显示隐私屏', ctypes.get_last_error(), ) + _user32.DestroyWindow(hwnd) + return None + + _user32.ShowWindow(hwnd, SW_SHOWNOACTIVATE) + _user32.UpdateWindow(hwnd) + log.info('隐私屏已覆盖虚拟桌面 (%s,%s) %sx%s', vx, vy, vw, vh) # 修饰键固定 Ctrl+Alt+Shift，主键来自设置；解析失败时回退默认值。 # MOD_NOREPEAT：按住不连发，避免一次按键把隐私屏来回切换。 @@ -498,9 +502,9 @@ def start(self) -> bool: if self._thread is not None and self._thread.is_alive(): hwnd = self._hwnd if hwnd: - # 窗口线程正在等消息，同步发过去，返回时黑屏已经盖上。 - _user32.SendMessageW(hwnd, WM_PRIVACY_SHOW, 0, 0) - return True + # 同步获取窗口线程的显示结果，排除捕获失败时不能报告成功。 + return bool(_user32.SendMessageW(hwnd, WM_PRIVACY_SHOW, 0, 0)) + return False self._hwnd = None self._ready.clear() diff --git a/tests/test_privacy_screen.py b/tests/test_privacy_screen.py new file mode 100644 index 0000000..fd3036f --- /dev/null +++ b/tests/test_privacy_screen.py @@ -0,0 +1,232 @@ +import ctypes +import importlib.util +import sys +import unittest +from pathlib import Path +from types import ModuleType +from unittest.mock import Mock, call, patch + + +def _load_privacy_screen(): + path = ( + Path(__file__).resolve().parents[1] + / "src/script_chainer/utils/privacy_screen.py" + ) + spec = importlib.util.spec_from_file_location("privacy_screen_under_test", path) + module = importlib.util.module_from_spec(spec) + runner_logging = ModuleType("script_chainer.win_exe.runner_logging") + runner_logging.log = Mock() + # Load the real module without opening Win32 DLLs or importing runner dependencies. + with ( + patch.dict(sys.modules, {runner_logging.__name__: runner_logging}), + patch.object( + ctypes, "WinDLL", side_effect=lambda *a, **kw: Mock(), create=True + ), + patch.object(ctypes, "WINFUNCTYPE", ctypes.CFUNCTYPE, create=True), + ): + spec.loader.exec_module(module) + return module + + +privacy_screen = _load_privacy_screen() + + +class PrivacyScreenTests(unittest.TestCase): + def setUp(self): + self.hwnd = 1234 + self.user32 = Mock() + self.kernel32 = Mock() + self.log = Mock() + self.user32.GetSystemMetrics.side_effect = { + privacy_screen.SM_XVIRTUALSCREEN: 0, + privacy_screen.SM_YVIRTUALSCREEN: 0, + privacy_screen.SM_CXVIRTUALSCREEN: 1920, + privacy_screen.SM_CYVIRTUALSCREEN: 1080, + }.__getitem__ + self.user32.CreateWindowExW.return_value = self.hwnd + self.user32.SetWindowDisplayAffinity.return_value = 1 + self.user32.RegisterHotKey.return_value = 1 + self.user32.SetThreadDpiAwarenessContext.return_value = 1 + self.user32.GetMessageW.return_value = 0 + self.enterContext( + patch.multiple( + privacy_screen, + _user32=self.user32, + _kernel32=self.kernel32, + _register_window_class=Mock(return_value=True), + _enable_click_through=Mock(return_value=True), + log=self.log, + ) + ) + self.enterContext(patch.object(ctypes, "set_last_error", create=True)) + self.enterContext( + patch.object(ctypes, "get_last_error", return_value=5, create=True) + ) + + def test_create_sets_affinity_before_showing_window(self): + self.assertEqual(privacy_screen._create_privacy_window("f12"), self.hwnd) + + calls = self.user32.mock_calls + affinity = calls.index( + call.SetWindowDisplayAffinity( + self.hwnd, + privacy_screen.WDA_EXCLUDEFROMCAPTURE, + ) + ) + show = calls.index(call.ShowWindow(self.hwnd, privacy_screen.SW_SHOWNOACTIVATE)) + update = calls.index(call.UpdateWindow(self.hwnd)) + self.assertLess(affinity, show) + self.assertLess(show, update) + self.user32.DestroyWindow.assert_not_called() + style = self.user32.CreateWindowExW.call_args.args[3] + self.assertFalse(style & privacy_screen.WS_VISIBLE) + + def test_create_destroys_window_without_showing_it_when_affinity_fails(self): + self.user32.SetWindowDisplayAffinity.return_value = 0 + + self.assertIsNone(privacy_screen._create_privacy_window("f12")) + + self.user32.DestroyWindow.assert_called_once_with(self.hwnd) + self.user32.ShowWindow.assert_not_called() + self.user32.UpdateWindow.assert_not_called() + self.user32.RegisterHotKey.assert_not_called() + self.user32.SetTimer.assert_not_called() + self.log.error.assert_called_once() + self.assertEqual(self.log.error.call_args.args[1], 5) + + def test_create_still_cleans_up_when_click_through_fails(self): + privacy_screen._enable_click_through.return_value = False + + self.assertIsNone(privacy_screen._create_privacy_window("f12")) + + self.user32.DestroyWindow.assert_called_once_with(self.hwnd) + self.user32.SetWindowDisplayAffinity.assert_not_called() + self.user32.ShowWindow.assert_not_called() + + def test_reveal_sets_affinity_before_showing_window(self): + self.assertTrue(privacy_screen._reveal_privacy_window(self.hwnd)) + + self.assertEqual( + self.user32.mock_calls, + [ + call.SetWindowDisplayAffinity( + self.hwnd, privacy_screen.WDA_EXCLUDEFROMCAPTURE + ), + call.ShowWindow(self.hwnd, privacy_screen.SW_SHOWNOACTIVATE), + call.SetWindowPos( + self.hwnd, + privacy_screen.HWND_TOPMOST, + 0, + 0, + 0, + 0, + privacy_screen.SWP_NOMOVE + | privacy_screen.SWP_NOSIZE + | privacy_screen.SWP_NOACTIVATE + | privacy_screen.SWP_SHOWWINDOW, + ), + ], + ) + + def test_reveal_keeps_window_hidden_when_affinity_fails(self): + self.user32.SetWindowDisplayAffinity.return_value = 0 + + self.assertFalse(privacy_screen._reveal_privacy_window(self.hwnd)) + + self.user32.ShowWindow.assert_not_called() + self.user32.SetWindowPos.assert_not_called() + self.user32.DestroyWindow.assert_not_called() + self.log.error.assert_called_once() + self.assertEqual(self.log.error.call_args.args[1], 5) + self.log.info.assert_not_called() + + def test_show_message_reports_reveal_result(self): + self.user32.IsWindowVisible.return_value = 0 + for success in (0, 1): + with self.subTest(success=success): + self.user32.SetWindowDisplayAffinity.return_value = success + result = privacy_screen._window_proc( + self.hwnd, + privacy_screen.WM_PRIVACY_SHOW, + 0, + 0, + ) + self.assertEqual(result, success) + + def test_show_message_reports_success_for_already_visible_window(self): + self.user32.IsWindowVisible.return_value = 1 + + result = privacy_screen._window_proc( + self.hwnd, + privacy_screen.WM_PRIVACY_SHOW, + 0, + 0, + ) + + self.assertEqual(result, 1) + self.user32.SetWindowDisplayAffinity.assert_not_called() + self.user32.ShowWindow.assert_not_called() + + def test_hotkey_does_not_show_window_when_affinity_fails(self): + self.user32.IsWindowVisible.return_value = 0 + self.user32.SetWindowDisplayAffinity.return_value = 0 + + privacy_screen._window_proc( + self.hwnd, + privacy_screen.WM_HOTKEY, + privacy_screen.EMERGENCY_HOTKEY_ID, + 0, + ) + + self.user32.ShowWindow.assert_not_called() + self.user32.SetWindowPos.assert_not_called() + + def test_start_reports_failure_when_initial_affinity_fails(self): + self.user32.SetWindowDisplayAffinity.return_value = 0 + screen = privacy_screen.PrivacyScreen() + + self.assertFalse(screen.start()) + screen._thread.join(timeout=1) + + self.assertFalse(screen._thread.is_alive()) + self.assertIsNone(screen._hwnd) + self.user32.DestroyWindow.assert_called_once_with(self.hwnd) + self.user32.ShowWindow.assert_not_called() + + def test_start_reports_success_when_initial_affinity_succeeds(self): + screen = privacy_screen.PrivacyScreen() + + self.assertTrue(screen.start()) + screen._thread.join(timeout=1) + + self.assertEqual(screen._hwnd, self.hwnd) + self.user32.ShowWindow.assert_called_once_with( + self.hwnd, + privacy_screen.SW_SHOWNOACTIVATE, + ) + + def test_start_existing_window_reports_reveal_result(self): + self.user32.IsWindowVisible.return_value = 0 + self.user32.SendMessageW.side_effect = privacy_screen._window_proc + for success in (0, 1): + with self.subTest(success=success): + self.user32.SetWindowDisplayAffinity.return_value = success + screen = privacy_screen.PrivacyScreen() + screen._thread = Mock() + screen._thread.is_alive.return_value = True + screen._hwnd = self.hwnd + + self.assertEqual(screen.start(), bool(success)) + + def test_start_does_not_report_success_without_a_window(self): + screen = privacy_screen.PrivacyScreen() + screen._thread = Mock() + screen._thread.is_alive.return_value = True + + self.assertFalse(screen.start()) + + self.user32.SendMessageW.assert_not_called() + + +if __name__ == "__main__": + unittest.main() From 1922e55fbdcdfbe103f5d21eca0051ad38bbe9fe Mon Sep 17 00:00:00 2001 From: antecanis8 <42382878+antecanis8@users.noreply.github.com> Date: Fri, 16 Oct 2026 13:03:30 +0800 Subject: [PATCH 4/4] =?UTF-8?q?fix(privacy-screen):=20=E4=BF=AE=E5=A4=8D?= =?UTF-8?q?=E5=90=AF=E5=8A=A8=E8=B6=85=E6=97=B6=E4=B8=8E=E5=81=9C=E6=AD=A2?= =?UTF-8?q?=E7=AB=9E=E6=80=81=E5=AF=BC=E8=87=B4=E7=9A=84=E9=BB=91=E5=B1=8F?= =?UTF-8?q?=E6=AE=8B=E7=95=99?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/script_chainer/utils/privacy_screen.py | 56 ++++- tests/test_privacy_screen.py | 245 +++++++++++++++++++++ 2 files changed, 290 insertions(+), 11 deletions(-) diff --git a/src/script_chainer/utils/privacy_screen.py b/src/script_chainer/utils/privacy_screen.py index 01dfe71..fb4a59b 100644 --- a/src/script_chainer/utils/privacy_screen.py +++ b/src/script_chainer/utils/privacy_screen.py @@ -492,6 +492,7 @@ def __init__(self, close_key: str = DEFAULT_CLOSE_KEY) -> None: self._thread: threading.Thread | None = None self._hwnd: int | None = None self._ready = threading.Event() + self._stop_requested = threading.Event() def start(self) -> bool: """在后台线程拉起隐私屏，返回是否成功盖住屏幕。 @@ -499,7 +500,11 @@ def start(self) -> bool: 已经在运行但被快捷键隐藏时，会重新显示。 """ with self._lock: - if self._thread is not None and self._thread.is_alive(): + if ( + self._thread is not None + and self._thread.is_alive() + and not self._stop_requested.is_set() + ): hwnd = self._hwnd if hwnd: # 同步获取窗口线程的显示结果，排除捕获失败时不能报告成功。 @@ -507,23 +512,40 @@ def start(self) -> bool: return False self._hwnd = None - self._ready.clear() + # 每次启动独立持有事件，迟到的旧线程不能唤醒或覆盖新一轮启动。 + ready = self._ready = threading.Event() + stop_requested = self._stop_requested = threading.Event() self._thread = threading.Thread( - target=self._run, name='privacy_screen', daemon=True + target=self._run, + args=(ready, stop_requested), + name='privacy_screen', + daemon=True, ) self._thread.start() - self._ready.wait(START_TIMEOUT_SECONDS) + + # 等待时不持锁，让 stop() 能取消尚未完成的建窗。 + is_ready = ready.wait(START_TIMEOUT_SECONDS) + with self._lock: + if self._stop_requested is not stop_requested or stop_requested.is_set(): + return False hwnd = self._hwnd + if is_ready and hwnd is not None: + return True - if hwnd is None: - log.error('隐私屏启动失败') - return False + stop_requested.set() + self._hwnd = None - return True + # 超时与句柄发布可能同时发生，此时窗口线程已错过取消检查。 + if hwnd: + _user32.PostMessageW(hwnd, WM_CLOSE, 0, 0) + log.error('隐私屏启动失败') + return False def stop(self) -> None: """关闭隐私屏，恢复屏幕显示。""" with self._lock: + self._stop_requested.set() + self._ready.set() thread = self._thread hwnd = self._hwnd self._thread = None @@ -541,7 +563,9 @@ def stop(self) -> None: else: log.info('隐私屏已关闭') - def _run(self) -> None: + def _run( + self, ready: threading.Event, stop_requested: threading.Event + ) -> None: """窗口线程：建窗、显示，然后跑消息循环直到窗口销毁。""" old_dpi = _set_thread_dpi_awareness() try: @@ -554,12 +578,22 @@ def _run(self) -> None: # 范围缩到最小（窗口本身的属性在建窗时已锁定）。 _restore_thread_dpi_awareness(old_dpi) - self._hwnd = hwnd - self._ready.set() + with self._lock: + cancelled = ( + self._stop_requested is not stop_requested or stop_requested.is_set() + ) + if not cancelled: + self._hwnd = hwnd + ready.set() if hwnd is None: return + if cancelled: + # 建窗期间可能已停止甚至重启；在窗口所属线程直接回收迟到的窗口。 + _user32.DestroyWindow(hwnd) + return + msg = wintypes.MSG() while _user32.GetMessageW(ctypes.byref(msg), None, 0, 0) > 0: _user32.TranslateMessage(ctypes.byref(msg)) diff --git a/tests/test_privacy_screen.py b/tests/test_privacy_screen.py index fd3036f..36f3ca0 100644 --- a/tests/test_privacy_screen.py +++ b/tests/test_privacy_screen.py @@ -1,6 +1,7 @@ import ctypes import importlib.util import sys +import threading import unittest from pathlib import Path from types import ModuleType @@ -227,6 +228,250 @@ def test_start_does_not_report_success_without_a_window(self): self.user32.SendMessageW.assert_not_called() + def test_start_timeout_destroys_late_window_without_stop(self): + self._check_timed_out_window_is_destroyed(stop_before_creation=False) + + def test_stop_after_start_timeout_still_destroys_late_window(self): + self._check_timed_out_window_is_destroyed(stop_before_creation=True) + + def _check_timed_out_window_is_destroyed(self, stop_before_creation): + creating = threading.Event() + release_creation = threading.Event() + destroyed_by = [] + screen = privacy_screen.PrivacyScreen() + self.user32.DestroyWindow.side_effect = ( + lambda hwnd: destroyed_by.append(threading.current_thread()) + ) + + def create_window(close_key): + creating.set() + if not release_creation.wait(2): + raise RuntimeError("Test did not release window creation") + return self.hwnd + + self.enterContext( + patch.object(privacy_screen, "_create_privacy_window", create_window) + ) + self.enterContext(patch.object(privacy_screen, "START_TIMEOUT_SECONDS", 0)) + self.enterContext(patch.object(privacy_screen, "STOP_TIMEOUT_SECONDS", 0)) + worker = None + try: + self.assertFalse(screen.start()) + worker = screen._thread + self.assertTrue(creating.wait(1)) + self.assertTrue(screen._stop_requested.is_set()) + if stop_before_creation: + screen.stop() + release_creation.set() + worker.join(timeout=1) + + self.assertFalse(worker.is_alive()) + self.assertIsNone(screen._hwnd) + self.user32.DestroyWindow.assert_called_once_with(self.hwnd) + self.assertEqual(destroyed_by, [worker]) + self.user32.GetMessageW.assert_not_called() + finally: + release_creation.set() + if worker is not None: + worker.join(timeout=2) + + def test_start_timeout_closes_window_published_at_timeout_boundary(self): + screen = privacy_screen.PrivacyScreen() + ready = threading.Event() + closed = threading.Event() + wait_until_ready = ready.wait + threading_proxy = Mock(wraps=threading) + threading_proxy.Event.side_effect = [ready, threading.Event()] + + def timeout_after_publication(timeout): + self.assertTrue(wait_until_ready(1)) + return False + + def get_message(*args): + closed.wait(2) + return 0 + + self.user32.GetMessageW.side_effect = get_message + self.user32.PostMessageW.side_effect = lambda *args: closed.set() + with ( + patch.object(privacy_screen, "threading", threading_proxy), + patch.object(ready, "wait", side_effect=timeout_after_publication), + ): + try: + self.assertFalse(screen.start()) + screen._thread.join(timeout=1) + + self.assertFalse(screen._thread.is_alive()) + self.assertIsNone(screen._hwnd) + self.assertTrue(screen._stop_requested.is_set()) + self.user32.PostMessageW.assert_called_once_with( + self.hwnd, privacy_screen.WM_CLOSE, 0, 0 + ) + finally: + closed.set() + screen._thread.join(timeout=2) + + def test_stop_during_creation_cancels_start_and_destroys_late_window(self): + creating = threading.Event() + release_creation = threading.Event() + screen = privacy_screen.PrivacyScreen() + results = [] + + def create_window(close_key): + creating.set() + if not release_creation.wait(2): + raise RuntimeError("Test did not release window creation") + return self.hwnd + + self.enterContext( + patch.object(privacy_screen, "_create_privacy_window", create_window) + ) + self.enterContext(patch.object(privacy_screen, "STOP_TIMEOUT_SECONDS", 0)) + starter = threading.Thread(target=lambda: results.append(screen.start())) + worker = None + starter.start() + try: + self.assertTrue(creating.wait(1)) + worker = screen._thread + screen.stop() + starter.join(timeout=1) + + self.assertFalse(starter.is_alive()) + self.assertEqual(results, [False]) + self.assertTrue(worker.is_alive()) + self.assertIsNone(screen._thread) + self.user32.PostMessageW.assert_not_called() + + release_creation.set() + worker.join(timeout=1) + screen.stop() + + self.assertFalse(worker.is_alive()) + self.assertIsNone(screen._hwnd) + self.user32.DestroyWindow.assert_called_once_with(self.hwnd) + self.user32.GetMessageW.assert_not_called() + finally: + release_creation.set() + starter.join(timeout=2) + if worker is not None: + worker.join(timeout=2) + + def test_stop_posts_close_for_published_window(self): + closed = threading.Event() + screen = privacy_screen.PrivacyScreen() + + def get_message(*args): + closed.wait(2) + return 0 + + self.user32.GetMessageW.side_effect = get_message + self.user32.PostMessageW.side_effect = lambda *args: closed.set() + try: + self.assertTrue(screen.start()) + worker = screen._thread + screen.stop() + screen.stop() + + self.assertFalse(worker.is_alive()) + self.user32.PostMessageW.assert_called_once_with( + self.hwnd, privacy_screen.WM_CLOSE, 0, 0 + ) + self.assertIsNone(screen._thread) + self.assertIsNone(screen._hwnd) + finally: + closed.set() + if screen._thread is not None: + screen._thread.join(timeout=2) + + def test_old_creation_cannot_publish_handle_or_signal_restarted_screen(self): + for old_hwnd in (self.hwnd, None): + with self.subTest(old_hwnd=old_hwnd): + self._check_restart_after_old_creation(old_hwnd) + + def _check_restart_after_old_creation(self, old_hwnd): + self.user32.reset_mock() + creating_old = threading.Event() + creating_new = threading.Event() + release_old = threading.Event() + release_new = threading.Event() + closed = threading.Event() + screen = privacy_screen.PrivacyScreen() + new_hwnd = 5678 + results = [] + + def create_window(close_key): + if not creating_old.is_set(): + creating_old.set() + if not release_old.wait(2): + raise RuntimeError("Test did not release old creation") + return old_hwnd + creating_new.set() + if not release_new.wait(2): + raise RuntimeError("Test did not release new creation") + return new_hwnd + + def get_message(*args): + closed.wait(2) + return 0 + + self.user32.GetMessageW.side_effect = get_message + self.user32.PostMessageW.side_effect = lambda *args: closed.set() + with patch.object( + privacy_screen, "_create_privacy_window", create_window + ): + with patch.object(privacy_screen, "START_TIMEOUT_SECONDS", 0): + self.assertFalse(screen.start()) + old_worker = screen._thread + old_ready = screen._ready + old_stop = screen._stop_requested + starter = threading.Thread( + target=lambda: results.append(screen.start()) + ) + try: + self.assertTrue(creating_old.wait(1)) + starter.start() + self.assertTrue(creating_new.wait(1)) + new_worker = screen._thread + self.assertIsNot(screen._ready, old_ready) + self.assertIsNot(screen._stop_requested, old_stop) + self.assertFalse(screen._stop_requested.is_set()) + + release_old.set() + old_worker.join(timeout=1) + + self.assertFalse(old_worker.is_alive()) + self.assertTrue(old_ready.is_set()) + self.assertFalse(screen._ready.is_set()) + self.assertIsNone(screen._hwnd) + self.assertEqual(results, []) + if old_hwnd is not None: + self.user32.DestroyWindow.assert_called_once_with(old_hwnd) + else: + self.user32.DestroyWindow.assert_not_called() + + release_new.set() + starter.join(timeout=1) + + self.assertFalse(starter.is_alive()) + self.assertEqual(results, [True]) + self.assertEqual(screen._hwnd, new_hwnd) + self.assertIs(screen._thread, new_worker) + screen.stop() + + self.assertFalse(new_worker.is_alive()) + self.user32.PostMessageW.assert_called_once_with( + new_hwnd, privacy_screen.WM_CLOSE, 0, 0 + ) + finally: + release_old.set() + release_new.set() + closed.set() + old_worker.join(timeout=2) + if starter.ident is not None: + starter.join(timeout=2) + if screen._thread is not None: + screen._thread.join(timeout=2) + if __name__ == "__main__": unittest.main()