Agent/examples/test_browser/diagnose_cdp.py

"""
底层 CDP 诊断脚本
直接测试 Chrome 的 CDP 端点响应
"""
import subprocess
import time
import httpx
import asyncio


def start_chrome_with_cdp(port=9222, headless=True):
    """启动 Chrome 并返回进程"""
    cmd = [
        '/Applications/Google Chrome.app/Contents/MacOS/Google Chrome',
        f'--remote-debugging-port={port}',
        '--user-data-dir=/tmp/test-cdp-diagnosis',
        '--no-first-run',
        '--no-default-browser-check',
    ]

    if headless:
        cmd.append('--headless=new')

    cmd.append('about:blank')

    print(f"启动命令: {' '.join(cmd)}")

    proc = subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE
    )

    return proc


async def test_cdp_endpoint(port=9222, max_retries=10):
    """测试 CDP 端点是否响应"""
    url = f'http://localhost:{port}/json/version'

    print(f"\n测试 CDP 端点: {url}")
    print("="*60)

    async with httpx.AsyncClient(timeout=10.0) as client:
        for i in range(max_retries):
            try:
                print(f"尝试 {i+1}/{max_retries}...", end=' ')
                resp = await client.get(url)

                print(f"✅ 状态码: {resp.status_code}")
                print(f"内容长度: {len(resp.content)} 字节")
                print(f"内容: {resp.text[:200]}")

                if resp.status_code == 200 and len(resp.content) > 0:
                    data = resp.json()
                    print(f"\n✅ CDP 端点正常响应")
                    print(f"Browser: {data.get('Browser')}")
                    print(f"WebSocket URL: {data.get('webSocketDebuggerUrl')}")
                    return True

            except httpx.TimeoutException:
                print("❌ 超时")
            except httpx.ConnectError:
                print("❌ 连接失败")
            except Exception as e:
                print(f"❌ 错误: {e}")

            await asyncio.sleep(1)

    print("\n❌ CDP 端点无法响应")
    return False


async def main():
    print("Chrome CDP 底层诊断")
    print("="*60)

    # 测试 1: headless=True
    print("\n【测试 1】headless=True")
    proc1 = start_chrome_with_cdp(port=9222, headless=True)
    print("等待 Chrome 启动...")
    time.sleep(3)

    result1 = await test_cdp_endpoint(port=9222)

    print("\n终止 Chrome...")
    proc1.terminate()
    proc1.wait(timeout=5)

    # 清理
    subprocess.run(['rm', '-rf', '/tmp/test-cdp-diagnosis'],
                   stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

    await asyncio.sleep(2)

    # 测试 2: headless=False
    print("\n" + "="*60)
    print("【测试 2】headless=False")
    proc2 = start_chrome_with_cdp(port=9223, headless=False)
    print("等待 Chrome 启动...")
    time.sleep(3)

    result2 = await test_cdp_endpoint(port=9223)

    print("\n终止 Chrome...")
    proc2.terminate()
    proc2.wait(timeout=5)

    # 清理
    subprocess.run(['rm', '-rf', '/tmp/test-cdp-diagnosis'],
                   stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

    # 总结
    print("\n" + "="*60)
    print("诊断总结")
    print("="*60)
    print(f"headless=True:  {'✅ CDP 正常' if result1 else '❌ CDP 失败'}")
    print(f"headless=False: {'✅ CDP 正常' if result2 else '❌ CDP 失败'}")
    print("="*60)

    if not result1 and not result2:
        print("\n⚠️  问题分析:")
        print("Chrome 的 CDP 端点无法响应 HTTP 请求")
        print("\n可能原因:")
        print("1. Chrome 被防火墙或安全软件阻止")
        print("2. macOS 权限问题")
        print("3. Chrome 版本与 CDP 协议不兼容")
        print("4. 系统资源不足导致 Chrome 启动缓慢")
        print("\n建议:")
        print("1. 检查系统偏好设置 > 安全性与隐私 > 防火墙")
        print("2. 尝试重新安装 Chrome")
        print("3. 检查是否有杀毒软件拦截")


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