diff --git a/CHANGELOG.md b/CHANGELOG.md index 26bb076..24c1956 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,28 @@ --- +## v5.3.0 (2026-05-09) + +### ✨ 新增功能 + +- **跨平台进程保护模块** (`xcli_core/process_protection.py`) + - **单实例保护** — 防止重复启动，Windows 使用命名 Mutex，macOS/Linux 使用 fcntl 文件锁（自动清理残留锁文件） + - **进程优先级提升** — Windows: `SetPriorityClass` (ABOVE_NORMAL/HIGH/REALTIME)，macOS/Linux: `os.nice()` + `renice` + - **Windows PPL 风格保护** — `SetErrorMode` 抑制系统错误弹窗 (SEM_FAILCRITICALERRORS | SEM_NOGPFAULTERRORBOX)、`faulthandler` 崩溃堆栈捕获 + - **POSIX 信号保护** — macOS/Linux 忽略 SIGHUP（终端关闭不退出）、注册 SIGTERM 优雅关闭、macOS 忽略 SIGPIPE + - **看门狗监控** — 后台线程监控父进程状态，父进程退出时自动清理并退出 + - **关闭钩子系统** — 注册自定义清理函数，进程退出时自动执行 + - **一键启用** — `enable_process_protection()` 单函数调用启用全部保护 + - **`/protect` 命令** — CLI 中查看保护状态，`/protect test` 重新测试所有保护 + +### 改进 + +- `platform_utils.py` 新增 `is_process_running()`、`get_process_priority()`、`set_process_priority()` 跨平台工具函数 +- `cli_base.py` 启动时自动初始化进程保护，退出时自动清理 +- `cli_core.py` 启动 banner 显示保护状态图标（单实例/优先级/PPL/看门狗） + +--- + ## v5.2.0 (2026-05-08) ### ✨ 新增功能 diff --git a/README.md b/README.md index 8496daa..8b7f845 100644 --- a/README.md +++ b/README.md @@ -9,7 +9,7 @@

- version + version python license tests @@ -35,6 +35,7 @@ - **定时任务** — 支持相对时间、绝对时间、重复任务，到期自动激活 AI - **系统通知** — 跨平台任务完成通知（Windows / macOS / Linux） - **MCP 协议** — 19 个插件全部实现 MCP 定义，支持 Function Calling 原生调用 +- **进程保护** — 跨平台单实例保护 + 进程优先级提升 + Windows PPL 风格保护 + 看门狗监控 + 优雅关闭 - **19 个插件 161+ 个操作** — 代码编辑、Git、浏览器、工程化、文件管理、图像生成、记忆等 - **三模式界面** — TUI 图形化 + 传统命令行 + Web UI 浏览器界面 - **150+ 个自动化测试** — pytest 覆盖全部插件和核心模块 diff --git a/platform_utils.py b/platform_utils.py index a019876..1bf792a 100644 --- a/platform_utils.py +++ b/platform_utils.py @@ -121,4 +121,79 @@ def get_config_directory(): elif platform == 'macos': return os.path.join(os.path.expanduser("~"), 'Library', 'Application Support', 'Xiaoli Pro-CLI 3') else: # Linux - return os.path.join(os.path.expanduser('~'), '.config', 'xiaoli-pro-cli-3') \ No newline at end of file + return os.path.join(os.path.expanduser('~'), '.config', 'xiaoli-pro-cli-3') + +def is_process_running(pid): + """检查指定 PID 的进程是否在运行""" + platform = get_platform() + + try: + if platform == 'windows': + import ctypes + kernel32 = ctypes.windll.kernel32 + handle = kernel32.OpenProcess(0x1000, False, pid) # PROCESS_QUERY_LIMITED_INFORMATION + if handle: + kernel32.CloseHandle(handle) + return True + return False + else: + os.kill(pid, 0) + return True + except (OSError, ProcessLookupError): + return False + + +def get_process_priority(): + """获取当前进程优先级""" + platform = get_platform() + + try: + if platform == 'windows': + import ctypes + kernel32 = ctypes.windll.kernel32 + handle = kernel32.GetCurrentProcess() + return kernel32.GetPriorityClass(handle) + else: + import resource + return resource.getpriority(resource.PRIO_PROCESS, 0) + except Exception: + return None + + +def set_process_priority(level="above_normal"): + """ + 设置进程优先级 + + Args: + level: "below_normal" | "normal" | "above_normal" | "high" + + Returns: + 是否设置成功 + """ + platform = get_platform() + + try: + if platform == 'windows': + import ctypes + kernel32 = ctypes.windll.kernel32 + priority_map = { + "below_normal": 0x00004000, + "normal": 0x00000020, + "above_normal": 0x00008000, + "high": 0x00000080, + } + priority = priority_map.get(level, 0x00008000) + handle = kernel32.GetCurrentProcess() + return bool(kernel32.SetPriorityClass(handle, priority)) + else: + nice_map = { + "below_normal": 10, + "normal": 0, + "above_normal": -10, + "high": -15, + } + nice_val = nice_map.get(level, -10) + os.nice(nice_val) + return True + except Exception: + return False diff --git a/xcli_core/cli_base.py b/xcli_core/cli_base.py index 31f551f..6ad3af1 100644 --- a/xcli_core/cli_base.py +++ b/xcli_core/cli_base.py @@ -11,6 +11,7 @@ from .constants import ( ) from .config import get_system_config, set_system_config, logger from .plugin_manager import LiuginManager +from .process_protection import get_protection, enable_process_protection # 可选导入 try: @@ -31,6 +32,14 @@ class BaseAICLI: # 初始化colorama init(autoreset=True) + # 初始化进程保护（单实例 + 优先级 + PPL保护 + 看门狗） + self.protection_results = enable_process_protection( + priority_level="above_normal", + watchdog=True, + restart_on_crash=False, + ) + self._protection = get_protection() + # 确保插件管理器使用正确的路径 project_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) liugins_dir = os.path.join(project_dir, "plugins") @@ -114,6 +123,12 @@ class BaseAICLI: def cleanup_resources(self): """清理资源，确保在程序退出时正确关闭所有资源""" print(f"{Fore.YELLOW}正在清理资源...{Style.RESET_ALL}") + # 清理进程保护 + if hasattr(self, '_protection') and self._protection: + try: + self._protection.disable_all() + except Exception: + pass self._stop_clawli_monitor() if CLAWLI_SERVER_AVAILABLE and clawli_server: clawli_server.stop_server() diff --git a/xcli_core/cli_core.py b/xcli_core/cli_core.py index 6a505fc..84d5cd6 100644 --- a/xcli_core/cli_core.py +++ b/xcli_core/cli_core.py @@ -20,6 +20,7 @@ from .cli_display import DisplayMixin from .cli_history import HistoryMixin from .plugin_market import PluginMarketMixin from .notification import notify_task_complete, get_notification_manager +from .process_protection import get_protection_status class AICLI(BaseAICLI, ClawliMixin, ToolMixin, CodeExecMixin, DisplayMixin, HistoryMixin, PluginMarketMixin): @@ -1020,6 +1021,7 @@ multi 操作支持一次修改多处： print(f"{Fore.GREEN}输入 '/chat open <名称>' 加载聊天记录{Style.RESET_ALL}") print(f"{Fore.GREEN}输入 '/remote' 查看远程连接帮助{Style.RESET_ALL}") print(f"{Fore.GREEN}输入 '/plugin' 管理插件市场 (安装/卸载/搜索){Style.RESET_ALL}") + print(f"{Fore.GREEN}输入 '/protect' 查看进程保护状态{Style.RESET_ALL}") current_engine_name = getattr(self.current_engine, 'name', '未设置') if self.current_engine else '未设置' print(f"{Fore.GREEN}当前使用的AI引擎: {current_engine_name}{Style.RESET_ALL}") loaded_engines = list(self.engines.keys()) @@ -1027,6 +1029,15 @@ multi 操作支持一次修改多处： print(f"{Fore.GREEN}已加载的AI引擎: {', '.join(loaded_engines)}{Style.RESET_ALL}") print(f"{Fore.GREEN}{'-' * 50}{Style.RESET_ALL}") + # 显示进程保护状态 + if hasattr(self, 'protection_results') and self.protection_results: + r = self.protection_results + si_icon = '' if r.get('single_instance') else '' + prio_icon = '' if r.get('priority') else '' + ppl_icon = '' if r.get('ppl_protection') else '' + wd_icon = '' if r.get('watchdog') else '' + print(f"{Fore.CYAN} 进程保护: {si_icon}单实例 {prio_icon}优先级 {ppl_icon}PPL保护 {wd_icon}看门狗{Style.RESET_ALL}") + # 询问 diff 弹窗模式 self._ask_diff_popup_mode() @@ -1171,6 +1182,29 @@ multi 操作支持一次修改多处： print(result) continue + if user_input.startswith('/protect'): + parts = user_input.split() + if len(parts) > 1 and parts[1] == 'test': + # 重新测试所有保护 + from .process_protection import enable_process_protection + results = enable_process_protection() + print(f"\n{Fore.CYAN} 进程保护测试结果:{Style.RESET_ALL}") + for key, val in results.items(): + icon = '' if val else '' + print(f" {icon} {key}: {val}") + else: + status = get_protection_status() + print(f"\n{Fore.CYAN} 进程保护状态:{Style.RESET_ALL}") + print(f" 平台: {status.get('platform', '未知')}") + icon = '' if status.get('is_protected') else '' + print(f" {icon} 保护已启用: {status.get('is_protected', False)}") + icon = '' if status.get('single_instance_held') else '' + print(f" {icon} 单实例锁: {status.get('single_instance_held', False)}") + icon = '' if status.get('watchdog_running') else '' + print(f" {icon} 看门狗: {status.get('watchdog_running', False)}") + print(f" 关闭钩子: {status.get('shutdown_hooks_count', 0)} 个") + continue + if user_input.startswith('/remote'): remote_command = user_input[7:].strip() self._handle_remote_command(remote_command) diff --git a/xcli_core/process_protection.py b/xcli_core/process_protection.py new file mode 100644 index 0000000..fde7b3f --- /dev/null +++ b/xcli_core/process_protection.py @@ -0,0 +1,659 @@ +""" +跨平台进程保护模块 +提供单实例保护、进程优先级提升、看门狗监控、优雅关闭等功能 + +支持 Windows / macOS / Linux +Windows 特有: PPL (Protected Process Light) 风格保护、错误模式抑制 +""" +import os +import sys +import time +import signal +import atexit +import threading +import subprocess +from typing import Optional, Callable + + +# ── 平台检测 ── +PLATFORM = ( + 'windows' if os.name == 'nt' + else 'macos' if sys.platform == 'darwin' + else 'linux' +) + + +class ProcessProtection: + """跨平台进程保护器""" + + def __init__(self, app_name: str = "xiaoli-cli"): + self.app_name = app_name + self._lock_file_path: Optional[str] = None + self._lock_fd = None + self._watchdog_thread: Optional[threading.Thread] = None + self._watchdog_running = threading.Event() + self._shutdown_hooks: list[Callable] = [] + self._is_protected = False + self._original_priority = None + self._restart_on_crash = False + self._restart_script: Optional[str] = None + + # ══════════════════════════════════════════════════ + # 单实例保护 + # ══════════════════════════════════════════════════ + + def acquire_single_instance(self) -> bool: + """ + 获取单实例锁，防止重复启动 + 返回 True 表示成功（当前是唯一实例），False 表示已有实例在运行 + """ + if PLATFORM == 'windows': + return self._acquire_windows_mutex() + else: + return self._acquire_posix_lock() + + def release_single_instance(self): + """释放单实例锁""" + if PLATFORM == 'windows': + self._release_windows_mutex() + else: + self._release_posix_lock() + + def _acquire_windows_mutex(self) -> bool: + """Windows: 使用命名 Mutex 实现单实例""" + try: + import ctypes + from ctypes import wintypes + + kernel32 = ctypes.windll.kernel32 + mutex_name = f"Global\\{self.app_name}_single_instance" + + # 创建命名 Mutex + handle = kernel32.CreateMutexW(None, True, mutex_name) + last_error = kernel32.GetLastError() + + ERROR_ALREADY_EXISTS = 183 + if last_error == ERROR_ALREADY_EXISTS: + kernel32.CloseHandle(handle) + return False + + self._win_mutex_handle = handle + atexit.register(self.release_single_instance) + return True + + except Exception: + # 降级到文件锁 + return self._acquire_posix_lock() + + def _release_windows_mutex(self): + """Windows: 释放 Mutex""" + try: + if hasattr(self, '_win_mutex_handle') and self._win_mutex_handle: + import ctypes + ctypes.windll.kernel32.CloseHandle(self._win_mutex_handle) + self._win_mutex_handle = None + except Exception: + pass + + def _acquire_posix_lock(self) -> bool: + """POSIX (macOS/Linux): 使用文件锁实现单实例""" + import tempfile + + lock_dir = tempfile.gettempdir() + self._lock_file_path = os.path.join(lock_dir, f"{self.app_name}.lock") + + try: + # 尝试创建并锁定文件 + if sys.platform == 'darwin': + # macOS: 使用 fcntl 文件锁 + import fcntl + self._lock_fd = open(self._lock_file_path, 'w') + try: + fcntl.flock(self._lock_fd.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except (IOError, OSError): + # 文件已被锁定 — 已有实例在运行 + # 检查锁文件中的 PID 是否仍在运行 + try: + old_pid = self._lock_fd.read().strip() + if old_pid and self._is_process_running(int(old_pid)): + self._lock_fd.close() + self._lock_fd = None + return False + else: + # 旧进程已死，清理残留锁 + self._lock_fd.seek(0) + self._lock_fd.truncate() + except (ValueError, OSError): + pass + + self._lock_fd.seek(0) + self._lock_fd.truncate() + self._lock_fd.write(str(os.getpid())) + self._lock_fd.flush() + atexit.register(self.release_single_instance) + return True + else: + # Linux: 使用 fcntl 文件锁 + import fcntl + self._lock_fd = open(self._lock_file_path, 'w') + try: + fcntl.flock(self._lock_fd.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except (IOError, OSError): + # 检查旧进程是否仍在运行 + try: + old_pid = self._lock_fd.read().strip() + if old_pid and self._is_process_running(int(old_pid)): + self._lock_fd.close() + self._lock_fd = None + return False + else: + self._lock_fd.seek(0) + self._lock_fd.truncate() + except (ValueError, OSError): + pass + + self._lock_fd.seek(0) + self._lock_fd.truncate() + self._lock_fd.write(str(os.getpid())) + self._lock_fd.flush() + atexit.register(self.release_single_instance) + return True + + except Exception: + # 锁获取失败，但不阻止程序运行 + return True + + def _release_posix_lock(self): + """POSIX: 释放文件锁""" + try: + if self._lock_fd: + if sys.platform == 'darwin': + import fcntl + fcntl.flock(self._lock_fd.fileno(), fcntl.LOCK_UN) + else: + import fcntl + fcntl.flock(self._lock_fd.fileno(), fcntl.LOCK_UN) + self._lock_fd.close() + self._lock_fd = None + if self._lock_file_path and os.path.exists(self._lock_file_path): + try: + os.unlink(self._lock_file_path) + except OSError: + pass + except Exception: + pass + + @staticmethod + def _is_process_running(pid: int) -> bool: + """检查指定 PID 的进程是否在运行""" + try: + if PLATFORM == 'windows': + import ctypes + kernel32 = ctypes.windll.kernel32 + PROCESS_QUERY_LIMITED_INFORMATION = 0x1000 + handle = kernel32.OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, False, pid) + if handle: + kernel32.CloseHandle(handle) + return True + return False + else: + os.kill(pid, 0) + return True + except (OSError, ProcessLookupError): + return False + + # ══════════════════════════════════════════════════ + # 进程优先级提升 + # ══════════════════════════════════════════════════ + + def elevate_priority(self, level: str = "high") -> bool: + """ + 提升进程优先级 + + Args: + level: "above_normal" | "high" | "realtime"(仅Windows) + + Returns: + 是否成功提升 + """ + try: + if PLATFORM == 'windows': + return self._elevate_windows(level) + elif PLATFORM == 'macos': + return self._elevate_macos(level) + else: + return self._elevate_linux(level) + except Exception: + return False + + def _elevate_windows(self, level: str) -> bool: + """Windows: 使用 SetPriorityClass 提升优先级""" + try: + import ctypes + kernel32 = ctypes.windll.kernel32 + + # 优先级映射 + priority_map = { + "above_normal": 0x00008000, # ABOVE_NORMAL_PRIORITY_CLASS + "high": 0x00000080, # HIGH_PRIORITY_CLASS + "realtime": 0x00000100, # REALTIME_PRIORITY_CLASS + } + + priority = priority_map.get(level, 0x00000080) + + # 获取当前进程句柄 + handle = kernel32.GetCurrentProcess() + + # 记录原始优先级 + self._original_priority = kernel32.GetPriorityClass(handle) + + # 设置新优先级 + result = kernel32.SetPriorityClass(handle, priority) + if result: + self._is_protected = True + return True + return False + + except Exception: + return False + + def _elevate_macos(self, level: str) -> bool: + """macOS: 使用 renice 提升优先级""" + try: + nice_map = { + "above_normal": -10, + "high": -15, + "realtime": -20, + } + nice_val = nice_map.get(level, -10) + os.nice(nice_val) + self._is_protected = True + return True + except Exception: + return False + + def _elevate_linux(self, level: str) -> bool: + """Linux: 使用 renice 提升优先级""" + try: + nice_map = { + "above_normal": -10, + "high": -15, + "realtime": -20, + } + nice_val = nice_map.get(level, -10) + + # os.nice() 是相对值，需要先获取当前 nice 值 + try: + import resource + current = resource.getpriority(resource.PRIO_PROCESS, 0) + target = nice_val + delta = target - current + if delta != 0: + os.nice(delta) + except (ImportError, OSError): + # fallback: 直接使用 subprocess + subprocess.run( + ["renice", str(nice_val), "-p", str(os.getpid())], + capture_output=True, timeout=5 + ) + + self._is_protected = True + return True + except Exception: + return False + + def restore_priority(self): + """恢复原始进程优先级""" + if not self._is_protected: + return + + try: + if PLATFORM == 'windows' and self._original_priority: + import ctypes + kernel32 = ctypes.windll.kernel32 + handle = kernel32.GetCurrentProcess() + kernel32.SetPriorityClass(handle, self._original_priority) + elif PLATFORM in ('macos', 'linux'): + # 恢复到默认 nice 值 (0) + try: + import resource + current = resource.getpriority(resource.PRIO_PROCESS, 0) + os.nice(-current) + except (ImportError, OSError): + subprocess.run( + ["renice", "0", "-p", str(os.getpid())], + capture_output=True, timeout=5 + ) + self._is_protected = False + except Exception: + pass + + # ══════════════════════════════════════════════════ + # Windows PPL 风格保护 + # ══════════════════════════════════════════════════ + + def enable_ppl_protection(self) -> bool: + """ + Windows: 启用 PPL (Protected Process Light) 风格保护 + - 抑制错误弹窗 + - 设置进程缓解策略 + - 禁用 DEP 异常弹窗 + + macOS/Linux: 启用等效保护 + - 设置信号处理 + - 忽略 SIGHUP + """ + if PLATFORM == 'windows': + return self._enable_windows_ppl() + else: + return self._enable_posix_protection() + + def _enable_windows_ppl(self) -> bool: + """Windows: PPL 风格保护""" + try: + import ctypes + kernel32 = ctypes.windll.kernel32 + + # 1. 抑制系统错误弹窗 (SEM_FAILCRITICALERRORS | SEM_NOGPFAULTERRORBOX) + SEM_FAILCRITICALERRORS = 0x0001 + SEM_NOGPFAULTERRORBOX = 0x0002 + SEM_NOOPENFILEERRORBOX = 0x8000 + old_mode = kernel32.SetErrorMode( + SEM_FAILCRITICALERRORS | SEM_NOGPFAULTERRORBOX | SEM_NOOPENFILEERRORBOX + ) + self._original_error_mode = old_mode + + # 2. 设置进程为后台优先级模式（减少对用户交互的影响） + # PROCESS_MODE_BACKGROUND_BEGIN = 0x00100000 + # kernel32.SetPriorityClass(kernel32.GetCurrentProcess(), 0x00100000) + + # 3. 注册异常处理（防止未处理异常导致崩溃弹窗） + try: + import faulthandler + faulthandler.enable() + except Exception: + pass + + self._is_protected = True + return True + + except Exception: + return False + + def _enable_posix_protection(self) -> bool: + """macOS/Linux: 等效保护""" + try: + # 1. 忽略 SIGHUP（终端关闭时不停止进程） + signal.signal(signal.SIGHUP, signal.SIG_IGN) + + # 2. 注册优雅关闭信号 + def _graceful_shutdown(signum, frame): + self._execute_shutdown_hooks() + sys.exit(0) + + signal.signal(signal.SIGTERM, _graceful_shutdown) + + # 3. macOS 特有：忽略 SIGPIPE + if PLATFORM == 'macos': + signal.signal(signal.SIGPIPE, signal.SIG_IGN) + + # 4. 启用 faulthandler（崩溃时输出堆栈） + try: + import faulthandler + faulthandler.enable() + except Exception: + pass + + self._is_protected = True + return True + + except Exception: + return False + + # ══════════════════════════════════════════════════ + # 看门狗监控 + # ══════════════════════════════════════════════════ + + def start_watchdog(self, check_interval: float = 5.0, + restart_on_crash: bool = False, + restart_script: Optional[str] = None): + """ + 启动看门狗线程，监控父进程状态 + 当父进程退出时，自动清理并退出 + + Args: + check_interval: 检查间隔（秒） + restart_on_crash: 崩溃时是否自动重启 + restart_script: 重启脚本路径 + """ + self._restart_on_crash = restart_on_crash + self._restart_script = restart_script + + if self._watchdog_thread and self._watchdog_thread.is_alive(): + return + + self._watchdog_running.set() + self._watchdog_thread = threading.Thread( + target=self._watchdog_loop, + args=(check_interval,), + daemon=True, + name="process-watchdog" + ) + self._watchdog_thread.start() + + def stop_watchdog(self): + """停止看门狗""" + self._watchdog_running.clear() + if self._watchdog_thread: + self._watchdog_thread.join(timeout=3) + self._watchdog_thread = None + + def _watchdog_loop(self, check_interval: float): + """看门狗主循环""" + parent_pid = os.getppid() + + while self._watchdog_running.is_set(): + try: + # 检查父进程是否仍在运行 + if not self._is_process_running(parent_pid): + # 父进程已退出，执行清理 + self._execute_shutdown_hooks() + if self._restart_on_crash and self._restart_script: + self._do_restart() + else: + os._exit(0) + break + + time.sleep(check_interval) + + except Exception: + time.sleep(check_interval) + + def _do_restart(self): + """执行重启""" + try: + if self._restart_script and os.path.exists(self._restart_script): + if PLATFORM == 'windows': + subprocess.Popen( + ["python", self._restart_script], + creationflags=subprocess.CREATE_NEW_PROCESS_GROUP + ) + else: + subprocess.Popen( + ["python3", self._restart_script], + start_new_session=True + ) + except Exception: + pass + finally: + os._exit(0) + + # ══════════════════════════════════════════════════ + # 关闭钩子 + # ══════════════════════════════════════════════════ + + def register_shutdown_hook(self, hook: Callable): + """注册关闭钩子，在进程退出时执行""" + if hook not in self._shutdown_hooks: + self._shutdown_hooks.append(hook) + + def _execute_shutdown_hooks(self): + """执行所有关闭钩子""" + for hook in reversed(self._shutdown_hooks): + try: + hook() + except Exception: + pass + + # ══════════════════════════════════════════════════ + # 一键启用全部保护 + # ══════════════════════════════════════════════════ + + def enable_all(self, priority_level: str = "above_normal", + watchdog: bool = True, + restart_on_crash: bool = False, + restart_script: Optional[str] = None) -> dict: + """ + 一键启用所有进程保护 + + Args: + priority_level: 优先级 ("above_normal" | "high") + watchdog: 是否启用看门狗 + restart_on_crash: 崩溃时是否自动重启 + restart_script: 重启脚本路径 + + Returns: + 各项保护的启用状态字典 + """ + results = { + "single_instance": False, + "priority": False, + "ppl_protection": False, + "watchdog": False, + "platform": PLATFORM, + } + + # 1. 单实例保护 + results["single_instance"] = self.acquire_single_instance() + + # 2. 进程优先级提升 + results["priority"] = self.elevate_priority(priority_level) + + # 3. PPL 风格保护 + results["ppl_protection"] = self.enable_ppl_protection() + + # 4. 看门狗 + if watchdog: + self.start_watchdog( + restart_on_crash=restart_on_crash, + restart_script=restart_script + ) + results["watchdog"] = True + + # 注册清理 + atexit.register(self.disable_all) + + return results + + def disable_all(self): + """禁用所有保护并清理""" + self.stop_watchdog() + self.restore_priority() + self.release_single_instance() + self._execute_shutdown_hooks() + self._is_protected = False + + def get_status(self) -> dict: + """获取当前保护状态""" + return { + "platform": PLATFORM, + "is_protected": self._is_protected, + "single_instance_held": ( + self._lock_fd is not None or + hasattr(self, '_win_mutex_handle') and self._win_mutex_handle is not None + ), + "watchdog_running": ( + self._watchdog_thread is not None and + self._watchdog_thread.is_alive() + ), + "shutdown_hooks_count": len(self._shutdown_hooks), + } + + +# ── 全局单例 ── +_protection: Optional[ProcessProtection] = None + + +def get_protection() -> ProcessProtection: + """获取全局进程保护器单例""" + global _protection + if _protection is None: + _protection = ProcessProtection() + return _protection + + +def enable_process_protection(priority_level: str = "above_normal", + watchdog: bool = True, + restart_on_crash: bool = False) -> dict: + """ + 快捷函数：启用进程保护 + + Returns: + 各项保护状态字典 + """ + return get_protection().enable_all( + priority_level=priority_level, + watchdog=watchdog, + restart_on_crash=restart_on_crash, + ) + + +def disable_process_protection(): + """快捷函数：禁用进程保护""" + get_protection().disable_all() + + +def get_protection_status() -> dict: + """快捷函数：获取保护状态""" + return get_protection().get_status() + + +# ── CLI 入口（可独立运行测试） ── +if __name__ == "__main__": + import json + + print(f"平台: {PLATFORM}") + print(f"PID: {os.getpid()}") + print() + + prot = ProcessProtection("xiaoli-cli-test") + + # 测试单实例 + is_first = prot.acquire_single_instance() + print(f"单实例锁: {'获取成功 (首个实例)' if is_first else '获取失败 (已有实例)'}") + + if is_first: + # 测试优先级提升 + prio_ok = prot.elevate_priority("above_normal") + print(f"优先级提升: {'成功' if prio_ok else '失败'}") + + # 测试 PPL 保护 + ppl_ok = prot.enable_ppl_protection() + print(f"PPL 保护: {'启用' if ppl_ok else '失败'}") + + # 测试看门狗 + prot.start_watchdog(check_interval=2.0) + print(f"看门狗: 已启动") + + # 显示状态 + status = prot.get_status() + print(f"\n状态: {json.dumps(status, indent=2, ensure_ascii=False)}") + + print("\n按 Ctrl+C 退出...") + try: + while True: + time.sleep(1) + except KeyboardInterrupt: + print("\n正在清理...") + prot.disable_all() + print("完成")