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

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")
    
    browser.close()

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)

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)

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)

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)

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

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

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

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

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)

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->>Playwright: route.abort()
        Playwright->>Browser: 中止请求
    end

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"

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)}")

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

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

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

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()

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