Playwright实现网络流量监控与修改指南

目录

• 一、基本网络监控
• 1. 记录所有请求
• 2. 获取请求详细信息
• 二、网络请求修改
• 1. 修改请求头
• 2. 修改请求体
• 3. 拦截并返回模拟响应
• 三、高级网络控制
• 1. 延迟响应
• 2. 阻止特定请求
• 3. 修改响应
• 四、底层运行原理
• 1. 整体架构
• 2. 请求生命周期
• 3. 关键技术点
• 五、实际应用示例
• 1. API 测试模拟
• 2. 性能分析
• 3. 认证处理
• 六、最佳实践

Playwright 提供了强大的网络流量控制能力，可以拦截、修改和分析所有 HTTP/HTTPS 请求。下面我将详细介绍各种使用方法和底层原理。

一、基本网络监控

1. 记录所有请求

from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    browser = p.chromium.launch()
    page = browser.new_page()
    
    # 监听请求和响应事件
    def print_request(request):
        print(f">> Request: {request.method} {request.url}")
    
    def print_response(response):
        print(f"<< Response: {response.status} {response.url}")
    
    page.on("request", print_request)
    page.on("response", print_response)
    
    page.goto("https://example.com"lmxtb)
    
    browser.close()

2. 获取请求详细信息

def log_request(request):
    print(f"""
    Request: {request.method} {request.url}
    Headers: {request.headers}
    Post Data: {request.post_data}
    Resource Type: {request.resource_type}
    Navigation: {request.is_navigation_request()}
    """)

page.on("request", log_request)

二、网络请求修改

1. 修改请求头

async def modify_headers(route):
    headers = route.request.headers.copy()
    headers["x-custom-header"] = "my-value"
    await route.continue_(headers=headers)

await page.route("**/*", modify_headers)

2. 修改请求体

async def modify_post_data(route):
    if route.request.method == "POST":
        post_data = route.request.post_data
        modified_data = post_data.replace("old", "new")
        await route.continue_(post_data=modified_data)
    else:
        await route.continue_()

await page.route("**/api/**", modify_post_data)

3. 拦截并返回模拟响应

async def mock_response(route):
    await route.fulfill(
        status=200,
        content_type="application/json",
        body=json.dumps({"data": "mocked"})
    )

await page.route("**/api/data", mock_response)

三、高级网络控制

1. 延迟响应

async def delay_response(route):
    await asyncio.sleep(2)  # 延迟2秒
    await route.continue_()

await page.route("**/*.css", delay_response)

2. 阻止特定请求

async def block_analytics(route):
    if "google-analytics" in route.request.url:
        await route.abort()
    else:
        await route.continue_()

await page.route("**/*", block_analytics)

3. 修改响应

async def modify_response(route):
    response = await route.fetch()
    body = await response.text()
    modified_body = body.replace("Original", "Modified")
    await route.fulfill(
        response=response,
        body=modified_body
    )

await page.route("**/api/content", modify_response)

四、底层运行原理

1. 整体架构

+-------------------+     +-------------------+     +-------------------+

|   Playwright      |     |   Browser         |     |   Remote Server   |

|   python Client   |<--->|   (Chromium/etc)  |<--->|   (example.com)   |

+-------------------+     +-------------------+     +-------------------+

        |                       ^    ^

        | CDP (Chrome DevTools) |    |

        | WebSocket Connection |    | Actual Network Traffic

        v     js                  |    |

+-------------------+           |    |

|   Network Proxy   |-----------+    |

|   Layer           |----------------+

+-------------------+

2. 请求生命周期

1.初始化阶段:

• 当调用 page.route() 时，Playwright 会在浏览器中设置请求拦截器
• 通过 CDP 的 Fetch 域启用请求拦截

2.请求拦截流程:

sequenceDiagram
    participant Page
    participant Playwright
    participant Browser
    
    Page->>Browser: 发起请求
    Browser->>Playwright: 触发请求拦截 (Fetch.requestPaused)
    Playwright->>Python: 调用注册的路由处理器
    alt 继续请求
        Python->>Playwright: route.continue_()
        Playwright->>Browser: 继续请求到服务器
    else 模拟响应
        Python->>Playwright: route.fulfill()
        Playwright->>Browser: 返回模拟响应
    else 中止请求
        Python->>Pl编程客栈aywright: route.abort()
        Playwright->>Browser: 中止请求
    end

3.修改请求流程:

Playwright 使用中间人技术拦截请求

可以修改的部分包括:

• URL
• 方法 (GET/POST等)
• 请求头
• 请求体
• 重定向行为

4.响应处理流程:

对于 route.fetch() 操作:

• Playwright 会实际发送请求到服务器
• 获取响应后允许修改再返回给页面

对于 route.fulfill() 操作:

• 直接构造响应返回给浏览器
• 不接触真实服务器

3. 关键技术点

1.请求匹配系统:

• 使用 glob 模式或正则表达式匹配 URL
• 可以按资源类型 (XHR, stylesheet, image等) 过滤

2.内存管理:

• 每个路由处理器都保持引用直到手动取消
• 需要调用 page.unroute() 避免内存泄漏

3.安全机制:

• HTTPS 拦截需要 Playwright 安装自己的 CA 证书
• 修改后的请求仍保持安全特性
• 遵循同源策略和 CORS 规则

4.性能优化:

• 拦截器运行在浏览器进程中
• 最小化 Python 和浏览器之间的通信
• 批量处理多个请求

五、实际应用示例

1. API 测试模拟

async def test_api(page):
    async def handle_api(route):
        if route.request.method == "POST":
            await route.fulfill(
                status=201,
                json={"id": 123, "status": "created"}
            )
        else:
            await route.continue_()
    
    await page.route("**/api/users*", handle_api)
    
    # 测试代码
    await page.goto("https://app.example.com")
    await page.click("#create-user")
    assert await page.text_content(".status") == "User 123 created"

2. 性能分析

async def analyze_performance(page):
    requests = []
    
    def log_request(request):
        requests.append({
            "url": request.url,
            "start": time.time(),
            "type": request.resource_type
        })
    
    def log_response(response):
        for req in requests:
            if req["url"] == response.url:
                req["duration"] = time.time() - req["start"]
                req["status"] = response.status
    
    page.on("request", log_request)
    page.on("response", log_response)
    
    await page.goto("https://example.com")
    
    # 输出性能报告
    slow_requests = [r for r in requests if r.get("duration", 0) > 1]
    print(f"Slow requests: {len(slow_requests)}/{len(requests)}")

3. 认证处理

async def handle_auth(page):
    await page.rolmxtbute("**/api/**", lambda route: route.continue_(
        headers={**route.request.headers, "Authorization": "Bearer xyz"}
    ))
    
    await page.goto("https://app.example.com")
    # 所有API请求会自动带上认证头

六、最佳实践

1.精确路由匹配:

• 避免使用太宽泛的 **/* 模式
• 结合 URL 和资源类型进行过滤

2.清理路由:

# 添加路由
await page.route("**/api/**", handler)

# 测试完成后移除
await page.unroute("**/api/**", handler)

3.错误处理:

async def safe_handler(route):
    try:
        await route.continue_()
    except Exception as e:
        print(f"Failed to handle {route.request.url}: {e}")
        await route.abort()

4.性能敏感操作:

• 对于大量请求，考虑在 Pythonpython 端实现缓存
• 避免在路由处理器中进行复杂计算

5.调试技巧:

# 打印未处理的请求
page.on("request", lambda r: print("Unhandled:", r.url))

Playwright 的网络拦截 API 提供了对浏览器网络活动的完全控制，使得测试复杂 web 应用变得更加容易，特别是在需要模拟各种网络条件或测试边缘情况时。理解其底层原理有助于更有效地使用这些功能。

到此这篇关于Playwright实现网络流量监控与修改指南的文章就介绍到这了,更多相关Playwright网络流量监控内容请搜索编程客栈(www.devze.com)以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程客栈(www.devze.com)！