别再手动抓包了！AI 时代的前端调试神器——chrome-devtools-mcp 完全评测

传统方式：开发者 → 手动操作浏览器 → 观察结果 → 修改代码 → 循环
AI 时代：开发者 → AI 助手操控浏览器 → AI 观察结果 → AI 修复问题

python3 --version
# 预期输出：Python 3.10.x 或更高版本

# 如果版本过低，需要升级 Python
# macOS 用户可以使用 Homebrew：
brew install python@3.11

pip install chrome-devtools-mcp

pip show chrome-devtools-mcp
# 应该看到类似输出：
# Name: chrome-devtools-mcp
# Version: x.x.x
# Summary: MCP server for Chrome DevTools

# 关闭所有 Chrome 窗口，然后执行：
/Applications/Google\ Chrome.app/Contents/MacOS/Google\ Chrome \
    --remote-debugging-port=9222 \
    --user-data-dir=/tmp/chrome-debug

google-chrome \
    --remote-debugging-port=9222 \
    --user-data-dir=/tmp/chrome-debug

"C:\Program Files\Google\Chrome\Application\chrome.exe" `
    --remote-debugging-port=9222 `
    --user-data-dir="C:\temp\chrome-debug"

open http://localhost:9222/json

[
  {
    "id": "xxx",
    "type": "page",
    "title": "新标签页",
    "url": "chrome://newtab/",
    "webSocketDebuggerUrl": "ws://localhost:9222/devtools/page/xxx"
  }
]

chrome-devtools-mcp
├── 页面控制
│   ├── 导航控制（打开、刷新、前进、后退）
│   ├── 页面截图
│   └── 页面信息获取
├── 网络控制
│   ├── 网络请求监听
│   └── 请求拦截
├── JavaScript 执行
│   ├── 代码注入
│   └── 控制台交互
├── DOM 操作
│   ├── 元素查询
│   ├── 属性修改
│   └── 事件监听
└── 控制台交互
    ├── 日志读取
    └── 消息发送

# 导航到指定网页
await page.navigate("https://example.com")

# 刷新当前页面
await page.reload()

# 获取完整页面截图
screenshot = await page.screenshot(full_page=True)

# 获取视口截图
screenshot = await page.screenshot(full_page=False)

# 在页面中执行 JavaScript
result = await page.evaluate("""
    () => {
        // 获取页面标题
        return document.title;
    }
""")

# 修改页面元素
await page.evaluate("""
    () => {
        const heading = document.querySelector('h1');
        if (heading) {
            heading.textContent = 'Modified by MCP!';
            heading.style.color = 'red';
        }
    }
""")

# 点击按钮
await page.evaluate("""
    () => {
        const button = document.querySelector('#submit-btn');
        if (button) button.click();
    }
""")

# 填写表单
await page.evaluate("""
    () => {
        const input = document.querySelector('input[name="username"]');
        if (input) {
            input.value = 'testuser';
            input.dispatchEvent(new Event('input', { bubbles: true }));
        }
    }
""")

# 开启网络监控
await network.enable()

# 监听请求完成事件
async def handle_request_finished(params):
    request = params.get('request')
    print(f"Request: {request['url']}")
    print(f"Method: {request['method']}")
    print(f"Status: {params.get('response', {}).get('status')}")

# 注册事件处理器
session.on('Network.requestFinished', handle_request_finished)

# 监听控制台日志
async def handle_console_message(params):
    message_type = params.get('type')
    message_text = params.get('args', [])

    if message_type == 'log':
        print(f"[Console Log] {message_text}")
    elif message_type == 'error':
        print(f"[Console Error] {message_text}")

# 注册控制台消息处理器
session.on('Runtime.consoleAPICalled', handle_console_message)

# 使用 CSS 选择器查询元素
from devtools_puppeteer import Page

page = Page(ws_url)
await page.query_selector("div.container")

login-automation/
├── main.py              # 主程序入口
├── config.py            # 配置文件
├── tests/
│   └── test_login.py    # 登录测试用例
├── reports/
│   └── screenshots/     # 截图保存目录
└── requirements.txt     # 依赖文件

# config.py
"""
配置文件
包含测试所需的各项配置参数
"""

# Chrome 调试端口
CHROME_DEBUG_PORT = 9222

# 测试目标网站
TEST_URLS = {
    "login_page": "https://example.com/login",
    "dashboard": "https://example.com/dashboard",
}

# 测试账号
TEST_CREDENTIALS = {
    "username": "testuser@example.com",
    "password": "TestPass123!",
}

# 超时设置（秒）
TIMEOUT = {
    "page_load": 30,
    "element_wait": 10,
    "network_idle": 5,
}

# 截图保存路径
SCREENSHOT_DIR = "./reports/screenshots"

# browser.py
"""
浏览器连接管理模块
处理 Chrome DevTools Protocol 的连接和初始化
"""

import asyncio
import json
from typing import Optional, Dict, Any
from urllib.parse import urljoin

try:
    import websockets
except ImportError:
    print("请安装 websockets 库：pip install websockets")
    raise


class ChromeConnection:
    """
    Chrome 连接管理类
    负责与 Chrome DevTools Protocol 通信
    """

    def __init__(self, debug_port: int = 9222):
        self.debug_port = debug_port
        self.ws_url: Optional[str] = None
        self.ws: Optional[websockets.WebSocketClientProtocol] = None
        self.message_id = 0
        self.pending_responses: Dict[int, asyncio.Future] = {}

    async def connect(self) -> None:
        """
        连接到 Chrome 调试实例
        """
        # 获取 WebSocket URL
        debug_url = f"http://localhost:{self.debug_port}/json"

        async with websockets.connect(debug_url) as ws:
            self.ws = ws
            self.ws_url = ws.url

            # 获取页面列表
            pages = await self.receive_message()
            print(f"发现 {len(pages)} 个浏览器标签页")

            # 选择第一个页面（或创建新页面）
            if pages:
                self.ws_url = pages[0].get('webSocketDebuggerUrl')
                print(f"已连接到页面：{pages[0].get('title')}")

    async def send_command(self, method: str, params: Optional[Dict] = None) -> Any:
        """
        发送 CDP 命令

        Args:
            method: CDP 方法名
            params: 命令参数

        Returns:
            命令响应结果
        """
        self.message_id += 1
        message = {
            "id": self.message_id,
            "method": method,
            "params": params or {}
        }

        await self.ws.send(json.dumps(message))
        response = await self.receive_response(self.message_id)

        return response

    async def receive_response(self, msg_id: int) -> Dict:
        """
        等待并接收指定 ID 的响应
        """
        while True:
            response = await self.receive_message()

            if isinstance(response, dict) and response.get('id') == msg_id:
                return response.get('result', {})

            # 处理事件通知
            if isinstance(response, dict) and 'method' in response:
                await self.handle_event(response)

    async def receive_message(self) -> Any:
        """
        接收 WebSocket 消息
        """
        raw_message = await self.ws.recv()
        return json.loads(raw_message)

    async def handle_event(self, event: Dict) -> None:
        """
        处理事件通知
        子类可以重写此方法处理特定事件
        """
        pass

    async def close(self) -> None:
        """
        关闭连接
        """
        if self.ws:
            await self.ws.close()

# page.py
"""
页面操作模块
封装常用的页面操作方法
"""

import base64
import asyncio
from typing import Optional, Callable, List, Dict, Any
from pathlib import Path


class Page:
    """
    页面操作类
    提供与页面交互的高级 API
    """

    def __init__(self, connection):
        self.connection = connection
        self.target_id: Optional[str] = None

    async def navigate(self, url: str) -> Dict:
        """
        导航到指定 URL

        Args:
            url: 目标网页地址

        Returns:
            导航结果
        """
        result = await self.connection.send_command(
            "Page.navigate",
            {"url": url}
        )

        # 等待页面加载完成
        await self.wait_for_load_state("load")

        return result

    async def reload(self, ignore_cache: bool = False) -> Dict:
        """
        刷新页面

        Args:
            ignore_cache: 是否忽略缓存

        Returns:
            刷新结果
        """
        return await self.connection.send_command(
            "Page.reload",
            {"ignoreCache": ignore_cache}
        )

    async def screenshot(self, path: Optional[str] = None, 
                        full_page: bool = True,
                        format: str = "png") -> Optional[bytes]:
        """
        页面截图

        Args:
            path: 保存路径，为空则返回图片数据
            full_page: 是否截取完整页面
            format: 图片格式 (png/jpeg)

        Returns:
            截图数据或保存路径
        """
        # 启用 Page 模块
        await self.connection.send_command("Page.enable")

        # 执行截图
        result = await self.connection.send_command(
            "Page.captureScreenshot",
            {
                "format": format,
                "captureBeyondViewport": full_page,
                "fromSurface": True
            }
        )

        # 解码图片
        if "data" in result:
            image_data = base64.b64decode(result["data"])

            if path:
                Path(path).parent.mkdir(parents=True, exist_ok=True)
                with open(path, "wb") as f:
                    f.write(image_data)
                print(f"截图已保存：{path}")
                return None
            else:
                return image_data

        return None

    async def evaluate(self, expression: str, 
                       await_promise: bool = False) -> Any:
        """
        在页面上下文中执行 JavaScript

        Args:
            expression: JavaScript 表达式或函数
            await_promise: 是否等待 Promise 解决

        Returns:
            执行结果
        """
        result = await self.connection.send_command(
            "Runtime.evaluate",
            {
                "expression": expression,
                "returnByValue": True,
                "awaitPromise": await_promise
            }
        )

        if result.get("exceptionDetails"):
            print(f"JavaScript 执行错误：{result['exceptionDetails']}")
            return None

        return result.get("result", {}).get("value")

    async def query_selector(self, selector: str) -> Optional[Dict]:
        """
        查询单个元素

        Args:
            selector: CSS 选择器

        Returns:
            元素信息
        """
        result = await self.evaluate(f"""
            () => {{
                const element = document.querySelector('{selector}');
                if (!element) return null;

                return {{
                    tagName: element.tagName,
                    textContent: element.textContent?.trim(),
                    innerHTML: element.innerHTML,
                    attributes: Array.from(element.attributes).reduce(
                        (acc, attr) => {{
                            acc[attr.name] = attr.value;
                            return acc;
                        }}, {{}}
                    ),
                    boundingBox: element.getBoundingClientRect()
                }};
            }}
        """)

        return result

    async def fill_input(self, selector: str, value: str) -> bool:
        """
        填写输入框

        Args:
            selector: 输入框选择器
            value: 填写的值

        Returns:
            是否成功
        """
        script = f"""
            () => {{
                const element = document.querySelector('{selector}');
                if (!element) return false;

                // 聚焦元素
                element.focus();

                // 清除现有值
                element.value = '';

                // 设置新值
                const nativeInputValueSetter = 
                    Object.getOwnPropertyDescriptor(
                        window.HTMLInputElement.prototype, 
                        'value'
                    ).set;
                nativeInputValueSetter.call(element, '{value}');

                // 触发输入事件
                element.dispatchEvent(new Event('input', {{ bubbles: true }}));
                element.dispatchEvent(new Event('change', {{ bubbles: true }}));

                return true;
            }}
        """

        return await self.evaluate(script) or False

    async def click(self, selector: str) -> bool:
        """
        点击元素

        Args:
            selector: 元素选择器

        Returns:
            是否成功
        """
        script = f"""
            () => {{
                const element = document.querySelector('{selector}');
                if (!element) return false;

                // 创建并触发点击事件
                const event = new MouseEvent('click', {{
                    view: window,
                    bubbles: true,
                    cancelable: true
                }});
                element.dispatchEvent(event);

                return true;
            }}
        """

        return await self.evaluate(script) or False

    async def wait_for_selector(self, selector: str, timeout: int = 10000) -> bool:
        """
        等待元素出现

        Args:
            selector: CSS 选择器
            timeout: 超时时间（毫秒）

        Returns:
            元素是否出现
        """
        start_time = asyncio.get_event_loop().time()

        while (asyncio.get_event_loop().time() - start_time) * 1000 < timeout:
            exists = await self.evaluate(f"""
                () => document.querySelector('{selector}') !== null
            """)

            if exists:
                return True

            await asyncio.sleep(0.1)

        return False

    async def wait_for_load_state(self, state: str = "load", timeout: int = 30000) -> None:
        """
        等待页面加载状态

        Args:
            state: 加载状态 (load/domcontentloaded/networkidle)
            timeout: 超时时间（毫秒）
        """
        await self.connection.send_command("Page.setLifecycleEventsEnabled", {"enabled": True})

        event_received = asyncio.Event()

        async def handler(params):
            if params.get("name") == state:
                event_received.set()

        # 临时保存事件处理器
        original_handler = self.connection.handle_event
        self.connection.handle_event = handler

        try:
            await asyncio.wait_for(event_received.wait(), timeout=timeout/1000)
        except asyncio.TimeoutError:
            print(f"等待 {state} 状态超时")
        finally:
            self.connection.handle_event = original_handler

    async def get_console_messages(self) -> List[Dict]:
        """
        获取控制台消息

        Returns:
            控制台消息列表
        """
        result = await self.connection.send_command(
            "Runtime.getConsoleMessages",
            {}
        )

        return result.get("messages", [])

    async def clear_console(self) -> None:
        """
        清空控制台
        """
        await self.connection.send_command("Runtime.discardConsoleEntries", {})

# tests/test_login.py
"""
登录功能测试用例
演示如何使用 chrome-devtools-mcp 进行自动化测试
"""

import asyncio
import sys
from pathlib import Path

# 添加项目根目录到路径
sys.path.insert(0, str(Path(__file__).parent.parent))

from browser import ChromeConnection
from page import Page
from config import (
    TEST_URLS, 
    TEST_CREDENTIALS, 
    TIMEOUT,
    SCREENSHOT_DIR
)


class LoginAutomation:
    """
    登录自动化测试类
    """

    def __init__(self):
        self.connection: ChromeConnection = None
        self.page: Page = None
        self.test_results = []

    async def setup(self):
        """
        测试初始化
        """
        print("=" * 50)
        print("初始化浏览器连接...")
        print("=" * 50)

        # 创建连接
        self.connection = ChromeConnection(debug_port=9222)
        await self.connection.connect()

        # 创建页面对象
        self.page = Page(self.connection)

        print("浏览器连接成功！\n")

    async def teardown(self):
        """
        测试清理
        """
        if self.connection:
            await self.connection.close()
            print("\n浏览器连接已关闭")

    async def take_screenshot(self, name: str) -> str:
        """
        截取屏幕截图

        Args:
            name: 截图文件名

        Returns:
            截图保存路径
        """
        # 确保目录存在
        Path(SCREENSHOT_DIR).mkdir(parents=True, exist_ok=True)

        # 截取页面
        filename = f"{SCREENSHOT_DIR}/{name}.png"
        await self.page.screenshot(filename, full_page=True)

        return filename

    async def test_open_login_page(self) -> bool:
        """
        测试用例 1：打开登录页面
        """
        print("\n[测试 1] 打开登录页面")
        print("-" * 30)

        try:
            # 导航到登录页面
            await self.page.navigate(TEST_URLS["login_page"])

            # 等待页面加载
            await asyncio.sleep(2)

            # 截图保存
            await self.take_screenshot("01_login_page_loaded")

            # 验证页面标题
            title = await self.page.evaluate("() => document.title")
            print(f"页面标题：{title}")

            self.test_results.append({
                "name": "打开登录页面",
                "status": "PASS",
                "message": f"成功加载页面，标题：{title}"
            })

            return True

        except Exception as e:
            self.test_results.append({
                "name": "打开登录页面",
                "status": "FAIL",
                "message": str(e)
            })
            print(f"错误：{e}")
            return False

    async def test_fill_login_form(self) -> bool:
        """
        测试用例 2：填写登录表单
        """
        print("\n[测试 2] 填写登录表单")
        print("-" * 30)

        try:
            # 填写用户名
            username_filled = await self.page.fill_input(
                "input[name='username'], input[type='email'], input[id='username']",
                TEST_CREDENTIALS["username"]
            )

            if username_filled:
                print(f"✓ 用户名已填写：{TEST_CREDENTIALS['username']}")
            else:
                print("✗ 用户名输入失败，尝试备用选择器...")
                # 备用方案：直接通过 JavaScript 填写
                await self.page.evaluate(f"""
                    () => {{
                        const inputs = document.querySelectorAll('input');
                        for (const input of inputs) {{
                            if (input.type !== 'password') {{
                                input.value = '{TEST_CREDENTIALS["username"]}';
                                input.dispatchEvent(new Event('input', {{ bubbles: true }}));
                                break;
                            }}
                        }}
                    }}
                """)

            # 填写密码
            await asyncio.sleep(0.5)
            await self.page.evaluate(f"""
                () => {{
                    const inputs = document.querySelectorAll('input[type="password"]');
                    if (inputs.length > 0) {{
                        inputs[0].value = '{TEST_CREDENTIALS["password"]}';
                        inputs[0].dispatchEvent(new Event('input', {{ bubbles: true }}));
                    }}
                }}
            """)
            print("✓ 密码已填写")

            # 截图保存
            await self.take_screenshot("02_form_filled")

            self.test_results.append({
                "name": "填写登录表单",
                "status": "PASS",
                "message": "表单已成功填写"
            })

            return True

        except Exception as e:
            self.test_results.append({
                "name": "填写登录表单",
                "status": "FAIL",
                "message": str(e)
            })
            print(f"错误：{e}")
            return False

    async def test_submit_login(self) -> bool:
        """
        测试用例 3：提交登录表单
        """
        print("\n[测试 3] 提交登录表单")
        print("-" * 30)

        try:
            # 查找并点击登录按钮
            clicked = await self.page.evaluate("""
                () => {
                    // 尝试多种选择器
                    const selectors = [
                        'button[type="submit"]',
                        'button:contains("登录")',
                        'button:contains("登录")',
                        'input[type="submit"]',
                        '.login-btn',
                        '#login-btn',
                        '[data-action="login"]'
                    ];

                    for (const selector of selectors) {
                        try {
                            const btn = document.querySelector(selector);
                            if (btn) {
                                btn.click();
                                return true;
                            }
                        } catch (e) {
                            continue;
                        }
                    }
                    return false;
                }
            """)

            if clicked:
                print("✓ 登录按钮已点击")
            else:
                print("✗ 未找到登录按钮，尝试按下 Enter 键...")
                await self.page.evaluate("""
                    () => {
                        const form = document.querySelector('form');
                        if (form) {
                            form.dispatchEvent(new Event('submit', { bubbles: true, cancelable: true }));
                        }
                    }
                """)

            # 等待页面响应
            await asyncio.sleep(3)

            # 截图保存
            await self.take_screenshot("03_after_login_submit")

            # 验证登录结果
            current_url = await self.page.evaluate("() => window.location.href")
            print(f"当前 URL：{current_url}")

            # 检查是否跳转到仪表盘
            if "dashboard" in current_url.lower() or "welcome" in current_url.lower():
                self.test_results.append({
                    "name": "提交登录表单",
                    "status": "PASS",
                    "message": "登录成功，已跳转到仪表盘"
                })
                return True
            else:
                self.test_results.append({
                    "name": "提交登录表单",
                    "status": "PARTIAL",
                    "message": f"表单已提交，当前 URL：{current_url}"
                })
                return True

        except Exception as e:
            self.test_results.append({
                "name": "提交登录表单",
                "status": "FAIL",
                "message": str(e)
            })
            print(f"错误：{e}")
            return False

    async def test_verify_dashboard(self) -> bool:
        """
        测试用例 4：验证仪表盘页面
        """
        print("\n[测试 4] 验证仪表盘页面")
        print("-" * 30)

        try:
            # 获取页面信息
            page_info = await self.page.evaluate("""
                () => {
                    return {
                        title: document.title,
                        url: window.location.href,
                        hasContent: document.body.children.length > 0,
                        mainContent: document.querySelector('main, .content, #content')?.textContent?.trim()?.substring(0, 200)
                    };
                }
            """)

            print(f"页面标题：{page_info.get('title')}")
            print(f"页面 URL：{page_info.get('url')}")
            print(f"主要内容：{page_info.get('mainContent', '无')}...")

            # 截图保存
            await self.take_screenshot("04_dashboard_verified")

            self.test_results.append({
                "name": "验证仪表盘",
                "status": "PASS",
                "message": f"成功进入仪表盘页面"
            })

            return True

        except Exception as e:
            self.test_results.append({
                "name": "验证仪表盘",
                "status": "FAIL",
                "message": str(e)
            })
            print(f"错误：{e}")
            return False

    async def test_console_output(self) -> bool:
        """
        测试用例 5：检查控制台输出
        """
        print("\n[测试 5] 检查控制台输出")
        print("-" * 30)

        try:
            # 获取控制台消息
            messages = await self.page.get_console_messages()

            print(f"共捕获 {len(messages)} 条控制台消息：\n")

            for msg in messages[:10]:  # 只显示前 10 条
                msg_type = msg.get('type', 'log')
                msg_text = msg.get('parameters', [])

                if msg_text and len(msg_text) > 0:
                    text = msg_text[0].get('value', str(msg_text))
                    print(f"  [{msg_type.upper()}] {text}")

            self.test_results.append({
                "name": "控制台检查",
                "status": "PASS",
                "message": f"捕获 {len(messages)} 条消息"
            })

            return True

        except Exception as e:
            self.test_results.append({
                "name": "控制台检查",
                "status": "FAIL",
                "message": str(e)
            })
            print(f"错误：{e}")
            return False

    def print_test_summary(self):
        """
        打印测试摘要
        """
        print("\n" + "=" * 50)
        print("测试摘要")
        print("=" * 50)

        passed = sum(1 for r in self.test_results if r["status"] == "PASS")
        failed = sum(1 for r in self.test_results if r["status"] == "FAIL")
        partial = sum(1 for r in self.test_results if r["status"] == "PARTIAL")

        print(f"\n总计：{len(self.test_results)} 个测试")
        print(f"通过：{passed}")
        print(f"失败：{failed}")
        print(f"部分通过：{partial}")

        print("\n详细结果：")
        print("-" * 50)

        for result in self.test_results:
            status_icon = "✓" if result["status"] == "PASS" else ("△" if result["status"] == "PARTIAL" else "✗")
            print(f"{status_icon} {result['name']}: {result['message']}")

        print("\n" + "=" * 50)

    async def run(self):
        """
        运行所有测试
        """
        try:
            # 初始化
            await self.setup()

            # 运行测试用例
            await self.test_open_login_page()
            await self.test_fill_login_form()
            await self.test_submit_login()
            await self.test_verify_dashboard()
            await self.test_console_output()

            # 打印摘要
            self.print_test_summary()

        finally:
            # 清理
            await self.teardown()


async def main():
    """
    主函数入口
    """
    print("\n" + "=" * 50)
    print("chrome-devtools-mcp 自动化测试")
    print("=" * 50 + "\n")

    # 创建并运行测试
    automation = LoginAutomation()
    await automation.run()


if __name__ == "__main__":
    asyncio.run(main())

# 确保 Chrome 调试模式已启动
# 然后运行测试脚本
python tests/test_login.py

==================================================
chrome-devtools-mcp 自动化测试
==================================================

==================================================
初始化浏览器连接...
==================================================
浏览器连接成功！

[测试 1] 打开登录页面
------------------------------
页面标题：登录
截图已保存：./reports/screenshots/01_login_page_loaded.png
✓ 用户名已填写：testuser@example.com

[测试 2] 填写登录表单
------------------------------
✓ 密码已填写
截图已保存：./reports/screenshots/02_form_filled.png

[测试 3] 提交登录表单
------------------------------
✓ 登录按钮已点击
当前 URL：https://example.com/dashboard
截图已保存：./reports/screenshots/03_after_login_submit.png

[测试 4] 验证仪表盘页面
------------------------------
页面标题：仪表盘
页面 URL：https://example.com/dashboard
截图已保存：./reports/screenshots/04_dashboard_verified.png

[测试 5] 检查控制台输出
------------------------------
共捕获 5 条控制台消息：

  [LOG] 页面加载完成
  [LOG] API 初始化成功
  [LOG] 用户会话已创建

==================================================
测试摘要
==================================================

总计：5 个测试
通过：5
失败：0
部分通过：0

详细结果：
------------------------------
✓ 打开登录页面: 成功加载页面，标题：登录
✓ 填写登录表单: 表单已成功填写
✓ 提交登录表单: 登录成功，已跳转到仪表盘
✓ 验证仪表盘: 成功进入仪表盘页面
✓ 控制台检查: 捕获 5 条消息

==================================================

浏览器连接已关闭

"""
网页数据抓取示例
"""

async def scrape_dynamic_content():
    """
    抓取需要 JavaScript 渲染的页面内容
    """
    connection = ChromeConnection()
    await connection.connect()
    page = Page(connection)

    # 导航到目标页面
    await page.navigate("https://example.com/data-table")

    # 等待数据加载
    await page.wait_for_selector(".data-table tbody tr")

    # 提取数据
    data = await page.evaluate("""
        () => {
            const rows = document.querySelectorAll('.data-table tbody tr');
            return Array.from(rows).map(row => {
                const cells = row.querySelectorAll('td');
                return {
                    id: cells[0]?.textContent,
                    name: cells[1]?.textContent,
                    value: cells[2]?.textContent
                };
            });
        }
    """)

    print(f"抓取到 {len(data)} 条数据")
    for item in data:
        print(f"  {item['id']}: {item['name']} = {item['value']}")

    await connection.close()

"""
UI 测试示例
"""

async def test_ui_components():
    """
    测试 UI 组件的交互功能
    """
    connection = ChromeConnection()
    await connection.connect()
    page = Page(connection)

    # 打开测试页面
    await page.navigate("https://example.com/ui-test")

    # 测试模态框
    print("测试模态框...")
    await page.click(".open-modal-btn")
    await page.wait_for_selector(".modal.show")
    modal_visible = await page.evaluate("""
        () => document.querySelector('.modal.show') !== null
    """)
    print(f"  模态框显示：{'✓' if modal_visible else '✗'}")

    # 测试下拉菜单
    print("测试下拉菜单...")
    await page.click(".dropdown-toggle")
    await page.wait_for_selector(".dropdown-menu.show")
    await page.click(".dropdown-item:nth-child(2)")
    selected = await page.evaluate("""
        () => document.querySelector('.dropdown-toggle').textContent
    """)
    print(f"  选中项：{selected}")

    # 测试表单验证
    print("测试表单验证...")
    await page.fill_input("#email-input", "invalid-email")
    await page.click("#submit-btn")
    error_shown = await page.evaluate("""
        () => document.querySelector('.error-message') !== null
    """)
    print(f"  错误提示显示：{'✓' if error_shown else '✗'}")

    await connection.close()

"""
性能监控示例
"""

async def monitor_page_performance():
    """
    监控页面性能指标
    """
    connection = ChromeConnection()
    await connection.connect()
    page = Page(connection)

    # 启用性能监控
    await connection.send_command("Performance.enable")

    # 导航并测量
    await page.navigate("https://example.com")

    # 获取性能指标
    metrics = await page.evaluate("""
        () => {
            const timing = performance.timing;
            return {
                // 页面加载时间
                pageLoadTime: timing.loadEventEnd - timing.navigationStart,
                // DOM 解析时间
                domParseTime: timing.domContentLoadedEventEnd - timing.domLoading,
                // 首字节时间
                ttfb: timing.responseStart - timing.requestStart,
                // 资源加载时间
                resourceLoadTime: performance.now(),
                // 关键性能指标
                metrics: {
                    fcp: null, // First Contentful Paint
                    lcp: null, // Largest Contentful Paint
                    fid: null, // First Input Delay
                    cls: null  // Cumulative Layout Shift
                }
            };
        }
    """)

    print("性能指标：")
    print(f"  页面加载时间：{metrics['pageLoadTime']}ms")
    print(f"  DOM 解析时间：{metrics['domParseTime']}ms")
    print(f"  首字节时间：{metrics['ttfb']}ms")

    # 获取 Network 请求统计
    await connection.send_command("Network.enable")
    await asyncio.sleep(2)  # 等待网络请求完成

    await connection.close()

// MCP 配置文件 (mcp.json)
{
    "mcpServers": {
        "chrome-devtools": {
            "command": "npx",
            "args": ["-y", "chrome-devtools-mcp"]
        }
    }
}

"""
AI 助手指令示例
"""

AI_COMMANDS = """
你可以使用以下命令控制浏览器：

1. 页面导航
   page.navigate(url) - 打开指定网页
   page.reload() - 刷新页面
   page.back() - 后退
   page.forward() - 前进

2. 元素操作
   page.click(selector) - 点击元素
   page.fill_input(selector, value) - 填写输入框
   page.hover(selector) - 鼠标悬停

3. 内容获取
   page.screenshot(path) - 截取页面
   page.get_text(selector) - 获取文本内容
   page.get_html(selector) - 获取 HTML 内容

4. JavaScript 执行
   page.evaluate(script) - 执行 JavaScript 代码

5. 等待操作
   page.wait_for_selector(selector) - 等待元素出现
   page.wait_for_navigation() - 等待导航完成
"""

print(AI_COMMANDS)

"""
高级错误处理示例
"""

import asyncio
from typing import Optional, Callable, Any
from functools import wraps


class BrowserAutomationError(Exception):
    """浏览器自动化基础异常"""
    pass


class ElementNotFoundError(BrowserAutomationError):
    """元素未找到异常"""
    pass


class TimeoutError(BrowserAutomationError):
    """操作超时异常"""
    pass


async def retry_on_failure(
    max_retries: int = 3,
    delay: float = 1.0,
    exceptions: tuple = (Exception,)
):
    """
    重试装饰器

    Args:
        max_retries: 最大重试次数
        delay: 重试间隔（秒）
        exceptions: 需要重试的异常类型
    """
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        async def wrapper(*args, **kwargs) -> Any:
            last_exception = None

            for attempt in range(max_retries):
                try:
                    return await func(*args, **kwargs)
                except exceptions as e:
                    last_exception = e
                    if attempt < max_retries - 1:
                        print(f"尝试 {attempt + 1}/{max_retries} 失败: {e}")
                        print(f"等待 {delay} 秒后重试...")
                        await asyncio.sleep(delay)
                    else:
                        print(f"已达到最大重试次数 {max_retries}")

            raise last_exception

        return wrapper
    return decorator


class RobustPage(Page):
    """
    增强版的页面操作类
    提供更稳定的操作方法
    """

    async def safe_click(self, selector: str, timeout: int = 10) -> bool:
        """
        安全点击元素，带重试机制

        Args:
            selector: CSS 选择器
            timeout: 超时时间

        Returns:
            是否成功
        """
        @retry_on_failure(max_retries=3, delay=1.0)
        async def _click():
            # 先滚动到元素可见
            await self.evaluate(f"""
                () => {{
                    const el = document.querySelector('{selector}');
                    if (el) {{
                        el.scrollIntoView({{ behavior: 'smooth', block: 'center' }}));
                    }}
                }}
            """)
            await asyncio.sleep(0.5)

            # 执行点击
            result = await self.click(selector)
            if not result:
                raise ElementNotFoundError(f"无法点击元素: {selector}")

            return result

        return await _click()

    async def safe_fill(self, selector: str, value: str, 
                       clear_first: bool = True) -> bool:
        """
        安全填写输入框

        Args:
            selector: CSS 选择器
            value: 填写值
            clear_first: 是否先清空

        Returns:
            是否成功
        """
        @retry_on_failure(max_retries=3, delay=0.5)
        async def _fill():
            if clear_first:
                # 清空输入框
                await self.evaluate(f"""
                    () => {{
                        const el = document.querySelector('{selector}');
                        if (el) {{
                            el.value = '';
                            el.dispatchEvent(new Event('input', {{ bubbles: true }}));
                        }}
                    }}
                """)
                await asyncio.sleep(0.2)

            return await self.fill_input(selector, value)

        return await _fill()

    async def wait_and_get(
        self, 
        selector: str, 
        attribute: Optional[str] = None,
        timeout: int = 10
    ) -> Any:
        """
        等待元素出现并获取其内容

        Args:
            selector: CSS 选择器
            attribute: 要获取的属性，为空则获取文本
            timeout: 超时时间

        Returns:
            元素内容
        """
        # 等待元素
        found = await self.wait_for_selector(selector, timeout * 1000)

        if not found:
            raise ElementNotFoundError(f"元素未找到: {selector}")

        # 获取内容
        if attribute:
            return await self.evaluate(f"""
                () => document.querySelector('{selector}')?.getAttribute('{attribute}')
            """)
        else:
            return await self.evaluate(f"""
                () => document.querySelector('{selector}')?.textContent?.trim()
            """)

"""
并发操作示例
"""

import asyncio
from typing import List, Dict


async def batch_navigate(pages: List[Page], urls: List[str]) -> List[Dict]:
    """
    批量并发导航

    Args:
        pages: 页面对象列表
        urls: URL 列表

    Returns:
        每个页面的导航结果
    """
    async def navigate_single(page: Page, url: str) -> Dict:
        try:
            await page.navigate(url)
            title = await page.evaluate("() => document.title")
            return {"url": url, "status": "success", "title": title}
        except Exception as e:
            return {"url": url, "status": "error", "error": str(e)}

    # 使用 gather 并发执行所有导航
    tasks = [navigate_single(page, url) 
             for page, url in zip(pages, urls)]

    results = await asyncio.gather(*tasks, return_exceptions=True)

    return results


async def scrape_multiple_sites(sites: List[str]) -> Dict[str, List]:
    """
    并发抓取多个网站

    Args:
        sites: 网站列表

    Returns:
        抓取结果
    """
    connection = ChromeConnection()
    await connection.connect()

    # 创建多个标签页
    page_objects = []
    for _ in range(len(sites)):
        page = Page(connection)
        page_objects.append(page)

    # 并发抓取
    results = await batch_navigate(page_objects, sites)

    # 汇总数据
    scraped_data = {}
    for result in results:
        if result.get("status") == "success":
            scraped_data[result["url"]] = {
                "title": result["title"],
                "links": await page_objects[results.index(result)]
                    .evaluate("() => Array.from(document.querySelectorAll('a')).slice(0, 10).map(a => a.href)")
            }

    await connection.close()
    return scraped_data

"""
日志系统示例
"""

import logging
import json
from datetime import datetime
from pathlib import Path
from typing import Optional


class AutomationLogger:
    """
    自动化日志记录器
    """

    def __init__(self, log_dir: str = "./logs"):
        self.log_dir = Path(log_dir)
        self.log_dir.mkdir(parents=True, exist_ok=True)

        # 创建日志文件名（按日期）
        log_file = self.log_dir / f"automation_{datetime.now().strftime('%Y%m%d')}.log"

        # 配置日志
        self.logger = logging.getLogger("BrowserAutomation")
        self.logger.setLevel(logging.DEBUG)

        # 文件处理器
        file_handler = logging.FileHandler(log_file, encoding="utf-8")
        file_handler.setLevel(logging.DEBUG)

        # 控制台处理器
        console_handler = logging.StreamHandler()
        console_handler.setLevel(logging.INFO)

        # 格式化
        formatter = logging.Formatter(
            "%(asctime)s [%(levelname)s] %(message)s",
            datefmt="%Y-%m-%d %H:%M:%S"
        )
        file_handler.setFormatter(formatter)
        console_handler.setFormatter(formatter)

        self.logger.addHandler(file_handler)
        self.logger.addHandler(console_handler)

        # 截图日志
        self.screenshot_log = self.log_dir / "screenshots.json"

    def info(self, message: str):
        self.logger.info(message)

    def debug(self, message: str):
        self.logger.debug(message)

    def warning(self, message: str):
        self.logger.warning(message)

    def error(self, message: str, exc_info: Optional[Exception] = None):
        if exc_info:
            self.logger.error(f"{message}\n{str(exc_info)}", exc_info=True)
        else:
            self.logger.error(message)

    def log_step(self, step_name: str, details: dict = None):
        """记录测试步骤"""
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "step": step_name,
            "details": details or {}
        }
        self.logger.info(f"步骤: {step_name} | 详情: {json.dumps(details, ensure_ascii=False)}")

        # 追加到截图日志
        self._append_screenshot_log(log_entry)

    def _append_screenshot_log(self, entry: dict):
        """追加截图日志"""
        try:
            if self.screenshot_log.exists():
                with open(self.screenshot_log, "r", encoding="utf-8") as f:
                    logs = json.load(f)
            else:
                logs = []

            logs.append(entry)

            with open(self.screenshot_log, "w", encoding="utf-8") as f:
                json.dump(logs, f, ensure_ascii=False, indent=2)
        except Exception as e:
            self.logger.warning(f"无法记录截图日志: {e}")


# 使用示例
async def logged_automation():
    """
    带日志的自动化示例
    """
    logger = AutomationLogger()

    logger.info("开始自动化测试")

    try:
        logger.log_step("初始化浏览器", {"port": 9222})
        # 连接浏览器...

        logger.log_step("打开目标页面", {"url": "https://example.com"})
        # 导航...

        logger.log_step("执行操作", {"action": "click", "selector": "#submit"})
        # 执行点击...

        logger.info("测试完成")

    except Exception as e:
        logger.error("测试失败", exc_info=e)
        raise

"""
配置管理示例
"""

import json
from pathlib import Path
from typing import Optional, Dict, Any
from dataclasses import dataclass, field


@dataclass
class Environment:
    """环境配置"""
    name: str
    base_url: str
    api_url: str
    debug_port: int
    timeout: int = 30
    headless: bool = False


@dataclass
class Credentials:
    """凭证配置"""
    username: str
    password: str
    api_key: Optional[str] = None


@dataclass
class TestConfig:
    """测试配置"""
    environment: Environment
    credentials: Credentials
    screenshot_on_error: bool = True
    retry_count: int = 3
    retry_delay: float = 1.0
    custom_options: Dict[str, Any] = field(default_factory=dict)

    @classmethod
    def from_json(cls, path: str) -> "TestConfig":
        """从 JSON 文件加载配置"""
        with open(path, "r", encoding="utf-8") as f:
            data = json.load(f)

        env_data = data.get("environment", {})
        env = Environment(
            name=env_data.get("name", "default"),
            base_url=env_data.get("base_url", ""),
            api_url=env_data.get("api_url", ""),
            debug_port=env_data.get("debug_port", 9222),
            timeout=env_data.get("timeout", 30),
            headless=env_data.get("headless", False)
        )

        cred_data = data.get("credentials", {})
        creds = Credentials(
            username=cred_data.get("username", ""),
            password=cred_data.get("password", ""),
            api_key=cred_data.get("api_key")
        )

        return cls(
            environment=env,
            credentials=creds,
            screenshot_on_error=data.get("screenshot_on_error", True),
            retry_count=data.get("retry_count", 3),
            retry_delay=data.get("retry_delay", 1.0),
            custom_options=data.get("custom_options", {})
        )

    def to_json(self, path: str):
        """保存配置到 JSON 文件"""
        data = {
            "environment": {
                "name": self.environment.name,
                "base_url": self.environment.base_url,
                "api_url": self.environment.api_url,
                "debug_port": self.environment.debug_port,
                "timeout": self.environment.timeout,
                "headless": self.environment.headless
            },
            "credentials": {
                "username": self.credentials.username,
                "password": self.credentials.password,
                "api_key": self.credentials.api_key
            },
            "screenshot_on_error": self.screenshot_on_error,
            "retry_count": self.retry_count,
            "retry_delay": self.retry_delay,
            "custom_options": self.custom_options
        }

        Path(path).parent.mkdir(parents=True, exist_ok=True)
        with open(path, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)


# 配置文件示例 (config.json)
CONFIG_EXAMPLE = """
{
    "environment": {
        "name": "测试环境",
        "base_url": "https://test.example.com",
        "api_url": "https://api.test.example.com",
        "debug_port": 9222,
        "timeout": 30,
        "headless": false
    },
    "credentials": {
        "username": "testuser@example.com",
        "password": "TestPass123!",
        "api_key": null
    },
    "screenshot_on_error": true,
    "retry_count": 3,
    "retry_delay": 1.0,
    "custom_options": {
        "window_size": [1920, 1080],
        "user_agent": "Mozilla/5.0 (compatible; Bot/1.0)"
    }
}
"""

# 使用配置
async def run_with_config(config_path: str):
    """
    使用配置文件运行测试
    """
    config = TestConfig.from_json(config_path)

    print(f"运行配置：{config.environment.name}")
    print(f"目标地址：{config.environment.base_url}")

    connection = ChromeConnection(debug_port=config.environment.debug_port)
    await connection.connect()

    page = Page(connection)
    await page.navigate(config.environment.base_url)

    # 使用配置的凭证
    await page.fill_input("#username", config.credentials.username)
    await page.fill_input("#password", config.credentials.password)

    await connection.close()

ConnectionRefusedError: [Errno 61] Connection refused

# 1. 确认 Chrome 调试端口是否开启
curl http://localhost:9222/json

# 2. 如果端口被占用，找出并关闭占用进程
lsof -i :9222

# 3. 重新启动 Chrome 调试模式
# macOS
/Applications/Google\ Chrome.app/Contents/MacOS/Google\ Chrome \
    --remote-debugging-port=9223 \
    --user-data-dir=/tmp/chrome-debug

# 确保元素可见并可交互
await page.evaluate("""
    () => {
        const el = document.querySelector('YOUR_SELECTOR');
        el.scrollIntoView({ behavior: 'smooth', block: 'center' });
    }
""")

# 使用 JavaScript 直接触发点击
await page.evaluate("""
    () => {
        const el = document.querySelector('YOUR_SELECTOR');
        el.dispatchEvent(new MouseEvent('click', {
            view: window,
            bubbles: true,
            cancelable: true
        }));
    }
""")

TimeoutError: waiting for selector failed

# 增加超时时间
await page.wait_for_selector(
    "YOUR_SELECTOR",
    timeout=60000  # 60 秒
)

# 或者使用轮询方式
import asyncio

async def wait_for_content(page: Page, condition_fn, timeout: int = 60):
    """等待条件满足"""
    start = asyncio.get_event_loop().time()

    while (asyncio.get_event_loop().time() - start) < timeout:
        result = await page.evaluate(condition_fn)
        if result:
            return True
        await asyncio.sleep(1)

    raise TimeoutError("条件未在超时时间内满足")

result = {'exceptionDetails': {...}}

# 添加错误处理
result = await page.evaluate("""
    () => {
        try {
            return {
                success: true,
                data: document.querySelector('YOUR_SELECTOR').textContent
            };
        } catch (e) {
            return {
                success: false,
                error: e.message
            };
        }
    }
""")

if result.get("success"):
    print(result["data"])
else:
    print(f"JavaScript 错误: {result['error']}")

# 确保文件保存为 UTF-8 编码
# 在代码开头添加
import sys
sys.stdout.reconfigure(encoding='utf-8')

# 读取包含中文的配置
with open("config.json", "r", encoding="utf-8") as f:
    config = json.load(f)

npm install puppeteer

pip install playwright
playwright install

pip install selenium

pip install DrissionPage

npx create-mcp-server