diff --git a/plugins/browser_auto.py b/plugins/browser_auto.py new file mode 100644 index 0000000..b266af5 --- /dev/null +++ b/plugins/browser_auto.py @@ -0,0 +1,549 @@ +""" +浏览器自动化插件 - 无头浏览器操控 +基于 Playwright,支持页面导航、元素交互、截图、JS 执行 +""" +import os +import json +import base64 +import time +from typing import Optional + + +class Plugin: + """浏览器自动化 - 网页操控、表单填写、数据抓取、截图""" + + def __init__(self): + self.usage = """浏览器自动化工具 + +连接: + start - 启动浏览器实例 + stop - 关闭浏览器 + status - 查看浏览器状态 + +导航: + open - 打开网页 + back - 后退 + forward - 前进 + reload - 刷新页面 + title - 获取页面标题 + url - 获取当前 URL + +交互: + click <选择器> - 点击元素 + fill <选择器> <文本> - 填写输入框 + select <选择器> <值> - 下拉选择 + check <选择器> - 勾选复选框 + uncheck <选择器> - 取消勾选 + hover <选择器> - 鼠标悬停 + press <按键> - 按键 (Enter/Tab/Escape 等) + +提取: + text [选择器] - 获取元素文本 (默认 body) + html [选择器] - 获取元素 HTML + attr <选择器> <属性名> - 获取元素属性 + value <选择器> - 获取输入框值 + eval - 执行 JavaScript + query <选择器> - 查询匹配元素列表 + +截图: + screenshot [路径] - 页面截图 (默认 screenshot.png) + screenshot --selector <选择器> - 元素截图 + +等待: + wait <选择器> [超时秒数] - 等待元素出现 + wait-gone <选择器> [超时秒数] - 等待元素消失 + wait-url [超时秒数] - 等待 URL 变化 + +高级: + pdf [路径] - 保存页面为 PDF + cookies - 获取所有 cookies + set-cookie - 设置 cookie + viewport <宽> <高> - 设置视口大小 + download [保存路径] - 下载文件 +""" + self.cli = None + self._browser = None + self._context = None + self._page = None + self._pw = None + + def set_cli(self, cli): + self.cli = cli + + def get_tool_info(self): + return { + "name": "browser_auto", + "description": "浏览器自动化 - 网页导航、元素交互、表单填写、数据抓取、截图、JS执行", + "keywords": ["浏览器", "browser", "网页", "点击", "截图", "screenshot", "爬虫", + "playwright", "自动化", "表单", "抓取", "导航", "puppeteer"], + "usage": self.usage + } + + def get_mcp_definition(self): + return { + "name": "browser_auto", + "description": "浏览器自动化工具", + "inputSchema": { + "type": "object", + "properties": { + "operation": {"type": "string", "description": "操作类型"}, + "url": {"type": "string", "description": "网页 URL"}, + "selector": {"type": "string", "description": "CSS 选择器"}, + "value": {"type": "string", "description": "输入值"}, + "script": {"type": "string", "description": "JavaScript 代码"}, + "path": {"type": "string", "description": "保存路径"} + }, + "required": ["operation"] + } + } + + def convert_mcp_args(self, arguments): + op = arguments.get("operation", "") + url = arguments.get("url", "") + selector = arguments.get("selector", "") + return f"{op} {url} {selector}".strip() + + # ── 路由 ── + + def handle(self, args: str) -> str: + try: + parts = args.strip().split(maxsplit=1) + if not parts: + return "错误:请提供操作类型" + + op = parts[0].lower() + rest = parts[1] if len(parts) > 1 else "" + + handlers = { + "start": self._start, + "stop": self._stop, + "status": self._status, + "open": self._open, + "back": self._back, + "forward": self._forward, + "reload": self._reload, + "title": self._title, + "url": self._url, + "click": self._click, + "fill": self._fill, + "select": self._select, + "check": self._check, + "uncheck": self._uncheck, + "hover": self._hover, + "press": self._press, + "text": self._text, + "html": self._html, + "attr": self._attr, + "value": self._value, + "eval": self._eval, + "query": self._query, + "screenshot": self._screenshot, + "wait": self._wait, + "wait-gone": self._wait_gone, + "wait-url": self._wait_url, + "pdf": self._pdf, + "cookies": self._cookies, + "set-cookie": self._set_cookie, + "viewport": self._viewport, + "download": self._download, + } + + handler = handlers.get(op) + if not handler: + return f"错误:不支持的操作 '{op}'" + return handler(rest) + + except Exception as e: + return f"浏览器自动化错误: {str(e)}" + + # ── 浏览器生命周期 ── + + def _ensure_playwright(self): + """确保 Playwright 已安装并可用""" + try: + from playwright.sync_api import sync_playwright + return sync_playwright + except ImportError: + return None + + def _start(self, args: str) -> str: + if self._browser and self._page: + return "浏览器已在运行中" + + sync_playwright = self._ensure_playwright() + if not sync_playwright: + return ("❌ 未安装 Playwright。安装步骤:\n" + " pip install playwright\n" + " playwright install chromium") + + try: + self._pw = sync_playwright().start() + self._browser = self._pw.chromium.launch(headless=True) + self._context = self._browser.new_context( + viewport={"width": 1280, "height": 720}, + user_agent="Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" + ) + self._page = self._context.new_page() + return "✅ 浏览器已启动 (headless Chromium, 1280×720)" + except Exception as e: + return f"❌ 启动失败: {e}" + + def _stop(self, args: str) -> str: + try: + if self._context: + self._context.close() + if self._browser: + self._browser.close() + if self._pw: + self._pw.stop() + except Exception: + pass + self._page = None + self._context = None + self._browser = None + self._pw = None + return "✅ 浏览器已关闭" + + def _status(self, args: str) -> str: + if self._page and not self._page.is_closed(): + url = self._page.url + title = self._page.title() + return f"🟢 运行中\n URL: {url}\n 标题: {title}" + return "🔴 未运行" + + def _require_page(self): + """确保页面可用,否则自动启动""" + if self._page and not self._page.is_closed(): + return self._page + self._start("") + return self._page + + # ── 导航 ── + + def _open(self, args: str) -> str: + url = args.strip() + if not url: + return "错误:请提供 URL" + if not url.startswith(("http://", "https://", "file://")): + url = "https://" + url + + page = self._require_page() + try: + page.goto(url, wait_until="domcontentloaded", timeout=30000) + title = page.title() + return f"✅ 已打开: {url}\n 标题: {title}" + except Exception as e: + return f"❌ 打开失败: {e}" + + def _back(self, args: str) -> str: + page = self._require_page() + page.go_back() + return f"✅ 后退 → {page.url}" + + def _forward(self, args: str) -> str: + page = self._require_page() + page.go_forward() + return f"✅ 前进 → {page.url}" + + def _reload(self, args: str) -> str: + page = self._require_page() + page.reload() + return f"✅ 已刷新: {page.url}" + + def _title(self, args: str) -> str: + page = self._require_page() + return page.title() or "(无标题)" + + def _url(self, args: str) -> str: + page = self._require_page() + return page.url + + # ── 交互 ── + + def _click(self, selector: str) -> str: + selector = selector.strip() + if not selector: + return "错误:请提供 CSS 选择器" + page = self._require_page() + try: + page.click(selector, timeout=10000) + return f"✅ 已点击: {selector}" + except Exception as e: + return f"❌ 点击失败: {e}" + + def _fill(self, args: str) -> str: + parts = args.strip().split(maxsplit=1) + if len(parts) < 2: + return "错误:格式: fill <选择器> <文本>" + selector, text = parts[0], parts[1] + page = self._require_page() + try: + page.fill(selector, text, timeout=10000) + return f"✅ 已填写: {selector}" + except Exception as e: + return f"❌ 填写失败: {e}" + + def _select(self, args: str) -> str: + parts = args.strip().split(maxsplit=1) + if len(parts) < 2: + return "错误:格式: select <选择器> <值>" + selector, value = parts[0], parts[1] + page = self._require_page() + try: + page.select_option(selector, value, timeout=10000) + return f"✅ 已选择: {selector} = {value}" + except Exception as e: + return f"❌ 选择失败: {e}" + + def _check(self, selector: str) -> str: + page = self._require_page() + try: + page.check(selector.strip(), timeout=10000) + return f"✅ 已勾选: {selector}" + except Exception as e: + return f"❌ 勾选失败: {e}" + + def _uncheck(self, selector: str) -> str: + page = self._require_page() + try: + page.uncheck(selector.strip(), timeout=10000) + return f"✅ 已取消勾选: {selector}" + except Exception as e: + return f"❌ 取消勾选失败: {e}" + + def _hover(self, selector: str) -> str: + page = self._require_page() + try: + page.hover(selector.strip(), timeout=10000) + return f"✅ 悬停: {selector}" + except Exception as e: + return f"❌ 悬停失败: {e}" + + def _press(self, key: str) -> str: + page = self._require_page() + page.keyboard.press(key.strip()) + return f"✅ 按键: {key}" + + # ── 提取 ── + + def _text(self, selector: str) -> str: + selector = selector.strip() or "body" + page = self._require_page() + try: + el = page.query_selector(selector) + if el: + text = el.inner_text() + if len(text) > 3000: + text = text[:3000] + f"\n... (截断,共 {len(text)} 字符)" + return text + return f"未找到元素: {selector}" + except Exception as e: + return f"❌ 获取文本失败: {e}" + + def _html(self, selector: str) -> str: + selector = selector.strip() or "body" + page = self._require_page() + try: + el = page.query_selector(selector) + if el: + html = el.inner_html() + if len(html) > 5000: + html = html[:5000] + "\n... (截断)" + return html + return f"未找到元素: {selector}" + except Exception as e: + return f"❌ 获取 HTML 失败: {e}" + + def _attr(self, args: str) -> str: + parts = args.strip().split(maxsplit=1) + if len(parts) < 2: + return "错误:格式: attr <选择器> <属性名>" + selector, attr_name = parts[0], parts[1] + page = self._require_page() + try: + el = page.query_selector(selector) + if el: + val = el.get_attribute(attr_name) + return f"{attr_name} = {val}" + return f"未找到元素: {selector}" + except Exception as e: + return f"❌ 获取属性失败: {e}" + + def _value(self, selector: str) -> str: + page = self._require_page() + try: + el = page.query_selector(selector.strip()) + if el: + return el.input_value() + return f"未找到元素: {selector}" + except Exception as e: + return f"❌ 获取值失败: {e}" + + def _eval(self, script: str) -> str: + script = script.strip() + if not script: + return "错误:请提供 JavaScript 代码" + page = self._require_page() + try: + result = page.evaluate(script) + if isinstance(result, (dict, list)): + return json.dumps(result, ensure_ascii=False, indent=2) + return str(result) + except Exception as e: + return f"❌ JS 执行失败: {e}" + + def _query(self, selector: str) -> str: + selector = selector.strip() + if not selector: + return "错误:请提供 CSS 选择器" + page = self._require_page() + try: + elements = page.query_selector_all(selector) + if not elements: + return f"未找到匹配 '{selector}' 的元素" + + result = [f"找到 {len(elements)} 个元素:\n"] + for i, el in enumerate(elements[:20]): + tag = el.evaluate("el => el.tagName.toLowerCase()") + text = el.inner_text()[:80].replace('\n', ' ').strip() + result.append(f" [{i}] <{tag}> {text}") + if len(elements) > 20: + result.append(f" ... 还有 {len(elements) - 20} 个") + return '\n'.join(result) + except Exception as e: + return f"❌ 查询失败: {e}" + + # ── 截图 ── + + def _screenshot(self, args: str) -> str: + page = self._require_page() + + # 解析参数 + path = "screenshot.png" + selector = None + parts = args.strip().split() + i = 0 + while i < len(parts): + if parts[i] == "--selector" and i + 1 < len(parts): + selector = parts[i + 1] + i += 2 + elif not parts[i].startswith("-"): + path = parts[i] + i += 1 + else: + i += 1 + + try: + if selector: + el = page.query_selector(selector) + if not el: + return f"未找到元素: {selector}" + el.screenshot(path=path) + else: + page.screenshot(path=path, full_page=True) + + size = os.path.getsize(path) + return f"✅ 截图已保存: {path} ({size/1024:.1f} KB)" + except Exception as e: + return f"❌ 截图失败: {e}" + + # ── 等待 ── + + def _wait(self, args: str) -> str: + parts = args.strip().split() + if not parts: + return "错误:请提供选择器" + selector = parts[0] + timeout = int(parts[1]) * 1000 if len(parts) > 1 else 30000 + page = self._require_page() + try: + page.wait_for_selector(selector, timeout=timeout) + return f"✅ 元素已出现: {selector}" + except Exception: + return f"⏱️ 等待超时: {selector}" + + def _wait_gone(self, args: str) -> str: + parts = args.strip().split() + if not parts: + return "错误:请提供选择器" + selector = parts[0] + timeout = int(parts[1]) * 1000 if len(parts) > 1 else 30000 + page = self._require_page() + try: + page.wait_for_selector(selector, state="hidden", timeout=timeout) + return f"✅ 元素已消失: {selector}" + except Exception: + return f"⏱️ 等待超时: {selector}" + + def _wait_url(self, args: str) -> str: + parts = args.strip().split() + if not parts: + return "错误:请提供 URL 片段" + url_part = parts[0] + timeout = int(parts[1]) * 1000 if len(parts) > 1 else 30000 + page = self._require_page() + try: + page.wait_for_url(f"**{url_part}**", timeout=timeout) + return f"✅ URL 已变化: {page.url}" + except Exception: + return f"⏱️ 等待超时: URL 未变化" + + # ── 高级 ── + + def _pdf(self, args: str) -> str: + path = args.strip() or "page.pdf" + page = self._require_page() + try: + page.pdf(path=path) + size = os.path.getsize(path) + return f"✅ PDF 已保存: {path} ({size/1024:.1f} KB)" + except Exception as e: + return f"❌ PDF 生成失败: {e}" + + def _cookies(self, args: str) -> str: + page = self._require_page() + cookies = page.context.cookies() + if not cookies: + return "无 cookies" + result = [f"Cookies ({len(cookies)} 个):\n"] + for c in cookies[:20]: + result.append(f" {c['name']}: {c['value'][:50]} (domain: {c.get('domain', '-')})") + return '\n'.join(result) + + def _set_cookie(self, args: str) -> str: + try: + cookie = json.loads(args.strip()) + except json.JSONDecodeError: + return "错误:cookie 必须是 JSON 格式: {\"name\": \"...\", \"value\": \"...\", \"domain\": \"...\"}" + page = self._require_page() + page.context.add_cookies([cookie]) + return f"✅ 已设置 cookie: {cookie.get('name', '?')}" + + def _viewport(self, args: str) -> str: + parts = args.strip().split() + if len(parts) < 2: + return "错误:格式: viewport <宽> <高>" + try: + w, h = int(parts[0]), int(parts[1]) + except ValueError: + return "错误:宽高必须是整数" + page = self._require_page() + page.set_viewport_size({"width": w, "height": h}) + return f"✅ 视口已设置: {w}×{h}" + + def _download(self, args: str) -> str: + parts = args.strip().split() + if not parts: + return "错误:格式: download [保存路径]" + url = parts[0] + save_path = parts[1] if len(parts) > 1 else os.path.basename(url) or "download" + page = self._require_page() + try: + with page.expect_download() as download_info: + page.evaluate(f"window.open('{url}')") + download = download_info.value + download.save_as(save_path) + size = os.path.getsize(save_path) + return f"✅ 已下载: {save_path} ({size/1024:.1f} KB)" + except Exception as e: + return f"❌ 下载失败: {e}" diff --git a/tests/test_plugins.py b/tests/test_plugins.py index 8c01ea3..3d7eef3 100644 --- a/tests/test_plugins.py +++ b/tests/test_plugins.py @@ -322,6 +322,37 @@ class TestTaskManager: assert "已清除" in result +# ══════════════════════════════════════ +# 引擎加载测试 +# ══════════════════════════════════════ + +class TestBrowserAuto: + """browser_auto 功能测试""" + + @pytest.fixture + def browser(self): + mod = load_module("browser_auto", os.path.join(PLUGINS_DIR, "browser_auto.py")) + return mod.Plugin() + + def test_status_stopped(self, browser): + result = browser.handle("status") + assert "未运行" in result + + def test_invalid_operation(self, browser): + result = browser.handle("nonexistent") + assert "错误" in result + + def test_empty_args(self, browser): + result = browser.handle("") + assert "错误" in result + + def test_tool_info(self, browser): + info = browser.get_tool_info() + assert info["name"] == "browser_auto" + assert "浏览器" in info["description"] + assert len(info["keywords"]) > 0 + + # ══════════════════════════════════════ # 引擎加载测试 # ══════════════════════════════════════