569 lines
17 KiB
Python
569 lines
17 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Browser Tool - Anti-Detection Browser Automation for OpenClaw
|
||
|
||
使用 Playwright 调用本地 Chrome，规避爬虫检测。
|
||
|
||
功能：
|
||
- navigate(url) - 打开网页并返回页面快照
|
||
- click(ref) - 点击元素
|
||
- type(ref, text) - 在输入框填入文本
|
||
- snapshot() - 获取当前页面快照
|
||
- scroll(direction) - 上下滚动
|
||
- close() - 关闭浏览器会话
|
||
|
||
两种模式：
|
||
1. 原生启动模式：直接启动本地 Chrome（带反检测参数）
|
||
2. CDP 连接模式：连接已运行的 Chrome 实例（通过 --remote-debugging-port=9222）
|
||
|
||
依赖：
|
||
pip install playwright
|
||
playwright install chromium
|
||
|
||
作者：Catalyst (封装)
|
||
"""
|
||
|
||
import json
|
||
import re
|
||
import time
|
||
import subprocess
|
||
import os
|
||
import tempfile
|
||
from pathlib import Path
|
||
from typing import Optional, Dict, Any, Literal
|
||
from dataclasses import dataclass, field
|
||
|
||
# Playwright 导入
|
||
try:
|
||
from playwright.sync_api import sync_playwright, Page, Browser, BrowserContext
|
||
except ImportError:
|
||
raise ImportError(
|
||
"Playwright not installed. Run:\n"
|
||
" pip install playwright\n"
|
||
" playwright install chromium"
|
||
)
|
||
|
||
|
||
# ============================================================================
|
||
# Configuration
|
||
# ============================================================================
|
||
|
||
CHROME_PATH = "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome"
|
||
DEFAULT_VIEWPORT = {"width": 1280, "height": 720}
|
||
DEFAULT_USER_AGENT = (
|
||
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
|
||
"AppleWebKit/537.36 (KHTML, like Gecko) "
|
||
"Chrome/120.0.0.0 Safari/537.36"
|
||
)
|
||
|
||
# 反检测启动参数
|
||
STEALTH_ARGS = [
|
||
"--disable-blink-features=AutomationControlled",
|
||
"--disable-web-security",
|
||
"--disable-features=IsolateOrigins,site-per-process",
|
||
"--disable-site-isolation-trials",
|
||
"--disable-dev-shm-usage",
|
||
"--no-sandbox",
|
||
"--disable-setuid-sandbox",
|
||
"--disable-accelerated-2d-canvas",
|
||
"--disable-gpu",
|
||
"--window-size=1280,720",
|
||
"--start-maximized",
|
||
]
|
||
|
||
# 禁用 webdriver 标记的脚本
|
||
HIDE_WEBDRIVER_SCRIPT = """
|
||
Object.defineProperty(navigator, 'webdriver', {
|
||
get: () => undefined
|
||
});
|
||
Object.defineProperty(navigator, 'plugins', {
|
||
get: () => [1, 2, 3, 4, 5]
|
||
});
|
||
window.chrome = { runtime: {} };
|
||
Object.defineProperty(navigator, 'languages', {
|
||
get: () => ['zh-CN', 'zh', 'en']
|
||
});
|
||
"""
|
||
|
||
|
||
# ============================================================================
|
||
# Data Classes
|
||
# ============================================================================
|
||
|
||
@dataclass
|
||
class ElementInfo:
|
||
"""页面元素信息"""
|
||
ref: str # @e1, @e2, etc.
|
||
tag: str # a, button, input, etc.
|
||
text: str
|
||
attributes: Dict[str, str] = field(default_factory=dict)
|
||
|
||
|
||
@dataclass
|
||
class PageSnapshot:
|
||
"""页面快照"""
|
||
url: str
|
||
title: str
|
||
elements: list
|
||
text_content: str
|
||
screenshot_path: Optional[str] = None
|
||
|
||
|
||
# ============================================================================
|
||
# Browser Tool Class
|
||
# ============================================================================
|
||
|
||
class BrowserTool:
|
||
"""
|
||
反检测浏览器自动化工具
|
||
|
||
使用本地 Chrome，规避爬虫检测
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
mode: Literal["launch", "cdp"] = "launch",
|
||
cdp_url: str = "http://localhost:9222",
|
||
headless: bool = False,
|
||
chrome_path: Optional[str] = None,
|
||
):
|
||
"""
|
||
初始化浏览器工具
|
||
|
||
Args:
|
||
mode: "launch" 直接启动 Chrome，"cdp" 连接已运行的 Chrome
|
||
cdp_url: CDP 连接地址（mode="cdp" 时使用）
|
||
headless: 是否无头模式（建议 False 规避检测）
|
||
chrome_path: Chrome 可执行文件路径
|
||
"""
|
||
self.mode = mode
|
||
self.cdp_url = cdp_url
|
||
self.headless = headless
|
||
self.chrome_path = chrome_path or CHROME_PATH
|
||
|
||
self._playwright = None
|
||
self._browser: Optional[Browser] = None
|
||
self._context: Optional[BrowserContext] = None
|
||
self._page: Optional[Page] = None
|
||
self._element_map: Dict[str, ElementInfo] = {}
|
||
self._closed = False
|
||
|
||
def _ensure_started(self):
|
||
"""确保浏览器已启动"""
|
||
if self._closed:
|
||
raise RuntimeError("Browser session has been closed")
|
||
if self._page is None:
|
||
raise RuntimeError("Browser not started. Call navigate() first.")
|
||
|
||
def _start_browser(self):
|
||
"""启动或连接浏览器"""
|
||
if self._playwright is None:
|
||
self._playwright = sync_playwright().start()
|
||
|
||
if self._browser is not None:
|
||
return
|
||
|
||
if self.mode == "cdp":
|
||
# CDP 连接模式
|
||
self._browser = self._playwright.chromium.connect_over_cdp(self.cdp_url)
|
||
contexts = self._browser.contexts
|
||
if contexts:
|
||
self._context = contexts[0]
|
||
else:
|
||
self._context = self._browser.new_context(
|
||
viewport=DEFAULT_VIEWPORT,
|
||
user_agent=DEFAULT_USER_AGENT,
|
||
)
|
||
else:
|
||
# 原生启动模式
|
||
if not os.path.exists(self.chrome_path):
|
||
raise FileNotFoundError(
|
||
f"Chrome not found at {self.chrome_path}\n"
|
||
"Please install Google Chrome or specify correct path."
|
||
)
|
||
|
||
self._browser = self._playwright.chromium.launch(
|
||
executable_path=self.chrome_path,
|
||
headless=self.headless,
|
||
args=STEALTH_ARGS,
|
||
)
|
||
self._context = self._browser.new_context(
|
||
viewport=DEFAULT_VIEWPORT,
|
||
user_agent=DEFAULT_USER_AGENT,
|
||
)
|
||
|
||
# 注入反检测脚本
|
||
self._context.add_init_script(HIDE_WEBDRIVER_SCRIPT)
|
||
|
||
# 创建新页面
|
||
self._page = self._context.new_page()
|
||
|
||
def _build_element_map(self) -> Dict[str, ElementInfo]:
|
||
"""构建页面元素映射表"""
|
||
self._ensure_started()
|
||
|
||
# 获取所有可交互元素
|
||
elements = self._page.query_selector_all(
|
||
'a, button, input, textarea, select, [role="button"], [onclick]'
|
||
)
|
||
|
||
self._element_map = {}
|
||
for idx, elem in enumerate(elements, start=1):
|
||
try:
|
||
tag = elem.evaluate("el => el.tagName.toLowerCase()")
|
||
text = elem.inner_text().strip()[:100] # 限制长度
|
||
|
||
# 获取关键属性
|
||
attrs = {}
|
||
for attr in ["href", "id", "name", "placeholder", "type", "value"]:
|
||
val = elem.get_attribute(attr)
|
||
if val:
|
||
attrs[attr] = val[:100]
|
||
|
||
ref = f"@e{idx}"
|
||
self._element_map[ref] = ElementInfo(
|
||
ref=ref,
|
||
tag=tag,
|
||
text=text,
|
||
attributes=attrs
|
||
)
|
||
except:
|
||
continue
|
||
|
||
return self._element_map
|
||
|
||
def _format_snapshot(self) -> str:
|
||
"""格式化页面快照为文本"""
|
||
lines = [
|
||
f"URL: {self._page.url}",
|
||
f"Title: {self._page.title()}",
|
||
"",
|
||
"=== Interactive Elements ===",
|
||
"",
|
||
]
|
||
|
||
for ref, elem in self._element_map.items():
|
||
attr_str = " ".join([f'{k}="{v}"' for k, v in elem.attributes.items()])
|
||
if attr_str:
|
||
attr_str = f" {attr_str}"
|
||
|
||
text = elem.text.replace("\n", " ")[:80]
|
||
lines.append(f"[{ref}] <{elem.tag}>{attr_str}> {text}")
|
||
|
||
return "\n".join(lines)
|
||
|
||
def _get_element_by_ref(self, ref: str):
|
||
"""通过 ref ID 获取元素"""
|
||
self._ensure_started()
|
||
|
||
if ref not in self._element_map:
|
||
# 刷新元素映射
|
||
self._build_element_map()
|
||
|
||
if ref not in self._element_map:
|
||
raise ValueError(f"Element {ref} not found. Call snapshot() to refresh.")
|
||
|
||
# 重新获取元素（Playwright 句柄会过期）
|
||
idx = int(ref.replace("@e", "")) - 1
|
||
elements = self._page.query_selector_all(
|
||
'a, button, input, textarea, select, [role="button"], [onclick]'
|
||
)
|
||
|
||
if idx >= len(elements):
|
||
raise ValueError(f"Element {ref} no longer exists. Call snapshot() to refresh.")
|
||
|
||
return elements[idx]
|
||
|
||
# ========================================================================
|
||
# Public API
|
||
# ========================================================================
|
||
|
||
def navigate(self, url: str, wait_until: str = "networkidle") -> str:
|
||
"""
|
||
导航到指定 URL
|
||
|
||
Args:
|
||
url: 目标网址
|
||
wait_until: 等待条件 (load, domcontentloaded, networkidle)
|
||
|
||
Returns:
|
||
页面快照文本
|
||
"""
|
||
self._start_browser()
|
||
|
||
self._page.goto(url, wait_until=wait_until)
|
||
|
||
# 等待页面稳定
|
||
self._page.wait_for_timeout(1000)
|
||
|
||
# 构建元素映射
|
||
self._build_element_map()
|
||
|
||
return self._format_snapshot()
|
||
|
||
def snapshot(self, full: bool = False) -> str:
|
||
"""
|
||
获取当前页面快照
|
||
|
||
Args:
|
||
full: 是否返回完整页面内容
|
||
|
||
Returns:
|
||
页面快照文本
|
||
"""
|
||
self._ensure_started()
|
||
|
||
if full:
|
||
# 返回完整页面文本
|
||
text = self._page.inner_text("body")
|
||
return f"URL: {self._page.url}\nTitle: {self._page.title()}\n\n{text[:8000]}"
|
||
|
||
# 刷新元素映射
|
||
self._build_element_map()
|
||
|
||
return self._format_snapshot()
|
||
|
||
def click(self, ref: str, wait_for_navigation: bool = False):
|
||
"""
|
||
点击元素
|
||
|
||
Args:
|
||
ref: 元素引用 ID (如 @e5)
|
||
wait_for_navigation: 是否等待页面跳转
|
||
"""
|
||
self._ensure_started()
|
||
|
||
element = self._get_element_by_ref(ref)
|
||
|
||
if wait_for_navigation:
|
||
with self._page.expect_navigation():
|
||
element.click()
|
||
else:
|
||
element.click()
|
||
|
||
# 等待一下让页面响应
|
||
self._page.wait_for_timeout(500)
|
||
|
||
def type(self, ref: str, text: str, clear: bool = True):
|
||
"""
|
||
在输入框填入文本
|
||
|
||
Args:
|
||
ref: 元素引用 ID (如 @e3)
|
||
text: 要输入的文本
|
||
clear: 是否先清空输入框
|
||
"""
|
||
self._ensure_started()
|
||
|
||
element = self._get_element_by_ref(ref)
|
||
|
||
if clear:
|
||
element.fill(text)
|
||
else:
|
||
element.type(text)
|
||
|
||
def scroll(self, direction: Literal["up", "down"] = "down", amount: int = 500):
|
||
"""
|
||
滚动页面
|
||
|
||
Args:
|
||
direction: up 或 down
|
||
amount: 滚动像素数
|
||
"""
|
||
self._ensure_started()
|
||
|
||
delta = -amount if direction == "up" else amount
|
||
self._page.evaluate(f"window.scrollBy(0, {delta})")
|
||
|
||
# 等待滚动完成
|
||
self._page.wait_for_timeout(300)
|
||
|
||
def screenshot(self, path: Optional[str] = None) -> str:
|
||
"""
|
||
截取页面截图
|
||
|
||
Args:
|
||
path: 截图保存路径（默认自动命名）
|
||
|
||
Returns:
|
||
截图文件路径
|
||
"""
|
||
self._ensure_started()
|
||
|
||
if path is None:
|
||
timestamp = int(time.time())
|
||
path = f"/tmp/browser_screenshot_{timestamp}.png"
|
||
|
||
self._page.screenshot(path=path, full_page=True)
|
||
return path
|
||
|
||
def press(self, key: str):
|
||
"""
|
||
按下键盘按键
|
||
|
||
Args:
|
||
key: 按键名称 (Enter, Tab, Escape, ArrowDown, etc.)
|
||
"""
|
||
self._ensure_started()
|
||
self._page.keyboard.press(key)
|
||
|
||
def back(self):
|
||
"""返回上一页"""
|
||
self._ensure_started()
|
||
self._page.go_back()
|
||
|
||
def close(self):
|
||
"""关闭浏览器会话"""
|
||
if self._closed:
|
||
return
|
||
|
||
try:
|
||
if self._context:
|
||
self._context.close()
|
||
if self._browser:
|
||
self._browser.close()
|
||
if self._playwright:
|
||
self._playwright.stop()
|
||
except:
|
||
pass
|
||
|
||
self._page = None
|
||
self._context = None
|
||
self._browser = None
|
||
self._playwright = None
|
||
self._closed = True
|
||
|
||
def __enter__(self):
|
||
return self
|
||
|
||
def __exit__(self, exc_type, exc_val, exc_tb):
|
||
self.close()
|
||
|
||
# ========================================================================
|
||
# Utility Methods
|
||
# ========================================================================
|
||
|
||
def check_detection(self) -> Dict[str, Any]:
|
||
"""
|
||
检查反检测效果
|
||
|
||
Returns:
|
||
检测结果字典
|
||
"""
|
||
if self._page is None:
|
||
self._start_browser()
|
||
|
||
# 访问检测页面
|
||
self._page.goto("https://bot.sannysoft.com/")
|
||
self._page.wait_for_timeout(3000)
|
||
|
||
# 获取关键检测结果
|
||
results = self._page.evaluate("""
|
||
() => ({
|
||
webdriver: navigator.webdriver,
|
||
plugins: navigator.plugins.length,
|
||
languages: navigator.languages,
|
||
userAgent: navigator.userAgent,
|
||
chrome: typeof window.chrome !== 'undefined',
|
||
})
|
||
""")
|
||
|
||
return results
|
||
|
||
|
||
# ============================================================================
|
||
# Helper Functions
|
||
# ============================================================================
|
||
|
||
def launch_chrome_debug_port(port: int = 9222) -> subprocess.Popen:
|
||
"""
|
||
启动 Chrome 并开启远程调试端口
|
||
|
||
Args:
|
||
port: 调试端口
|
||
|
||
Returns:
|
||
Chrome 进程对象
|
||
"""
|
||
if not os.path.exists(CHROME_PATH):
|
||
raise FileNotFoundError(f"Chrome not found at {CHROME_PATH}")
|
||
|
||
cmd = [
|
||
CHROME_PATH,
|
||
f"--remote-debugging-port={port}",
|
||
"--user-data-dir=/tmp/chrome_debug_profile",
|
||
"--no-first-run",
|
||
"--no-default-browser-check",
|
||
] + STEALTH_ARGS
|
||
|
||
proc = subprocess.Popen(
|
||
cmd,
|
||
stdout=subprocess.DEVNULL,
|
||
stderr=subprocess.DEVNULL,
|
||
)
|
||
|
||
# 等待 Chrome 启动
|
||
time.sleep(2)
|
||
|
||
return proc
|
||
|
||
|
||
def create_browser(
|
||
mode: Literal["launch", "cdp"] = "launch",
|
||
headless: bool = False,
|
||
) -> BrowserTool:
|
||
"""
|
||
创建浏览器工具实例（快捷函数）
|
||
|
||
Args:
|
||
mode: 启动模式
|
||
headless: 是否无头模式
|
||
|
||
Returns:
|
||
BrowserTool 实例
|
||
"""
|
||
return BrowserTool(mode=mode, headless=headless)
|
||
|
||
|
||
# ============================================================================
|
||
# CLI Interface
|
||
# ============================================================================
|
||
|
||
if __name__ == "__main__":
|
||
import argparse
|
||
|
||
parser = argparse.ArgumentParser(description="Browser Automation Tool")
|
||
parser.add_argument("command", choices=["navigate", "snapshot", "screenshot", "check"])
|
||
parser.add_argument("--url", help="URL for navigate")
|
||
parser.add_argument("--mode", default="launch", choices=["launch", "cdp"])
|
||
parser.add_argument("--headless", action="store_true")
|
||
parser.add_argument("--output", help="Output file path")
|
||
|
||
args = parser.parse_args()
|
||
|
||
with BrowserTool(mode=args.mode, headless=args.headless) as browser:
|
||
if args.command == "navigate":
|
||
if not args.url:
|
||
print("Error: --url required for navigate")
|
||
exit(1)
|
||
result = browser.navigate(args.url)
|
||
print(result)
|
||
|
||
elif args.command == "snapshot":
|
||
# 需要先 navigate
|
||
print("Error: Use navigate command first")
|
||
exit(1)
|
||
|
||
elif args.command == "screenshot":
|
||
if not args.url:
|
||
print("Error: --url required")
|
||
exit(1)
|
||
browser.navigate(args.url)
|
||
path = browser.screenshot(args.output)
|
||
print(f"Screenshot saved: {path}")
|
||
|
||
elif args.command == "check":
|
||
browser.navigate("https://bot.sannysoft.com/")
|
||
results = browser.check_detection()
|
||
print(json.dumps(results, indent=2))
|