""" aiskills/browser_stealth.py - DrissionPage 反检测浏览器技能 ========================================================== 基于 DrissionPage 的反检测浏览器自动化，专用于需要登录的网站操作。 核心能力: - 反自动化检测: 自动隐藏 webdriver 标志、CDP 痕迹等 - Profile 持久化: 每个网站独立 Profile，登录一次永久保持 - Cookie 管理: 保存/恢复/清除 Cookie - 人机协作: 遇到验证码/2FA 时暂停，等待用户手动操作 - 完整操作: 导航、点击、填写、截图、执行 JS、等待元素等 与 chromedev_mcp.py 的区别: - chromedev_mcp: 标准 CDP 协议，功能全面，但会被网站检测为自动化 - browser_stealth: 反检测优先，适合登录敏感网站（Google/X/微博/抖音等） 架构: DrissionPage (Python) ──CDP(隐藏)──> Chromium/Chrome ↑ User Data Directory (Profile) Cookie 持久化 依赖: - DrissionPage >= 4.1.0 - Chromium/Chrome (系统已有，或自动检测) 参考: https://drissionpage.cn/ """ from __future__ import annotations import asyncio import json import os import shutil import subprocess import threading import time from pathlib import Path from typing import Any, Dict, List, Optional from core.logger import get_logger from aiskills.base import Skill, SkillResult, SkillParameter logger = get_logger("myagent.skills.browser_stealth") # ── Xvfb 虚拟显示管理 ────────────────────────────────────── _xvfb_process: Optional[subprocess.Popen] = None _xvfb_display: Optional[str] = None _xvfb_lock = threading.Lock() def _has_display() -> bool: """检查当前环境是否有可用的 X11 显示""" display = os.environ.get("DISPLAY", "").strip() if not display: return False # DISPLAY 已设置，尝试检测是否真的可用 # 方法1: xdpyinfo try: result = subprocess.run( ["xdpyinfo"], capture_output=True, timeout=3, env={**os.environ}, ) if result.returncode == 0: return True except Exception: pass # 方法2: xdpyinfo 不可用时，尝试用 Xvfb 的 -displayfd 验证 # 如果 DISPLAY 变量指向一个活跃的 X server，即使 xdpyinfo 不可用， # 我们也可以认为有可用显示（例如 Xvfb 已经启动过了） try: # 尝试连接 X server（使用 xdotool 或 xprop 等轻量工具） for cmd in (["xprop", "-root", "_NET_SUPPORTING_WM_CHECK"], ["xdotool", "getdisplaywidth"]): result = subprocess.run( cmd, capture_output=True, timeout=3, env={**os.environ}, ) if result.returncode == 0: return True except Exception: pass # 方法3: 如果 DISPLAY 格式正确且 /tmp/.X11-unix/ 下有对应 socket，认为可用 try: if display.startswith(":"): num = display[1:].split(".")[0] socket_path = f"/tmp/.X11-unix/X{num}" if os.path.exists(socket_path): return True except Exception: pass return False def _start_xvfb(display_num: int = 99) -> Optional[str]: """ 启动 Xvfb 虚拟显示服务器。 [v1.47.10] 增强：如果指定 display 号被占用，自动尝试其他 display 号（:100-:109）。 僵尸 Xvfb 进程（OOM Kill 后残留锁文件）是 returncode=1 的常见原因。 Returns: DISPLAY 环境变量值（如 :99），失败返回 None """ global _xvfb_process, _xvfb_display with _xvfb_lock: # 如果已经启动，直接返回 if _xvfb_process and _xvfb_process.poll() is None: return _xvfb_display # 清理旧的 if _xvfb_process: try: _xvfb_process.terminate() _xvfb_process.wait(timeout=3) except Exception: try: _xvfb_process.kill() except Exception: pass _xvfb_process = None xvfb_path = shutil.which("Xvfb") if not xvfb_path: logger.warning("Xvfb 未安装，无法启动虚拟显示") return None # [v1.47.10] 尝试多个 display 号，避免锁文件冲突 # 优先用 :99（和 VNC 一致），失败则尝试 :100-:109 display_candidates = [display_num] + list(range(100, 110)) for dn in display_candidates: display_str = f":{dn}" try: # 先清理可能残留的 X11 锁文件 for lock_file in (f"/tmp/.X{dn}-lock", f"/tmp/.X11-unix/X{dn}"): if os.path.exists(lock_file): try: os.unlink(lock_file) except Exception: pass # 检查是否有 Xvfb 进程已经在用这个 display try: result = subprocess.run( ["pgrep", "-f", f"Xvfb :{dn}"], capture_output=True, text=True, timeout=3, ) if result.returncode == 0 and result.stdout.strip(): # 已有进程在用这个 display，复用它 _xvfb_display = display_str os.environ["DISPLAY"] = display_str logger.info(f"检测到已有 Xvfb 进程使用 display :{dn}，复用") return display_str except Exception: pass _xvfb_process = subprocess.Popen( [xvfb_path, display_str, "-screen", "0", "1920x1080x24", "-ac", "-nolisten", "tcp"], stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, ) # 等待 Xvfb 启动 time.sleep(0.8) if _xvfb_process.poll() is not None: stderr = "" try: stderr = _xvfb_process.stderr.read().decode("utf-8", errors="replace")[:300] except Exception: pass logger.debug(f"Xvfb :{dn} 启动失败 (rc={_xvfb_process.returncode}): {stderr}") _xvfb_process = None # 继续尝试下一个 display 号 continue _xvfb_display = display_str os.environ["DISPLAY"] = display_str logger.info(f"Xvfb 虚拟显示已启动: {display_str}") return display_str except Exception as e: logger.debug(f"启动 Xvfb :{dn} 异常: {e}") _xvfb_process = None continue logger.error("所有 display 号均无法启动 Xvfb") return None def _stop_xvfb() -> None: """停止 Xvfb 虚拟显示服务器""" global _xvfb_process, _xvfb_display with _xvfb_lock: if _xvfb_process: try: _xvfb_process.terminate() _xvfb_process.wait(timeout=5) except Exception: try: _xvfb_process.kill() except Exception: pass _xvfb_process = None _xvfb_display = None logger.info("Xvfb 虚拟显示已停止") def _ensure_display() -> Optional[Dict[str, Any]]: """ 确保有可用的 X11 显示，供浏览器有头模式使用。 Termux+Ubuntu 环境: 只支持 VNC，无其他 fallback 非 Termux 容器环境: VNC > 独立 Xvfb 降级 Returns: {"display": ":99", "vnc": True/False, "xvfb_standalone": True/False} 失败返回 None """ # 检测是否为 Termux 环境 _is_termux = False try: from core.env_detect import is_termux _is_termux = is_termux() except ImportError: pass # 1. 尝试复用 VNC 远程桌面 try: from core.vnc_manager import get_vnc_manager vnc = get_vnc_manager() # VNC 已在运行 → 直接复用其 DISPLAY if vnc.is_running: display = vnc.display # 默认 :99 os.environ["DISPLAY"] = display logger.info(f"VNC 远程桌面已在运行，复用显示: {display}") return {"display": display, "vnc": True, "xvfb_standalone": False} # [v1.47.13] VNC 未运行 → 尝试自动启动 # 重要修复: 不能在 ThreadPoolExecutor + asyncio.run 中调用 vnc.start()！ # 因为 vnc.start() 内部用 asyncio.Lock()，跨 event loop 会导致死锁。 # 正确做法：在单独线程中用 asyncio.run 启动，然后轮询等待 VNC 就绪。 try: import asyncio import concurrent.futures # 在独立线程中启动 VNC（新的 event loop，避免冲突） def _start_vnc_in_thread(): try: return asyncio.run(vnc.start()) except Exception as e: return {"success": False, "message": str(e)} with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool: future = pool.submit(_start_vnc_in_thread) # 轮询等待 VNC 就绪（同时等待 future 完成） max_wait = 60 # 最多等60秒 check_interval = 2 elapsed = 0 while elapsed < max_wait: # 先检查 VNC 是否已就绪 if vnc.is_running: display = vnc.display os.environ["DISPLAY"] = display logger.info(f"VNC 远程桌面已启动，复用显示: {display}") return {"display": display, "vnc": True, "xvfb_standalone": False} # 检查启动线程是否已完成（非阻塞） try: result = future.result(timeout=0.1) # 线程完成了，检查结果 if result.get("success") or vnc.is_running: display = vnc.display os.environ["DISPLAY"] = display logger.info(f"VNC 远程桌面已启动，显示: {display}") return {"display": display, "vnc": True, "xvfb_standalone": False} else: logger.warning(f"VNC 启动失败: {result.get('message', '')}") break except concurrent.futures.TimeoutError: # 线程还在运行，继续等待 pass import time time.sleep(check_interval) elapsed += check_interval # 超时后再做最后检查 if vnc.is_running: display = vnc.display os.environ["DISPLAY"] = display logger.info(f"VNC 远程桌面已启动（等待后），显示: {display}") return {"display": display, "vnc": True, "xvfb_standalone": False} except Exception as e: logger.warning(f"VNC 自动启动异常: {e}") except ImportError: logger.debug("VNC 管理器不可用，跳过 VNC 集成") except Exception as e: logger.warning(f"VNC 检测异常: {e}") # ── Termux+Ubuntu: VNC 是唯一方式，不允许其他 fallback ── if _is_termux: logger.error( "Termux+Ubuntu 环境仅支持通过 VNC 启动浏览器，" "VNC 启动失败。请检查 VNC 配置或手动启动 VNC 远程桌面。" ) return None # 2. 非 Termux 容器环境 → 独立启动 Xvfb 作为 fallback xvfb_display = _start_xvfb(display_num=100) if xvfb_display: return {"display": xvfb_display, "vnc": False, "xvfb_standalone": True} return None # ── 反检测浏览器管理器 ────────────────────────────────────── class StealthBrowser: """ 基于 DrissionPage 的反检测浏览器实例。 用法: browser = StealthBrowser(profile_name="google") await browser.start() await browser.navigate("https://accounts.google.com/") await browser.fill('input[type="email"]', "user@gmail.com") await browser.click('button:contains("下一步")') screenshot = await browser.screenshot() await browser.close() """ def __init__( self, profile_name: str = "default", headless: bool = False, user_data_dir: Optional[str] = None, use_system_profile: bool = False, ): """ Args: profile_name: Profile 名称（对应 BrowserProfileManager 中的 profile） headless: 是否无头模式（登录场景建议 False） user_data_dir: 自定义用户数据目录（覆盖 Profile 管理） use_system_profile: 使用系统 Chrome 的 Profile（保留所有登录状态） macOS: ~/Library/Application Support/Google/Chrome/ Linux: ~/.config/google-chrome/ Windows: %LOCALAPPDATA%\\Google\\Chrome\\User Data\\ 注意: 如果系统 Chrome 正在运行，需要先关闭，否则 Profile 会被锁定 """ self.profile_name = profile_name self._headless = headless self._custom_user_data_dir = user_data_dir self._use_system_profile = use_system_profile self._page = None # DrissionPage ChromiumPage 实例 self._browser = None # DrissionPage Chromium 实例 self._started = False self._user_data_dir = "" self._xvfb_started_by_us = False # 是否由本实例独立启动的 Xvfb self._vnc_used = False # 是否使用了 VNC 远程桌面的显示 # [v1.47.16] Firefox+VNC 模式：VNC 环境下直接用 Firefox，不走 DrissionPage/Chromium self._firefox_mode = False self._firefox_process: Optional[subprocess.Popen] = None self._firefox_profile_dir = "" # [v1.35.0] 浏览器实例级异步锁：防止多个 Agent 并发操作同一个浏览器实例 # 同一 profile 的浏览器共享 _page，并发操作会导致竞态（导航冲突、点击错位等） self._usage_lock = asyncio.Lock() self._current_user: Optional[str] = None # 当前持有锁的 agent_path self._acquired_at: float = 0.0 # ── [v1.35.0] 浏览器使用锁 API ── def is_busy(self) -> bool: """检查浏览器是否正在被其他 Agent 使用""" return self._usage_lock.locked() def get_lock_info(self) -> dict: """获取浏览器锁状态信息""" return { "busy": self._usage_lock.locked(), "profile": self.profile_name, "current_user": self._current_user, "acquired_at": self._acquired_at, "started": self._started, } async def acquire(self, agent_path: str = "", timeout: float = 120.0) -> bool: """ 异步获取浏览器使用权（带超时）。 如果浏览器被其他 Agent 占用，当前协程会异步等待，不会阻塞整个进程。 等待期间其他协程（如 1:1 聊天、其他 API 请求）可以正常执行。 Args: agent_path: 请求使用的 Agent 路径 timeout: 最长等待时间（秒），默认 120 秒 Returns: True 如果成功获取，False 如果超时 """ if self._usage_lock.locked() and self._current_user == agent_path: # 同一个 Agent 重复获取（幂等） return True try: await asyncio.wait_for(self._usage_lock.acquire(), timeout=timeout) self._current_user = agent_path self._acquired_at = time.time() if agent_path: logger.info(f"浏览器锁已获取: profile={self.profile_name}, agent={agent_path}") return True except asyncio.TimeoutError: holder = self._current_user or "未知" logger.warning(f"浏览器锁等待超时: profile={self.profile_name}, agent={agent_path}, 持有者={holder}, timeout={timeout}s") return False async def release(self, agent_path: str = ""): """释放浏览器使用权""" try: if self._usage_lock.locked(): self._current_user = None self._acquired_at = 0.0 self._usage_lock.release() if agent_path: logger.info(f"浏览器锁已释放: profile={self.profile_name}, agent={agent_path}") except RuntimeError: pass # 锁未被当前协程持有，忽略 def _get_user_data_dir(self) -> str: """获取用户数据目录路径""" logger.info(f"[_get_user_data_dir] profile_name={self.profile_name}, use_system={self._use_system_profile}, custom={self._custom_user_data_dir}") if self._custom_user_data_dir: logger.info(f"[_get_user_data_dir] 使用自定义路径: {self._custom_user_data_dir}") return self._custom_user_data_dir # 使用系统 Chrome 的 Profile（保留登录状态、Cookie、扩展等） if self._use_system_profile: system_dir = self._get_system_chrome_user_data_dir() if system_dir and os.path.isdir(system_dir): self._user_data_dir = system_dir logger.info(f"[_get_user_data_dir] 使用系统 Chrome Profile: {system_dir}") return self._user_data_dir else: logger.warning( f"未找到系统 Chrome Profile，回退到 myagent 隔离 Profile" ) from core.browser_profile import get_browser_profile_manager mgr = get_browser_profile_manager() logger.info(f"[_get_user_data_dir] BrowserProfileManager.base_dir={mgr.base_dir}") profile = mgr.get_profile(self.profile_name) profile.ensure_dirs() # DrissionPage 的 set_user_data_path 会自动在路径下创建 Default 目录 # 我们直接用 profile_dir 作为 user-data-dir，避免嵌套 self._user_data_dir = str(profile.profile_dir) # [DEBUG] 详细输出 Profile 目录状态 resolved = os.path.realpath(self._user_data_dir) logger.info(f"[_get_user_data_dir] profile_dir={profile.profile_dir}") logger.info(f"[_get_user_data_dir] resolved path={resolved}") logger.info(f"[_get_user_data_dir] dir exists={os.path.isdir(resolved)}") if os.path.isdir(resolved): entries = os.listdir(resolved) logger.info(f"[_get_user_data_dir] dir contents={entries}") # 检查 Chrome 的 Default 目录 default_dir = os.path.join(resolved, "Default") logger.info(f"[_get_user_data_dir] Default/ exists={os.path.isdir(default_dir)}") if os.path.isdir(default_dir): default_entries = os.listdir(default_dir) logger.info(f"[_get_user_data_dir] Default/ contents={default_entries}") # 检查 cookies.json cookies_json = os.path.join(resolved, "cookies.json") logger.info(f"[_get_user_data_dir] cookies.json exists={os.path.isfile(cookies_json)}") return self._user_data_dir @staticmethod def _get_system_chrome_user_data_dir() -> Optional[str]: """检测系统 Chrome 的用户数据目录路径""" import platform system = platform.system() home = os.path.expanduser("~") if system == "Darwin": # macOS path = os.path.join( home, "Library", "Application Support", "Google", "Chrome" ) elif system == "Windows": local_app_data = os.environ.get("LOCALAPPDATA", "") if local_app_data: path = os.path.join(local_app_data, "Google", "Chrome", "User Data") else: path = "" else: # Linux # XDG 配置目录优先，回退到默认路径 xdg_config = os.environ.get("XDG_CONFIG_HOME", "") if xdg_config: path = os.path.join(xdg_config, "google-chrome") else: path = os.path.join(home, ".config", "google-chrome") if path and os.path.isdir(path): return path return None @staticmethod def _is_system_chrome_running() -> bool: """检查系统 Chrome 是否正在运行（使用同一个 user-data-dir）""" import platform system = platform.system() try: if system == "Darwin": # macOS: 用 pgrep 检查 Google Chrome 进程 result = subprocess.run( ["pgrep", "-x", "Google Chrome"], capture_output=True, text=True, timeout=5, ) return result.returncode == 0 elif system == "Windows": # Windows: 用 tasklist 检查 result = subprocess.run( ["tasklist", "/FI", "IMAGENAME eq chrome.exe"], capture_output=True, text=True, timeout=5, ) return "chrome.exe" in result.stdout else: # Linux: 用 pgrep 检查 result = subprocess.run( ["pgrep", "-f", "google-chrome"], capture_output=True, text=True, timeout=5, ) return result.returncode == 0 except Exception: return False async def start(self) -> SkillResult: """启动反检测浏览器""" # [v1.47.16] VNC/Termux 模式下直接使用 Firefox，不走 DrissionPage/Chromium # Chromium 在 proot ARM64 下与 DrissionPage 不兼容， # 所以 VNC 模式下先检测环境，直接走 Firefox 路径，不需要 import DrissionPage _is_vnc_mode = False _is_termux_env = False try: from core.env_detect import is_termux _is_termux_env = is_termux() except ImportError: pass try: from core.env_detect import is_desktop if not is_desktop(): _is_vnc_mode = True except ImportError: if not _has_display(): _is_vnc_mode = True # VNC 模式：跳过所有 Chrome/DrissionPage 逻辑，直接用 Firefox if _is_vnc_mode: logger.info("VNC/非桌面环境: 直接启动 Firefox（跳过 Chromium 检测）") if not self._headless: display = _ensure_display() if display: self._vnc_used = display.get("vnc", False) self._xvfb_started_by_us = display.get("xvfb_standalone", False) if self._vnc_used: return self._start_firefox_in_vnc() # VNC/Xvfb 不可用 if _is_termux_env: return SkillResult( success=False, error=( "Termux+Ubuntu 环境仅支持通过 VNC 启动浏览器，" "VNC 启动失败。请先启动 VNC 远程桌面后再使用浏览器功能。" ), ) # 非 Termux：尝试 Xvfb if display and display.get("xvfb_standalone"): # 有 Xvfb 但没有 VNC → 用 Xvfb + Chromium（走下面的正常流程） pass else: # 没有 Xvfb 也没有 VNC → 降级 headless 或报错 logger.warning("无显示环境且 VNC/Xvfb 均不可用，尝试 Firefox headless 模式") return self._start_firefox_in_vnc() else: # headless=True 被显式请求 if _is_termux_env: logger.info("Termux+Ubuntu 环境: 忽略 headless 请求，强制使用 VNC 模式") self._headless = False display = _ensure_display() if display: self._vnc_used = display.get("vnc", False) self._xvfb_started_by_us = display.get("xvfb_standalone", False) if self._vnc_used: return self._start_firefox_in_vnc() return SkillResult( success=False, error=( "Termux+Ubuntu 环境仅支持通过 VNC 启动浏览器，" "VNC 启动失败。headless 模式在此环境下不可用。" ), ) # 非 Termux headless → 仍然尝试 Firefox return self._start_firefox_in_vnc() # ── 桌面环境：使用 DrissionPage + Chromium ── try: from DrissionPage import Chromium, ChromiumOptions except ImportError: return SkillResult( success=False, error="DrissionPage 未安装。请运行: pip install DrissionPage", ) try: user_data = self._get_user_data_dir() co = ChromiumOptions() # 设置用户数据目录（Profile 持久化） # 注意：只用 set_argument，不用 set_user_data_path，因为后者会 # 在路径下再套一层 Default/ 子目录，导致路径混乱 if user_data: resolved_dir = os.path.realpath(user_data) logger.info(f"[start] 设置 --user-data-dir={resolved_dir}") co.set_argument(f"--user-data-dir={resolved_dir}") # 使用系统 Profile 时，指定默认 Profile 目录 # Chrome user-data-dir 下有 Default/, Profile 1/ 等子目录 if self._use_system_profile: co.set_argument("--profile-directory=Default") # 检查系统 Chrome 是否正在运行 if self._is_system_chrome_running(): logger.warning( "系统 Chrome 正在运行，Profile 可能被锁定。" "建议先关闭系统 Chrome，或在系统 Chrome 中操作。" ) else: logger.warning("未设置 user-data-dir，浏览器将使用临时 Profile（登录状态不会持久化）") # [DEBUG] 打印 ChromiumOptions 中的所有 arguments try: all_args = co.arguments if hasattr(co, 'arguments') else [] logger.info(f"[start] ChromiumOptions.arguments={all_args}") except Exception as e: logger.info(f"[start] 无法读取 arguments: {e}") # 反检测核心参数 co.set_argument("--no-first-run") co.set_argument("--no-default-browser-check") # 注意: 不再使用 --disable-blink-features=AutomationControlled # Chrome 130+ 已移除该 blink 特性，使用会触发"不受支持的命令行标记"警告 # DrissionPage 内置了反检测机制（自动处理 webdriver 标志等），无需手动设置 # 另外在 JS 层通过 navigator.webdriver 注入实现等效效果 # 禁用自动化相关标志 co.set_argument("--disable-infobars") # 注意: 不再全局禁用扩展 --disable-extensions # 某些网站需要扩展才能正常登录（如密码管理器），改为仅在需要时禁用 # ── 容器环境参数 ── # 桌面环境 (Windows/Mac/Linux with real DISPLAY) 不需要这些参数， # 直接调用系统 Chrome 即可；容器/Termux 环境才需要。 try: from core.env_detect import needs_container_args _need_container_args = needs_container_args() except ImportError: _need_container_args = not _has_display() # 降级: 无 DISPLAY 则视为容器 if _need_container_args: co.set_argument("--no-sandbox") co.set_argument("--disable-setuid-sandbox") co.set_argument("--disable-dev-shm-usage") co.set_argument("--disable-gpu") logger.info("容器/Termux 环境，已添加 --no-sandbox 等容器参数") else: logger.info("桌面环境，使用系统 Chrome 原生参数") # ── 显示环境处理（仅桌面环境到达此代码路径）── # VNC/Termux 模式已在方法开头走 Firefox 分支，这里只处理桌面环境 if not self._headless: logger.info("桌面环境，直接使用系统浏览器，跳过 VNC/Xvfb") # 无头模式（co.headless() 内部设置 --headless=new） if self._headless: co.headless() # 自动检测浏览器路径 browser_path = self._detect_browser() logger.info(f"[start] 检测到浏览器路径: {browser_path}") if browser_path: co.set_browser_path(browser_path) else: return SkillResult( success=False, error=( "未找到 Chrome/Chromium 浏览器。" "请安装 Chrome 或设置 CHROME_PATH 环境变量。" ), ) # 设置窗口大小 co.set_argument("--window-size=1920,1080") # 端口管理：手动找空闲端口，设置 --remote-debugging-port # [v1.38.2] 不能用 co.auto_port()！它会覆盖 --user-data-dir 为临时目录 # (/tmp/DrissionPage/autoPortData/)，导致 Profile 每次都是全新的 import socket as _sock _debug_port = 0 try: with _sock.socket(_sock.AF_INET, _sock.SOCK_STREAM) as _s: _s.bind(('127.0.0.1', 0)) _debug_port = _s.getsockname()[1] co.set_argument(f"--remote-debugging-port={_debug_port}") # DrissionPage 需要 addr 才能正确连接到 Chrome # auto_port() 会自动设置，手动分配端口时需要显式设置 co.set_address(f"127.0.0.1:{_debug_port}") logger.info(f"[start] 手动分配调试端口: {_debug_port}") except Exception as port_e: logger.warning(f"[start] 手动分配端口失败，回退到 auto_port: {port_e}") try: co.auto_port() except Exception: pass # 旧版 DrissionPage 可能没有 auto_port # 创建浏览器实例 # 策略: # 1. 先尝试正常启动 DrissionPage # 2. 如果失败，尝试连接已运行的同 user-data-dir Chrome 实例 # 3. 重试时 NOT 杀 Chrome 进程（Chrome 已启动，可能只是连接慢） last_error = None for attempt in range(3): try: logger.info(f"[start] 第 {attempt + 1}/3 次尝试启动 Chromium(co)...") self._browser = Chromium(co) self._page = self._browser.latest_tab logger.info(f"[start] Chromium() 成功, browser={self._browser}, page={self._page}") # [DEBUG] 打印浏览器连接详情 + 实际进程命令行 try: if hasattr(self._browser, 'address'): logger.info(f"[start] browser.address={self._browser.address}") if hasattr(self._browser, 'user_data_dir'): logger.info(f"[start] browser.user_data_dir={self._browser.user_data_dir}") # DrissionPage 可能没有 process 属性，用 lsof 通过端口查 PID self._log_chrome_cmdline() except Exception as dbg_e: logger.info(f"[start] 读取浏览器详情异常: {dbg_e}") # 验证浏览器是否使用了正确的 Profile self._verify_profile(co) break except Exception as e: last_error = e logger.warning(f"浏览器启动第 {attempt + 1}/3 次尝试失败: {e}") # 尝试连接已有 Chrome 实例 # Chrome 可能已经在运行但调试端口不同， # 扫描常见端口尝试连接 if attempt == 0: connected = self._try_connect_existing_browser() if connected: logger.info("已连接到已运行的 Chrome 实例") break # 等待更长时间再重试（Chrome 可能正在启动中） # 不要杀进程！Chrome 已启动，只是 DrissionPage 连不上 time.sleep(3 * (attempt + 1)) else: raise last_error # 如果浏览器已崩溃或无法获取 tab if not self._page: return SkillResult( success=False, error="启动浏览器后无法获取页面标签（Chrome 可能未正确启动）", ) # 隐藏 webdriver 标志（额外注入） try: self._page.run_js( "Object.defineProperty(navigator,'webdriver',{get:()=>undefined});" "Object.defineProperty(navigator,'plugins',{get:()=>[1,2,3,4,5]});" "Object.defineProperty(navigator,'languages',{get:()=>['zh-CN','zh','en']});" "window.chrome={runtime:{}};" ) except Exception as e: logger.debug(f"反检测 JS 注入提示: {e}") # 等待浏览器完全就绪 # Bug Fix: 使用 asyncio.sleep 而非 time.sleep，避免阻塞事件循环 await asyncio.sleep(1) self._started = True logger.info( f"反检测浏览器已启动 (profile={self.profile_name}, " f"headless={self._headless})" ) return SkillResult( success=True, message=f"反检测浏览器已启动 (Profile: {self.profile_name})", data={"profile": self.profile_name, "headless": self._headless}, ) except Exception as e: logger.error(f"启动反检测浏览器失败: {e}") return SkillResult( success=False, error=f"启动反检测浏览器失败: {e}", ) def _start_firefox_in_vnc(self) -> SkillResult: """[v1.47.16] VNC 模式下直接启动 Firefox。 在 VNC/Termux 环境下，Chromium 与 DrissionPage 不兼容 （proot ARM64 下报 "browser executable file path cannot be found"）， 直接用 Firefox 在 VNC 中运行。 Firefox 模式下: - navigate: 通过 subprocess 打开 URL（Firefox 支持远程打开 URL） - screenshot: 通过 xdotool + import (ImageMagick) 截图 - click/fill: 通过 xdotool 发送鼠标/键盘事件 - cookie: 读取 Firefox profile 目录中的 cookies.sqlite """ self._firefox_mode = True display = os.environ.get("DISPLAY", ":99") # 1. 检测 Firefox 路径 firefox_path = None for candidate in ("firefox", "myagent-browser"): found = shutil.which(candidate) if found: # 跳过 snap 包装器 if self._is_snap_wrapper(found): logger.info(f"跳过 {candidate} ({found}) — snap 包装器，proot 下不可用") continue firefox_path = found break if not firefox_path: # 搜索常见路径 for p in ("/usr/bin/firefox", "/usr/local/bin/firefox", "/snap/bin/firefox"): if os.path.isfile(p) and os.access(p, os.X_OK): if self._is_snap_wrapper(p): continue firefox_path = p break if not firefox_path: return SkillResult( success=False, error="VNC 模式下未找到 Firefox 浏览器。请安装 Firefox: apt install firefox", ) logger.info(f"[_start_firefox_in_vnc] 使用 Firefox: {firefox_path}") # 2. 准备 Firefox Profile 目录 if self._custom_user_data_dir: self._firefox_profile_dir = self._custom_user_data_dir else: from core.browser_profile import get_browser_profile_manager mgr = get_browser_profile_manager() profile = mgr.get_profile(self.profile_name) profile.ensure_dirs() self._firefox_profile_dir = str(profile.profile_dir) # 确保目录存在 try: os.makedirs(self._firefox_profile_dir, exist_ok=True) except Exception as e: logger.warning(f"创建 Firefox Profile 目录失败: {e}") # 3. 检查是否已有 Firefox 运行 try: result = subprocess.run( ["pgrep", "-f", "firefox"], capture_output=True, text=True, timeout=5, ) if result.returncode == 0 and result.stdout.strip(): logger.info(f"检测到已有 Firefox 运行 (PID: {result.stdout.strip().split()[0]})，复用") self._started = True return SkillResult( success=True, message=f"Firefox 已在 VNC 中运行 (Profile: {self.profile_name})", data={"profile": self.profile_name, "mode": "firefox_vnc", "reused": True}, ) except Exception: pass # 4. 启动 Firefox env = {**os.environ, "DISPLAY": display} # proot 兼容环境变量 if not env.get("G_SLICE"): env["G_SLICE"] = "always-malloc" if not env.get("GSETTINGS_BACKEND"): env["GSETTINGS_BACKEND"] = "memory" if os.environ.get("DBUS_SESSION_BUS_ADDRESS"): env["DBUS_SESSION_BUS_ADDRESS"] = os.environ["DBUS_SESSION_BUS_ADDRESS"] 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: firefox_cmd = [ firefox_path, "--profile", self._firefox_profile_dir, "--width", "960", "--height", "540", "--no-remote", ] logger.info(f"[_start_firefox_in_vnc] 启动: {' '.join(firefox_cmd)}") self._firefox_process = subprocess.Popen( firefox_cmd, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env=env, preexec_fn=os.setpgrp, ) # 等待 Firefox 启动 time.sleep(3) if self._firefox_process.poll() is not None: exit_code = self._firefox_process.returncode self._firefox_process = None logger.error(f"Firefox 启动后立即退出 (exit code: {exit_code})") return SkillResult( success=False, error=f"Firefox 启动后立即退出 (exit code: {exit_code})", ) self._started = True logger.info( f"Firefox 已在 VNC 中启动 (PID={self._firefox_process.pid}, " f"profile={self.profile_name})" ) return SkillResult( success=True, message=f"Firefox 已在 VNC 中启动 (Profile: {self.profile_name})", data={"profile": self.profile_name, "mode": "firefox_vnc", "pid": self._firefox_process.pid}, ) except Exception as e: logger.error(f"Firefox 启动失败: {e}") return SkillResult( success=False, error=f"Firefox 启动失败: {e}", ) async def close(self) -> SkillResult: """关闭浏览器""" self._started = False # [v1.47.20] Firefox+VNC 模式：VNC 桌面的浏览器不能杀，只清理内部状态 if self._firefox_mode: try: if self._vnc_used: # VNC 模式：Firefox 由 vnc_manager 管理，不能杀进程 # 只清理本实例的内部状态 logger.info("Firefox+VNC 模式: 不关闭 VNC 浏览器（由 vnc_manager 管理），仅释放内部状态") elif self._firefox_process and self._firefox_process.poll() is None: # 非 VNC 模式（独立启动的 Firefox）：可以关闭 self._firefox_process.terminate() try: self._firefox_process.wait(timeout=5) except Exception: try: self._firefox_process.kill() except Exception: pass logger.info("Firefox 已关闭") else: # 可能是复用的非 VNC Firefox 进程 logger.info("Firefox 进程非本实例启动，跳过关闭") except Exception as e: logger.error(f"关闭 Firefox 异常: {e}") finally: self._firefox_process = None self._firefox_mode = False self._vnc_used = False return SkillResult(success=True, message="浏览器会话已释放（VNC 浏览器保持运行）") try: if self._browser: self._browser.quit() self._browser = None self._page = None logger.info("反检测浏览器已关闭") except Exception as e: logger.error(f"关闭浏览器异常: {e}") self._browser = None self._page = None # 如果由本实例独立启动了 Xvfb（非 VNC），关闭浏览器后停止虚拟显示 # 注意：VNC 管理自己的 Xvfb 生命周期，不在此停止 if self._xvfb_started_by_us: self._xvfb_started_by_us = False # 检查是否还有其他浏览器实例在使用 Xvfb with _browser_lock: other_active = any( b._started and b is not self for b in _browsers.values() ) if not other_active: _stop_xvfb() # VNC 显示不由浏览器管理，无需停止 self._vnc_used = False return SkillResult(success=True, message="浏览器已关闭") async def navigate(self, url: str, wait: float = 2.0) -> SkillResult: """导航到指定 URL""" if not self._ensure_page(): return SkillResult(success=False, error="浏览器未启动") # [v1.47.16] Firefox+VNC 模式：通过 subprocess 打开 URL if self._firefox_mode: return self._firefox_navigate(url, wait) try: self._page.get(url) if wait > 0: await asyncio.sleep(wait) title = self._page.title or "" current_url = self._page.url or "" # [DEBUG] 导航后检查 Cookie 是否从 Profile 加载 try: cookies = self._page.cookies() cookie_count = len(cookies) if cookies else 0 domains = set() for c in (cookies or []): if isinstance(c, dict): d = c.get('domain', '') if d: domains.add(d) logger.info( f"[navigate] 导航后 Cookie: count={cookie_count}, " f"domains={domains}" ) if cookie_count == 0 and self._user_data_dir: logger.warning( f"[navigate] Cookie 为 0! Profile ({self._user_data_dir}) " f"中的登录状态未被加载。可能 Chrome 使用了临时 Profile。" ) except Exception as cookie_err: logger.warning(f"[navigate] 导航后 Cookie 检查失败: {cookie_err}") return SkillResult( success=True, message=f"已导航到: {url}", data={"title": title, "url": current_url}, ) except Exception as e: return SkillResult(success=False, error=f"导航失败: {e}") def _find_element(self, selector: str, timeout: float = 10): """查找页面元素，支持多重回退策略。 DrissionPage 的 ele() 有自己的选择器语法（@attr=val, text=xxx 等）， 自动检测器会误解析 CSS 伪选择器（:first-child, :nth-child 等）， 导致 div[role="listitem"]:first-child 这样的 CSS 选择器匹配失败。 策略: 1. 直接用原始选择器（兼容 DrissionPage 原生语法如 text=xxx, @attr=val） 2. 加 css: 前缀重试（强制 CSS 模式） 3. 用 run_js + document.querySelector() 最终回退（兼容 Shadow DOM） """ # 策略 1: 原始选择器 ele = self._page.ele(selector, timeout=timeout) if ele: return ele # 策略 2: 加 css: 前缀（仅对看起来像 CSS 的选择器） # 跳过已有显式前缀或 DrissionPage 原生语法的选择器 _dp_prefixes = ('css:', 'xpath:', 'text:', 'tag:', '@@') _dp_native = ('text=', 'text:', '@') if not any(selector.strip().startswith(p) for p in _dp_prefixes) and \ not any(selector.strip().startswith(p) for p in _dp_native): css_selector = f'css:{selector}' logger.info(f"[_find_element] 原始选择器未找到，尝试 css: 前缀: {css_selector}") ele = self._page.ele(css_selector, timeout=min(timeout, 5)) if ele: return ele # 策略 3: 用 JS document.querySelector() 最终回退 logger.info(f"[_find_element] css: 前缀也未找到，尝试 JS querySelector: {selector}") try: # 用 JS 找元素并返回其 XPath，再用 DrissionPage 定位（确保后续操作可靠） js_result = self._page.run_js( f""" var el = document.querySelector({json.dumps(selector)}); if (!el) return null; // 获取唯一标识路径 function getPath(el) {{ if (el.id) return 'css:#' + el.id; if (el === document.body) return 'css:body'; var path = el.tagName.toLowerCase(); if (el.parentElement) {{ var siblings = Array.from(el.parentElement.children); var idx = siblings.indexOf(el) + 1; path = getPath(el.parentElement) + ' > ' + path + ':nth-child(' + idx + ')'; }} return 'css:' + path; }} return getPath(el); """ ) if js_result and js_result != 'null': logger.info(f"[_find_element] JS 找到元素路径: {js_result}") ele = self._page.ele(js_result, timeout=5) if ele: return ele except Exception as js_e: logger.debug(f"[_find_element] JS 回退失败: {js_e}") return None async def click(self, selector: str, wait: float = 1.0) -> SkillResult: """点击页面元素，支持虚拟滚动列表等不可见元素。""" if not self._ensure_page(): return SkillResult(success=False, error="浏览器未启动") # [v1.47.16] Firefox+VNC 模式：通过 xdotool 点击 if self._firefox_mode: return self._firefox_click(selector, wait) try: ele = self._find_element(selector, timeout=10) if not ele: return SkillResult( success=False, error=f"未找到元素: {selector}", ) # 尝试 DrissionPage 原生 click ele.click() if wait > 0: await asyncio.sleep(wait) return SkillResult(success=True, message=f"已点击: {selector}") except Exception as dp_err: err_msg = str(dp_err) # DrissionPage 报 "该元素没有位置及大小" — 元素在虚拟滚动列表中不可见 # 降级策略：先用文本/选择器找元素 → scrollIntoView → JS click if "位置及大小" in err_msg or "位置" in err_msg or "没有位置" in err_msg: logger.info(f"[click] 元素不可见（虚拟滚动?），尝试 JS click 回退: {selector}") try: # 判断 selector 是文本还是 CSS 选择器 _is_text = not any(c in selector for c in ['[', '#', '.', ':', '>', '+', '~', '*']) if _is_text: # 纯文本选择器：用文本匹配查找元素 js_code = f""" var all = document.querySelectorAll('span[data-thread-id], div[role="listitem"], tr, td, a, button, [role="button"], [role="link"], [role="option"]'); for (var i = 0; i < all.length; i++) {{ if (all[i].textContent.trim() === {json.dumps(selector.strip())}) {{ all[i].scrollIntoView({{block: 'center'}}); all[i].click(); return 'CLICKED:' + all[i].tagName + '#' + (all[i].id || 'no-id'); }} }} // 宽松匹配：包含该文本 for (var i = 0; i < all.length; i++) {{ if (all[i].textContent.trim().indexOf({json.dumps(selector.strip())}) >= 0) {{ all[i].scrollIntoView({{block: 'center'}}); all[i].click(); return 'CLICKED:' + all[i].tagName + '#' + (all[i].id || 'no-id'); }} }} return 'NOT_FOUND'; """ else: # CSS 选择器 js_code = f""" var el = document.querySelector({json.dumps(selector)}); if (!el) return 'NOT_FOUND'; el.scrollIntoView({{block: 'center'}}); el.click(); return 'CLICKED:' + el.tagName + '#' + (el.id || 'no-id'); """ js_result = self._page.run_js(js_code) if js_result and js_result.startswith('CLICKED'): logger.info(f"[click] JS click 成功: {js_result}") if wait > 0: await asyncio.sleep(wait) return SkillResult(success=True, message=f"已点击(JS): {selector} ({js_result})") elif js_result == 'NOT_FOUND': return SkillResult(success=False, error=f"JS click: 未找到元素: {selector}") else: if wait > 0: await asyncio.sleep(wait) return SkillResult(success=True, message=f"已点击(JS): {selector}") except Exception as js_e: return SkillResult(success=False, error=f"点击失败(DP+JS): DP={dp_err}, JS={js_e}") return SkillResult(success=False, error=f"点击失败: {dp_err}") async def fill( self, selector: str, value: str, clear: bool = True, wait: float = 0.5 ) -> SkillResult: """填写输入框""" if not self._ensure_page(): return SkillResult(success=False, error="浏览器未启动") # [v1.47.16] Firefox+VNC 模式：通过 xdotool 输入 if self._firefox_mode: return self._firefox_fill(selector, value, clear, wait) try: ele = self._find_element(selector, timeout=10) if not ele: return SkillResult( success=False, error=f"未找到元素: {selector}", ) if clear: ele.clear() ele.input(value) if wait > 0: await asyncio.sleep(wait) return SkillResult( success=True, message=f"已填写 {selector}: {value[:50]}{'...' if len(value) > 50 else ''}", ) except Exception as e: return SkillResult(success=False, error=f"填写失败: {e}") async def type_text( self, selector: str = "", text: str = "", clear: bool = True, delay_ms: int = 50, wait: float = 0.3, ) -> SkillResult: """模拟真实键盘输入。 与 fill() 不同，本方法使用 JS execCommand('insertText') 触发 真实的 InputEvent，适用于 React/Vue 等现代框架的内容编辑区域 （如 x.com 的发帖框、contenteditable div 等）。 Args: selector: 输入框选择器（CSS/文本）。为空时输入到当前聚焦元素。 text: 要输入的文本。 clear: 是否先清空现有内容。 delay_ms: 逐字符输入的间隔毫秒数（0=一次性输入）。 wait: 输入完成后等待时间（秒）。 """ if not self._ensure_page(): return SkillResult(success=False, error="浏览器未启动") # [v1.47.16] Firefox+VNC 模式：通过 xdotool type 输入 if self._firefox_mode: return self._firefox_fill(selector or "", text, clear, wait) try: if selector: ele = self._find_element(selector, timeout=10) if not ele: return SkillResult( success=False, error=f"未找到元素: {selector}", ) # 尝试聚焦元素 try: ele.click() await asyncio.sleep(0.2) except Exception: # 元素不可见，用 JS 聚焦 self._page.run_js(f""" var el = document.querySelector({json.dumps(selector)}); if (el) {{ el.focus(); el.click(); }} """) await asyncio.sleep(0.2) if clear: # 清空内容：全选 + 删除 self._page.run_js(""" var el = document.activeElement; if (el) { // 对 input/textarea 用 .value='' if (el.tagName === 'INPUT' || el.tagName === 'TEXTAREA') { el.value = ''; el.dispatchEvent(new Event('input', {bubbles: true})); el.dispatchEvent(new Event('change', {bubbles: true})); } else { // contenteditable: 全选后删除 el.focus(); document.execCommand('selectAll', false, null); document.execCommand('delete', false, null); } } """) await asyncio.sleep(0.15) if delay_ms > 0: # 逐字符输入（模拟真实打字） for i, char in enumerate(text): escaped = json.dumps(char) # 正确转义特殊字符 self._page.run_js(f""" var el = document.activeElement; if (!el) return false; el.focus(); document.execCommand('insertText', false, {escaped}); return true; """) # 最后一个字符不需要额外延迟 if i < len(text) - 1: await asyncio.sleep(delay_ms / 1000.0) else: # 一次性输入（速度快，适合长文本） self._page.run_js(f""" var el = document.activeElement; if (!el) return false; el.focus(); document.execCommand('insertText', false, {json.dumps(text)}); return true; """) if wait > 0: await asyncio.sleep(wait) display_text = text[:50] + ('...' if len(text) > 50 else '') return SkillResult( success=True, message=f"已输入: {display_text}", ) except Exception as e: return SkillResult(success=False, error=f"键盘输入失败: {e}") async def press_key( self, key: str, selector: str = "", wait: float = 0.5, ) -> SkillResult: """按下键盘按键。 支持常见按键: Enter, Tab, Escape, Backspace, Delete, ArrowUp, ArrowDown, ArrowLeft, ArrowRight, Home, End, PageUp, PageDown, F1-F12, Ctrl+A, Ctrl+C, Ctrl+V, Ctrl+Z 等。 组合键用 + 连接，如 "Ctrl+A", "Shift+Tab"。 """ if not self._ensure_page(): return SkillResult(success=False, error="浏览器未启动") # [v1.47.16] Firefox+VNC 模式：通过 xdotool key 按键 if self._firefox_mode: return self._firefox_press_key(key, selector, wait) try: if selector: ele = self._find_element(selector, timeout=10) if ele: try: ele.click() except Exception: self._page.run_js(f""" var el = document.querySelector({json.dumps(selector)}); if (el) {{ el.focus(); }} """) await asyncio.sleep(0.1) # 解析组合键 parts = key.strip().split('+') keys_js = [] modifiers = [] modifier_map = { 'ctrl': 'Control', 'shift': 'Shift', 'alt': 'Alt', 'meta': 'Meta', 'cmd': 'Meta', 'command': 'Meta', } special_keys = { 'enter': 'Enter', 'tab': 'Tab', 'escape': 'Escape', 'esc': 'Escape', 'backspace': 'Backspace', 'delete': 'Delete', 'del': 'Delete', 'arrowup': 'ArrowUp', 'arrowdown': 'ArrowDown', 'arrowleft': 'ArrowLeft', 'arrowright': 'ArrowRight', 'up': 'ArrowUp', 'down': 'ArrowDown', 'left': 'ArrowLeft', 'right': 'ArrowRight', 'home': 'Home', 'end': 'End', 'pageup': 'PageUp', 'pagedown': 'PageDown', 'space': ' ', ' ': ' ', } # F1-F12 for i in range(1, 13): special_keys[f'f{i}'] = f'F{i}' for part in parts: p = part.strip() if p.lower() in modifier_map: modifiers.append(modifier_map[p.lower()]) elif p.lower() in special_keys: keys_js.append(special_keys[p.lower()]) else: # 单个字符直接使用 keys_js.append(p) # 构建 JS: 按下修饰键 → 按下主键 → 释放主键 → 释放修饰键 key_arg = keys_js[0] if keys_js else 'Enter' mod_str = ','.join(f'"{m}"' for m in modifiers) js_code = f""" var el = document.activeElement; if (!el) el = document.body; el.focus(); var options = {{ bubbles: true, cancelable: true, key: "{key_arg}", code: "{key_arg}" }}; if ({mod_str}) {{ var mods = [{mod_str}]; options.ctrlKey = mods.indexOf('Control') >= 0; options.shiftKey = mods.indexOf('Shift') >= 0; options.altKey = mods.indexOf('Alt') >= 0; options.metaKey = mods.indexOf('Meta') >= 0; }} el.dispatchEvent(new KeyboardEvent('keydown', options)); el.dispatchEvent(new KeyboardEvent('keyup', options)); // 对于 input/textarea，也触发 input 事件以配合 React if (el.tagName === 'INPUT' || el.tagName === 'TEXTAREA') {{ el.dispatchEvent(new Event('input', {{bubbles: true}})); }} return 'PRESSED:{key_arg}'; """ self._page.run_js(js_code) if wait > 0: await asyncio.sleep(wait) return SkillResult( success=True, message=f"已按键: {key}", ) except Exception as e: return SkillResult(success=False, error=f"按键失败: {e}") async def screenshot(self, save_path: str = "") -> SkillResult: """页面截图""" if not self._ensure_page(): return SkillResult(success=False, error="浏览器未启动") # [v1.47.16] Firefox+VNC 模式：通过 ImageMagick import 截图 if self._firefox_mode: return self._firefox_screenshot(save_path) try: if not save_path: # 自动生成路径 from core.browser_profile import get_browser_profile_manager save_dir = get_browser_profile_manager().base_dir / "screenshots" save_dir.mkdir(parents=True, exist_ok=True) save_path = str( save_dir / f"{self.profile_name}_{int(time.time())}.png" ) self._page.get_screenshot(path=save_path, full_page=False) return SkillResult( success=True, message=f"截图已保存: {save_path}", files=[save_path], ) except Exception as e: return SkillResult(success=False, error=f"截图失败: {e}") async def evaluate(self, script: str) -> SkillResult: """执行 JavaScript，自动处理 return 返回值。 DrissionPage 的 run_js() 支持顶层 return，但 LLM 生成的脚本经常 没有 return 语句（只写了逻辑没有返回值）。此方法自动检测并补充。 """ if not self._ensure_page(): return SkillResult(success=False, error="浏览器未启动") # [v1.47.20] Firefox+VNC 模式：无法通过 CDP 执行 JS if self._firefox_mode: return SkillResult( success=False, error="Firefox+VNC 模式下不支持 JS 执行。请使用 stealth_browser_navigate " "导航页面，用 stealth_browser_content（截图+标签页信息）获取内容，" "用 xdotool 相关操作（click/fill/key）进行交互。", ) try: _script = script.strip() # 检测是否包含 return 语句 has_return = ( _script.startswith('return ') or _script.startswith('return\n') or '\nreturn ' in _script or '\nreturn\n' in _script ) if has_return: # 有 return：直接执行 result = self._page.run_js(_script) else: # 没有 return：提取最后一个变量名，包装成 IIFE + return lines = _script.rstrip().split('\n') last_line = lines[-1].strip() if lines else '' auto_var = None for prefix in ('const ', 'let ', 'var '): if last_line.startswith(prefix): var_part = last_line[len(prefix):].split('=')[0].strip() if var_part and not any(c in var_part for c in ['[', '{', ',']): auto_var = var_part break if auto_var: # 追加 return 最后一个变量 exec_script = f"(function() {{\n{_script}\nreturn {auto_var};\n}})()" else: exec_script = f"(function() {{\n{_script}\n}})()" result = self._page.run_js(exec_script) # 格式化输出 if result is None or result in ('undefined', 'null'): output = "" elif isinstance(result, bool): output = str(result) elif isinstance(result, (dict, list)): output = json.dumps(result, ensure_ascii=False) else: output = str(result) if not output: output = "(执行成功但无返回值。请在脚本末尾添加: return 变量名;)" return SkillResult( success=True, message=f"JS 执行完成", data={"result": output}, output=output, ) except Exception as e: return SkillResult(success=False, error=f"JS 执行失败: {e}") async def get_content(self) -> SkillResult: """获取页面文本内容""" if not self._ensure_page(): return SkillResult(success=False, error="浏览器未启动") # [v1.47.20] Firefox+VNC 模式：截图 + sessionstore 读取 if self._firefox_mode: return self._firefox_get_content() try: # Bug Fix: DrissionPage 没有 page.text 属性 # 需要通过 page.ele('tag:html').text 获取页面文本 try: text = self._page.ele('tag:html').text or "" except Exception: # 降级：用 BS4 从 html 提取文本 try: from bs4 import BeautifulSoup text = BeautifulSoup(self._page.html or "", "html.parser").get_text(strip=True, separator="\n") except Exception: text = "" title = self._page.title or "" url = self._page.url or "" # 限制长度避免返回过多数据 output_text = text[:10000] if len(text) > 10000 else text return SkillResult( success=True, message=f"页面内容: {title}", data={ "title": title, "url": url, "text": output_text, "text_length": len(text), "truncated": len(text) > 10000, }, output=output_text, ) except Exception as e: return SkillResult(success=False, error=f"获取页面内容失败: {e}") async def get_html(self) -> SkillResult: """获取页面 HTML""" if not self._ensure_page(): return SkillResult(success=False, error="浏览器未启动") # [v1.47.20] Firefox+VNC 模式：截图 + sessionstore 替代 if self._firefox_mode: return self._firefox_get_content() try: html = self._page.html or "" output_html = html[:50000] if len(html) > 50000 else html return SkillResult( success=True, message=f"HTML 获取完成 ({len(html)} 字符)", data={ "url": self._page.url or "", "html_length": len(html), "html": output_html, }, output=output_html, ) except Exception as e: return SkillResult(success=False, error=f"获取 HTML 失败: {e}") async def wait_for( self, selector: str, timeout: float = 10.0 ) -> SkillResult: """等待元素出现""" if not self._ensure_page(): return SkillResult(success=False, error="浏览器未启动") # [v1.47.20] Firefox+VNC 模式：sleep + 截图替代 if self._firefox_mode: await asyncio.sleep(min(timeout, 5)) # 等待后截图，确认页面状态 session_info = self._firefox_read_sessionstore() url = session_info.get("url", "") title = session_info.get("title", "") return SkillResult( success=True, message=f"Firefox+VNC: 已等待 {min(timeout, 5)}秒，当前页面: {title or url}", data={"url": url, "title": title}, ) try: ele = self._find_element(selector, timeout=timeout) if ele: return SkillResult( success=True, message=f"元素已出现: {selector}", ) else: return SkillResult( success=False, error=f"等待超时: {selector} ({timeout}s)", ) except Exception as e: return SkillResult(success=False, error=f"等待元素失败: {e}") async def get_cookies(self) -> SkillResult: """获取当前页面的 Cookie""" if not self._ensure_page(): return SkillResult(success=False, error="浏览器未启动") # [v1.47.16] Firefox+VNC 模式：读取 cookies.sqlite if self._firefox_mode: return self._firefox_get_cookies() try: # DrissionPage cookies() 返回 CookiesList（list 子类），每项是 dict cookies = self._page.cookies() cookie_list = [] if cookies: for c in cookies: # CookiesList 中的每项是 dict，但 key 可能不全 if isinstance(c, dict): cookie_list.append({ "name": c.get("name", ""), "value": c.get("value", ""), "domain": c.get("domain", ""), "path": c.get("path", "/"), }) else: cookie_list.append(str(c)) return SkillResult( success=True, message=f"获取到 {len(cookie_list)} 个 Cookie", data={"cookies": cookie_list, "count": len(cookie_list)}, ) except Exception as e: return SkillResult(success=False, error=f"获取 Cookie 失败: {e}") async def save_cookies(self) -> SkillResult: """保存当前 Cookie 到 Profile""" if not self._ensure_page(): return SkillResult(success=False, error="浏览器未启动") # [v1.47.16] Firefox+VNC 模式：Cookie 已由 Firefox 自动保存到 profile 目录 if self._firefox_mode: return SkillResult( success=True, message="Firefox+VNC 模式: Cookie 由 Firefox 自动保存到 profile 目录", ) try: cookies = self._page.cookies() cookie_list = [] if cookies: for c in cookies: # CookiesList 中每项是 dict，确保可序列化 if isinstance(c, dict): cookie_list.append(dict(c)) else: cookie_list.append(str(c)) from core.browser_profile import get_browser_profile_manager mgr = get_browser_profile_manager() # 保存时用 profile_name 对应的原始 BrowserProfile（不加 user_data 子路径） profile = mgr.get_profile(self.profile_name) profile.save_cookies(cookie_list) return SkillResult( success=True, message=f"已保存 {len(cookie_list)} 个 Cookie 到 Profile: {self.profile_name}", ) except Exception as e: return SkillResult(success=False, error=f"保存 Cookie 失败: {e}") async def load_cookies(self) -> SkillResult: """从 Profile 加载 Cookie""" if not self._ensure_page(): return SkillResult(success=False, error="浏览器未启动") # [v1.47.16] Firefox+VNC 模式：Cookie 由 Firefox 自动从 profile 目录加载 if self._firefox_mode: return SkillResult( success=True, message="Firefox+VNC 模式: Cookie 由 Firefox 自动从 profile 目录加载", ) try: from core.browser_profile import get_browser_profile_manager mgr = get_browser_profile_manager() profile = mgr.get_profile(self.profile_name) cookies = profile.load_cookies() if cookies: # Bug Fix: page.set.cookies 是 CookiesSetter 对象（可调用） # 直接调用 page.set.cookies(cookies_list) 即可 self._page.set.cookies(cookies) return SkillResult( success=True, message=f"已加载 {len(cookies)} 个 Cookie", ) else: return SkillResult( success=True, message="Profile 中没有保存的 Cookie", ) except Exception as e: return SkillResult(success=False, error=f"加载 Cookie 失败: {e}") async def clear_cookies(self) -> SkillResult: """清除当前页面和 Profile 中的 Cookie""" if not self._ensure_page(): return SkillResult(success=False, error="浏览器未启动") # [v1.47.16] Firefox+VNC 模式：删除 cookies.sqlite if self._firefox_mode: return self._firefox_clear_cookies() try: # Bug Fix: DrissionPage 没有 page.clear_cookies() 方法 # 正确方式是 page.set.cookies.clear() 或 page.clear_cache() try: self._page.set.cookies.clear() except Exception: self._page.clear_cache(cookies=True) from core.browser_profile import get_browser_profile_manager mgr = get_browser_profile_manager() profile = mgr.get_profile(self.profile_name) profile.clear_cookies() return SkillResult(success=True, message="Cookie 已清除") except Exception as e: return SkillResult(success=False, error=f"清除 Cookie 失败: {e}") async def wait_for_manual( self, reason: str = "验证码", timeout: float = 300.0 ) -> SkillResult: """ 等待用户手动操作（验证码/2FA 等）。 暂停自动化流程，等待指定时间让用户通过 VNC 或 Web Control 手动完成验证码、短信验证等操作。 Args: reason: 等待原因 timeout: 最长等待时间（秒），默认 5 分钟 """ message = ( f"⏸ 需要用户手动操作: {reason}\n" f"请通过 VNC 远程桌面或 Web Control 面板手动完成操作。\n" f"最多等待 {int(timeout)} 秒..." ) logger.info(f"[{self.profile_name}] 等待用户手动操作: {reason}") # Bug Fix: 使用 asyncio.sleep 避免阻塞事件循环 start = time.time() poll_interval = 3 # [v1.47.16] Firefox+VNC 模式下简化等待逻辑 if self._firefox_mode: while time.time() - start < timeout: await asyncio.sleep(poll_interval) # 检查 Firefox 是否仍在运行 if not self._ensure_page(): break elapsed = int(time.time() - start) return SkillResult( success=True, message=f"等待用户手动操作完成 ({reason})，耗时 {elapsed} 秒", data={"elapsed": elapsed}, ) last_url = self._page.url if self._page else "" while time.time() - start < timeout: await asyncio.sleep(poll_interval) if self._page: try: current_url = self._page.url or "" except Exception: break # 浏览器已关闭 # 如果 URL 发生变化，说明用户完成了操作 if current_url != last_url and current_url: elapsed = int(time.time() - start) return SkillResult( success=True, message=f"用户已完成手动操作 ({reason})，耗时 {elapsed} 秒", data={"elapsed": elapsed, "new_url": current_url}, ) last_url = current_url return SkillResult( success=False, error=f"等待用户操作超时 ({int(timeout)} 秒): {reason}", ) def _ensure_page(self) -> bool: """确保浏览器和页面可用""" # [v1.47.16] Firefox+VNC 模式：检查 Firefox 进程是否存活 if self._firefox_mode: if not self._started: return False if self._firefox_process and self._firefox_process.poll() is None: return True # 复用的 Firefox，检查进程是否存在 try: result = subprocess.run( ["pgrep", "-f", "firefox"], capture_output=True, text=True, timeout=3, ) if result.returncode == 0 and result.stdout.strip(): return True except Exception: pass return False if not self._started or not self._page: return False try: # 检查页面是否仍然有效（try 访问 url，如果浏览器已崩溃会抛异常） _ = self._page.url return True except Exception: # 不要在这里重置 _started，由 close() 方法负责 # 避免在异步上下文中误判 logger.debug(f"页面检查失败 (profile={self.profile_name})，浏览器可能已关闭") return False # ── [v1.47.16] Firefox+VNC 模式辅助方法 ────────────────────────── def _firefox_navigate(self, url: str, wait: float = 2.0) -> SkillResult: """Firefox+VNC 模式下导航到指定 URL。 Firefox 支持远程打开 URL：firefox 会在已运行的实例中打开新标签页。 [v1.47.19] 改用 Popen 非阻塞方式：firefox 在已运行实例中打开新标签页 后会阻塞等待窗口关闭，subprocess.run + timeout 会导致超时。 改用 Popen 后不等待命令完成，只确认启动即可。 """ display = os.environ.get("DISPLAY", ":99") env = {**os.environ, "DISPLAY": display} if not env.get("G_SLICE"): env["G_SLICE"] = "always-malloc" if not env.get("GSETTINGS_BACKEND"): env["GSETTINGS_BACKEND"] = "memory" if os.environ.get("DBUS_SESSION_BUS_ADDRESS"): env["DBUS_SESSION_BUS_ADDRESS"] = os.environ["DBUS_SESSION_BUS_ADDRESS"] try: firefox_path = shutil.which("firefox") or shutil.which("myagent-browser") if not firefox_path: return SkillResult(success=False, error="Firefox 未找到") # [v1.47.19] 使用 Popen 非阻塞方式启动 # firefox 在已有实例运行时会将 URL 发送给已有实例， # 然后阻塞等待（因为 Firefox 的 -no-remote 行为）。 # 用 Popen 启动后不等待完成，只等待进程启动成功。 process = subprocess.Popen( [firefox_path, "--profile", self._firefox_profile_dir, url], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env=env, start_new_session=True, ) # 等待短时间确认进程没有立即崩溃 time.sleep(0.5) if process.poll() is not None: # 进程已退出，可能是错误 exit_code = process.returncode # Firefox 在已有实例时，转发URL后以 exit code 0 退出（正常行为） if exit_code == 0: logger.info(f"Firefox 已转发 URL 到已有实例: {url}") else: return SkillResult( success=False, error=f"Firefox 打开 URL 失败 (exit code: {exit_code}): {url}", ) if wait > 0: time.sleep(wait) logger.info(f"Firefox 已打开 URL: {url}") return SkillResult( success=True, message=f"Firefox 已打开: {url}", data={"url": url}, ) except Exception as e: return SkillResult(success=False, error=f"Firefox 导航失败: {e}") def _firefox_click(self, selector: str, wait: float = 1.0) -> SkillResult: """Firefox+VNC 模式下通过 xdotool 点击元素。 限制：只能通过坐标点击，无法像 DrissionPage 精确匹配 CSS 选择器。 策略：先用 xdotool search 查找 Firefox 窗口，再尝试通过 xdotool key 模拟 Tab+Enter（适用于链接/按钮）。 """ display = os.environ.get("DISPLAY", ":99") env = {**os.environ, "DISPLAY": display} try: # 查找 Firefox 窗口 result = subprocess.run( ["xdotool", "search", "--onlyvisible", "--class", "firefox"], capture_output=True, text=True, timeout=5, env=env, start_new_session=True, ) if result.returncode != 0 or not result.stdout.strip(): return SkillResult( success=False, error="Firefox+VNC: 未找到 Firefox 窗口，无法点击。" "请在 VNC 中确认 Firefox 已打开。", ) window_id = result.stdout.strip().split()[0] # 激活窗口并点击 subprocess.run( ["xdotool", "windowactivate", "--sync", window_id], capture_output=True, timeout=5, env=env, start_new_session=True, ) # 如果 selector 是文本，用 Ctrl+F 搜索后按 Enter # 否则直接在窗口中心点击 subprocess.run( ["xdotool", "click", "--window", window_id, "1"], capture_output=True, timeout=5, env=env, start_new_session=True, ) if wait > 0: time.sleep(wait) return SkillResult( success=True, message=f"Firefox+VNC: 已在 Firefox 窗口点击 (selector={selector})。" f"VNC 模式下点击精度有限，建议手动操作。", ) except Exception as e: return SkillResult( success=False, error=f"Firefox+VNC 点击失败: {e}。建议在 VNC 中手动操作。", ) def _firefox_fill(self, selector: str, value: str, clear: bool = True, wait: float = 0.5) -> SkillResult: """Firefox+VNC 模式下通过 xdotool 填写输入框。 限制：无法精确定位输入框，需要用户先在 VNC 中点击目标输入框。 """ display = os.environ.get("DISPLAY", ":99") env = {**os.environ, "DISPLAY": display} try: # 查找 Firefox 窗口 result = subprocess.run( ["xdotool", "search", "--onlyvisible", "--class", "firefox"], capture_output=True, text=True, timeout=5, env=env, start_new_session=True, ) if result.returncode != 0 or not result.stdout.strip(): return SkillResult( success=False, error="Firefox+VNC: 未找到 Firefox 窗口。请在 VNC 中确认 Firefox 已打开。", ) window_id = result.stdout.strip().split()[0] subprocess.run( ["xdotool", "windowactivate", "--sync", window_id], capture_output=True, timeout=5, env=env, start_new_session=True, ) if clear: # Ctrl+A 全选，然后输入新内容替换 subprocess.run( ["xdotool", "key", "--window", window_id, "ctrl+a"], capture_output=True, timeout=3, env=env, start_new_session=True, ) time.sleep(0.2) # 使用 xdotool type 输入文本 subprocess.run( ["xdotool", "type", "--window", window_id, "--delay", "50", value], capture_output=True, timeout=15, env=env, start_new_session=True, ) if wait > 0: time.sleep(wait) return SkillResult( success=True, message=f"Firefox+VNC: 已输入文本到 Firefox。" f"请确认输入框已选中（先在 VNC 中点击目标输入框）。", ) except Exception as e: return SkillResult( success=False, error=f"Firefox+VNC 输入失败: {e}", ) def _firefox_press_key(self, key: str, selector: str = "", wait: float = 0.5) -> SkillResult: """[v1.47.16] Firefox+VNC 模式下通过 xdotool key 按键。""" display = os.environ.get("DISPLAY", ":99") env = {**os.environ, "DISPLAY": display} try: # 查找 Firefox 窗口 result = subprocess.run( ["xdotool", "search", "--onlyvisible", "--class", "firefox"], capture_output=True, text=True, timeout=5, env=env, start_new_session=True, ) if result.returncode != 0 or not result.stdout.strip(): return SkillResult( success=False, error="Firefox+VNC: 未找到 Firefox 窗口。请在 VNC 中确认 Firefox 已打开。", ) window_id = result.stdout.strip().split()[0] subprocess.run( ["xdotool", "windowactivate", "--sync", window_id], capture_output=True, timeout=5, env=env, start_new_session=True, ) # 将 JS 按键名称映射为 xdotool 按键名称 xdotool_key_map = { 'enter': 'Return', 'tab': 'Tab', 'escape': 'Escape', 'esc': 'Escape', 'backspace': 'BackSpace', 'delete': 'Delete', 'del': 'Delete', 'arrowup': 'Up', 'arrowdown': 'Down', 'arrowleft': 'Left', 'arrowright': 'Right', 'up': 'Up', 'down': 'Down', 'left': 'Left', 'right': 'Right', 'home': 'Home', 'end': 'End', 'pageup': 'Page_Up', 'pagedown': 'Page_Down', 'space': 'space', ' ': 'space', 'ctrl': 'ctrl', 'shift': 'shift', 'alt': 'alt', 'meta': 'super', 'cmd': 'super', 'command': 'super', } # F1-F12 for i in range(1, 13): xdotool_key_map[f'f{i}'] = f'F{i}' # 解析组合键 parts = key.strip().split('+') xdotool_keys = [] for part in parts: p = part.strip() mapped = xdotool_key_map.get(p.lower(), p) xdotool_keys.append(mapped) # 构建 xdotool key 参数 xdotool_key_str = '+'.join(xdotool_keys) subprocess.run( ["xdotool", "key", "--window", window_id, xdotool_key_str], capture_output=True, timeout=5, env=env, start_new_session=True, ) if wait > 0: time.sleep(wait) return SkillResult( success=True, message=f"Firefox+VNC: 已按键: {key}", ) except Exception as e: return SkillResult( success=False, error=f"Firefox+VNC 按键失败: {e}", ) def _firefox_screenshot(self, save_path: str = "") -> SkillResult: """Firefox+VNC 模式下通过 xdotool + import 截图。""" display = os.environ.get("DISPLAY", ":99") env = {**os.environ, "DISPLAY": display} try: if not save_path: from core.browser_profile import get_browser_profile_manager save_dir = get_browser_profile_manager().base_dir / "screenshots" os.makedirs(save_dir, exist_ok=True) timestamp = time.strftime("%Y%m%d_%H%M%S") save_path = str(save_dir / f"firefox_{timestamp}.png") # 方法1: 使用 ImageMagick import 截取整个屏幕 import_cmd = shutil.which("import") if import_cmd: result = subprocess.run( [import_cmd, "-window", "root", "-display", display, save_path], capture_output=True, timeout=10, env=env, start_new_session=True, ) if result.returncode == 0 and os.path.isfile(save_path): logger.info(f"Firefox 截图已保存: {save_path}") return SkillResult( success=True, message=f"Firefox+VNC 截图已保存", data={"path": save_path}, ) logger.warning( f"[_firefox_screenshot] import 截图失败 (rc={result.returncode}): " f"{result.stderr.decode(errors='ignore')[:200]}" ) # 方法2: 使用 xdotool + scrot scrot_cmd = shutil.which("scrot") if scrot_cmd: result = subprocess.run( [scrot_cmd, save_path], capture_output=True, timeout=10, env=env, start_new_session=True, ) if result.returncode == 0 and os.path.isfile(save_path): logger.info(f"Firefox 截图已保存 (scrot): {save_path}") return SkillResult( success=True, message=f"Firefox+VNC 截图已保存 (scrot)", data={"path": save_path}, ) logger.warning( f"[_firefox_screenshot] scrot 截图失败 (rc={result.returncode}): " f"{result.stderr.decode(errors='ignore')[:200]}" ) # 方法3: 使用 xwd + convert xwd_cmd = shutil.which("xwd") convert_cmd = shutil.which("convert") if xwd_cmd and convert_cmd: xwd_tmp = save_path + ".xwd" result = subprocess.run( [xwd_cmd, "-root", "-display", display, "-out", xwd_tmp], capture_output=True, timeout=10, env=env, start_new_session=True, ) if result.returncode == 0: subprocess.run( [convert_cmd, xwd_tmp, save_path], capture_output=True, timeout=10, env=env, start_new_session=True, ) try: os.unlink(xwd_tmp) except Exception: pass if os.path.isfile(save_path): logger.info(f"Firefox 截图已保存 (xwd+convert): {save_path}") return SkillResult( success=True, message=f"Firefox+VNC 截图已保存", data={"path": save_path}, ) logger.warning( f"[_firefox_screenshot] xwd+convert 截图失败 (rc={result.returncode}): " f"{result.stderr.decode(errors='ignore')[:200]}" ) # 方法4: 使用 xdotool + xwd + ffmpeg (proot 兼容) xdotool_cmd = shutil.which("xdotool") ffmpeg_cmd = shutil.which("ffmpeg") if xdotool_cmd and ffmpeg_cmd: xwd_tmp = save_path + ".xwd" # 用 xdotool 获取活动窗口并截图 result = subprocess.run( [xwd_cmd or "xwd", "-root", "-display", display, "-out", xwd_tmp], capture_output=True, timeout=10, env=env, start_new_session=True, ) if result.returncode == 0 and os.path.isfile(xwd_tmp): # 用 ffmpeg 转换 xwd 到 png result2 = subprocess.run( [ffmpeg_cmd, "-y", "-i", xwd_tmp, save_path], capture_output=True, timeout=10, env=env, start_new_session=True, ) try: os.unlink(xwd_tmp) except Exception: pass if result2.returncode == 0 and os.path.isfile(save_path): logger.info(f"Firefox 截图已保存 (xwd+ffmpeg): {save_path}") return SkillResult( success=True, message=f"Firefox+VNC 截图已保存", data={"path": save_path}, ) _available = [] if import_cmd: _available.append("import") if scrot_cmd: _available.append("scrot") if xwd_cmd and convert_cmd: _available.append("xwd+convert") if xdotool_cmd and ffmpeg_cmd: _available.append("xdotool+ffmpeg") _tried = ", ".join(_available) if _available else "无" return SkillResult( success=False, error=f"Firefox+VNC 截图失败 (尝试过: {_tried}) " f"— 需要 ImageMagick import / scrot / xwd+convert", ) except Exception as e: return SkillResult(success=False, error=f"Firefox+VNC 截图失败: {e}") def _firefox_read_sessionstore(self, max_wait: float = 5.0) -> dict: """[v1.47.20] 读取 Firefox sessionstore 获取当前标签页 URL/标题。 Firefox 运行时会将当前会话信息写入 sessionstore-backups/recovery.jsonlz4。 该文件使用 mozLz4 格式（8字节自定义头 + LZ4 压缩数据）。 Args: max_wait: 最长等待时间（秒），给 Firefox 时间写入 sessionstore Returns: dict: {"url": str, "title": str, "tabs": [{"url": str, "title": str}]} """ result_info = {"url": "", "title": "", "tabs": []} # 搜索可能的 sessionstore 路径 search_dirs = [] if self._firefox_profile_dir: search_dirs.append(self._firefox_profile_dir) # 也搜索 Firefox 默认 profile 和 vnc_manager 启动的 profile ff_base = os.path.expanduser("~/.mozilla/firefox") for extra in [ os.path.join(ff_base, "default"), ff_base, ]: if os.path.isdir(extra) and extra not in search_dirs: search_dirs.append(extra) # [v1.47.24] 搜索所有 Firefox profile 子目录（如 xxx.default-release） try: if os.path.isdir(ff_base): for entry in os.listdir(ff_base): entry_path = os.path.join(ff_base, entry) if os.path.isdir(entry_path) and entry_path not in search_dirs: # Firefox profile 目录通常包含 user.js 或 prefs.js if os.path.isfile(os.path.join(entry_path, "prefs.js")) or \ os.path.isfile(os.path.join(entry_path, "user.js")): search_dirs.append(entry_path) except Exception: pass # [v1.47.24] 等待 Firefox 写入 sessionstore（刚导航完可能还没写入） _start_time = time.time() while True: recovery_files = [] for base_dir in search_dirs: ss_dir = os.path.join(base_dir, "sessionstore-backups") if os.path.isdir(ss_dir): for fname in ("recovery.jsonlz4", "recovery.baklz4", "previous.jsonlz4"): fpath = os.path.join(ss_dir, fname) if os.path.isfile(fpath): recovery_files.append(fpath) # 也检查 base_dir 本身（有些 Firefox 版本） for fname in ("sessionstore.jsonlz4", "sessionstore-backups/recovery.jsonlz4"): fpath = os.path.join(base_dir, fname) if os.path.isfile(fpath): recovery_files.append(fpath) # 如果找到了文件，或者超时了，就跳出 if recovery_files or (time.time() - _start_time) >= max_wait: break # 等 0.5 秒再试 time.sleep(0.5) if not recovery_files: logger.warning( f"[_firefox_read_sessionstore] 未找到 sessionstore 文件 " f"(搜索了 {len(search_dirs)} 个目录: {[d.replace(os.path.expanduser('~'), '~') for d in search_dirs]})" ) return result_info # 读取 mozLz4 格式 for fpath in recovery_files: try: import struct with open(fpath, "rb") as f: data = f.read() if len(data) < 8: continue magic = struct.unpack(" SkillResult: """[v1.47.20] Firefox+VNC 模式下获取页面内容。 由于无法通过 CDP 获取 Firefox 页面文本，采用以下策略： 1. 截取当前屏幕截图（供 VLM 视觉理解） 2. 读取 Firefox sessionstore 获取 URL/标题 3. 返回截图路径和基本元信息 Agent 可以使用截图进行视觉理解，或使用 web_search/web_read 获取页面文本。 """ # 1. 截图 screenshot_result = self._firefox_screenshot() screenshot_path = "" if screenshot_result.success and screenshot_result.data: screenshot_path = screenshot_result.data.get("path", "") logger.info(f"[_firefox_get_content] 截图成功: {screenshot_path}") else: logger.warning( f"[_firefox_get_content] 截图失败: " f"{screenshot_result.error or '无数据'}" ) # 2. 读取 sessionstore session_info = self._firefox_read_sessionstore() logger.info( f"[_firefox_get_content] sessionstore: url={session_info.get('url', '')}, " f"title={session_info.get('title', '')}, tabs={len(session_info.get('tabs', []))}" ) # 3. 组合返回信息 url = session_info.get("url", "") title = session_info.get("title", "") tabs = session_info.get("tabs", []) tabs_summary = "" if tabs: tabs_lines = [] for i, tab in enumerate(tabs[:10]): # 最多10个标签页 tabs_lines.append(f" [{i+1}] {tab.get('title', '?')} - {tab.get('url', '?')}") tabs_summary = "\n".join(tabs_lines) content_parts = [] if title: content_parts.append(f"标题: {title}") if url: content_parts.append(f"URL: {url}") if tabs_summary: content_parts.append(f"标签页:\n{tabs_summary}") if screenshot_path: content_parts.append(f"截图: {screenshot_path}") if not content_parts: # 全部失败 → 至少给截图失败原因 fallback_msg = "Firefox+VNC 无法获取页面内容" if not screenshot_result.success: fallback_msg += f" (截图失败: {screenshot_result.error})" if not session_info.get("url"): fallback_msg += " (sessionstore 未读取到 URL)" logger.error(f"[_firefox_get_content] {fallback_msg}") content_text = fallback_msg else: content_text = "\n".join(content_parts) return SkillResult( success=True, message=f"Firefox+VNC 页面信息: {title or url or '未知'}", data={ "url": url, "title": title, "tabs": tabs, "screenshot_path": screenshot_path, "mode": "firefox_vnc", "note": "Firefox+VNC 模式无法直接获取页面文本，已提供截图和标签页信息。" "请使用截图进行视觉理解，或用 web_search/web_read 获取网页文本。", }, files=[screenshot_path] if screenshot_path else [], output=content_text, ) def _firefox_get_cookies(self) -> SkillResult: """Firefox+VNC 模式下读取 cookies.sqlite。""" try: import sqlite3 cookies_db = os.path.join(self._firefox_profile_dir, "cookies.sqlite") if not os.path.isfile(cookies_db): return SkillResult( success=True, message="Firefox Profile 中没有 cookies.sqlite", data={"cookies": []}, ) # Firefox 使用 WAL 模式，需要先 checkpoint 或复制文件再读取 # 直接读取可能会锁冲突，先复制到临时文件 import tempfile with tempfile.NamedTemporaryFile(suffix=".sqlite", delete=False) as tmp: tmp_path = tmp.name shutil.copy2(cookies_db, tmp_path) try: conn = sqlite3.connect(tmp_path) cursor = conn.cursor() cursor.execute("SELECT name, value, host, path FROM moz_cookies") rows = cursor.fetchall() conn.close() cookie_list = [] for row in rows: cookie_list.append({ "name": row[0], "value": row[1], "domain": row[2], "path": row[3], }) return SkillResult( success=True, message=f"读取到 {len(cookie_list)} 个 Cookie", data={"cookies": cookie_list}, ) finally: try: os.unlink(tmp_path) except Exception: pass except Exception as e: return SkillResult(success=False, error=f"Firefox 读取 Cookie 失败: {e}") def _firefox_clear_cookies(self) -> SkillResult: """Firefox+VNC 模式下清除 cookies.sqlite。""" try: cookies_db = os.path.join(self._firefox_profile_dir, "cookies.sqlite") if os.path.isfile(cookies_db): os.remove(cookies_db) logger.info("Firefox cookies.sqlite 已删除") return SkillResult(success=True, message="Firefox Cookie 已清除") except Exception as e: return SkillResult(success=False, error=f"Firefox 清除 Cookie 失败: {e}") def _log_chrome_cmdline(self) -> None: """[DEBUG] 获取 Chrome 实际进程的命令行，验证 --user-data-dir 是否正确。""" import re try: # 方法1: 如果 DrissionPage 有 process 属性 pid = None if hasattr(self._browser, 'process') and self._browser.process: try: pid = self._browser.process.pid except Exception: pass # 方法2: 通过 lsof 用调试端口查 PID if not pid and hasattr(self._browser, 'address'): addr = self._browser.address # e.g. "127.0.0.1:52905" port = addr.split(":")[-1] if ":" in addr else "" if port: try: lsof_result = subprocess.run( ["lsof", "-i", f":{port}", "-sTCP:LISTEN", "-t"], capture_output=True, text=True, timeout=5, ) pids = lsof_result.stdout.strip().split("\n") if pids and pids[0]: pid = pids[0] logger.info(f"[start] 通过 lsof 找到 Chrome PID: {pid}") except Exception as le: logger.debug(f"[start] lsof 查询失败: {le}") # 方法3: 用 pgrep 查找包含 user-data-dir 的 Chrome 进程 if not pid and self._user_data_dir: try: pgrep_result = subprocess.run( ["pgrep", "-f", f"user-data-dir={self._user_data_dir}"], capture_output=True, text=True, timeout=5, ) pids = pgrep_result.stdout.strip().split("\n") if pids and pids[0]: pid = pids[0] logger.info(f"[start] 通过 pgrep 找到 Chrome PID: {pid}") except Exception: pass if not pid: logger.warning("[start] 无法找到 Chrome 进程 PID，跳过命令行检查") return # 用 ps 获取命令行 cmd_result = subprocess.run( ["ps", "-p", str(pid), "-o", "command="], capture_output=True, text=True, timeout=5, ) actual_cmd = cmd_result.stdout.strip() if not actual_cmd: logger.warning(f"[start] 无法获取 PID {pid} 的命令行") return # 提取 --user-data-dir 参数 udd_match = re.search(r'--user-data-dir=(\S+)', actual_cmd) if udd_match: actual_udd = udd_match.group(1) expected_udd = os.path.realpath(self._user_data_dir) if os.path.realpath(actual_udd) == expected_udd: logger.info(f"[start] ✓ Chrome 实际 --user-data-dir 匹配: {actual_udd}") else: logger.warning( f"[start] ✗ Chrome 实际 --user-data-dir 不匹配!" f" 预期: {expected_udd}, 实际: {actual_udd}" ) else: logger.warning(f"[start] ✗ Chrome 命令行中未找到 --user-data-dir!") # 输出前 800 字符的命令行 logger.info(f"[start] Chrome 命令行 (前800字): {actual_cmd[:800]}") except Exception as e: logger.warning(f"[start] _log_chrome_cmdline 失败: {e}") def _verify_profile(self, co) -> None: """验证浏览器是否使用了正确的 Profile 目录。 通过 CDP 协议获取浏览器的 user-data-dir 路径， 与预期路径对比，确保 Profile 持久化正确工作。 """ if not self._page or not self._user_data_dir: logger.warning(f"[_verify_profile] 跳过: page={self._page is not None}, user_data_dir={self._user_data_dir}") return try: # 通过 JS 获取 navigator.userAgent 确认浏览器已就绪 ua = self._page.run_js("return navigator.userAgent") if ua: logger.info(f"[_verify_profile] UA: {ua[:120]}") except Exception as e: logger.warning(f"[_verify_profile] 获取 UA 失败: {e}") # [DEBUG] 通过 CDP 获取实际的 user-data-dir try: info = self._browser.user_data_dir if hasattr(self._browser, 'user_data_dir') else None expected = os.path.realpath(self._user_data_dir) actual = os.path.realpath(info) if info else None logger.info(f"[_verify_profile] expected={expected}, actual={actual}") if actual and actual != expected: logger.warning( f"Profile 路径不匹配! 预期: {expected}, " f"实际: {actual}。浏览器可能使用了临时 Profile。" ) elif actual: logger.info(f"[_verify_profile] Profile 路径匹配: {actual}") else: logger.warning(f"[_verify_profile] 无法获取 browser.user_data_dir (DrissionPage 版本不支持?)") except Exception as e: logger.warning(f"[_verify_profile] CDP 验证异常: {e}") # [DEBUG] 检查 Cookie 状态 try: cookies = self._page.cookies() if cookies: cookie_count = len(cookies) if cookies else 0 logger.info(f"[_verify_profile] 当前 Cookie 数量: {cookie_count}") # 列出域名 domains = set() for c in (cookies or []): if isinstance(c, dict): d = c.get('domain', '') if d: domains.add(d) if domains: logger.info(f"[_verify_profile] Cookie 域名: {domains}") else: logger.info(f"[_verify_profile] 当前 Cookie 数量: 0 (Profile 为空或未导航到网站)") except Exception as e: logger.warning(f"[_verify_profile] 获取 Cookie 失败: {e}") def _try_connect_existing_browser(self) -> bool: """尝试连接已运行的 Chrome 实例。 扫描常见调试端口（9222-9232），如果发现可用实例则直接连接。 优先查找使用相同 user-data-dir 的实例，确保 Profile 正确。 """ import socket from DrissionPage import Chromium for port in range(9222, 9233): try: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.settimeout(0.5) s.connect(("127.0.0.1", port)) # 端口有服务在监听，尝试用 DrissionPage 连接 try: browser = Chromium(f"127.0.0.1:{port}") tab = browser.latest_tab if tab: # 连接成功，验证是否使用了正确的 Profile if self._user_data_dir and hasattr(browser, 'user_data_dir'): actual_dir = browser.user_data_dir or "" expected_dir = os.path.realpath(self._user_data_dir) if actual_dir and os.path.realpath(actual_dir) != expected_dir: logger.warning( f"端口 {port} 的 Chrome 实例 Profile 不匹配 " f"(预期: {expected_dir}, 实际: {actual_dir})，跳过" ) try: browser.quit() except Exception: pass continue # Profile 匹配或无法验证，复用该实例 self._browser = browser self._page = tab logger.info(f"已连接到已运行的 Chrome 实例 (端口: {port})") return True except Exception as e: logger.debug(f"尝试连接 127.0.0.1:{port} 失败: {e}") except (ConnectionRefusedError, OSError): continue return False @staticmethod def _kill_stale_chrome(user_data_dir: str = "") -> None: """清理可能残留的 Chrome/Chromium 进程""" try: # 查找使用相同 user-data-dir 的残留 Chrome 进程 patterns = [ "chromium-browser.*--remote-debugging-port", "chromium.*--remote-debugging-port", "chrome.*--remote-debugging-port", "google-chrome.*--remote-debugging-port", "headless_shell.*--remote-debugging-port", ] for pattern in patterns: result = subprocess.run( ["pgrep", "-f", pattern], 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: pid = int(pid_str.strip()) os.kill(pid, 9) # SIGKILL logger.info(f"已清理残留 Chrome 进程: PID {pid}") except (ValueError, ProcessLookupError, PermissionError): pass except Exception as e: logger.debug(f"清理残留 Chrome 进程时出错: {e}") @staticmethod def _is_snap_wrapper(path: str) -> bool: """[v1.47.10] 检测浏览器路径是否是 snap 包装器。 Ubuntu 22.04+ 的 chromium-browser 和 firefox 可能是 snap 包装脚本， 在 proot 环境下不可用（snapd 需要 systemd）。 """ try: if not os.path.isfile(path): return False # snap 包装器通常是 shell 脚本，包含 "snap" 关键字 with open(path, "r", errors="replace") as f: content = f.read(2048) # 常见 snap 包装脚本特征 snap_signatures = [ "snap", # snap run chromium / snap run firefox "/snap/", # snap 路径 "SNAP_NAME", # snap 环境变量 "snapd", # snap daemon ] # 只有文本文件（脚本）才检查，二进制不需要 if content.startswith("#!") or content.startswith("/*"): for sig in snap_signatures: if sig in content: return True except Exception: pass return False @staticmethod def _detect_browser(skip_puppeteer: bool = False) -> Optional[str]: """自动检测 Chromium/Chrome 浏览器路径 [v1.47.10] 增加 snap 包装器检测：proot 下 snap 不可用， 跳过 snap 包装脚本，避免浏览器启动后无法连接。 [v1.47.16] 增加 skip_puppeteer 参数：VNC/Termux 模式下 Puppeteer Chrome 在 proot ARM64 下不可用（DrissionPage 报 "The browser executable file path cannot be found"）， 跳过以避免无效启动尝试。 """ # 1. 环境变量 for key in ("CHROME_PATH", "BROWSER_PATH", "CHROMIUM_PATH"): val = os.environ.get(key, "").strip() if val and os.path.isfile(val): return val # 2. PATH 中的浏览器命令 for cmd in ( "google-chrome", "google-chrome-stable", "chromium-browser", "chromium", "brave-browser", "microsoft-edge", ): found = shutil.which(cmd) if found: # [v1.47.10] 跳过 snap 包装器 if StealthBrowser._is_snap_wrapper(found): logger.info(f"跳过 {cmd} ({found}) — snap 包装器，proot 下不可用") continue return found # 3. 操作系统特定路径 import platform system = platform.system() home = os.path.expanduser("~") if system == "Darwin": # macOS 常见路径 for p in ( "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome", "/Applications/Chromium.app/Contents/MacOS/Chromium", "/Applications/Brave Browser.app/Contents/MacOS/Brave Browser", "/Applications/Microsoft Edge.app/Contents/MacOS/Microsoft Edge", ): if os.path.isfile(p) and os.access(p, os.X_OK): return p elif system == "Windows": # Windows 常见路径 for p in ( os.path.join(os.environ.get("PROGRAMFILES", "C:\\Program Files"), "Google", "Chrome", "Application", "chrome.exe"), os.path.join(os.environ.get("PROGRAMFILES(X86)", "C:\\Program Files (x86)"), "Google", "Chrome", "Application", "chrome.exe"), os.path.join(os.environ.get("LOCALAPPDATA", ""), "Google", "Chrome", "Application", "chrome.exe"), os.path.join(os.environ.get("PROGRAMFILES", "C:\\Program Files"), "Microsoft", "Edge", "Application", "msedge.exe"), os.path.join(os.environ.get("LOCALAPPDATA", ""), "BraveSoftware", "Brave-Browser", "Application", "brave.exe"), ): if os.path.isfile(p): return p else: # Linux 常见路径 for p in ( "/usr/bin/google-chrome", "/usr/bin/google-chrome-stable", "/usr/bin/chromium-browser", "/usr/bin/chromium", "/snap/bin/chromium", "/usr/bin/brave-browser", "/usr/bin/microsoft-edge", ): if os.path.isfile(p) and os.access(p, os.X_OK): # [v1.47.10] 跳过 snap 包装器 if StealthBrowser._is_snap_wrapper(p): logger.info(f"跳过 {p} — snap 包装器，proot 下不可用") continue return p # 4. Puppeteer / Playwright 缓存 # [v1.47.16] VNC/Termux 模式下跳过：Puppeteer Chrome 在 proot ARM64 下 # 与 DrissionPage 不兼容（报 "browser executable file path cannot be found"） if not skip_puppeteer: for cache in ( os.path.join(home, ".cache", "puppeteer", "chrome"), os.path.join(home, ".cache", "ms-playwright"), ): if os.path.isdir(cache): for root, _, files in os.walk(cache): for fname in files: if fname in ("chrome", "chromium", "headless_shell"): full = os.path.join(root, fname) if os.access(full, os.X_OK): return full else: logger.info("[_detect_browser] VNC/Termux 模式: 跳过 Puppeteer/Playwright 缓存浏览器 (proot ARM64 不可用)") return None # ── 全局浏览器实例管理 ────────────────────────────────────── _browsers: Dict[str, StealthBrowser] = {} _browser_lock = threading.Lock() # 仅保护 _browsers 字典的创建/销毁（同步操作） # [v1.35.0] 全局浏览器使用锁：防止不同 Agent 并发使用同一浏览器实例 # 这是一个 dict，key=profile_name, value=asyncio.Lock # 与 _browser_lock 的区别：_browser_lock 保护字典操作，_browser_usage_locks 保护浏览器使用 _browser_usage_locks: Dict[str, asyncio.Lock] = {} _browser_usage_mutex = threading.Lock() # 保护 _browser_usage_locks 字典本身 def _get_browser_usage_lock(profile_name: str) -> asyncio.Lock: """获取指定 profile 的浏览器使用锁（懒创建，线程安全）""" with _browser_usage_mutex: if profile_name not in _browser_usage_locks: _browser_usage_locks[profile_name] = asyncio.Lock() return _browser_usage_locks[profile_name] async def acquire_browser( profile_name: str = "default", agent_path: str = "", timeout: float = 120.0, ) -> Optional[StealthBrowser]: """ 异步获取浏览器使用权（整合了获取实例 + 获取使用锁）。 流程: 1. 获取浏览器实例（同步，线程安全） 2. 异步等待浏览器使用锁（如果被其他 Agent 占用则等待，不会阻塞整个进程） 3. 返回可用的浏览器实例 Args: profile_name: Profile 名称 agent_path: 请求使用的 Agent 路径（用于日志和调试） timeout: 最长等待时间（秒） Returns: StealthBrowser 实例（已获取使用锁），超时返回 None """ # Step 1: 同步获取浏览器实例 browser = get_stealth_browser(profile_name) if not browser: return None # Step 2: 异步等待使用锁 ok = await browser.acquire(agent_path=agent_path, timeout=timeout) if ok: return browser return None async def release_browser(browser: StealthBrowser, agent_path: str = ""): """释放浏览器使用权""" if browser: await browser.release(agent_path=agent_path) async def _acquire_browser_or_fail( profile_name: str = "default", agent_path: str = "", timeout: float = 120.0, ) -> tuple: """ 获取浏览器使用权（供技能类使用）。 Returns: (browser, error_msg) — browser 不为 None 时成功；error_msg 不为空时失败 """ browser = get_stealth_browser(profile_name) if not browser: return None, f"浏览器实例创建失败 (profile={profile_name})" ok = await browser.acquire(agent_path=agent_path, timeout=timeout) if not ok: holder = browser._current_user or "未知" held_since = "" if browser._acquired_at: held_since = f"，已占用 {int(time.time() - browser._acquired_at)} 秒" return None, ( f"浏览器被占用 (profile={profile_name})，当前使用者: {holder}{held_since}。" f"请稍后再试，或使用 browser_wait_status 工具查看浏览器状态。" ) return browser, "" def get_all_browser_lock_status() -> List[dict]: """获取所有浏览器实例的锁状态""" result = [] with _browser_lock: for key, browser in _browsers.items(): info = browser.get_lock_info() info["cache_key"] = key result.append(info) return result def get_stealth_browser( profile_name: str = "default", headless: bool = False, use_system_profile: bool = False, ) -> StealthBrowser: """ 获取反检测浏览器实例（按 profile_name 复用）。 同一 profile 名称返回相同实例，避免重复启动。 Bug Fix: 改用同步锁 + 同步返回，避免在非 async 上下文中死锁。 Args: profile_name: Profile 名称 headless: 是否无头模式 use_system_profile: 使用系统 Chrome Profile（保留登录状态） """ # 使用系统 profile 时，用特殊 key 避免和隔离 profile 冲突 cache_key = f"__system__:{profile_name}" if use_system_profile else profile_name with _browser_lock: if cache_key in _browsers: browser = _browsers[cache_key] # Bug Fix: 检测浏览器是否已崩溃，如果崩溃则清理并重建 if browser._started and browser._ensure_page(): return browser else: # 浏览器已崩溃或未启动，清理旧实例 logger.info(f"浏览器实例失效 (profile={cache_key})，正在重建...") try: if browser._browser: browser._browser.quit() except Exception: pass del _browsers[cache_key] browser = StealthBrowser( profile_name=profile_name, headless=headless, use_system_profile=use_system_profile, ) _browsers[cache_key] = browser return browser def close_stealth_browser(profile_name: str = "") -> None: """关闭浏览器实例（同步版本）""" with _browser_lock: if profile_name: # 同时查找隔离 profile 和系统 profile 的实例 # 系统 profile 的 cache key 为 __system__: keys_to_close = [] for key, browser in list(_browsers.items()): if key == profile_name or key == f"__system__:{profile_name}": keys_to_close.append(key) for key in keys_to_close: browser = _browsers.pop(key, None) if browser: browser._started = False try: if browser._browser: browser._browser.quit() except Exception: pass browser._browser = None browser._page = None elif not profile_name: for name, browser in _browsers.items(): try: browser._started = False if browser._browser: browser._browser.quit() except Exception: pass _browsers.clear() # 所有浏览器关闭后，仅在独立 Xvfb（非 VNC）时才停止 # VNC 管理自己的 Xvfb 生命周期 any_vnc = any(b._vnc_used for b in _browsers.values()) if _browsers else False if not any_vnc: _stop_xvfb() async def close_stealth_browser_async(profile_name: str = "") -> None: """关闭浏览器实例（异步版本，供 async 上下文调用）""" close_stealth_browser(profile_name) # ── 技能类（供 SkillRegistry 自动发现）───────────────────── class StealthBrowserStartSkill(Skill): """启动反检测浏览器""" name = "stealth_browser_start" description = ( "启动反检测浏览器 — 使用 DrissionPage 启动带有反检测能力的 Chromium 浏览器，" "适合需要登录的网站操作。每个网站使用独立 Profile，登录状态自动保持。" "使用 site-manage list 查看所有可用网站。" ) category = "browser_stealth" parameters = [ SkillParameter( name="profile", type="string", description="Profile 名称（使用 site-manage list 查看可用网站）", required=False, default="default", ), SkillParameter( name="headless", type="boolean", description="是否无头模式（登录场景建议 false，默认 false）", required=False, default=False, ), SkillParameter( name="use_system_profile", type="boolean", description="使用系统 Chrome Profile（保留所有登录状态和 Cookie）。如果系统 Chrome 正在运行，需要先关闭。", required=False, default=False, ), ] dangerous = False async def execute(self, profile: str = "default", headless: bool = False, use_system_profile: bool = False, _agent_path: str = "", **kw) -> SkillResult: agent_path = _agent_path or getattr(self, '_current_agent_path', '') browser, err = await _acquire_browser_or_fail(profile_name=profile, agent_path=agent_path) if err: return SkillResult(success=False, error=err) try: if use_system_profile and not browser._use_system_profile: # Need to recreate with system profile browser = get_stealth_browser(profile_name=profile, headless=headless, use_system_profile=True) return await browser.start() finally: await release_browser(browser, agent_path=agent_path) class StealthBrowserNavigateSkill(Skill): """反检测浏览器导航""" name = "stealth_browser_navigate" description = "反检测浏览器导航 — 导航到指定 URL" category = "browser_stealth" parameters = [ SkillParameter(name="url", type="string", description="目标 URL", required=True), SkillParameter(name="profile", type="string", description="Profile 名称", required=False, default="default"), ] async def execute(self, url: str, profile: str = "default", _agent_path: str = "", **kw) -> SkillResult: agent_path = _agent_path or "" browser, err = await _acquire_browser_or_fail(profile_name=profile, agent_path=agent_path) if err: return SkillResult(success=False, error=err) try: if not browser._started: r = await browser.start() if not r.success: return r return await browser.navigate(url) finally: await release_browser(browser, agent_path=agent_path) class StealthBrowserClickSkill(Skill): """反检测浏览器点击""" name = "stealth_browser_click" description = "反检测浏览器点击 — 点击页面元素（支持 CSS 选择器、文本匹配等）" category = "browser_stealth" parameters = [ SkillParameter(name="selector", type="string", description="元素选择器 (CSS/文本/XPath)", required=True), SkillParameter(name="profile", type="string", description="Profile 名称", required=False, default="default"), ] async def execute(self, selector: str, profile: str = "default", _agent_path: str = "", **kw) -> SkillResult: agent_path = _agent_path or "" browser, err = await _acquire_browser_or_fail(profile_name=profile, agent_path=agent_path) if err: return SkillResult(success=False, error=err) try: return await browser.click(selector) finally: await release_browser(browser, agent_path=agent_path) class StealthBrowserFillSkill(Skill): """反检测浏览器填写""" name = "stealth_browser_fill" description = "反检测浏览器填写 — 填写输入框（支持 CSS 选择器定位）" category = "browser_stealth" parameters = [ SkillParameter(name="selector", type="string", description="输入框选择器", required=True), SkillParameter(name="value", type="string", description="要输入的文本", required=True), SkillParameter(name="clear", type="boolean", description="是否先清空", required=False, default=True), SkillParameter(name="profile", type="string", description="Profile 名称", required=False, default="default"), ] async def execute( self, selector: str, value: str, clear: bool = True, profile: str = "default", _agent_path: str = "", **kw ) -> SkillResult: agent_path = _agent_path or "" browser, err = await _acquire_browser_or_fail(profile_name=profile, agent_path=agent_path) if err: return SkillResult(success=False, error=err) try: return await browser.fill(selector, value, clear=clear) finally: await release_browser(browser, agent_path=agent_path) class StealthBrowserTypeSkill(Skill): """反检测浏览器键盘输入""" name = "stealth_browser_type" description = ( "反检测浏览器键盘输入 — 模拟真实键盘输入文本到页面元素。" "使用 execCommand('insertText') 触发真实 InputEvent，适用于 React/Vue 框架的" " contenteditable 区域（如 x.com 发帖框、微博等）。" "与 stealth_browser_fill 不同，fill 仅设置 value 属性，对现代前端框架无效。" "如果 selector 为空，则输入到当前已聚焦的元素。" ) category = "browser_stealth" parameters = [ SkillParameter(name="text", type="string", description="要输入的文本", required=True), SkillParameter(name="selector", type="string", description="输入框选择器（CSS/文本）。为空则输入到当前聚焦元素", required=False, default=""), SkillParameter(name="clear", type="boolean", description="是否先清空现有内容", required=False, default=True), SkillParameter(name="delay_ms", type="integer", description="逐字符输入间隔毫秒数（0=一次性输入）", required=False, default=0), SkillParameter(name="profile", type="string", description="Profile 名称", required=False, default="default"), ] async def execute( self, text: str, selector: str = "", clear: bool = True, delay_ms: int = 0, profile: str = "default", _agent_path: str = "", **kw, ) -> SkillResult: agent_path = _agent_path or "" browser, err = await _acquire_browser_or_fail(profile_name=profile, agent_path=agent_path) if err: return SkillResult(success=False, error=err) try: return await browser.type_text( selector=selector, text=text, clear=clear, delay_ms=delay_ms, ) finally: await release_browser(browser, agent_path=agent_path) class StealthBrowserPressKeySkill(Skill): """反检测浏览器按键""" name = "stealth_browser_press_key" description = ( "反检测浏览器按键 — 模拟按下键盘按键。支持 Enter、Tab、Escape、Backspace 等按键，" "以及组合键如 Ctrl+A、Ctrl+C、Ctrl+V、Ctrl+Z。用于提交表单、切换焦点等操作。" ) category = "browser_stealth" parameters = [ SkillParameter(name="key", type="string", description="按键名称（如 Enter, Tab, Ctrl+A, Shift+Tab）", required=True), SkillParameter(name="selector", type="string", description="先聚焦此元素再按键（可选）", required=False, default=""), SkillParameter(name="profile", type="string", description="Profile 名称", required=False, default="default"), ] async def execute( self, key: str, selector: str = "", profile: str = "default", _agent_path: str = "", **kw, ) -> SkillResult: agent_path = _agent_path or "" browser, err = await _acquire_browser_or_fail(profile_name=profile, agent_path=agent_path) if err: return SkillResult(success=False, error=err) try: return await browser.press_key(key=key, selector=selector) finally: await release_browser(browser, agent_path=agent_path) class StealthBrowserScreenshotSkill(Skill): """反检测浏览器截图""" name = "stealth_browser_screenshot" description = "反检测浏览器截图 — 截取当前页面截图并返回文件路径" category = "browser_stealth" parameters = [ SkillParameter(name="save_path", type="string", description="保存路径（空=自动）", required=False, default=""), SkillParameter(name="profile", type="string", description="Profile 名称", required=False, default="default"), ] async def execute(self, save_path: str = "", profile: str = "default", _agent_path: str = "", **kw) -> SkillResult: agent_path = _agent_path or "" browser, err = await _acquire_browser_or_fail(profile_name=profile, agent_path=agent_path) if err: return SkillResult(success=False, error=err) try: return await browser.screenshot(save_path=save_path) finally: await release_browser(browser, agent_path=agent_path) class StealthBrowserEvalSkill(Skill): """反检测浏览器执行 JS""" name = "stealth_browser_eval" description = "反检测浏览器执行 JavaScript — 在页面中执行自定义 JS 代码" category = "browser_stealth" parameters = [ SkillParameter(name="script", type="string", description="JavaScript 代码", required=True), SkillParameter(name="profile", type="string", description="Profile 名称", required=False, default="default"), ] async def execute(self, script: str, profile: str = "default", _agent_path: str = "", **kw) -> SkillResult: agent_path = _agent_path or "" browser, err = await _acquire_browser_or_fail(profile_name=profile, agent_path=agent_path) if err: return SkillResult(success=False, error=err) try: return await browser.evaluate(script) finally: await release_browser(browser, agent_path=agent_path) class StealthBrowserGetContentSkill(Skill): """反检测浏览器获取内容""" name = "stealth_browser_content" description = "反检测浏览器获取页面内容 — 获取当前页面的文本内容（用于读取邮件/帖子等）" category = "browser_stealth" parameters = [ SkillParameter(name="profile", type="string", description="Profile 名称", required=False, default="default"), ] async def execute(self, profile: str = "default", _agent_path: str = "", **kw) -> SkillResult: agent_path = _agent_path or "" browser, err = await _acquire_browser_or_fail(profile_name=profile, agent_path=agent_path) if err: return SkillResult(success=False, error=err) try: return await browser.get_content() finally: await release_browser(browser, agent_path=agent_path) class StealthBrowserCookieSkill(Skill): """反检测浏览器 Cookie 管理""" name = "stealth_browser_cookies" description = ( "反检测浏览器 Cookie 管理 — 保存/加载/获取/清除 Cookie。" "save=保存当前Cookie到Profile, load=从Profile加载Cookie, " "get=获取当前Cookie列表, clear=清除Cookie" ) category = "browser_stealth" parameters = [ SkillParameter( name="action", type="string", description="操作: save/load/get/clear", required=True, enum=["save", "load", "get", "clear"], ), SkillParameter(name="profile", type="string", description="Profile 名称", required=False, default="default"), ] async def execute(self, action: str, profile: str = "default", _agent_path: str = "", **kw) -> SkillResult: agent_path = _agent_path or "" browser, err = await _acquire_browser_or_fail(profile_name=profile, agent_path=agent_path) if err: return SkillResult(success=False, error=err) try: if action == "save": return await browser.save_cookies() elif action == "load": return await browser.load_cookies() elif action == "get": return await browser.get_cookies() elif action == "clear": return await browser.clear_cookies() else: return SkillResult(success=False, error=f"未知操作: {action}") finally: await release_browser(browser, agent_path=agent_path) class StealthBrowserCloseSkill(Skill): """关闭反检测浏览器""" name = "stealth_browser_close" description = "关闭反检测浏览器 — 关闭指定 Profile 的浏览器实例" category = "browser_stealth" parameters = [ SkillParameter(name="profile", type="string", description="Profile 名称（空=关闭所有）", required=False, default=""), ] async def execute(self, profile: str = "", **kw) -> SkillResult: # [v1.47.20] VNC 模式下不关闭 Firefox，只释放会话状态 with _browser_lock: is_vnc = False if profile: for key, browser in _browsers.items(): if key == profile or key == f"__system__:{profile}": if browser._vnc_used: is_vnc = True break else: is_vnc = any(b._vnc_used for b in _browsers.values()) close_stealth_browser(profile_name=profile) if is_vnc: return SkillResult( success=True, message="VNC 模式: 浏览器会话已释放，Firefox 保持运行（VNC 远程桌面需要）", ) return SkillResult(success=True, message="浏览器已关闭") class StealthBrowserWaitManualSkill(Skill): """等待用户手动操作（验证码/2FA）""" name = "stealth_browser_wait_manual" description = ( "等待用户手动操作 — 遇到验证码/短信验证/2FA 等需要人工干预的场景时调用。" "暂停自动化流程，等待用户通过 VNC 或 Web Control 面板手动完成操作。" "检测到 URL 变化后自动恢复。" ) category = "browser_stealth" parameters = [ SkillParameter( name="reason", type="string", description="等待原因（如: 验证码/短信验证/2FA）", required=False, default="验证码", ), SkillParameter( name="timeout", type="integer", description="最长等待秒数（默认 300）", required=False, default=300, ), SkillParameter(name="profile", type="string", description="Profile 名称", required=False, default="default"), ] async def execute( self, reason: str = "验证码", timeout: int = 300, profile: str = "default", _agent_path: str = "", **kw ) -> SkillResult: agent_path = _agent_path or "" browser, err = await _acquire_browser_or_fail(profile_name=profile, agent_path=agent_path) if err: return SkillResult(success=False, error=err) try: return await browser.wait_for_manual(reason=reason, timeout=float(timeout)) finally: await release_browser(browser, agent_path=agent_path) class StealthBrowserWaitSkill(Skill): """等待元素出现""" name = "stealth_browser_wait" description = "反检测浏览器等待 — 等待页面元素出现" category = "browser_stealth" parameters = [ SkillParameter(name="selector", type="string", description="元素选择器", required=True), SkillParameter(name="timeout", type="integer", description="超时秒数（默认 10）", required=False, default=10), SkillParameter(name="profile", type="string", description="Profile 名称", required=False, default="default"), ] async def execute( self, selector: str, timeout: int = 10, profile: str = "default", _agent_path: str = "", **kw ) -> SkillResult: agent_path = _agent_path or "" browser, err = await _acquire_browser_or_fail(profile_name=profile, agent_path=agent_path) if err: return SkillResult(success=False, error=err) try: return await browser.wait_for(selector, timeout=float(timeout)) finally: await release_browser(browser, agent_path=agent_path) class BrowserWaitStatusSkill(Skill): """查看浏览器锁状态""" name = "browser_wait_status" description = ( "查看浏览器锁状态 — 当浏览器操作失败提示「浏览器被占用」时，" "使用此工具查看哪个 Agent 正在使用浏览器、已占用多久，以及预估等待时间。" "也可以用来主动等待浏览器空闲。" ) category = "browser_stealth" parameters = [ SkillParameter( name="profile", type="string", description="Profile 名称（空=查看所有）", required=False, default="", ), SkillParameter( name="wait", type="boolean", description="是否等待浏览器空闲（最多等待 120 秒）", required=False, default=False, ), SkillParameter( name="wait_timeout", type="integer", description="最长等待秒数（默认 120）", required=False, default=120, ), ] async def execute( self, profile: str = "", wait: bool = False, wait_timeout: int = 120, _agent_path: str = "", **kw ) -> SkillResult: agent_path = _agent_path or "" # Get all browser lock status all_status = get_all_browser_lock_status() if not all_status: return SkillResult( success=True, message="当前没有活跃的浏览器实例", output="当前没有活跃的浏览器实例", ) # Filter by profile if specified if profile: all_status = [s for s in all_status if s.get("cache_key", "").endswith(profile) or s.get("profile", "") == profile] if not all_status: return SkillResult( success=True, message=f"没有找到 profile '{profile}' 的浏览器实例", output=f"没有找到 profile '{profile}' 的浏览器实例", ) # Build status report lines = [] all_free = True for status in all_status: prof = status.get("profile", "?") busy = status.get("busy", False) user = status.get("current_user", "") acquired_at = status.get("acquired_at", 0) started = status.get("started", False) if not started: lines.append(f"- {prof}: 未启动") continue if busy: all_free = False held = int(time.time() - acquired_at) if acquired_at else 0 lines.append(f"- {prof}: 被占用 (使用者: {user}, 已占用 {held} 秒)") else: lines.append(f"- {prof}: 空闲") # If wait requested and some browser is busy if wait and not all_free: lines.append(f"\n正在等待浏览器空闲 (最长 {wait_timeout} 秒)...") # Find the first busy browser matching the profile target_browser = None for status in all_status: if status.get("busy", False): cache_key = status.get("cache_key", "") # Get the actual browser instance with _browser_lock: target_browser = _browsers.get(cache_key) if target_browser: break if target_browser: ok = await target_browser.acquire(agent_path=agent_path, timeout=float(wait_timeout)) if ok: lines.append(f"浏览器已空闲! (profile={target_browser.profile_name})") await target_browser.release(agent_path=agent_path) return SkillResult( success=True, message="浏览器已空闲", output="\n".join(lines), ) else: lines.append(f"等待超时 ({wait_timeout} 秒)，浏览器仍被占用") return SkillResult( success=False, error=f"等待浏览器空闲超时 ({wait_timeout} 秒)", output="\n".join(lines), ) status_text = "\n".join(lines) return SkillResult( success=True, message="所有浏览器空闲" if all_free else "有浏览器被占用", output=status_text, ) # ── Site Registry 管理技能 (v1.31.0) ──────────────────── class SiteManageListSkill(Skill): """列出所有已注册网站""" name = "site_manage_list" description = "列出所有已注册网站 — 显示网站名称、分类、登录 URL 和 Profile 状态" category = "browser_stealth" parameters = [ SkillParameter( name="category", type="string", description="按分类筛选（空=全部）", required=False, default="", ), ] async def execute(self, category: str = "", **kw) -> SkillResult: from core.site_registry import get_site_registry reg = get_site_registry() sites = reg.list_sites(category=category) summary = reg.to_summary() return SkillResult( success=True, message=f"共 {len(sites)} 个网站" + (f" (分类: {category})" if category else ""), data={"sites": sites, "summary": {k: v for k, v in summary.items() if k != "sites"}}, output=json.dumps(sites, ensure_ascii=False, indent=2), ) class SiteManageShowSkill(Skill): """显示网站详细配置""" name = "site_manage_show" description = "显示网站详细配置 — 查看网站的登录 URL、可用页面和操作提示" category = "browser_stealth" parameters = [ SkillParameter(name="name", type="string", description="网站名称", required=True), ] async def execute(self, name: str, **kw) -> SkillResult: from core.site_registry import get_site_registry reg = get_site_registry() site = reg.get_site(name) if not site: return SkillResult(success=False, error=f"网站不存在: {name}") return SkillResult( success=True, message=f"网站: {site.get('display_name', name)}", data=site, output=json.dumps(site, ensure_ascii=False, indent=2), ) class SiteManageAddSkill(Skill): """添加自定义网站""" name = "site_manage_add" description = ( "添加自定义网站 — 为新网站创建配置，后续可用 stealth-open 登录。" "需提供 name（英文标识）、login_url（登录页）和 display_name（显示名称）。" ) category = "browser_stealth" parameters = [ SkillParameter(name="name", type="string", description="网站名称（英文，如 my_site）", required=True), SkillParameter(name="display_name", type="string", description="显示名称", required=False, default=""), SkillParameter(name="login_url", type="string", description="登录页 URL", required=False, default=""), SkillParameter(name="home_url", type="string", description="首页 URL", required=False, default=""), SkillParameter(name="category", type="string", description="分类", required=False, default="custom"), ] async def execute( self, name: str, display_name: str = "", login_url: str = "", home_url: str = "", category: str = "custom", **kw ) -> SkillResult: from core.site_registry import get_site_registry reg = get_site_registry() try: site = reg.add_site({ "name": name, "display_name": display_name or name, "login_url": login_url, "home_url": home_url, "category": category, }) return SkillResult( success=True, message=f"已添加网站: {name}", data=site, ) except ValueError as e: return SkillResult(success=False, error=str(e)) class SiteManageRemoveSkill(Skill): """删除自定义网站""" name = "site_manage_remove" description = "删除自定义网站（内置网站不可删除）" category = "browser_stealth" parameters = [ SkillParameter(name="name", type="string", description="网站名称", required=True), ] async def execute(self, name: str, **kw) -> SkillResult: from core.site_registry import get_site_registry reg = get_site_registry() ok = reg.remove_site(name) if ok: return SkillResult(success=True, message=f"已删除网站: {name}") return SkillResult( success=False, error=f"删除失败: {name}（内置网站不可删除，或网站不存在）", ) class SiteManageInitSkill(Skill): """批量初始化网站 Profile""" name = "site_manage_init" description = "批量初始化所有网站的浏览器 Profile（仅创建目录，不自动登录）" category = "browser_stealth" parameters = [] async def execute(self, **kw) -> SkillResult: from core.site_registry import get_site_registry reg = get_site_registry() initialized = reg.init_profiles() return SkillResult( success=True, message=f"已初始化 {len(initialized)} 个网站的 Profile", data={"initialized": initialized, "count": len(initialized)}, ) # ── Profile 管理技能 ────────────────────────────────────── class BrowserProfileListSkill(Skill): """列出所有浏览器 Profile""" name = "browser_profile_list" description = "列出所有浏览器 Profile — 显示已创建的网站 Profile 及其状态" category = "browser_stealth" parameters = [] async def execute(self, **kw) -> SkillResult: from core.browser_profile import get_browser_profile_manager mgr = get_browser_profile_manager() profiles = mgr.list_profiles() presets = mgr.get_preset_names() uninitialized = [p for p in presets if not any(x["name"] == p for x in profiles)] info = { "profiles": profiles, "available_presets": presets, "uninitialized_presets": uninitialized, "total_size_mb": round(mgr.get_total_size_mb(), 2), } return SkillResult( success=True, message=f"共 {len(profiles)} 个 Profile", data=info, output=json.dumps(info, ensure_ascii=False, indent=2), ) class BrowserProfileCreateSkill(Skill): """创建浏览器 Profile""" name = "browser_profile_create" description = ( "创建浏览器 Profile — 为网站创建独立的浏览器 Profile，" "每个 Profile 有独立的 Cookie 和登录状态。" ) category = "browser_stealth" parameters = [ SkillParameter(name="name", type="string", description="Profile 名称", required=True), SkillParameter(name="display_name", type="string", description="显示名称", required=False, default=""), SkillParameter(name="login_url", type="string", description="登录页 URL", required=False, default=""), ] async def execute( self, name: str, display_name: str = "", login_url: str = "", **kw ) -> SkillResult: from core.browser_profile import get_browser_profile_manager mgr = get_browser_profile_manager() profile = mgr.create_profile( name=name, display_name=display_name, login_url=login_url, ) return SkillResult( success=True, message=f"Profile 已创建: {name}", data=profile.to_dict(), ) class BrowserProfileDeleteSkill(Skill): """删除浏览器 Profile""" name = "browser_profile_delete" description = "删除浏览器 Profile — 删除指定网站的 Profile（包括 Cookie 和用户数据）" category = "browser_stealth" parameters = [ SkillParameter(name="name", type="string", description="Profile 名称", required=True), ] async def execute(self, name: str, **kw) -> SkillResult: from core.browser_profile import get_browser_profile_manager mgr = get_browser_profile_manager() ok = mgr.delete_profile(name) if ok: return SkillResult(success=True, message=f"Profile 已删除: {name}") return SkillResult(success=False, error=f"删除失败: {name}")