""" core/vnc_manager.py - 远程桌面（VNC）管理器 ============================================ 为 MyAgent 提供 GUI 图形桌面功能，支持通过 Web 浏览器查看 agent 的操作界面。 架构: 用户浏览器 ──WebSocket──> noVNC (websockify) ──VNC──> x11vnc ──X11──> Xvfb (虚拟显示) ↓ Firefox (有头模式) [v1.34.0] 启动策略：桌面优先，依赖后台安装 - 核心服务（Xvfb + x11vnc + websockify）立即启动 - 非核心组件（icewm, Firefox 浏览器）在后台线程安装 - 即使安装失败/超时，桌面也必须能进入 组件: 1. Xvfb — X Virtual Framebuffer，虚拟 X11 显示服务器 2. x11vnc — VNC 服务器，将 Xvfb 的显示共享出去 3. websockify — WebSocket 代理，将 VNC 协议转为 WebSocket 供浏览器使用 4. noVNC — 浏览器端 VNC 客户端（HTML5 + Canvas） 5. icewm — 窗口管理器（后台安装，精简模式） 6. Firefox — Mozilla Firefox 浏览器（proot 专用安装，icewm隐藏标题栏） 依赖（自动检测并安装）: - xvfb (Xvfb 虚拟显示) — 核心依赖 - x11vnc (VNC 服务器) — 核心依赖 - websockify (WebSocket 代理，pip 安装) — 核心依赖 - python3-numpy (x11vnc 依赖) - icewm (窗口管理器，精简模式) — 非核心，后台安装 - firefox (Mozilla Firefox 浏览器) — 非核心，后台安装 - wmctrl (窗口最大化工具) — 非核心，后台安装 使用方法: manager = VNCManager() await manager.start() # 启动 Xvfb + x11vnc + websockify（后台安装其他组件） await manager.stop() # 停止所有服务 status = manager.status # 获取状态信息 """ from __future__ import annotations import asyncio import os import signal import shutil import subprocess import time from typing import Any, Dict, Optional from core.logger import get_logger logger = get_logger("myagent.vnc") # ── 默认配置 ────────────────────────────────────────────── DEFAULT_DISPLAY = ":99" DEFAULT_XVFB_RESOLUTION = "720x1280x24" # [v1.47.38] 默认竖屏（移动端为主），点"横"切换横屏 DEFAULT_X11VNC_PORT = 5900 DEFAULT_NOVNC_PORT = 6080 # WebSocket 代理端口（供 noVNC 客户端连接） class VNCManager: """ 远程桌面管理器 — 管理 Xvfb + x11vnc + websockify/noVNC 生命周期。 支持的操作: - start(): 启动虚拟显示 + VNC 服务器 + WebSocket 代理 - stop(): 停止所有服务 - status: 获取当前运行状态 - get_novnc_url(): 获取 noVNC 访问地址 - ensure_dependencies(): 检测并安装所需依赖 """ def __init__( self, display: str = DEFAULT_DISPLAY, resolution: str = DEFAULT_XVFB_RESOLUTION, x11vnc_port: int = DEFAULT_X11VNC_PORT, novnc_port: int = DEFAULT_NOVNC_PORT, ): self.display = display self.resolution = resolution self.x11vnc_port = x11vnc_port self.novnc_port = novnc_port # 进程引用 self._xvfb_process: Optional[subprocess.Popen] = None self._x11vnc_process: Optional[subprocess.Popen] = None self._websockify_process: Optional[subprocess.Popen] = None self._wm_process: Optional[subprocess.Popen] = None # [v1.23.45] 窗口管理器进程 self._browser_process: Optional[subprocess.Popen] = None # [v1.33.4] 浏览器进程 # 状态标记 self._running = False self._xvfb_attached = False # 是否附加到已有的 Xvfb 实例 self._xvfb_attached_pid: Optional[int] = None # 附加的 Xvfb PID self._wm_name: Optional[str] = None # [v1.23.45] 当前窗口管理器名称 self._lock = asyncio.Lock() # [v1.47.1] 屏幕旋转状态 ("normal" / "left" / "right" / "inverted") self._current_rotation: str = "normal" # [v1.47.36] xrandr --fb 能力缓存：None=未测试, True=支持, False=不支持 # Termux proot 下 Xvfb 的 RANDR 可能不支持动态调整帧缓冲区大小 self._xrandr_fb_supported: Optional[bool] = None # noVNC web 文件路径（用于 aiohttp 静态文件服务） self._novnc_web_dir: Optional[str] = None @property def is_running(self) -> bool: """VNC 服务是否正在运行""" if not self._running: # 即使内部标记为未运行，也检查系统上是否有 VNC 服务 # (myagent 重启后 _running 会重置为 False) if self._detect_existing_services(): self._running = True logger.info("检测到已有的 VNC 服务正在运行") else: return False # 检查各进程是否存活 # Xvfb: 如果是附加的实例，通过 PID 检查 if self._xvfb_attached: if self._xvfb_attached_pid and not self._is_pid_alive(self._xvfb_attached_pid): logger.warning("附加的 Xvfb 进程已退出") self._running = False return False elif self._xvfb_process and self._xvfb_process.poll() is not None: logger.warning(f"Xvfb 进程已退出 (exit code: {self._xvfb_process.returncode})") self._running = False return False for proc, name in [ (self._x11vnc_process, "x11vnc"), (self._websockify_process, "websockify"), ]: if proc and proc.poll() is not None: logger.warning(f"{name} 进程已退出 (exit code: {proc.returncode})") self._running = False return False return True def _is_pid_alive(self, pid: int) -> bool: """检查进程是否存活""" try: os.kill(pid, 0) # 发送信号 0，不实际杀死进程 return True except (OSError, ProcessLookupError): return False def _detect_existing_services(self) -> bool: """检测系统上是否已有 VNC 相关服务在运行（用于 myagent 重启后恢复状态）""" display_num = self.display.replace(":", "") socket_path = f"/tmp/.X11-unix/X{display_num}" # 检查 Xvfb socket 和进程 xvfb_ok = False if os.path.exists(socket_path): try: result = subprocess.run( ["pgrep", "-f", f"Xvfb {self.display}"], capture_output=True, text=True, timeout=5, ) if result.returncode == 0 and result.stdout.strip(): xvfb_ok = True except Exception: pass if not xvfb_ok: return False # 检查 x11vnc 是否在监听端口 x11vnc_ok = self._is_port_listening(self.x11vnc_port) if not x11vnc_ok: return False # 检查 websockify 是否在监听端口 websockify_ok = self._is_port_listening(self.novnc_port) if xvfb_ok and x11vnc_ok and websockify_ok: # 恢复附加状态 self._xvfb_attached = True try: result = subprocess.run( ["pgrep", "-f", f"Xvfb {self.display}"], capture_output=True, text=True, timeout=5, ) if result.returncode == 0: self._xvfb_attached_pid = int(result.stdout.strip().split()[0]) except Exception: pass return True return False @staticmethod def _is_port_listening(port: int) -> bool: """检查端口是否在监听""" try: import socket s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.settimeout(0.5) s.connect(("127.0.0.1", port)) s.close() return True except (OSError, ConnectionRefusedError): return False @staticmethod def _fix_dpkg_interrupted(sudo: list = None) -> None: """[v1.23.50] 修复 dpkg 中断状态 — proot/Android 环境下非常常见。 当之前的 apt install 被中断（进程被杀、proot 崩溃等）， dpkg 会留下锁文件和未完成的状态，导致后续所有 apt 操作失败： E: dpkg was interrupted, you must manually run 'dpkg --configure -a' [v1.33.5] 大幅增强 proot 兼容性： - 使用 policy-rc.d 阻止 dpkg configure 时启动服务（这是卡住的主因） - 添加 --force-confold --force-confdef 避免交互式配置提示 - 缩短超时防止长时间阻塞 """ if sudo is None: sudo = [] # proot 下 dpkg --configure -a 容易卡住（postinst 脚本启动服务） # 需要禁止服务启动，缩短超时 no_service_env = {**os.environ, "DEBIAN_FRONTEND": "noninteractive", "DPKG_OPTIONS": "--force-confold --force-confdef"} # 0. [v1.33.5] 创建 policy-rc.d 阻止服务启动 — 这是 dpkg configure 卡住的根源 # dpkg --configure -a 会触发 postinst 脚本，postinst 常用 invoke-rc.d 启动服务 # 在 proot 环境中，服务启动经常挂死（systemd 不可用、/run 不可写等） policy_rc_created = False policy_rc_path = "/usr/sbin/policy-rc.d" try: if not os.path.exists(policy_rc_path): with open(policy_rc_path, "w") as f: f.write("#!/bin/sh\nexit 101\n") os.chmod(policy_rc_path, 0o755) policy_rc_created = True logger.debug("已创建 policy-rc.d 阻止服务启动") except Exception: pass try: # 1. 先清理锁文件（最安全，不会卡住） for lock_file in [ "/var/lib/dpkg/lock", "/var/lib/dpkg/lock-frontend", "/var/lib/apt/lists/lock", "/var/cache/apt/archives/lock", ]: try: if os.path.exists(lock_file): os.remove(lock_file) logger.debug(f"已清理锁文件: {lock_file}") except Exception: pass # 2. 检查是否确实存在 dpkg 中断问题 need_fix = False try: result = subprocess.run( sudo + ["dpkg", "--audit"], capture_output=True, text=True, timeout=5, env=no_service_env, start_new_session=True, # [v1.40.0] ) # 如果 dpkg --audit 没有输出问题，跳过修复 if result.returncode == 0 and not result.stdout.strip(): return need_fix = True except Exception: # audit 超时也可能意味着有问题，继续尝试修复 need_fix = True if not need_fix: return # 3. 清理 dpkg 备份目录中的中断状态 try: for f in ["/var/lib/dpkg/status-bak", "/var/lib/dpkg/status-tmp", "/var/lib/dpkg/lock"]: try: if os.path.exists(f): os.remove(f) except Exception: pass # 确保 status 文件存在且有效 status_file = "/var/lib/dpkg/status" if os.path.exists(status_file): with open(status_file, "r") as sf: content = sf.read() # 如果 status 文件为空或截断，用备份恢复 if not content.strip(): backup = status_file + "-old" if os.path.exists(backup): import shutil as shutil_mod shutil_mod.copy2(backup, status_file) logger.info("已从 dpkg status-old 恢复 status 文件") except Exception as e: logger.debug(f"dpkg status 文件检查异常: {e}") # 4. 尝试修复 dpkg 中断 # [v1.33.5] 用 --no-triggers + --force 标志避免交互和卡住 logger.info("检测到 dpkg 中断状态，尝试自动修复...") try: # 先尝试轻量修复：只配置未完成的包，跳过触发器 result = subprocess.run( sudo + ["dpkg", "--configure", "-a", "--no-triggers", "--force-confold", "--force-confdef"], capture_output=True, text=True, timeout=15, env=no_service_env, start_new_session=True, # [v1.40.0] ) if result.returncode == 0: logger.info("dpkg 中断状态已修复 (--no-triggers)") return else: logger.warning(f"dpkg --configure -a --no-triggers 部分失败: {result.stderr[:200]}") except subprocess.TimeoutExpired: logger.warning("dpkg --configure -a --no-triggers 超时 (15s)") except FileNotFoundError: logger.debug("dpkg 命令不可用，跳过修复") return except Exception as e: logger.warning(f"dpkg 修复异常: {e}") # 回退：完整修复，但设短超时 try: result2 = subprocess.run( sudo + ["dpkg", "--configure", "-a", "--force-confold", "--force-confdef", "--force-all"], capture_output=True, text=True, timeout=30, env=no_service_env, start_new_session=True, # [v1.40.0] ) if result2.returncode == 0: logger.info("dpkg 中断状态已修复") else: logger.warning(f"dpkg --configure -a 失败: {result2.stderr[:200]}") except subprocess.TimeoutExpired: logger.warning("dpkg --configure -a 超时 (30s)，强制继续") except Exception as e: logger.warning(f"dpkg 修复异常: {e}") finally: # [v1.33.5] 清理 policy-rc.d（仅当我们创建的才删） if policy_rc_created: try: os.remove(policy_rc_path) except Exception: pass @staticmethod def _fix_apt_sources(sudo: list = None) -> None: """[v1.47.39] 修复 apt 源 — 确保 universe 仓库启用。 xvfb/x11vnc 在 universe 仓库中，Termux proot 重装后 sources.list 可能只有 main 或为空，导致找不到包。 修复策略： 1. 检测 Ubuntu 代号（从 lsb_release 或 /etc/os-release） 2. 确保 sources.list 至少有 main + universe 3. 重新 apt-get update """ if sudo is None: sudo = [] # 1. 检测 Ubuntu 代号 codename = "" try: result = subprocess.run( ["lsb_release", "-cs"], capture_output=True, text=True, timeout=5, ) if result.returncode == 0: codename = result.stdout.strip() except Exception: pass if not codename: try: with open("/etc/os-release") as f: for line in f: if line.startswith("VERSION_CODENAME="): codename = line.strip().split("=", 1)[1].strip('"') break elif line.startswith("UBUNTU_CODENAME="): codename = line.strip().split("=", 1)[1].strip('"') break except Exception: pass if not codename: # 常见代号回退 codename = "jammy" # Ubuntu 22.04 LTS logger.info(f"检测到 Ubuntu 代号: {codename}，修复 apt 源...") # 2. 检查并修复 sources.list sources_file = "/etc/apt/sources.list" sources_dir = "/etc/apt" try: existing = "" if os.path.exists(sources_file): with open(sources_file) as f: existing = f.read() # 检查是否已有 universe has_universe = "universe" in existing has_main = "main" in existing and codename in existing if has_universe and has_main and existing.strip(): logger.debug("sources.list 已包含 main + universe，无需修复") return # 写入最小化的 sources.list # 使用 http://ports.ubuntu.com（ARM64）和 archive.ubuntu.com（x86_64） import platform arch = platform.machine() if arch in ("aarch64", "armv7l"): base_url = "http://ports.ubuntu.com/ubuntu-ports" else: base_url = "http://archive.ubuntu.com/ubuntu" new_sources = ( f"deb {base_url} {codename} main universe\n" f"deb {base_url} {codename}-updates main universe\n" f"deb {base_url} {codename}-security main universe\n" ) # 检查是否需要更新（避免无谓写入） if new_sources.strip() == existing.strip(): logger.debug("sources.list 内容一致，无需更新") return logger.info(f"写入 sources.list: {new_sources.strip()}") # 先备份 try: if os.path.exists(sources_file) and existing.strip(): with open(sources_file + ".bak", "w") as f: f.write(existing) except Exception: pass # 写入新源 — 先确保 /etc/apt 目录存在 write_ok = False try: os.makedirs(sources_dir, exist_ok=True) with open(sources_file, "w") as f: f.write(new_sources) write_ok = True logger.info("sources.list 已更新（添加 universe 仓库）") except (PermissionError, FileNotFoundError, OSError): pass # 直接写失败，用 tee 写入 if not write_ok: try: subprocess.run( sudo + ["sh", "-c", f"mkdir -p {sources_dir} && cat > {sources_file}"], input=new_sources, capture_output=True, text=True, timeout=10, start_new_session=True, ) write_ok = True logger.info("sources.list 已通过 sh 写入") except Exception as e: logger.warning(f"无法写入 sources.list: {e}") if not write_ok: logger.error("sources.list 写入失败，apt 将无法找到包") except Exception as e: logger.warning(f"修复 apt 源异常: {e}") # 3. 重新 apt-get update logger.info("重新 apt-get update...") try: subprocess.run( sudo + ["apt-get", "update"], capture_output=True, text=True, timeout=90, start_new_session=True, ) except Exception as e: logger.debug(f"apt-get update 异常: {e}") @property def display_env(self) -> Dict[str, str]: """返回 DISPLAY 环境变量字典，供子进程使用""" return {"DISPLAY": self.display} @property def status(self) -> Dict[str, Any]: """获取当前状态信息""" running = self.is_running # Xvfb: 附加模式 vs 自启动模式 if self._xvfb_attached: xvfb_alive = self._xvfb_attached_pid is not None and self._is_pid_alive(self._xvfb_attached_pid) xvfb_pid = self._xvfb_attached_pid if xvfb_alive else None else: xvfb_alive = self._xvfb_process is not None and self._xvfb_process.poll() is None xvfb_pid = self._xvfb_process.pid if xvfb_alive else None vnc_alive = self._x11vnc_process is not None and self._x11vnc_process.poll() is None ws_alive = self._websockify_process is not None and self._websockify_process.poll() is None # [v1.47.2] 提供环境信息（前端用于决定是否显示VNC按钮） try: from core.env_detect import is_termux, is_desktop _is_termux = is_termux() _is_desktop = is_desktop() except ImportError: _is_termux = False _is_desktop = False return { "running": running, "display": self.display, "resolution": self.resolution, "rotation": self._current_rotation, # [v1.47.1] 当前屏幕旋转方向 "is_termux": _is_termux, # [v1.47.2] 前端用于决定是否显示VNC按钮 "is_desktop": _is_desktop, # [v1.47.2] "xvfb": { "alive": xvfb_alive or running, # 如果整体在运行，Xvfb 肯定是活的 "pid": xvfb_pid, "attached": self._xvfb_attached, }, "x11vnc": { "alive": vnc_alive or running, "port": self.x11vnc_port, "pid": self._x11vnc_process.pid if vnc_alive else None, }, "websockify": { "alive": ws_alive or running, "port": self.novnc_port, "pid": self._websockify_process.pid if ws_alive else None, }, # [v1.33.0] 浏览器和窗口管理器信息 "browser": { "available": bool(shutil.which("myagent-browser") or shutil.which("firefox")), "path": shutil.which("myagent-browser") or shutil.which("firefox") or None, }, "window_manager": self._wm_name, "novnc_url": f"/vnc/vnc.html?autoconnect=true&resize=remote" if running else None, } def get_novnc_url(self, base_url: str = "") -> Optional[str]: """获取 noVNC 客户端访问 URL""" if not self.is_running: return None return f"{base_url}/vnc/vnc.html?autoconnect=true&resize=remote" def rotate_display(self, direction: str = "toggle") -> Dict[str, Any]: """[v1.47.3] 旋转X11显示 — 通过交换Xvfb分辨率实现服务端旋转。 实现方式：解析当前resolution "WxHxD"，交换W和H， 然后用xrandr --fb修改Xvfb帧缓冲区大小。 Xvfb的RANDR扩展不支持--rotate，但支持--fb调整分辨率。 如果xrandr --fb失败（部分Xvfb版本不支持），则重启整个VNC堆栈 以新分辨率启动。 Args: direction: "toggle"=切换横竖, "portrait"=竖屏, "landscape"=横屏 Returns: {"success": bool, "rotation": str, "resolution": str, "error": str} """ if not self.is_running: return {"success": False, "error": "VNC未运行", "rotation": self._current_rotation} # 解析当前分辨率 "WxHxD" try: parts = self.resolution.split("x") cur_w, cur_h = int(parts[0]), int(parts[1]) depth = parts[2] if len(parts) > 2 else "24" except (ValueError, IndexError): return {"success": False, "error": f"无法解析分辨率: {self.resolution}", "rotation": self._current_rotation} # 确定目标分辨率 if direction == "toggle": # 切换横竖：当前横屏→竖屏，当前竖屏→横屏 if cur_w > cur_h: new_w, new_h = cur_h, cur_w new_rotation = "portrait" else: new_w, new_h = cur_h, cur_w new_rotation = "landscape" elif direction == "portrait": new_w, new_h = min(cur_w, cur_h), max(cur_w, cur_h) new_rotation = "portrait" elif direction == "landscape": new_w, new_h = max(cur_w, cur_h), min(cur_w, cur_h) new_rotation = "landscape" else: new_rotation = direction if direction == "normal" or direction == "landscape": new_w, new_h = max(cur_w, cur_h), min(cur_w, cur_h) else: new_w, new_h = min(cur_w, cur_h), max(cur_w, cur_h) new_resolution = f"{new_w}x{new_h}x{depth}" # 如果分辨率没变，无需操作 if new_resolution == self.resolution: return {"success": True, "rotation": new_rotation, "resolution": self.resolution} env = {**os.environ, "DISPLAY": self.display} # ── 方案1: 尝试xrandr --fb在线调整分辨率（不需要重启Xvfb） ── # [v1.47.36] 使用缓存的 xrandr 能力 xrandr_ok = False if self._xrandr_fb_supported is not False and shutil.which("xrandr"): try: result = subprocess.run( ["xrandr", "--fb", f"{new_w}x{new_h}"], capture_output=True, text=True, timeout=5, env=env, start_new_session=True, ) if result.returncode == 0: xrandr_ok = True self._xrandr_fb_supported = True logger.info(f"xrandr --fb {new_w}x{new_h} 成功") else: self._xrandr_fb_supported = False logger.debug(f"xrandr --fb 失败: {(result.stderr or '')[:100]}") except Exception as e: self._xrandr_fb_supported = False logger.debug(f"xrandr --fb 异常: {e}") if xrandr_ok: self.resolution = new_resolution self._current_rotation = new_rotation # 通知窗口管理器重新适配 self._notify_wm_resize(new_w, new_h, env) return {"success": True, "rotation": new_rotation, "resolution": new_resolution} # ── 方案2: 重启VNC堆栈（xrandr不支持时） ── logger.info(f"xrandr --fb 不支持，将通过重启VNC堆栈切换分辨率到 {new_resolution}") self.resolution = new_resolution self._current_rotation = new_rotation # 用同步方式重启（在后台线程中执行，避免阻塞async事件循环） import threading restart_done = threading.Event() restart_result = {"success": False} def _do_restart(): try: # 同步停止和启动 loop = asyncio.new_event_loop() try: loop.run_until_complete(self._restart_for_rotation()) restart_result["success"] = True finally: loop.close() except Exception as e: restart_result["error"] = str(e) logger.warning(f"VNC重启旋转异常: {e}") finally: restart_done.set() t = threading.Thread(target=_do_restart, daemon=True) t.start() # 等待最多30秒 restart_done.wait(timeout=30) if restart_result.get("success"): return {"success": True, "rotation": new_rotation, "resolution": new_resolution} else: return {"success": False, "error": restart_result.get("error", "重启超时"), "rotation": self._current_rotation, "resolution": self.resolution} async def _restart_for_rotation(self) -> None: """[v1.47.3] 为旋转重启VNC堆栈 — 停止所有服务并用新分辨率重新启动""" # 停止所有服务 await self.stop() # 短暂等待 await asyncio.sleep(1) # 用新分辨率重新启动 result = await self.start() if not result.get("success"): logger.error(f"旋转后重启VNC失败: {result.get('message')}") raise RuntimeError(result.get("message", "重启失败")) def _notify_wm_resize(self, w: int, h: int, env: dict) -> None: """[v1.47.3] 通知窗口管理器屏幕尺寸变化，触发重新布局""" # icewm: 发送 _NET_WM_FULLSCREEN_MONITORS 或通过 wmctrl 触发重排 if shutil.which("wmctrl"): try: subprocess.run( ["wmctrl", "-r", ":ACTIVE:", "-b", "remove,maximized_vert,maximized_horz"], capture_output=True, timeout=3, env=env, start_new_session=True, ) # 然后重新最大化当前窗口 subprocess.run( ["wmctrl", "-r", ":ACTIVE:", "-b", "add,maximized_vert,maximized_horz"], capture_output=True, timeout=3, env=env, start_new_session=True, ) except Exception: pass # 通用：刷新桌面 if shutil.which("xdotool"): try: subprocess.run( ["xdotool", "key", "super"], capture_output=True, timeout=2, env=env, start_new_session=True, ) time.sleep(0.2) subprocess.run( ["xdotool", "key", "Escape"], capture_output=True, timeout=2, env=env, start_new_session=True, ) except Exception: pass def resize_display(self, width: int, height: int) -> Dict[str, Any]: """[v1.47.33] 自适应调整 VNC 分辨率匹配客户端视口。 使用 xrandr --fb 在线调整 Xvfb 分辨率，不需要重启 VNC 服务。 Firefox 最大化后自动适配新分辨率。 [v1.47.36] 添加 xrandr 能力缓存：如果 xrandr --fb 曾失败， 后续调用直接跳过，避免在 Termux proot 下反复尝试。 Args: width: 目标宽度(px) height: 目标高度(px) Returns: {"success": bool, "resolution": str, "error": str} """ if not self.is_running: return {"success": False, "error": "VNC未运行", "resolution": self.resolution} # 限制范围 width = max(320, min(4096, width)) height = max(240, min(4096, height)) try: parts = self.resolution.split("x") depth = parts[2] if len(parts) > 2 else "24" except (ValueError, IndexError): depth = "24" new_resolution = f"{width}x{height}x{depth}" if new_resolution == self.resolution: return {"success": True, "resolution": self.resolution} env = {**os.environ, "DISPLAY": self.display} # [v1.47.36] 使用缓存的 xrandr 能力，避免反复尝试失败 if self._xrandr_fb_supported is not False and shutil.which("xrandr"): try: result = subprocess.run( ["xrandr", "--fb", f"{width}x{height}"], capture_output=True, text=True, timeout=5, env=env, start_new_session=True, ) if result.returncode == 0: self._xrandr_fb_supported = True self.resolution = new_resolution self._notify_wm_resize(width, height, env) logger.info(f"分辨率已调整: {new_resolution}") return {"success": True, "resolution": new_resolution} else: self._xrandr_fb_supported = False logger.debug(f"xrandr --fb 失败: {(result.stderr or '')[:100]}") except Exception as e: self._xrandr_fb_supported = False logger.debug(f"xrandr --fb 异常: {e}") # xrandr 不可用或已失败，只更新记录（下次重启生效） self.resolution = new_resolution logger.info(f"分辨率将在下次重启时生效: {new_resolution}") return {"success": True, "resolution": new_resolution, "note": "下次重启生效"} def ensure_dependencies(self) -> tuple: """检测并安装 VNC 所需依赖。 [v1.34.0] 重构：此方法仅用于后台安装，不再阻塞 start()。 核心依赖（Xvfb, x11vnc）在 start() 中通过 _install_core_deps() 快速安装。 非核心依赖在此方法中安装，运行在后台线程中。 Returns: (success: bool, message: str) """ # macOS 和 Windows 桌面环境不需要 VNC/Xvfb，直接跳过 import platform if platform.system() in ("Darwin", "Windows"): logger.info(f"{platform.system()} 桌面环境不需要 VNC/Xvfb，跳过依赖安装") return True, f"{platform.system()} 桌面环境不需要 VNC" warnings = [] sudo = ["sudo"] if os.getuid() != 0 and shutil.which("sudo") else [] # [v1.34.0] 收集需要安装的包 packages_to_install = [] # [v1.47.31] 精简模式 — 只装 icewm，不装主题/文件管理器/虚拟键盘 _has_icewm = shutil.which("icewm-session") or shutil.which("icewm") if not _has_icewm: packages_to_install.append("icewm") # 不装 icewm-themes — 精简模式不需要主题 # [v1.47.1] x11-xserver-utils — 提供 xrandr 命令（屏幕旋转） if not shutil.which("xrandr"): packages_to_install.append("x11-xserver-utils") # [v1.47.31] wmctrl — Firefox 窗口最大化（精简模式必需） if not shutil.which("wmctrl"): packages_to_install.append("wmctrl") # [v1.47.33] 中文字体支持 — CJK 字体 + emoji + locale # 检查是否已安装 CJK 字体（通过文件存在性判断，比 which 更准确） _has_cjk_font = os.path.isfile("/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc") or \ os.path.isfile("/usr/share/fonts/truetype/noto-cjk/NotoSansCJK-Regular.ttc") or \ os.path.isdir("/usr/share/fonts/opentype/noto") if not _has_cjk_font: packages_to_install.extend(["fonts-noto-cjk", "fonts-noto-color-emoji"]) # 浏览器安装由 _install_browser_background() 统一管理 if not packages_to_install: return True, "所有 VNC 依赖已就绪" # [v1.34.0] 后台安装 — 不再阻塞桌面 logger.info(f"后台安装非核心依赖: {packages_to_install}") self._fix_dpkg_interrupted(sudo) try: result = subprocess.run( sudo + ["apt-get", "install", "-y", "--fix-broken", "--no-install-recommends"] + packages_to_install, capture_output=True, text=True, timeout=180, start_new_session=True, # [v1.40.0] 进程隔离 ) if result.returncode == 0: logger.info(f"后台安装成功: {packages_to_install}") else: error_msg = result.stderr[-300:] if result.stderr else result.stdout[-300:] if "dpkg was interrupted" in error_msg or "dpkg --configure" in error_msg: self._fix_dpkg_interrupted(sudo) try: result2 = subprocess.run( sudo + ["apt-get", "install", "-y", "--fix-broken", "--no-install-recommends"] + packages_to_install, capture_output=True, text=True, timeout=180, start_new_session=True, ) if result2.returncode == 0: logger.info(f"后台安装成功 (修复后): {packages_to_install}") else: logger.warning(f"后台安装仍失败: {result2.stderr[:200]}") except subprocess.TimeoutExpired: logger.warning("后台安装超时，跳过") else: logger.warning(f"后台安装失败: {error_msg[:200]}") # 尝试逐个安装 self._install_packages_one_by_one(sudo, packages_to_install) except subprocess.TimeoutExpired: logger.warning("后台安装超时 (3分钟)，跳过") # 尝试逐个安装 self._install_packages_one_by_one(sudo, packages_to_install) except Exception as e: logger.warning(f"后台安装异常: {e}，跳过") # 安装 x11vnc 额外依赖（非关键，忽略错误） if not shutil.which("x11vnc"): extra_libs = ["python3-numpy", "libxtst6", "libxext6", "libxfixes3", "libxrandr2", "libxrender1", "libxinerama1", "libfontconfig1"] try: subprocess.run( sudo + ["apt-get", "install", "-y", "--no-install-recommends"] + extra_libs, capture_output=True, text=True, timeout=90, start_new_session=True, ) except Exception: pass # websockify pip 安装 if not shutil.which("websockify"): self._install_websockify_quick() msg = "VNC 依赖安装完成（后台）" return True, msg def _install_core_deps(self, missing: list) -> None: """[v1.34.0] 快速安装核心依赖（Xvfb, x11vnc），超时短，不阻塞桌面。 核心依赖是 VNC 桌面能运行的前提，必须先装好。 但即使失败也不完全阻止 — 让 start() 决定是否继续。 [v1.47.37] 重装时 apt 源可能为空，先 apt-get update 再安装； 超时从 60s 增加到 120s，proot 下下载较慢。 """ sudo = ["sudo"] if os.getuid() != 0 and shutil.which("sudo") else [] # 映射命令名到包名 cmd_to_pkg = {"Xvfb": "xvfb", "x11vnc": "x11vnc"} packages = [cmd_to_pkg.get(c, c.lower()) for c in missing] if not packages: return # 先修复 dpkg self._fix_dpkg_interrupted(sudo) # [v1.47.37] 先更新 apt 源 — 重装/新装时源可能为空， # 不 update 直接 install 会导致 "Unable to locate package" 或卡住超时 # [v1.47.39] 不用 -qq（会压制错误），检查返回码，失败时修复源 update_ok = False logger.info("更新 apt 源...") try: upd = subprocess.run( sudo + ["apt-get", "update"], capture_output=True, text=True, timeout=90, start_new_session=True, ) if upd.returncode == 0: update_ok = True else: logger.warning(f"apt-get update 失败: {(upd.stderr or '')[:200]}") except subprocess.TimeoutExpired: logger.warning("apt-get update 超时 (90s)") except Exception as e: logger.debug(f"apt-get update 异常: {e}") # [v1.47.39] 如果 update 失败，尝试修复 sources.list（确保 universe 仓库启用） if not update_ok: self._fix_apt_sources(sudo) install_cmd = sudo + ["apt-get", "install", "-y", "--fix-broken", "--no-install-recommends"] + packages logger.info(f"安装核心依赖: {packages}") try: result = subprocess.run( install_cmd, capture_output=True, text=True, timeout=120, start_new_session=True, # [v1.40.0] 进程隔离 ) if result.returncode == 0: logger.info(f"核心依赖安装成功: {packages}") else: error_msg = result.stderr[-300:] if result.stderr else result.stdout[-300:] if "dpkg was interrupted" in error_msg: self._fix_dpkg_interrupted(sudo) try: result2 = subprocess.run( install_cmd, capture_output=True, text=True, timeout=120, start_new_session=True, ) if result2.returncode == 0: logger.info(f"核心依赖安装成功 (修复后): {packages}") else: logger.warning(f"核心依赖安装失败: {result2.stderr[:200]}") except subprocess.TimeoutExpired: logger.warning("核心依赖安装超时 (120s)") elif "Unable to locate package" in error_msg: # [v1.47.39] 找不到包 → 修复源 + 重试 update + install logger.warning(f"找不到包，修复源并重试: {error_msg[:200]}") self._fix_apt_sources(sudo) try: subprocess.run( sudo + ["apt-get", "update"], capture_output=True, text=True, timeout=90, start_new_session=True, ) except Exception: pass try: result3 = subprocess.run( install_cmd, capture_output=True, text=True, timeout=120, start_new_session=True, ) if result3.returncode == 0: logger.info(f"核心依赖安装成功 (修复源后): {packages}") else: logger.warning(f"核心依赖安装失败: {result3.stderr[:200]}") except subprocess.TimeoutExpired: logger.warning("核心依赖安装超时 (120s)") else: logger.warning(f"核心依赖安装失败: {error_msg[:200]}") except subprocess.TimeoutExpired: logger.warning("核心依赖安装超时 (120s)") except Exception as e: logger.warning(f"核心依赖安装异常: {e}") def _install_websockify_quick(self) -> None: """[v1.34.0] 快速安装 websockify（pip）""" try: logger.info("pip install websockify") pip_cmd = shutil.which("pip3") or shutil.which("pip") if pip_cmd: result = subprocess.run( [pip_cmd, "install", "websockify"], capture_output=True, text=True, timeout=60, start_new_session=True, ) if result.returncode == 0: logger.info("websockify 安装成功") else: logger.warning(f"websockify 安装失败: {result.stderr[-200:]}") except Exception as e: logger.warning(f"websockify 安装异常: {e}") def _install_packages_one_by_one(self, sudo: list, packages: list) -> None: """[v1.34.0] 逐个安装包 — 批量安装失败时的回退策略。 某些包可能因依赖问题失败，但不应该阻止其他包安装。 每个包单独安装，超时 45 秒。 """ for pkg in packages: try: logger.info(f"尝试单独安装: {pkg}") result = subprocess.run( sudo + ["apt-get", "install", "-y", "--fix-broken", "--no-install-recommends", pkg], capture_output=True, text=True, timeout=45, start_new_session=True, ) if result.returncode == 0: logger.info(f"单独安装成功: {pkg}") else: logger.warning(f"单独安装失败: {pkg} — {result.stderr[:100]}") except subprocess.TimeoutExpired: logger.warning(f"单独安装超时: {pkg}") except Exception as e: logger.warning(f"单独安装异常: {pkg} — {e}") def _install_browser_background(self) -> None: """安装浏览器。 下载 Mozilla Firefox 预编译二进制文件安装到 /opt/firefox/。 """ # 检查是否已有可用浏览器 for name in ["myagent-browser", "firefox-esr", "firefox"]: existing = shutil.which(name) if existing and not self._is_snap_wrapper(existing): logger.info(f"已有可用浏览器: {existing}，跳过安装") return # snap 包装器不可用，需要安装真正的 Firefox snap_firefox = shutil.which("firefox") if snap_firefox and self._is_snap_wrapper(snap_firefox): logger.info("已有 Firefox 是 snap 包装器，需要下载安装真正的 Firefox") sudo = [] if os.getuid() == 0 or not shutil.which("sudo") else ["sudo"] # 检测架构 arch = "arm64" try: arch = subprocess.run( ["dpkg", "--print-architecture"], capture_output=True, text=True, timeout=5, start_new_session=True, ).stdout.strip() except Exception: pass logger.info("下载安装 Firefox（约需 10 分钟）...") browser_binary = self._install_firefox_proot(sudo, arch) if browser_binary: self._create_browser_wrapper(browser_binary) logger.info("Firefox 安装完成") else: logger.warning("Firefox 安装失败") def _create_titlebar_button_pixmaps(self, theme_dir: str) -> None: """[v1.47.5] 创建icewm标题栏完整XPM pixmap文件。 Look=pixmap模式下，icewm需要XPM文件来绘制标题栏背景和按钮符号。 icewm绘制按钮的流程： 1. 先将按钮区域清为黑色 2. 再绘制按钮符号XPM（close.xpm, minimize.xpm等）到按钮区域 [v1.47.3] 修复1：添加button.xpm/buttonA.xpm按钮背景 [v1.47.5] 修复2：符号XPM尺寸必须=按钮区域尺寸（72×72） - 如果符号XPM(48×48) < 按钮区域(72×72)，icewm只绘制48×48区域 - 剩余24px保持黑色，形成黑长条挡住按钮2/3高度 - 解决：符号XPM用72×72，符号居中绘制，四周填充背景色 需要的XPM文件： - titleBar.xpm / titleBarA.xpm — 标题栏背景（水平平铺） - button.xpm / buttonA.xpm — 按钮区背景（[v1.47.3] 关键！） - menuButton.xpm / menuButtonA.xpm — 窗口菜单按钮 - close.xpm / closeA.xpm — 关闭按钮符号 - minimize.xpm / minimizeA.xpm — 最小化按钮符号 - maximize.xpm / maximizeA.xpm — 最大化按钮符号 - restore.xpm / restoreA.xpm — 还原按钮符号 """ # 按钮尺寸 — 匹配TitleBarHeight=72的按钮区域 # icewm在pixmap模式下，按钮区域是正方形，边长=TitleBarHeight titlebar_h = 72 # 匹配TitleBarHeight=72 button_size = titlebar_h # [v1.47.3] 按钮背景必须与标题栏同高 # [v1.47.5] 符号XPM必须与按钮区域同尺寸（72×72） # icewm在pixmap模式下绘制按钮流程：先清黑底 → 再绘制符号XPM # 如果符号XPM尺寸<按钮区域，未覆盖部分保持黑色，形成黑长条 # 解决：符号XPM=按钮区域大小，符号居中，四周填充标题栏背景色 symbol_size = button_size # 颜色定义 # 普通窗口: 浅米背景 + 深棕符号 normal_bg = "#E8E0D5" normal_fg = "#3D3229" # 活跃窗口: 暖橙背景 + 白色符号 active_bg = "#D97757" active_fg = "#FFFFFF" # ── [v1.47.2] 标题栏背景XPM（解决黑色横条问题） ── # Look=pixmap模式下，没有titleBar XPM时icewm用黑色填充标题栏 # 创建4px宽×72px高的纯色XPM，icewm会水平平铺 self._write_solid_xpm(os.path.join(theme_dir, "titleBar.xpm"), 4, titlebar_h, normal_bg) self._write_solid_xpm(os.path.join(theme_dir, "titleBarA.xpm"), 4, titlebar_h, active_bg) # ── [v1.47.3] 按钮区背景XPM — 解决黑色横条挡住按钮 ── # icewm为每个按钮绘制button.xpm背景，然后叠加符号XPM # 没有button.xpm时，icewm用黑色填充，导致按钮被黑条挡住 # 按钮背景是正方形 button_size×button_size，与标题栏同高 self._write_solid_xpm(os.path.join(theme_dir, "button.xpm"), button_size, button_size, normal_bg) self._write_solid_xpm(os.path.join(theme_dir, "buttonA.xpm"), button_size, button_size, active_bg) # ── 窗口菜单按钮(≡): 三条横线 ── menu_grid = self._xpm_make_menu(symbol_size, 3, 4) self._write_xpm(os.path.join(theme_dir, "menuButton.xpm"), symbol_size, menu_grid, normal_bg, normal_fg) self._write_xpm(os.path.join(theme_dir, "menuButtonA.xpm"), symbol_size, menu_grid, active_bg, active_fg) # ── 关闭按钮(×): 两条交叉对角线 ── close_grid = self._xpm_make_cross(symbol_size, 4) self._write_xpm(os.path.join(theme_dir, "close.xpm"), symbol_size, close_grid, normal_bg, normal_fg) self._write_xpm(os.path.join(theme_dir, "closeA.xpm"), symbol_size, close_grid, active_bg, active_fg) # ── 最小化按钮(-): 底部水平线 ── minimize_grid = self._xpm_make_hline(symbol_size, 4, offset_y=6) self._write_xpm(os.path.join(theme_dir, "minimize.xpm"), symbol_size, minimize_grid, normal_bg, normal_fg) self._write_xpm(os.path.join(theme_dir, "minimizeA.xpm"), symbol_size, minimize_grid, active_bg, active_fg) # ── 最大化按钮(□): 矩形边框 ── maximize_grid = self._xpm_make_rect(symbol_size, 5) self._write_xpm(os.path.join(theme_dir, "maximize.xpm"), symbol_size, maximize_grid, normal_bg, normal_fg) self._write_xpm(os.path.join(theme_dir, "maximizeA.xpm"), symbol_size, maximize_grid, active_bg, active_fg) # ── 还原按钮(❐): 两个重叠矩形 ── restore_grid = self._xpm_make_restore(symbol_size, 4) self._write_xpm(os.path.join(theme_dir, "restore.xpm"), symbol_size, restore_grid, normal_bg, normal_fg) self._write_xpm(os.path.join(theme_dir, "restoreA.xpm"), symbol_size, restore_grid, active_bg, active_fg) logger.info(f"标题栏XPM已创建(含背景+按钮背景): {theme_dir}") @staticmethod def _xpm_make_cross(size: int, thickness: int) -> list: """生成×符号的像素网格（对角交叉线）""" grid = [[0] * size for _ in range(size)] margin = max(size // 5, 4) for i in range(size): for t in range(thickness): # 主对角线 (左上→右下) r1 = i + t - thickness // 2 if margin <= i < size - margin and margin <= r1 < size - margin: if 0 <= r1 < size: grid[r1][i] = 1 # 副对角线 (右上→左下) r2 = (size - 1 - i) + t - thickness // 2 if margin <= i < size - margin and margin <= r2 < size - margin: if 0 <= r2 < size: grid[r2][i] = 1 return grid @staticmethod def _xpm_make_hline(size: int, thickness: int, offset_y: int = 0) -> list: """生成-符号的像素网格（水平线）""" grid = [[0] * size for _ in range(size)] margin = max(size // 5, 4) y_start = size - margin - thickness - offset_y for y in range(y_start, min(y_start + thickness, size - margin)): for x in range(margin, size - margin): if 0 <= y < size: grid[y][x] = 1 return grid @staticmethod def _xpm_make_rect(size: int, thickness: int) -> list: """生成□符号的像素网格（矩形边框）""" grid = [[0] * size for _ in range(size)] margin = max(size // 5, 4) for y in range(margin, size - margin): for x in range(margin, size - margin): # 上边 if y < margin + thickness: grid[y][x] = 1 # 下边 elif y >= size - margin - thickness: grid[y][x] = 1 # 左边 elif x < margin + thickness: grid[y][x] = 1 # 右边 elif x >= size - margin - thickness: grid[y][x] = 1 return grid @staticmethod def _xpm_make_restore(size: int, thickness: int) -> list: """生成还原符号的像素网格（两个重叠矩形）""" grid = [[0] * size for _ in range(size)] margin = max(size // 5, 4) gap = max(size // 8, 3) # 后面的矩形（左上偏移） for y in range(margin, margin + (size - 2 * margin) // 2 + thickness): for x in range(margin, margin + (size - 2 * margin) // 2 + thickness): if y < margin + thickness or y >= margin + (size - 2 * margin) // 2: if 0 <= y < size and 0 <= x < size: grid[y][x] = 1 elif x < margin + thickness or x >= margin + (size - 2 * margin) // 2: if 0 <= y < size and 0 <= x < size: grid[y][x] = 1 # 前面的矩形（右下偏移） r_margin = margin + gap for y in range(r_margin, size - margin): for x in range(r_margin, size - margin): if y < r_margin + thickness or y >= size - margin - thickness: grid[y][x] = 1 elif x < r_margin + thickness or x >= size - margin - thickness: grid[y][x] = 1 return grid @staticmethod def _write_xpm(filepath: str, size: int, grid: list, bg_color: str, fg_color: str) -> None: """将像素网格写入XPM文件（正方形，2色）""" lines = [ "/* XPM */", f"static char * button_xpm[] = {{", f'"{size} {size} 2 1",', f'" c {bg_color}",', f'". c {fg_color}",', ] for row in grid: lines.append('"' + ''.join('.' if p else ' ' for p in row) + '",') # 最后一行不要逗号 lines[-1] = lines[-1][:-1] + '"' lines.append("};") with open(filepath, "w") as f: f.write("\n".join(lines) + "\n") @staticmethod def _write_solid_xpm(filepath: str, width: int, height: int, color: str) -> None: """[v1.47.2] 写入纯色XPM文件（用于标题栏背景平铺）""" lines = [ "/* XPM */", f"static char * titlebar_xpm[] = {{", f'"{width} {height} 1 1",', f'" c {color}",', ] for _ in range(height): lines.append('"' + ' ' * width + '",') # 最后一行不要逗号 lines[-1] = lines[-1][:-1] + '"' lines.append("};") with open(filepath, "w") as f: f.write("\n".join(lines) + "\n") @staticmethod def _xpm_make_menu(size: int, num_lines: int, thickness: int) -> list: """[v1.47.2] 生成≡符号的像素网格（三条水平线=汉堡菜单）""" grid = [[0] * size for _ in range(size)] margin = max(size // 5, 4) usable = size - 2 * margin gap = usable // (num_lines + 1) for n in range(num_lines): y_center = margin + gap * (n + 1) for t in range(thickness): y = y_center - thickness // 2 + t if margin <= y < size - margin: for x in range(margin, size - margin): grid[y][x] = 1 return grid def _ensure_dbus_session(self) -> None: """[v1.43.3] 确保 D-Bus 会话守护进程运行。 在 proot/Android 环境中，D-Bus 通常没有自动启动。 但 dbus-launch 在 proot 下有严重问题： - dbus-daemon 使用 --shm-helper 参数，proot 路径翻译破坏绝对路径 报错: "error: expected absolute path: --shm-helper" - dbus-launch 内部使用 futex 系统调用，proot 不支持 报错: "The futex facility returned an unexpected error code. Aborted" [v1.43.3] 修复策略: - proot 环境下：直接启动 dbus-daemon，指定 unix socket 地址 绕过 dbus-launch（它内部启动 dbus-daemon 时会带 --shm-helper） - 非 proot 环境：仍用 dbus-launch --sh-syntax（标准方式） """ # 确保 DISPLAY 已设置 if not os.environ.get("DISPLAY"): try: x11_dir = "/tmp/.X11-unix" if os.path.isdir(x11_dir): for f in os.listdir(x11_dir): if f.startswith("X"): os.environ["DISPLAY"] = f":{f[1:]}" break except Exception: pass if not os.environ.get("DISPLAY"): os.environ["DISPLAY"] = self.display # 检查是否已有 D-Bus 会话 if os.environ.get("DBUS_SESSION_BUS_ADDRESS"): logger.debug("D-Bus 会话已存在") return is_proot = self._is_proot_env() or self._detect_proot_environment() if is_proot: # ── proot 环境：跳过 D-Bus 启动 ── # [v1.47.41] proot 下 dbus-daemon 始终无法启动（exit code 1）， # dbus-launch 也因 --shm-helper 和 futex 问题不可用。 # 浏览器在无 D-Bus 环境下仍可正常工作 # （MOZ_FAKE_NO_SANDBOX=1 + MOZ_DBUS_REMOTE=1 已处理兼容性）， # 因此跳过此步骤以避免无意义的 WARNING 日志。 logger.debug("proot 环境下跳过 D-Bus 会话启动（已知不可用，浏览器无需 D-Bus）") else: # ── 非 proot 环境：用标准 dbus-launch ── self._start_dbus_launch_standard() def _start_dbus_launch_standard(self) -> None: """[v1.43.3] 非 proot 环境下用 dbus-launch 标准方式启动 D-Bus。""" dbus_launch = shutil.which("dbus-launch") if not dbus_launch: logger.debug("dbus-launch 不可用，跳过 D-Bus 会话启动") return try: dbus_env = {**os.environ, "DISPLAY": os.environ.get("DISPLAY", self.display)} result = subprocess.run( [dbus_launch, "--sh-syntax"], capture_output=True, text=True, timeout=10, env=dbus_env, start_new_session=True, ) if result.returncode == 0 and result.stdout: for line in result.stdout.strip().split(";"): line = line.strip() if line.startswith("export "): line = line[len("export "):] if "=" in line: key, _, value = line.partition("=") key = key.strip() value = value.strip().strip("'\"") if key and value: os.environ[key] = value logger.debug(f"D-Bus 环境变量: {key}={value[:50]}...") if os.environ.get("DBUS_SESSION_BUS_ADDRESS"): logger.info("D-Bus 会话已启动 (标准模式)") else: logger.warning("D-Bus 会话启动但未获取到 DBUS_SESSION_BUS_ADDRESS") else: logger.warning(f"dbus-launch 启动失败: {result.stderr[:200]}") except subprocess.TimeoutExpired: logger.warning("dbus-launch 超时 (10s)") except Exception as e: logger.warning(f"D-Bus 会话启动异常: {e}") def _setup_chinese_locale(self) -> None: """[v1.47.33] 配置中文 locale — 让 VNC 桌面正确显示中文。 设置 zh_CN.UTF-8 locale 并导出环境变量， 确保 Firefox 和 icewm 能正确渲染中文字符。 字体由 ensure_dependencies() 安装 (fonts-noto-cjk)。 """ sudo = ["sudo"] if os.getuid() != 0 and shutil.which("sudo") else [] # 1. 确保 locale 已生成 locale_gen_ok = False try: # 检查 zh_CN.UTF-8 是否已生成 result = subprocess.run( ["locale", "-a"], capture_output=True, text=True, timeout=10, ) if result.returncode == 0 and "zh_CN.utf8" in result.stdout.lower(): locale_gen_ok = True logger.info("zh_CN.UTF-8 locale 已存在") except Exception: pass if not locale_gen_ok: # 确保 locales 包已安装 try: subprocess.run( sudo + ["apt-get", "install", "-y", "--no-install-recommends", "locales"], capture_output=True, text=True, timeout=60, start_new_session=True, ) except Exception: pass # 生成 zh_CN.UTF-8 locale try: # 先在 /etc/locale.gen 中取消注释 locale_gen_file = "/etc/locale.gen" if os.path.isfile(locale_gen_file): # sed 取消注释 zh_CN.UTF-8 subprocess.run( sudo + ["sed", "-i", "s/# zh_CN.UTF-8 UTF-8/zh_CN.UTF-8 UTF-8/", locale_gen_file], capture_output=True, text=True, timeout=5, ) # locale-gen result = subprocess.run( sudo + ["locale-gen", "zh_CN.UTF-8"], capture_output=True, text=True, timeout=30, ) if result.returncode == 0: locale_gen_ok = True logger.info("zh_CN.UTF-8 locale 已生成") else: logger.warning(f"locale-gen 失败: {(result.stderr or '')[:200]}") except Exception as e: logger.warning(f"locale-gen 异常: {e}") # 2. 设置环境变量 — 影响当前进程和子进程 os.environ["LANG"] = "zh_CN.UTF-8" os.environ["LANGUAGE"] = "zh_CN:zh" os.environ["LC_ALL"] = "zh_CN.UTF-8" logger.info("中文 locale 环境变量已设置: LANG=zh_CN.UTF-8") # 3. 刷新字体缓存 try: subprocess.run( ["fc-cache", "-f"], capture_output=True, text=True, timeout=30, start_new_session=True, ) logger.info("字体缓存已刷新") except Exception as e: logger.debug(f"fc-cache 刷新异常: {e}") def _maximize_browser_window(self) -> None: """[v1.47.34] 最大化浏览器窗口 — icewm 隐藏标题栏后最大化占满屏幕。 使用 wmctrl 或 xdotool 将浏览器窗口最大化。 icewm 精简配置下标题栏高度=1且任务栏已隐藏， 最大化后浏览器几乎占满整个屏幕（地址栏保留）。 """ env = {**os.environ, "DISPLAY": self.display} time.sleep(2) # 等待浏览器窗口创建 # 方案1: wmctrl - 最大化 if shutil.which("wmctrl"): try: result = subprocess.run( ["wmctrl", "-l"], capture_output=True, text=True, timeout=5, env=env, start_new_session=True, ) if result.returncode == 0: for line in result.stdout.strip().split("\n"): if any(kw in line.lower() for kw in ["firefox", "navigator"]): win_id = line.split()[0] subprocess.run( ["wmctrl", "-i", "-r", win_id, "-b", "add,maximized_vert,maximized_horz"], capture_output=True, timeout=3, env=env, start_new_session=True, ) logger.info(f"浏览器窗口已最大化 (wmctrl, win_id={win_id})") return except Exception as e: logger.debug(f"wmctrl 最大化失败: {e}") # 方案2: xdotool if shutil.which("xdotool"): try: for search_name in ["Firefox"]: result = subprocess.run( ["xdotool", "search", "--name", search_name], capture_output=True, text=True, timeout=5, env=env, start_new_session=True, ) if result.returncode == 0 and result.stdout.strip(): win_id = result.stdout.strip().split("\n")[0] subprocess.run( ["xdotool", "windowactivate", "--sync", win_id], capture_output=True, timeout=3, env=env, start_new_session=True, ) logger.info(f"浏览器窗口已激活 (xdotool, win_id={win_id})") return except Exception as e: logger.debug(f"xdotool 最大化失败: {e}") logger.info("未找到 wmctrl/xdotool，跳过窗口最大化") def _start_browser_sync(self) -> None: """[v1.38.0] 同步版浏览器启动（在后台线程中调用）。 使用 myagent-browser 包装器作为主入口，包装器内部已包含 proot 兼容参数和 D-Bus/DISPLAY 环境初始化。 如果包装器启动失败，按优先级尝试直接启动原生浏览器。 """ env = {**os.environ, "DISPLAY": self.display} # [v1.47.37] proot/Termux 环境下 Firefox 兼容: # MOZ_FAKE_NO_SANDBOX=1: 让 Firefox 跳过沙盒检测，不弹"安全沙盒已禁用"警告 # （MOZ_DISABLE_CONTENT_SANDBOX 只是禁用沙盒，仍会弹警告） # 参考: github.com/termux/proot/issues/139, #162 env["MOZ_FAKE_NO_SANDBOX"] = "1" # [v1.47.42] 不设置 MOZ_DBUS_REMOTE: proot 下 D-Bus 不可用， # 强制使用 D-Bus 会导致 Firefox 无法传递 URL 给已有实例。 # 让 Firefox 使用默认的 X11 远程检测即可。 # 确保 D-Bus 环境变量已设置（传递给子进程） if os.environ.get("DBUS_SESSION_BUS_ADDRESS"): env["DBUS_SESSION_BUS_ADDRESS"] = os.environ["DBUS_SESSION_BUS_ADDRESS"] if not env.get("GSETTINGS_BACKEND"): env["GSETTINGS_BACKEND"] = "memory" # [v1.47.12] proot 兼容: 避免浏览器启动时 dbus --shm-helper 错误 if not env.get("G_SLICE"): env["G_SLICE"] = "always-malloc" xdg_runtime = env.get("XDG_RUNTIME_DIR") or f"/run/user/{os.getuid()}" env["XDG_RUNTIME_DIR"] = xdg_runtime try: os.makedirs(xdg_runtime, exist_ok=True) os.chmod(xdg_runtime, 0o700) except Exception: pass # 检查是否已有浏览器运行 try: result = subprocess.run( ["pgrep", "-f", "(firefox)"], capture_output=True, text=True, timeout=5, start_new_session=True, ) if result.returncode == 0 and result.stdout.strip(): logger.info("检测到已有浏览器运行，跳过") return except Exception: pass # [v1.47.41] 浏览器启动候选列表 — 只用 Firefox（proot 下唯一可靠） browser_candidates = [] # 候选1: myagent-browser 包装器 myagent_browser = shutil.which("myagent-browser") if myagent_browser: browser_candidates.append(("myagent-browser", [myagent_browser])) # 候选2-N: 直接启动 Firefox for name, cmd in [ ("firefox", ["firefox"]), ("firefox-esr", ["firefox-esr"]), ]: browser_path = shutil.which(cmd[0]) if browser_path: if self._is_snap_wrapper(browser_path): logger.debug(f"跳过 {cmd[0]} (snap 包装器)") continue if not any(name in (bc[0] or "") for bc in browser_candidates): browser_candidates.append((name, [browser_path])) if not browser_candidates: logger.warning("未找到可用浏览器，启动终端作为替代") for term_cmd in [["xterm"], ["lxterminal"], ["xfce4-terminal"]]: if shutil.which(term_cmd[0]): try: subprocess.Popen( term_cmd, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env=env, preexec_fn=os.setpgrp, ) except Exception: pass return return # ── 按候选顺序尝试启动浏览器 ── for browser_name, browser_cmd in browser_candidates: is_myagent_wrapper = "myagent-browser" in (browser_name or "") is_firefox = "firefox" in (browser_name or "") if is_myagent_wrapper: # 包装器已包含所有参数，直接启动 arg_sets = [browser_cmd] elif is_firefox: # [v1.47.34] 不用 --kiosk（会隐藏地址栏），正常启动+icewm隐藏标题栏 arg_sets = [ browser_cmd + ["--width", "960", "--height", "540"], browser_cmd, ] else: arg_sets = [browser_cmd] # 尝试每组参数 for i, cmd in enumerate(arg_sets): try: cmd_str = ' '.join(cmd) logger.info(f"启动浏览器: {browser_name} {cmd_str}") self._browser_process = subprocess.Popen( cmd, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, env=env, preexec_fn=os.setpgrp, ) # 等待浏览器启动（4秒 — GNOME 程序启动较慢） time.sleep(4) if self._browser_process.poll() is None: logger.info(f"浏览器 {browser_name} 已启动 (PID={self._browser_process.pid}, 第{i+1}组参数)") return else: exit_code = self._browser_process.returncode stderr_output = "" try: stderr_output = self._browser_process.stderr.read().decode("utf-8", errors="replace")[:500] except Exception: pass logger.warning(f"浏览器 {browser_name} 退出 (exit={exit_code}, 第{i+1}组): {stderr_output}") self._browser_process = None except Exception as e: logger.warning(f"启动浏览器 {browser_name} 失败 (第{i+1}组): {e}") self._browser_process = None # 当前候选浏览器的所有参数组合都失败，尝试下一个候选 logger.info(f"浏览器 {browser_name} 所有参数组合均失败，尝试下一个候选...") logger.warning("所有浏览器候选均启动失败") # 回退：启动 xterm for term_cmd in [["xterm"], ["lxterminal"]]: if shutil.which(term_cmd[0]): try: subprocess.Popen( term_cmd, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env=env, preexec_fn=os.setpgrp, ) except Exception: pass return def _find_novnc_dir(self) -> Optional[str]: """查找 noVNC web 文件目录 搜索顺序: 1. 项目内置 noVNC (data/novnc/) 2. /usr/share/novnc (apt 安装) 3. /usr/local/share/novnc 4. pip 安装的 websockify 自带 novnc """ search_paths = [] # [v1.20.9] 优先搜索项目内置 noVNC _builtin_novnc = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'data', 'novnc') if os.path.isdir(_builtin_novnc): search_paths.append(_builtin_novnc) search_paths.extend([ "/usr/share/novnc", "/usr/local/share/novnc", "/opt/novnc", ]) # 也搜索 pip 安装路径 try: result = subprocess.run( [shutil.which("python3") or shutil.which("python"), "-c", "import websockify; import os; print(os.path.dirname(websockify.__file__))"], capture_output=True, text=True, timeout=10, ) if result.returncode == 0: pkg_dir = result.stdout.strip() novnc_in_pkg = os.path.join(os.path.dirname(pkg_dir), "novnc", "vnc.html") if os.path.isfile(novnc_in_pkg): search_paths.insert(0, os.path.dirname(novnc_in_pkg)) except Exception: pass for path in search_paths: if os.path.isdir(path): # 检查是否有 vnc.html（完整 noVNC 安装） for sub in ["", "/vnc_lite", "/vnc"]: html = os.path.join(path + sub, "vnc.html") vnc_lite = os.path.join(path + sub, "vnc_lite.html") if os.path.isfile(html) or os.path.isfile(vnc_lite): return path + sub if sub else path # [v1.20.9] 检查是否有 lib/rfb.js（内置精简 noVNC） rfb_js = os.path.join(path, "lib", "rfb.js") if os.path.isfile(rfb_js): return path return None def _ensure_dev_shm(self) -> None: """[v1.20.5] 检查并修复 /dev/shm 挂载（Termux proot-distro 环境下常见问题）。 Android 对 SysV IPC / 共享内存有限制，导致 x11vnc 的 shmget() 失败。 即使加了 -noshm，某些 x11vnc 版本仍会尝试访问 /dev/shm， 所以提前确保目录存在且有正确权限。 """ shm_path = "/dev/shm" try: if not os.path.exists(shm_path): logger.info(f"/dev/shm 不存在，尝试创建...") os.makedirs(shm_path, exist_ok=True) try: os.chmod(shm_path, 0o1777) logger.info("/dev/shm 创建成功，权限设为 1777") except OSError as e: logger.warning(f"/dev/shm chmod 失败 (proot 环境可能无权限): {e}") else: # 检查权限 st = os.stat(shm_path) if st.st_mode & 0o777 != 0o777: try: os.chmod(shm_path, 0o1777) logger.info(f"/dev/shm 权限已修正为 1777") except OSError as e: logger.warning(f"/dev/shm chmod 失败: {e}") except Exception as e: logger.warning(f"/dev/shm 检查异常: {e}") def _kill_port_users(self, port: int) -> None: """[v1.20.5] 强制清理占用指定端口的残留进程（Termux 环境下进程管理不彻底）。 依次尝试: fuser -k → pkill -9 → fallback pgrep+kill """ # 方法1: fuser -k（最彻底，直接杀掉占用端口的进程） try: result = subprocess.run( ["fuser", "-k", f"{port}/tcp"], capture_output=True, text=True, timeout=5, ) if result.returncode == 0: logger.info(f"fuser -k {port}/tcp 执行成功") except (FileNotFoundError, Exception) as e: logger.debug(f"fuser -k 不可用或失败: {e}") def _kill_process_by_name(self, name: str) -> None: """[v1.20.5] 按进程名强制杀死所有匹配进程（pkill -9）。""" try: subprocess.run( ["pkill", "-9", name], capture_output=True, text=True, timeout=5, ) logger.debug(f"pkill -9 {name} 已执行") except (FileNotFoundError, Exception) as e: logger.debug(f"pkill -9 {name} 失败: {e}") async def start(self) -> Dict[str, Any]: """启动 VNC 远程桌面服务。 [v1.47.38] 先装好再启动：所有依赖安装完毕、浏览器就绪后， 才启动 VNC 服务返回成功，用户进入即看到完整桌面。 启动顺序: 1. 安装所有依赖（核心 + 非核心 + 浏览器） 2. 启动 Xvfb + 窗口管理器 + x11vnc + websockify 3. 启动浏览器（最大化） 4. 返回成功 — 桌面已完全就绪 Returns: {"success": bool, "message": str, "status": dict} """ async with self._lock: # 检查是否已在运行（包括检测系统上已有的服务） if self._running or self._detect_existing_services(): self._running = True return { "success": True, "message": "VNC 服务已在运行中", "status": self.status, } # ── Phase 1: 安装所有依赖（同步，装完再启动VNC） ── # 1a. 核心依赖（Xvfb, x11vnc） core_missing = [c for c in ["Xvfb", "x11vnc"] if not shutil.which(c)] if core_missing: logger.warning(f"核心依赖缺失: {core_missing}，尝试快速安装...") self._install_core_deps(core_missing) still_missing = [c for c in core_missing if not shutil.which(c)] if still_missing: logger.error(f"核心依赖安装失败: {still_missing}，无法启动 VNC") return {"success": False, "message": f"核心依赖缺失: {still_missing}", "status": self.status} # websockify 通过 pip 安装 if not shutil.which("websockify"): self._install_websockify_quick() # 1b. 非核心依赖（icewm, wmctrl, 字体, locale） logger.info("安装桌面组件（icewm/字体/locale）...") try: self.ensure_dependencies() except Exception as e: logger.warning(f"非核心依赖安装异常: {e}") # 1c. 浏览器 logger.info("安装浏览器...") self._install_browser_background() # 1d. proot 环境修复（不依赖 X11 显示） self._setup_chinese_locale() # ── Phase 2: 启动 VNC 服务 ── # [v1.47.42] 增大栈大小：proot 下 Firefox 需要更大的栈空间， # 默认 8MB 可能导致崩溃，增大到 32MB try: import resource soft, hard = resource.getrlimit(resource.RLIMIT_STACK) if soft < 32768 * 1024: new_soft = min(32768 * 1024, hard) resource.setrlimit(resource.RLIMIT_STACK, (new_soft, hard)) logger.info(f"栈大小已调整: {soft // 1024}KB → {new_soft // 1024}KB") except Exception as e: logger.debug(f"调整栈大小失败（不影响启动）: {e}") # 检查 /dev/shm 挂载（Termux proot-distro 兼容） self._ensure_dev_shm() # 启动或附加 Xvfb xvfb_ok = await self._start_xvfb() if not xvfb_ok: await self.stop() return {"success": False, "message": "Xvfb 启动失败", "status": self.status} # 等待 Xvfb 就绪 await asyncio.sleep(1) # 中文 locale — 在X客户端启动前设置，所有子进程继承 if not os.environ.get("LANG") or os.environ.get("LANG") == "C": os.environ["LANG"] = "zh_CN.UTF-8" os.environ["LANGUAGE"] = "zh_CN:zh" os.environ["LC_ALL"] = "zh_CN.UTF-8" # proot 兼容环境变量 — 全局设置，所有子进程继承 os.environ["G_SLICE"] = "always-malloc" if not os.environ.get("GSETTINGS_BACKEND"): os.environ["GSETTINGS_BACKEND"] = "memory" # 时区跟随手机系统 if not os.environ.get("TZ"): try: result = subprocess.run( ["getprop", "persist.sys.timezone"], capture_output=True, text=True, timeout=3, ) tz = result.stdout.strip() if result.returncode == 0 else "" if tz: os.environ["TZ"] = tz logger.info(f"时区已从Android系统获取: {tz}") except Exception: pass if not os.environ.get("TZ"): try: tz_file = "/etc/timezone" if os.path.exists(tz_file): with open(tz_file, "r") as f: tz = f.read().strip() if tz: os.environ["TZ"] = tz except Exception: pass # 加载 Xresources try: xrdb_env = {**os.environ, "DISPLAY": self.display} xrdb_file = os.path.expanduser("~/.Xresources") if os.path.exists(xrdb_file): subprocess.run( ["xrdb", "-merge", xrdb_file], capture_output=True, timeout=5, env=xrdb_env, start_new_session=True, ) except Exception as e: logger.debug(f"加载Xresources异常: {e}") # 启动窗口管理器（icewm 应已安装） await self._start_window_manager() # 启动 x11vnc vnc_ok = await self._start_x11vnc() if not vnc_ok: await self.stop() return {"success": False, "message": "x11vnc 启动失败", "status": self.status} await asyncio.sleep(1) # 启动 websockify ws_ok = await self._start_websockify() if not ws_ok: await self.stop() return {"success": False, "message": "websockify 启动失败", "status": self.status} # 查找 noVNC web 文件 self._novnc_web_dir = self._find_novnc_dir() if self._novnc_web_dir: logger.info(f"noVNC web 文件: {self._novnc_web_dir}") else: logger.warning("未找到 noVNC web 文件，将使用内置的轻量 VNC 客户端") # ── Phase 3: 启动桌面应用 ── # 杀掉可能的旧文件管理器进程 try: subprocess.run(["pkill", "-f", "(pcmanfm|thunar|nautilus)"], capture_output=True, timeout=3) except Exception: pass # 启动浏览器（不在 VNC 启动时自动打开，用户可从 icewm 菜单手动启动） # self._start_browser_sync() # self._maximize_browser_window() self._running = True logger.info(f"VNC 远程桌面已完全就绪 (display={self.display}, " f"vnc_port={self.x11vnc_port}, ws_port={self.novnc_port})") return { "success": True, "message": f"远程桌面已就绪 (DISPLAY={self.display})", "status": self.status, } async def stop(self) -> Dict[str, Any]: """停止所有 VNC 服务。 Returns: {"success": bool, "message": str} """ async with self._lock: errors = [] # 按逆序停止 if self._websockify_process: errors.extend(await self._kill_process(self._websockify_process, "websockify")) self._websockify_process = None if self._x11vnc_process: errors.extend(await self._kill_process(self._x11vnc_process, "x11vnc")) self._x11vnc_process = None # [v1.23.45] 停止窗口管理器 if self._wm_process: errors.extend(await self._kill_process(self._wm_process, f"wm-{self._wm_name}")) self._wm_process = None self._wm_name = None # [v1.33.4] 停止浏览器 if self._browser_process: errors.extend(await self._kill_process(self._browser_process, "browser")) self._browser_process = None # Xvfb: 只有自己启动的才杀，附加的不杀 if self._xvfb_process: errors.extend(await self._kill_process(self._xvfb_process, "Xvfb")) self._xvfb_process = None elif self._xvfb_attached: logger.info(f"Xvfb {self.display} 是附加到已有实例，不停止 (PID={self._xvfb_attached_pid})") self._xvfb_attached = False self._xvfb_attached_pid = None # 清理锁文件和 socket (只清理自己启动的) if not self._xvfb_attached: display_num = self.display.replace(":", "") for f in [f"/tmp/.X{display_num}-lock", f"/tmp/.X11-unix/X{display_num}"]: try: if os.path.exists(f): os.remove(f) except OSError: pass self._running = False self._xvfb_attached = False if errors: return {"success": True, "message": f"VNC 已停止 (部分警告: {'; '.join(errors)})"} return {"success": True, "message": "VNC 远程桌面已停止"} async def restart(self) -> Dict[str, Any]: """重启 VNC 服务""" await self.stop() await asyncio.sleep(1) return await self.start() async def take_screenshot(self, output_path: Optional[str] = None) -> Optional[str]: """截取虚拟显示的截图。 Args: output_path: 截图保存路径，为 None 时自动生成 Returns: 截图文件路径，失败返回 None """ if not self.is_running: logger.error("VNC 未运行，无法截图") return None if not output_path: import tempfile output_path = tempfile.mktemp(suffix=".png", prefix="vnc_screenshot_") # 方法1: 使用 xwd + convert (ImageMagick) # 方法2: 使用 import (ImageMagick) # 方法3: 使用 xdpyinfo + xdotool # 方法4: 使用 ffmpeg # 最简单: xdotool + scrot 或 xwd # 尝试 xdotool 获取窗口信息，然后用 scrot 或 xwd env = {**os.environ, "DISPLAY": self.display} # 使用 xwd 转 PNG try: # 先尝试用 import (ImageMagick) if shutil.which("import"): result = subprocess.run( ["import", "-window", "root", "-display", self.display, output_path], capture_output=True, text=True, timeout=15, env=env, ) if result.returncode == 0 and os.path.isfile(output_path): logger.info(f"截图成功 (ImageMagick import): {output_path}") return output_path # 尝试 xwd + pnmtopng if shutil.which("xwd") and shutil.which("pnmtopng"): xwd_path = output_path + ".xwd" subprocess.run( ["xwd", "-root", "-display", self.display, "-out", xwd_path], capture_output=True, text=True, timeout=15, env=env, ) subprocess.run( ["xwdtopnm", xwd_path], stdout=open(output_path + ".pnm", "wb"), capture_output=True, text=True, timeout=15, ) subprocess.run( ["pnmtopng", output_path + ".pnm"], stdout=open(output_path, "wb"), capture_output=True, text=True, timeout=15, ) if os.path.isfile(output_path): logger.info(f"截图成功 (xwd): {output_path}") return output_path except Exception as e: logger.warning(f"截图异常: {e}") # 尝试 ffmpeg try: if shutil.which("ffmpeg"): result = subprocess.run( ["ffmpeg", "-f", "x11grab", "-video_size", "1920x1080", "-i", self.display, "-vframes", "1", "-y", output_path], capture_output=True, text=True, timeout=15, env=env, ) if result.returncode == 0 and os.path.isfile(output_path): logger.info(f"截图成功 (ffmpeg): {output_path}") return output_path except Exception as e: logger.warning(f"ffmpeg 截图异常: {e}") logger.error(f"截图失败: 无可用的截图工具 (需要 ImageMagick, netpbm 或 ffmpeg)") return None # ── 内部方法 ────────────────────────────────────────── async def _start_xvfb(self) -> bool: """启动或附加到 Xvfb 虚拟显示服务器""" try: display_num = self.display.replace(":", "") socket_path = f"/tmp/.X11-unix/X{display_num}" # 检查 Xvfb 是否已在运行 if os.path.exists(socket_path): try: result = subprocess.run( ["pgrep", "-f", f"Xvfb {self.display}"], capture_output=True, text=True, timeout=5, ) if result.returncode == 0 and result.stdout.strip(): # Xvfb 已在运行，附加到它 pid = int(result.stdout.strip().split()[0]) self._xvfb_attached = True self._xvfb_attached_pid = pid logger.info(f"Xvfb {self.display} 已在运行 (PID={pid})，附加到现有实例") return True else: # 锁文件/socket 存在但进程不在，清理 logger.info(f"Xvfb {self.display} socket 存在但进程不在，清理残留文件") lock_file = f"/tmp/.X{display_num}-lock" try: os.remove(lock_file) except OSError: pass try: os.remove(socket_path) except OSError: pass except Exception as e: logger.warning(f"检查 Xvfb 状态异常: {e}") cmd = [ "Xvfb", self.display, "-screen", "0", self.resolution, "-ac", # 禁用访问控制 "+extension", "GLX", "+extension", "RANDR", "+extension", "RENDER", "-nolisten", "tcp", "-dpi", "96", # [v1.46.3] 恢复96DPI（192导致Firefox UI过大） ] env = {**os.environ} # 移除可能冲突的 DISPLAY env.pop("DISPLAY", None) logger.info(f"启动 Xvfb: {' '.join(cmd)}") self._xvfb_process = subprocess.Popen( cmd, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, env=env, preexec_fn=os.setpgrp, ) # 等待并检查进程 await asyncio.sleep(0.5) if self._xvfb_process.poll() is not None: stderr = "" try: stderr = self._xvfb_process.stderr.read().decode("utf-8", errors="replace")[:500] except Exception: pass logger.error(f"Xvfb 启动失败: {stderr}") return False # 验证 X socket 是否已创建 for _ in range(10): if os.path.exists(socket_path): break await asyncio.sleep(0.3) else: logger.warning(f"Xvfb 启动但 socket {socket_path} 未创建，继续尝试...") logger.info(f"Xvfb 已启动 (PID={self._xvfb_process.pid}, display={self.display})") return True except FileNotFoundError: logger.error("Xvfb 未安装。请运行: sudo apt install xvfb") return False except Exception as e: logger.error(f"Xvfb 启动异常: {e}") return False async def _start_window_manager(self) -> None: """[v1.23.45] 自动检测并启动窗口管理器，解决 VNC 黑屏问题。 [v1.23.50] 优先 icewm（Windows风格），配合 thunar 文件管理。 [v1.34.0] 不再在关键路径中安装 WM — 如果没有已安装的 WM，用 twm 兜底， icewm 的安装放到后台线程中完成。 优先级: icewm(Windows风格) > xfce4-session > fluxbox > openbox > jwm > twm """ # 1. 检查是否已有窗口管理器在目标 DISPLAY 上运行 env = {**os.environ, "DISPLAY": self.display} try: result = subprocess.run( ["pgrep", "-f", f"(icewm|fluxbox|openbox|xfce4-session|jwm|lxsession|gnome-session|startkde|mutter|kwin)"], capture_output=True, text=True, timeout=5, ) if result.returncode == 0 and result.stdout.strip(): pids = result.stdout.strip().split() logger.info(f"检测到已有窗口管理器运行 (PIDs={pids[:3]})，跳过自动启动") self._wm_name = "existing" return except Exception: pass # 2. 按优先级尝试启动已安装的 WM（不安装新的） _wm_commands = [ ("icewm", ["icewm-session"]), ("xfce4-session", ["startxfce4"]), ("fluxbox", ["fluxbox"]), ("openbox", ["openbox", "--sm-disable"]), ("jwm", ["jwm"]), ("twm", ["twm"]), # twm 是 X11 自带的最小 WM ] wm_name = None wm_cmd = None for name, cmd in _wm_commands: if name == "icewm": if shutil.which("icewm-session") or shutil.which("icewm"): wm_name = name wm_cmd = ["icewm-session"] if shutil.which("icewm-session") else ["icewm"] break elif shutil.which(cmd[0]): wm_name = name wm_cmd = cmd break if wm_name and wm_cmd: try: if wm_name == "icewm": self._configure_icewm_theme() logger.info(f"启动窗口管理器: {wm_name} ({' '.join(wm_cmd)})") self._wm_process = subprocess.Popen( wm_cmd, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env=env, preexec_fn=os.setpgrp, ) self._wm_name = wm_name await asyncio.sleep(0.8) if self._wm_process.poll() is None: logger.info(f"窗口管理器 {wm_name} 已启动 (PID={self._wm_process.pid})") try: subprocess.run( ["xsetroot", "-solid", "#008080"], capture_output=True, timeout=5, env=env, ) except Exception: pass return else: logger.warning(f"窗口管理器 {wm_name} 启动后立即退出") except Exception as e: logger.warning(f"启动窗口管理器 {wm_name} 失败: {e}") # [v1.34.0] 没有可用的 WM → 不再阻塞安装，icewm 将在后台安装 # 设置背景色，让用户知道 VNC 是连上的 try: subprocess.run( ["xsetroot", "-solid", "#008080"], capture_output=True, timeout=5, env=env, ) except Exception: pass logger.info("未找到已安装的窗口管理器，icewm 将在后台安装后自动启动") def _configure_icewm_theme(self) -> None: """[v1.43.0] 配置 icewm 暖色调扁平主题 — Claude Code 日间模式风格。 创建自带的 "MyAgent-Warm" 主题（Look=flat），特点： - 扁平化设计，无 3D 凸起/凹陷效果 - Claude Code 日间模式暖色调（奶油白底 + 暖橙强调色） - 大标题栏、大按钮、大字体，适合移动端触摸 - 任务栏始终可见，双倍高度，方便触摸点击 - 菜单项加大，图标加大，适合手指操作 """ icewm_dir = os.path.expanduser("~/.icewm") try: os.makedirs(icewm_dir, exist_ok=True) except Exception: return # [v1.43.0] 清理旧版深色主题 MyAgent-Flat old_theme_dir = os.path.join(icewm_dir, "themes", "MyAgent-Flat") if os.path.isdir(old_theme_dir): try: shutil.rmtree(old_theme_dir) logger.info("已清理旧版 MyAgent-Flat 深色主题") except Exception: pass # ── 创建 MyAgent-Warm 自定义主题 ── # [v1.46.6] Look=warp3 经典3D渲染，标题栏按钮始终可见（有凸起3D边框） # Claude Code 日间模式暖色调: 奶油白底 + 暖橙强调 + 深棕文字 theme_name = "MyAgent-Warm" theme_dir = os.path.join(icewm_dir, "themes", theme_name) try: os.makedirs(theme_dir, exist_ok=True) except Exception: theme_dir = None if theme_dir: theme_file = os.path.join(theme_dir, "default.theme") try: theme_content = '\n'.join([ '# MyAgent-Warm — Claude Code 日间模式暖色调（移动端适配）', '# Theme for IceWM with Look=warp3, warm cream/orange, large touch targets', 'ThemeDescription="MyAgent Warm - Claude Code light theme for mobile"', 'ThemeAuthor="MyAgent"', '', '# ── 渲染模式 ──', '# [v1.47.1] Look=pixmap 使用XPM pixmap绘制按钮，确保符号可见', '# 不设Look时icewm默认渲染的符号颜色与标题栏背景相近导致不可见', 'Look=pixmap', '', '# ── 尺寸 — [v1.47.31] 精简模式：标题栏最小化 ──', 'TitleBarHeight=1', # [v1.47.31] 标题栏最小 — Firefox 最大化时几乎看不到 'TitleBarJustify=50', 'TitleBarHorzOffset=0', 'TitleBarVertOffset=0', 'BorderSizeX=0', # [v1.47.31] 精简模式 — 无边框 'BorderSizeY=0', 'DlgBorderSizeX=0', 'DlgBorderSizeY=0', 'CornerSizeX=24', 'CornerSizeY=24', 'ScrollBarX=48', 'ScrollBarY=48', 'MenuIconSize=48', 'SmallIconSize=44', 'LargeIconSize=80', 'HugeIconSize=104', '', '# ── 标题栏按钮布局 ──', 'TitleButtonsLeft="s"', 'TitleButtonsRight="xmi"', 'TitleButtonsSupported="sxmi"', 'RolloverButtonsSupported=0', # [v1.46.6] 禁用悬停才显示，按钮始终可见 'ShowMenuButtonIcon=0', # [v1.47.3] 禁用图标，只显示文字"开始" '', '# ── 任务栏开始菜单按钮 ──', '# [v1.47.3] 开始菜单按钮：使用ASCII空格加宽 + 中文字符', '# icewm的MenuButtonName用空格填充宽度', 'MenuButtonName=" 开 始 "', '', '# ── 字体 — 超大字号 2x 适合移动端触摸 ──', 'TitleFontNameXft="sans-serif:size=30:bold"', 'MenuFontNameXft="sans-serif:size=28:bold"', 'StatusFontNameXft="monospace:size=26:bold"', 'QuickSwitchFontNameXft="monospace:size=28:bold"', 'NormalButtonFontNameXft="sans-serif:size=28"', 'ActiveButtonFontNameXft="sans-serif:size=28:bold"', 'NormalTaskBarFontNameXft="sans-serif:size=22"', 'ActiveTaskBarFontNameXft="sans-serif:size=22:bold"', # [v1.47.3] 任务栏字体缩小(22)，开始按钮更大更醒目 'MinimizedWindowFontNameXft="sans-serif:size=26"', 'ListBoxFontNameXft="sans-serif:size=28"', 'ToolTipFontNameXft="sans-serif:size=26"', 'ClockFontNameXft="sans-serif:size=14"', # [v1.46.3] 时钟字体缩小 'TempFontNameXft="monospace:size=26"', 'ApmFontNameXft="monospace:size=26"', 'InputFontNameXft="monospace:size=28"', 'LabelFontNameXft="sans-serif:size=28"', 'ToolButtonFontNameXft="sans-serif:size=26:bold"', # [v1.47.3] 开始按钮字体：bold醒目 'NormalWorkspaceFontNameXft="sans-serif:size=28"', 'ActiveWorkspaceFontNameXft="sans-serif:size=28:bold"', '', '# ── 颜色 — Claude Code 日间模式暖色调 ──', '# 主色调: 暖橙 #D97757 → 活跃窗口标题栏/强调色', '# 背景色: 奶油白 #FAF7F2 / 暖灰白 #F0EBE3 / 浅米 #E8E0D5', '# 文字色: 深棕 #3D3229 / 暖灰 #6B5E52', '# 边框色: 暖浅灰 #D4CCC2', '', '# 窗口边框', 'ColorActiveBorder="rgb:D9/77/57"', 'ColorNormalBorder="rgb:D4/CC/C2"', '', '# 标题栏 — 活跃窗口用暖橙，普通窗口用浅米', 'ColorActiveTitleBar="rgb:D9/77/57"', 'ColorActiveTitleBarText="rgb:FF/FF/FF"', 'ColorActiveTitleBarShadow=""', 'ColorNormalTitleBar="rgb:E8/E0/D5"', 'ColorNormalTitleBarText="rgb:3D/32/29"', 'ColorNormalTitleBarShadow=""', '', '# 标题栏按钮 — [v1.46.9] 高对比度按钮，确保符号可见', '# 正常窗口：深色按钮背景 + 白色符号（与浅米标题栏强烈对比）', 'ColorNormalTitleButton="rgb:3D/32/29"', 'ColorNormalTitleButtonText="rgb:FF/FF/FF"', '# 活跃窗口：深橙按钮背景 + 白色符号', 'ColorActiveTitleButton="rgb:C0/5E/3C"', 'ColorActiveTitleButtonText="rgb:FF/FF/FF"', '', '# 按钮', 'ColorNormalButton="rgb:F0/EB/E3"', 'ColorNormalButtonText="rgb:3D/32/29"', 'ColorActiveButton="rgb:D9/77/57"', 'ColorActiveButtonText="rgb:FF/FF/FF"', 'ColorButtonPixmap=""', '', '# 对话框', 'ColorDialog="rgb:FA/F7/F2"', '', '# 菜单 — 浅色底 + 深色字 + 暖橙选中', 'ColorNormalMenu="rgb:FA/F7/F2"', 'ColorNormalMenuItemText="rgb:3D/32/29"', 'ColorActiveMenuItem="rgb:D9/77/57"', 'ColorActiveMenuItemText="rgb:FF/FF/FF"', 'ColorDisabledMenuItemText="rgb:B0/A8/9E"', 'ColorDisabledMenuItemShadow=""', '', '# 任务栏 — 暖白底 + 深棕文字', 'ColorDefaultTaskBar="rgb:F0/EB/E3"', 'ColorNormalTaskBarApp="rgb:E8/E0/D5"', 'ColorNormalTaskBarAppText="rgb:3D/32/29"', 'ColorActiveTaskBarApp="rgb:D9/77/57"', 'ColorActiveTaskBarAppText="rgb:FF/FF/FF"', 'ColorMinimizedTaskBarApp="rgb:E0/D8/CC"', 'ColorMinimizedTaskBarAppText="rgb:6B/5E/52"', 'ColorInvisibleTaskBarApp="rgb:F0/EB/E3"', 'ColorInvisibleTaskBarAppText="rgb:B0/A8/9E"', '', '# [v1.47.3] 开始按钮专用颜色 — 暖橙醒目', 'ColorToolButton="rgb:D9/77/57"', 'ColorToolButtonText="rgb:FF/FF/FF"', 'ColorActiveToolButton="rgb:C0/5E/3C"', 'ColorActiveToolButtonText="rgb:FF/FF/FF"', '', '# 时钟', 'ColorClock="rgb:F0/EB/E3"', 'ColorClockText="rgb:3D/32/29"', '', '# 滚动条', 'ColorScrollBar="rgb:F0/EB/E3"', 'ColorScrollBarSlider="rgb:D4/CC/C2"', 'ColorScrollBarButton="rgb:E8/E0/D5"', 'ColorScrollBarButtonArrow="rgb:3D/32/29"', 'ColorScrollBarInactiveArrow="rgb:B0/A8/9E"', '', '# 列表', 'ColorListBox="rgb:FA/F7/F2"', 'ColorListBoxText="rgb:3D/32/29"', 'ColorListBoxSelection="rgb:D9/77/57"', 'ColorListBoxSelectionText="rgb:FF/FF/FF"', '', '# 工具提示', 'ColorToolTip="rgb:3D/32/29"', 'ColorToolTipText="rgb:FA/F7/F2"', '', '# 输入框', 'ColorInput="rgb:FF/FF/FF"', 'ColorInputText="rgb:3D/32/29"', 'ColorInputSelection="rgb:D9/77/57"', 'ColorInputSelectionText="rgb:FF/FF/FF"', '', '# 标签', 'ColorLabel="rgb:FA/F7/F2"', 'ColorLabelText="rgb:3D/32/29"', '', '# Alt+Tab 切换器', 'ColorQuickSwitch="rgb:FA/F7/F2"', 'ColorQuickSwitchText="rgb:3D/32/29"', 'ColorQuickSwitchActive="rgb:D9/77/57"', '', '# 移动/缩放状态', 'ColorMoveSizeStatus="rgb:D9/77/57"', 'ColorMoveSizeStatusText="rgb:FF/FF/FF"', '', '# CPU/MEM/网络 状态色 — 暖色系', 'ColorCPUStatusUser="rgb:D9/77/57"', 'ColorCPUStatusSystem="rgb:C0/5E/3C"', 'ColorCPUStatusInterrupts="rgb:E8/A8/6E"', 'ColorCPUStatusIoWait="rgb:8B/6B/4E"', 'ColorCPUStatusSoftIrq="rgb:B8/86/5E"', 'ColorCPUStatusNice="rgb:6B/8E/6B"', 'ColorCPUStatusIdle="rgb:F0/EB/E3"', 'ColorCPUStatusSteal="rgb:A0/70/50"', 'ColorCPUStatusTemp="rgb:9B/7B/5E"', 'ColorMEMStatusUser="rgb:D9/77/57"', 'ColorMEMStatusBuffers="rgb:B8/86/5E"', 'ColorMEMStatusCached="rgb:E8/A8/6E"', 'ColorMEMStatusFree="rgb:F0/EB/E3"', 'ColorNetSend="rgb:E8/A8/6E"', 'ColorNetReceive="rgb:C0/5E/3C"', 'ColorNetIdle="rgb:F0/EB/E3"', '', '# 电池', 'ColorApm="rgb:F0/EB/E3"', 'ColorApmText="rgb:3D/32/29"', 'ColorApmBattary="rgb:E8/A8/6E"', 'ColorApmLine="rgb:6B/8E/6B"', 'ColorApmGraphBg="rgb:F0/EB/E3"', '', '# 任务栏时钟用字体（非 LCD）', 'TaskBarClockLeds=0', ]) with open(theme_file, "w", encoding="utf-8") as f: f.write(theme_content + "\n") logger.info(f"MyAgent-Warm 主题已创建: {theme_dir}") except Exception as e: logger.warning(f"MyAgent-Warm 主题创建失败: {e}") # [v1.47.1] 创建标题栏按钮XPM pixmap文件 # Look=pixmap模式需要XPM文件来绘制按钮符号 # 之前的XPM因为格式错误被删除（渲染成黑色），现在用正确格式重建 # 关键：XPM颜色必须用 "c #RRGGBB" 格式，不能用 "s name" 符号色 if theme_dir: self._create_titlebar_button_pixmaps(theme_dir) # ── 检查是否有可用的系统主题 ── available_themes = [] for td in [ "/usr/share/icewm/themes", "/usr/local/share/icewm/themes", os.path.join(icewm_dir, "themes"), ]: if os.path.isdir(td): for d in os.listdir(td): if os.path.isdir(os.path.join(td, d)): available_themes.append(d) # 优先使用 MyAgent-Warm 主题 selected_theme = None if theme_name in available_themes or os.path.isdir(theme_dir or ""): selected_theme = theme_name else: # 回退：优先选择明亮风格主题 preferred = [ "Ypsilon", "NanoBlue", "blue", "Blue", "crystal", "Crystal", "default", "Default", ] for pref in preferred: if pref in available_themes: selected_theme = pref break if not selected_theme and available_themes: selected_theme = available_themes[0] # ── 写入 icewm preferences ── # [v1.45.0] Claude Code 暖色调 + 移动端全面优化 prefs_path = os.path.join(icewm_dir, "preferences") try: config_lines = [ "# MyAgent icewm 配置 — Claude Code 暖色调 + 移动端全面优化", f'Theme="{selected_theme}"' if selected_theme else 'Theme="default/default.theme"', "", "# ── 渲染模式 ──", "# [v1.47.1] Look=pixmap 使用XPM绘制按钮符号，确保可见", "Look=pixmap", "", "# ── 任务栏 — [v1.47.31] 精简模式：隐藏任务栏 ──", "ShowTaskBar=0", # [v1.47.31] 隐藏任务栏 — Firefox 全屏独占 "TaskBarAutoHide=0", "", "# ── 桌面 — 纯黑背景（Firefox 全屏时不可见）──", 'DesktopBackgroundColor="#000000"', 'DesktopBackgroundImage=""', "DesktopCenterImage=0", "", "# ── 窗口行为 — 移动端优化 ──", "ClickToFocus=1", "FocusOnMap=1", "RaiseOnFocus=1", "MoveMouseResistance=1", "", "# ── 菜单 — 扁平大图标适合触摸 ──", "ShowMenuButtonIcon=0", # [v1.47.3] 禁用图标，只显示文字"开始" 'MenuButtonName=" 开 始 "', # [v1.47.3] 开始菜单按钮更宽（更多空格填充） "", "# ── 边缘切换工作区（移动端禁用）──", "EdgeSwitch=0", "", "# ── 窗口移动/缩放 ──", "OpaqueMove=1", "OpaqueResize=1", "", "# ── 快捷键 ──", "Win95Keys=1", "ModSuperIsCtrlAlt=0", ] with open(prefs_path, "w", encoding="utf-8") as f: f.write("\n".join(config_lines) + "\n") logger.info(f"icewm 配置已写入: theme={selected_theme}, {prefs_path}") except Exception as e: logger.warning(f"icewm 配置写入失败: {e}") # ── [v1.47.32] icewm winoptions — Firefox 无边框无标题栏 ── winopts_path = os.path.join(icewm_dir, "winoptions") try: winopts_lines = [ "# MyAgent icewm winoptions — Firefox 隐藏标题栏", "# Firefox 主窗口 — 无标题栏、无边框、无按钮、不可最小化", "firefox.Navigator.nonICCCMconfigureRequests: 1", "firefox.Navigator.ignoreWinList: 0", "firefox.Navigator.ignoreTaskBar: 0", "firefox.Navigator.ignoreQuickSwitch: 0", "firefox.Navigator.dTitleBar: 0", "firefox.Navigator.dBorder: 0", "firefox.Navigator.dResize: 0", "firefox.Navigator.dClose: 0", "firefox.Navigator.dMinimize: 0", "firefox.Navigator.dMaximize: 0", "firefox.Navigator.dSysMenu: 0", "firefox.Navigator.noFocusOnMapRaise: 0", "", "# Chromium/Chrome — 同样无装饰", "chromium-browser.Chromium-browser.dTitleBar: 0", "chromium-browser.Chromium-browser.dBorder: 0", "chromium-browser.Chromium-browser.dResize: 0", "chromium-browser.Chromium-browser.dClose: 0", "chromium-browser.Chromium-browser.dMinimize: 0", "chromium-browser.Chromium-browser.dMaximize: 0", "", "google-chrome.Google-chrome.dTitleBar: 0", "google-chrome.Google-chrome.dBorder: 0", "google-chrome.Google-chrome.dResize: 0", "google-chrome.Google-chrome.dClose: 0", "google-chrome.Google-chrome.dMinimize: 0", "google-chrome.Google-chrome.dMaximize: 0", ] with open(winopts_path, "w", encoding="utf-8") as f: f.write("\n".join(winopts_lines) + "\n") logger.info(f"icewm winoptions 已写入: {winopts_path}") except Exception as e: logger.warning(f"icewm winoptions 写入失败: {e}") # ── [v1.45.0] Xresources — DPI 设置 ── xresources_path = os.path.expanduser("~/.Xresources") try: xr_lines = [ "! MyAgent Xresources — DPI 设置", "Xft.dpi: 96", # [v1.46.3] 恢复96（192导致Firefox标签页过大） "Xft.autohint: 1", "Xft.antialias: 1", "Xft.hinting: 1", "Xft.hintstyle: hintfull", "Xft.rgba: rgb", "Xft.lcdfilter: lcddefault", ] with open(xresources_path, "w", encoding="utf-8") as f: f.write("\n".join(xr_lines) + "\n") logger.info(f"Xresources 已写入: {xresources_path}") except Exception as e: logger.warning(f"Xresources 写入失败: {e}") # ── [v1.45.0] GTK3 设置 — 大字体 + 大图标 ── gtk3_dir = os.path.expanduser("~/.config/gtk-3.0") try: os.makedirs(gtk3_dir, exist_ok=True) gtk3_settings = os.path.join(gtk3_dir, "settings.ini") gtk3_lines = [ "[Settings]", "# MyAgent GTK3 设置 — 移动端大字体大图标", "gtk-font-name=sans-serif 20", "gtk-icon-sizes=gtk-menu=48,48:gtk-button=48,48:gtk-small-toolbar=48,48:gtk-large-toolbar=64,64:gtk-dialog=64,64:gtk-dnd=64,64", "gtk-toolbar-style=GTK_TOOLBAR_ICONS", "gtk-toolbar-icon-size=GTK_ICON_SIZE_LARGE_TOOLBAR", "gtk-button-images=1", "gtk-menu-images=1", "gtk-enable-animations=0", "gtk-primary-button-warps-slider=1", "gtk-application-prefer-dark-theme=0", ] with open(gtk3_settings, "w", encoding="utf-8") as f: f.write("\n".join(gtk3_lines) + "\n") logger.info(f"GTK3 设置已写入: {gtk3_settings}") except Exception as e: logger.warning(f"GTK3 设置写入失败: {e}") # ── [v1.45.0] GTK4 设置 — 同 GTK3 ── gtk4_dir = os.path.expanduser("~/.config/gtk-4.0") try: os.makedirs(gtk4_dir, exist_ok=True) gtk4_settings = os.path.join(gtk4_dir, "settings.ini") gtk4_lines = [ "[Settings]", "# MyAgent GTK4 设置 — 移动端大字体", "gtk-font-name=sans-serif 20", "gtk-enable-animations=0", "gtk-application-prefer-dark-theme=0", ] with open(gtk4_settings, "w", encoding="utf-8") as f: f.write("\n".join(gtk4_lines) + "\n") logger.info(f"GTK4 设置已写入: {gtk4_settings}") except Exception as e: logger.warning(f"GTK4 设置写入失败: {e}") # [v1.47.31] 精简模式 — 不配置 Thunar/XFCE（不装文件管理器） # [v1.47.35] 不配置 Firefox — 使用默认设置 # 写入 icewm 菜单 — [v1.47.31] 精简模式 menu_path = os.path.join(icewm_dir, "menu") try: menu_lines = ['# MyAgent icewm 菜单 — 精简模式'] # 浏览器 if shutil.which('myagent-browser'): menu_lines.append('prog "浏览器" web-browser myagent-browser') elif shutil.which('firefox'): menu_lines.append('prog "Firefox" web-browser firefox') menu_lines.extend([ 'prog "终端" terminal xterm', 'separator', 'prog "截图" camera import -window root -display :99', ]) with open(menu_path, "w", encoding="utf-8") as f: f.write("\n".join(menu_lines) + "\n") logger.info(f"icewm 菜单已写入: {menu_path}") except Exception as e: logger.warning(f"icewm 菜单写入失败: {e}") # 写入 icewm 启动脚本 — 自动启动文件管理器和 D-Bus startup_path = os.path.join(icewm_dir, "startup") try: startup_lines = [ '#!/bin/bash', '# MyAgent icewm 启动脚本', '', '# [v1.38.0] 确保 DISPLAY 已设置（D-Bus autolaunch 需要）', 'if [ -z "$DISPLAY" ]; then', ' for xsocket in /tmp/.X11-unix/X*; do', ' if [ -S "$xsocket" ]; then', ' export DISPLAY=$(basename "$xsocket" | sed "s/X/:/")', ' break', ' fi', ' done', ' [ -z "$DISPLAY" ] && export DISPLAY=":99"', 'fi', '', '# XDG 运行时目录', 'export XDG_RUNTIME_DIR="${XDG_RUNTIME_DIR:-/run/user/$(id -u)}"', 'mkdir -p "$XDG_RUNTIME_DIR" 2>/dev/null', '', '# [v1.47.33] 中文 locale — 确保Firefox等程序正确显示中文', 'if [ -z "$LANG" ] || [ "$LANG" = "C" ]; then', ' export LANG="zh_CN.UTF-8"', ' export LANGUAGE="zh_CN:zh"', ' export LC_ALL="zh_CN.UTF-8"', 'fi', '', '# [v1.46.6] 时区跟随手机系统 — 从Android属性或/etc/timezone读取', 'if [ -z "$TZ" ]; then', ' # 优先从Android系统属性获取时区（Termux proot环境）', ' _android_tz=$(getprop persist.sys.timezone 2>/dev/null)', ' if [ -n "$_android_tz" ]; then', ' export TZ="$_android_tz"', ' elif [ -f /etc/timezone ]; then', ' export TZ=$(cat /etc/timezone 2>/dev/null)', ' fi', 'fi', '', '# [v1.46.3] 不设置 GDK_SCALE/GDK_DPI_SCALE/QT_SCALE_FACTOR', '# 这些缩放环境变量会导致 Firefox UI 过大', '', '# 加载 Xresources（DPI 设置）', 'if [ -f "$HOME/.Xresources" ]; then', ' xrdb -merge "$HOME/.Xresources" 2>/dev/null || true', 'fi', '', '# [v1.47.41] D-Bus 在 proot 下无法启动，跳过', '# 非 proot 环境用 dbus-launch 标准方式', 'if [ -z "$DBUS_SESSION_BUS_ADDRESS" ]; then', ' IS_PROOT=0', ' if [ -f /proc/version ] && grep -qi "proot" /proc/version 2>/dev/null; then', ' IS_PROOT=1', ' elif [ -d /system/bin ] && [ -f /system/bin/toybox ]; then', ' IS_PROOT=1', ' fi', ' if [ "$IS_PROOT" = "1" ]; then', ' # proot: dbus-daemon 始终无法启动，跳过（浏览器无需 D-Bus）', ' :', ' elif [ -x /usr/bin/dbus-launch ]; then', ' # 非 proot: 标准方式', ' eval $(dbus-launch --sh-syntax 2>/dev/null) || true', ' export DBUS_SESSION_BUS_ADDRESS', ' fi', 'fi', '', '# GSettings 内存后端', 'export GSETTINGS_BACKEND=memory', '', '# [v1.47.0] proot 兼容环境变量', '# G_SLICE=always-malloc — 禁用 glib2 共享内存分配器，避免 --shm-helper 错误', 'export G_SLICE=always-malloc', '# G_DEBUG=fatal-criticals — 让 glib 在关键错误时崩溃而不是继续（调试用）', '# export G_DEBUG=fatal-criticals', '', '# [v1.39.0] WebKit/Eiphany 已移除 — proot 下 bwrap 不可用', '', '# [v1.47.31] 精简模式 — 不启动文件管理器/桌面图标', '', '# [v1.47.34] 精简模式 — 自动启动浏览器（不使用--kiosk，保留地址栏）', '# 延迟3秒启动，等待 icewm 完全初始化', '(sleep 3 && {', ' if which myagent-browser 2>/dev/null; then', ' myagent-browser &', ' elif which firefox 2>/dev/null; then', ' firefox &', ' fi', '}) &', '', '# [v1.44.0] 浏览器由后台自动安装（apt install firefox）', '', '# 启动 polkit 认证代理（可选）', '[ -x /usr/lib/policykit-1-gnome/polkit-gnome-authentication-agent-1 ] && \\', ' /usr/lib/policykit-1-gnome/polkit-gnome-authentication-agent-1 &', ] with open(startup_path, "w", encoding="utf-8") as f: f.write("\n".join(startup_lines) + "\n") os.chmod(startup_path, 0o755) logger.info(f"icewm 启动脚本已写入: {startup_path}") except Exception as e: logger.warning(f"icewm 启动脚本写入失败: {e}") # ── [v1.47.0] icewm 工具栏 — 任务栏快速启动按钮 ── # [v1.47.0] 修复黑色图标问题：使用真实图标文件路径，而非freedesktop图标名 # freedesktop图标名在最小安装下可能无法解析，导致显示黑色方块 toolbar_path = os.path.join(icewm_dir, "toolbar") try: toolbar_lines = [ '# MyAgent icewm 工具栏 — 快速启动按钮', ] # [v1.47.0] 辅助函数：查找真实图标文件路径 def _find_toolbar_icon(icon_names, search_subdirs=None): """查找图标文件路径，优先在指定子目录中查找""" if search_subdirs is None: search_subdirs = ["128x128/apps", "64x64/apps", "48x48/apps", ""] for icon_name in icon_names: for sd in search_subdirs: for base in ["/usr/share/icons/hicolor", "/usr/share/pixmaps"]: if sd: path = os.path.join(base, sd, f"{icon_name}.png") else: path = os.path.join(base, f"{icon_name}.png") if os.path.isfile(path): return path return None # [v1.47.31] 精简模式 — 工具栏只保留浏览器按钮 # 浏览器按钮 browser_cmd = (shutil.which("myagent-browser") or shutil.which("firefox") or None) if browser_cmd: br_icon = _find_toolbar_icon(["firefox", "firefox-esr", "web-browser"]) br_icon_str = br_icon if br_icon else "folder" toolbar_lines.append(f'prog "浏览器" {br_icon_str} {browser_cmd}') # 虚拟键盘按钮 if shutil.which("onboard"): kb_icon = _find_toolbar_icon(["onboard", "input-keyboard"]) kb_icon_str = kb_icon if kb_icon else "folder" toolbar_lines.append(f'prog "键盘" {kb_icon_str} onboard') # 终端按钮 if shutil.which("xterm"): term_icon = _find_toolbar_icon(["xterm", "utilities-terminal"]) term_icon_str = term_icon if term_icon else "folder" toolbar_lines.append(f'prog "终端" {term_icon_str} xterm') with open(toolbar_path, "w", encoding="utf-8") as f: f.write("\n".join(toolbar_lines) + "\n") logger.info(f"icewm 工具栏已写入: {toolbar_path}") except Exception as e: logger.warning(f"icewm 工具栏写入失败: {e}") # [v1.47.34] 精简模式 — 不配置桌面图标/pcmanfm/onboard（代码已删除） def _is_proot_env(self) -> bool: """检测是否在 proot 环境下运行（Android Termux 等）""" # 方法1: 检查 /proc/version 是否包含 proot try: with open("/proc/version", "r") as f: if "proot" in f.read().lower(): return True except Exception: pass # 方法2: 检查 Android 特征 if os.path.exists("/system/bin/app_process") or os.path.exists("/system/build.prop"): return True # 方法3: 检查 LD_PRELOAD 中有 proot ld_preload = os.environ.get("LD_PRELOAD", "") if "proot" in ld_preload.lower(): return True return False @staticmethod def _detect_proot_environment() -> bool: """[v1.39.1] 检测是否运行在 proot 环境中。 proot 环境特征： - /proc/self/maps 包含 'proot' 字样 - 无法读取 /proc/sys/kernel/overflowuid（bwrap 需要此文件） - 存在 PROOT_TMP 或 PROOT_DEBUG 环境变量 - Android + Linux 混合环境（Termux proot-distro） proot 下的限制：snap/bwrap 不可用，apt install firefox 会装成 snap 包装器。 """ # 方法1: 检查 /proc/self/maps try: with open("/proc/self/maps", "r") as f: maps_content = f.read(4096) if "proot" in maps_content.lower(): return True except Exception: pass # 方法2: 检查 proot 特有的环境变量 for env_var in ["PROOT_TMP", "PROOT_DEBUG", "PROOT_NO_SECCOMP"]: if os.environ.get(env_var): return True # 方法3: 检查 /proc/sys/kernel/overflowuid 是否可读 # bwrap 需要读取此文件，proot 下无法读取 try: with open("/proc/sys/kernel/overflowuid", "r") as f: f.read() except (PermissionError, FileNotFoundError, OSError): # 无法读取 = 很可能在 proot 中 return True except Exception: pass # 方法4: 检查 Android 特征（Termux proot-distro） try: if os.path.isfile("/system/bin/toybox") or os.path.isfile("/system/bin/app_process"): return True except Exception: pass # 方法5: 检查 uname 是否包含 Android try: result = subprocess.run( ["uname", "-r"], capture_output=True, text=True, timeout=3, ) if "android" in result.stdout.lower() or "termux" in result.stdout.lower(): return True except Exception: pass return False @staticmethod def _is_snap_wrapper(binary_path: str) -> bool: """[v1.23.50] 检测二进制是否为 snap 包装器脚本。 Ubuntu 24.04+ 的 chromium-browser 只是一个 snap 引导脚本， 在 proot 环境下 snap 不可用，执行会报： Command '/usr/bin/chromium-browser' requires the chromium snap to be installed. 检测方法：读取文件内容，如果包含 'snap' 关键字则为包装器。 """ try: # 检查是否为脚本文件（而非真正的 ELF 二进制） with open(binary_path, "r", errors="replace") as f: head = f.read(2048) # snap 包装器脚本的特征 snap_keywords = ["snap", "SNAP", "snapd"] if any(kw in head for kw in snap_keywords): return True # 也检查 shebang 是不是 shell 脚本但文件名含 chromium if head.startswith("#!") and "chromium" in binary_path.lower(): return True except (UnicodeDecodeError, PermissionError, OSError): # 二进制文件读不了文本 = 不是脚本 = 真正的浏览器 return False return False def _install_firefox_proot(self, sudo: list, arch: str = "arm64") -> Optional[str]: """下载 Mozilla Firefox 预编译二进制文件安装到 /opt/firefox/。 直接从 download.mozilla.org 下载官方预编译二进制， 解压到 /opt/firefox/，创建 /usr/local/bin/firefox 符号链接。 Returns: Firefox 二进制路径，失败返回 None """ import tempfile firefox_install_dir = "/opt/firefox" firefox_bin = os.path.join(firefox_install_dir, "firefox") link_path = "/usr/local/bin/firefox" # ── Step 0: 已存在则直接使用 ── if os.path.isfile(firefox_bin) and os.access(firefox_bin, os.X_OK): logger.info(f"Firefox 已存在: {firefox_bin}") if not os.path.exists(link_path): try: os.symlink(firefox_bin, link_path) except Exception: pass return firefox_bin # ── Step 1: 移除 snap 版 Firefox ── snap_firefox = shutil.which("firefox") if snap_firefox and self._is_snap_wrapper(snap_firefox): logger.info("[1/4] 移除 snap 版 Firefox...") subprocess.run( sudo + ["apt-get", "remove", "-y", "firefox"], capture_output=True, text=True, timeout=60, start_new_session=True, ) else: logger.info("[1/4] 无需移除旧版 Firefox") # ── Step 2: 安装依赖库 ── logger.info("[2/4] 安装 Firefox 依赖库...") subprocess.run( sudo + ["apt-get", "install", "-y", "--no-install-recommends", "libgtk-3-0", "libdbus-glib-1-2", "libxt6", "libasound2", "libatk1.0-0", "libatk-bridge2.0-0", "libcups2", "libxdamage1", "libxrandr2", "libxcomposite1", "libpango-1.0-0", "libcairo2", "libx11-xcb1"], capture_output=True, text=True, timeout=120, start_new_session=True, ) # ── Step 3: 下载 Firefox ── # 构建下载 URL: product=firefox-latest, lang=zh-CN if arch in ("arm64", "aarch64"): download_url = "https://download.mozilla.org/?product=firefox-latest&os=linux64-aarch64&lang=zh-CN" else: download_url = "https://download.mozilla.org/?product=firefox-latest&os=linux64&lang=zh-CN" tar_path = os.path.join(tempfile.gettempdir(), "firefox.tar.bz2") download_ok = False # 清理之前的下载残留 if os.path.exists(tar_path): try: os.remove(tar_path) except Exception: pass logger.info(f"[3/4] 下载 Firefox: {download_url[:70]}...") for dl_cmd in ["curl", "wget"]: if not shutil.which(dl_cmd): continue try: if dl_cmd == "wget": subprocess.run( ["wget", "-L", "--timeout=60", "--tries=2", "-O", tar_path, download_url], capture_output=True, text=True, timeout=600, start_new_session=True, ) else: subprocess.run( ["curl", "-L", "--max-time", "300", "--retry", "2", "-o", tar_path, download_url], capture_output=True, text=True, timeout=600, start_new_session=True, ) if not os.path.exists(tar_path): continue file_size = os.path.getsize(tar_path) # 验证：有效压缩包至少 5MB（xz 或 bzip2） # xz 魔术字节: \xfd7zXZ\x00 # bzip2 魔术字节: BZ is_valid = False try: with open(tar_path, "rb") as f: magic = f.read(6) is_valid = (magic[:2] == b'BZ' or magic[:6] == b'\xfd7zXZ\x00') except Exception: pass if file_size > 5_000_000 and is_valid: download_ok = True break else: logger.info(f"下载文件无效 (size={file_size}, valid={is_valid})") try: os.remove(tar_path) except Exception: pass except subprocess.TimeoutExpired: logger.warning(f"{dl_cmd} 下载超时") except Exception as e: logger.debug(f"{dl_cmd} 下载异常: {e}") if not download_ok: logger.warning("Firefox 下载失败") return None # ── Step 4: 解压并安装 ── logger.info("[4/4] 解压 Firefox 到 /opt/firefox/...") # 移除旧安装 if os.path.isdir(firefox_install_dir): try: shutil.rmtree(firefox_install_dir) except Exception: pass # tar -xf 自动检测压缩格式（xz/bzip2/gzip） result = subprocess.run( ["tar", "xf", tar_path, "-C", "/opt/"], capture_output=True, text=True, timeout=120, start_new_session=True, ) # 清理下载文件 try: os.remove(tar_path) except Exception: pass if result.returncode != 0: logger.warning(f"Firefox 解压失败: {result.stderr[:200]}") return None # 验证并创建符号链接 if os.path.isfile(firefox_bin) and os.access(firefox_bin, os.X_OK): try: if os.path.exists(link_path) or os.islink(link_path): os.remove(link_path) os.symlink(firefox_bin, link_path) except Exception: pass logger.info(f"Firefox 安装成功: {firefox_bin}") return firefox_bin logger.warning("Firefox 解压后未找到可执行文件") return None def _create_browser_wrapper(self, browser_binary: str) -> None: """[v1.38.0] 创建 /usr/local/bin/myagent-browser 启动包装脚本。 [v1.47.41] 简化：只支持 Firefox（proot 下唯一可靠的浏览器）。 """ wrapper_path = "/usr/local/bin/myagent-browser" wrapper_content = '''#!/bin/bash # MyAgent Browser Launcher — Firefox proot/Termux 兼容 BROWSER="{browser}" # [v1.47.42] 增大栈大小：proot 下 Firefox 需要更大的栈空间，默认 8MB 可能崩溃 ulimit -s 32768 2>/dev/null # 确保 DISPLAY 已设置 export DISPLAY="${{DISPLAY:-:99}}" # XDG 运行时目录 export XDG_RUNTIME_DIR="${{XDG_RUNTIME_DIR:-/run/user/$(id -u)}}" mkdir -p "$XDG_RUNTIME_DIR" 2>/dev/null chmod 700 "$XDG_RUNTIME_DIR" 2>/dev/null # [v1.47.38] MOZ_FAKE_NO_SANDBOX=1: 跳过沙盒检测，不弹"安全沙盒已禁用"警告 export MOZ_FAKE_NO_SANDBOX=1 export MOZ_DISABLE_GPU_SANDBOX=1 # [v1.47.12] proot 兼容环境变量 export G_SLICE=always-malloc export GSETTINGS_BACKEND=memory # [v1.47.41] D-Bus 在 proot 下无法启动，跳过 # 非 proot 环境用 dbus-launch 标准方式 if [ -z "$DBUS_SESSION_BUS_ADDRESS" ]; then IS_PROOT=0 grep -qi "proot" /proc/version 2>/dev/null && IS_PROOT=1 if [ "$IS_PROOT" = "1" ]; then # proot: dbus-daemon 始终无法启动，跳过（浏览器无需 D-Bus） : elif [ -x /usr/bin/dbus-launch ]; then eval $(dbus-launch --sh-syntax 2>/dev/null) || true export DBUS_SESSION_BUS_ADDRESS fi fi exec "$BROWSER" "$@" '''.format(browser=browser_binary) try: with open(wrapper_path, "w") as f: f.write(wrapper_content) os.chmod(wrapper_path, 0o755) logger.info(f"浏览器启动包装脚本已创建: {wrapper_path} -> {browser_binary}") except Exception as e: logger.warning(f"创建浏览器包装脚本失败: {e}") async def _start_x11vnc(self) -> bool: """启动 x11vnc VNC 服务器""" try: # [v1.20.5] 增强端口清理：fuser -k + pkill -9（Termux 环境下进程管理不彻底） if self._is_port_listening(self.x11vnc_port): logger.warning(f"端口 {self.x11vnc_port} 已被占用，清理残留进程...") self._kill_port_users(self.x11vnc_port) self._kill_process_by_name("x11vnc") self._kill_process_by_name("Xvfb") await asyncio.sleep(1.0) # 二次确认 if self._is_port_listening(self.x11vnc_port): # 回退到 pgrep + kill 方式 try: _result = subprocess.run( ["pgrep", "-f", f"x11vnc.*{self.display}"], capture_output=True, text=True, timeout=5, ) if _result.returncode == 0 and _result.stdout.strip(): for _pid_str in _result.stdout.strip().split("\n"): try: os.kill(int(_pid_str.strip()), signal.SIGKILL) logger.info(f"SIGKILL 残留 x11vnc PID={_pid_str.strip()}") except (ValueError, ProcessLookupError): pass await asyncio.sleep(0.5) except Exception as _e: logger.debug(f"pgrep 清理异常: {_e}") if self._is_port_listening(self.x11vnc_port): logger.warning(f"端口 {self.x11vnc_port} 仍被占用，可能被其他服务占用") # [v1.20.5] x11vnc 启动参数 # 关键: -noshm 禁用共享内存 — Android/Termux proot-distro 环境下 # SysV IPC (shmget) 不受支持，不加此参数会导致 shmget failed 错误 # 注意: 参数是 -noshm（无横杠分隔），不是 -no-shm # [v1.47.36] 移除 -nowcr -nocursorshape -randr：这些参数在 Termux proot 下 # 导致 x11vnc fork 后子进程崩溃（首次启动必定失败），必须重试才能成功 # 不使用 -randr：x11vnc 的 RANDR 代理在此环境下不可靠， # 分辨率调整通过 /api/vnc/resize → xrandr --fb 或重启 VNC 实现 cmd = [ "x11vnc", "-display", self.display, "-noshm", # [v1.20.5] 禁用共享内存（Termux proot-distro 必需） "-forever", # 持续运行 "-shared", # 允许多个客户端同时连接 "-nopw", # 无密码（本地使用） "-rfbport", str(self.x11vnc_port), "-listen", "127.0.0.1", # 仅本地监听（安全） "-xkb", # 使用 XKB 扩展 "-noxdamage", # 禁用 XDAMAGE（兼容性更好） "-nowf", # 禁用 wireframe "-deferupdate", "5", # 延迟更新（降低带宽） # [v1.47.33] 移除 -scale — 使用 RandR 自适应分辨率，不需要缩放 ] # [v1.18.5] 不使用 -threads 模式：与 -scale 组合在 0.9.17 中 # 会导致父进程 fork 后立即退出，端口延迟监听 # [v1.20.5] 移除 shm-helper 逻辑 — 改用 -noshm 完全跳过共享内存 # 旧版本 (v1.17.2 ~ v1.20.4) 尝试查找 shm-helper 并传绝对路径， # 但在 Termux 环境下找不到该文件会导致 "expected absolute path" 错误 env = {**os.environ, "DISPLAY": self.display} # [v1.18.0] proot/Termux 兼容 cmd.append("-nobell") env["X11VNC_NO_UNIXPW"] = "1" logger.info(f"启动 x11vnc: {' '.join(cmd)}") self._x11vnc_process = subprocess.Popen( cmd, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, env=env, preexec_fn=os.setpgrp, ) # [v1.18.5] x11vnc 0.9.17 可能 fork 到后台，需要更长等待 # [v1.18.10] 增加首次等待时间和重试次数 await asyncio.sleep(3) if self._x11vnc_process.poll() is not None: # [v1.18.4] x11vnc 可能 fork 到后台运行，父进程退出但子进程仍在监听端口 # 等待更长时间让 fork 的子进程完成端口绑定 if self._is_port_listening(self.x11vnc_port): logger.warning(f"x11vnc 父进程已退出但端口 {self.x11vnc_port} 仍在监听（fork 到后台），视为启动成功") return True # 端口未监听，再等几秒（某些系统 fork 后子进程初始化慢） logger.info("x11vnc 端口尚未就绪，额外等待 5 秒...") await asyncio.sleep(5) if self._is_port_listening(self.x11vnc_port): logger.warning(f"x11vnc fork 延迟启动成功 (port={self.x11vnc_port})") return True # [v1.18.10] 额外重试一次：彻底杀掉残留，重新启动 x11vnc logger.warning("x11vnc 首次启动失败，尝试清理并重新启动...") await self._kill_process(self._x11vnc_process, "x11vnc") self._x11vnc_process = None await asyncio.sleep(1) # 第二次尝试启动 return await self._retry_x11vnc() # 验证端口是否在监听 for _ in range(15): try: import socket s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.settimeout(0.5) s.connect(("127.0.0.1", self.x11vnc_port)) s.close() logger.info(f"x11vnc 已启动 (PID={self._x11vnc_process.pid}, port={self.x11vnc_port})") return True except (OSError, ConnectionRefusedError): await asyncio.sleep(0.3) # 端口未就绪，但进程还活着，可能还在初始化 if self._x11vnc_process.poll() is None: logger.warning("x11vnc 进程运行中但端口尚未就绪，继续...") return True logger.error("x11vnc 端口监听超时") return False except FileNotFoundError: logger.error("x11vnc 未安装。请运行: sudo apt install x11vnc") return False except Exception as e: logger.error(f"x11vnc 启动异常: {e}") return False async def _retry_x11vnc(self) -> bool: """[v1.18.10] x11vnc 首次启动失败后重试（简化参数，增加兼容性）""" try: # 先杀掉所有残留的 x11vnc 进程 try: _result = subprocess.run( ["pgrep", "-f", f"x11vnc.*{self.display}"], capture_output=True, text=True, timeout=5, ) if _result.returncode == 0 and _result.stdout.strip(): for _pid_str in _result.stdout.strip().split("\n"): try: os.kill(int(_pid_str.strip()), signal.SIGKILL) except (ValueError, ProcessLookupError): pass await asyncio.sleep(1) except Exception: pass # 第二次尝试使用更保守的参数（去掉 -scale 等可能的问题参数，保留 -noshm） cmd = [ "x11vnc", "-display", self.display, "-noshm", # [v1.20.5] Termux proot-distro 必需 "-forever", "-shared", "-nopw", "-rfbport", str(self.x11vnc_port), "-listen", "127.0.0.1", "-xkb", "-noxdamage", "-nowf", "-deferupdate", "5", "-nobell", # [v1.47.36] 不加 -randr：与主命令一致，此参数在 Termux 下导致 x11vnc 崩溃 ] env = {**os.environ, "DISPLAY": self.display} env["X11VNC_NO_UNIXPW"] = "1" logger.info(f"x11vnc 重试启动: {' '.join(cmd)}") self._x11vnc_process = subprocess.Popen( cmd, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, env=env, preexec_fn=os.setpgrp, ) await asyncio.sleep(5) if self._is_port_listening(self.x11vnc_port): logger.info(f"x11vnc 重试启动成功 (port={self.x11vnc_port})") return True if self._x11vnc_process.poll() is not None: stderr = "" try: stderr = self._x11vnc_process.stderr.read().decode("utf-8", errors="replace")[:3000] except Exception: pass logger.error(f"x11vnc 重试启动失败: {stderr}") return False # 进程还活着但端口未就绪，再等一轮 for _ in range(15): try: import socket s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.settimeout(0.5) s.connect(("127.0.0.1", self.x11vnc_port)) s.close() logger.info(f"x11vnc 重试后启动成功 (port={self.x11vnc_port})") return True except (OSError, ConnectionRefusedError): await asyncio.sleep(0.3) logger.error("x11vnc 重试后端口仍超时，放弃") return False except Exception as e: logger.error(f"x11vnc 重试异常: {e}") return False async def _start_websockify(self) -> bool: """启动 websockify WebSocket 代理（VNC → WebSocket）""" try: # 检查 noVNC web 目录 novnc_dir = self._find_novnc_dir() cmd = [ "websockify", "--daemon", # 后台运行... 不对，Popen 自己管理 str(self.novnc_port), f"localhost:{self.x11vnc_port}", ] # 如果找到了 noVNC web 目录，也服务这些文件 if novnc_dir: cmd.extend(["--web", novnc_dir]) self._novnc_web_dir = novnc_dir # 移除 --daemon（我们用 Popen 管理进程） cmd = [c for c in cmd if c != "--daemon"] logger.info(f"启动 websockify: {' '.join(cmd)}") self._websockify_process = subprocess.Popen( cmd, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, # 合并 stderr 到 stdout preexec_fn=os.setpgrp, ) # 后台线程读取 websockify 的 stdout（避免管道阻塞） if self._websockify_process.stdout: loop = asyncio.get_event_loop() loop.run_in_executor(None, self._read_websockify_output) # 等待 WebSocket 端口就绪 await asyncio.sleep(1) if self._websockify_process.poll() is not None: logger.error(f"websockify 启动失败") return False for _ in range(10): try: import socket s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.settimeout(0.5) s.connect(("127.0.0.1", self.novnc_port)) s.close() logger.info(f"websockify 已启动 (PID={self._websockify_process.pid}, ws_port={self.novnc_port})") return True except (OSError, ConnectionRefusedError): await asyncio.sleep(0.3) if self._websockify_process.poll() is None: logger.warning("websockify 进程运行中但端口尚未就绪，继续...") return True logger.error("websockify 端口监听超时") return False except FileNotFoundError: logger.error("websockify 未安装。请运行: pip install websockify") return False except Exception as e: logger.error(f"websockify 启动异常: {e}") return False def _read_websockify_output(self): """后台线程读取 websockify 的 stdout，防止管道阻塞""" if not self._websockify_process or not self._websockify_process.stdout: return try: for line in self._websockify_process.stdout: line_str = line.decode("utf-8", errors="replace").strip() if line_str and "connecting" in line_str.lower(): logger.debug(f"websockify: {line_str[:200]}") except Exception: pass @staticmethod async def _kill_process(proc: subprocess.Popen, name: str) -> list: """终止进程（先 SIGTERM，再 SIGKILL）""" errors = [] if proc.poll() is not None: return errors # 已退出 try: # 先尝试 SIGTERM proc.terminate() try: proc.wait(timeout=5) logger.info(f"{name} 已停止 (PID={proc.pid})") except subprocess.TimeoutExpired: # SIGKILL proc.kill() proc.wait(timeout=3) logger.warning(f"{name} 已强制停止 (PID={proc.pid})") except Exception as e: errors.append(f"{name} 停止异常: {e}") try: proc.kill() except Exception: pass return errors # ── 全局单例 ────────────────────────────────────────────── _vnc_manager: Optional[VNCManager] = None def get_vnc_manager() -> VNCManager: """获取全局 VNC 管理器单例""" global _vnc_manager if _vnc_manager is None: _vnc_manager = VNCManager() return _vnc_manager