文章目录
- 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)
- +——————-+ +——————-+ +——————-+| Playwright | | Browser | | Remote Server || Python Client |<—>| (Chromium/etc) |<—>| (example.com) |+——————-+ +——————-+ +——————-+ | ^ ^ | CDP (Chrome DevTools) | | | WebSocket Connection | | Actual Network Traffic v | |+——————-+ | || Network Proxy |———–+ || Layer |—————-++——————-+
- 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->>Playwright: route.abort() Playwright->>Browser: 中止请求 end 3.修改请求流程: Playwright 使用中间人技术拦截请求 可以修改的部分包括: URL 方法 (GET/POST等) 请求头 请求体 重定向行为 4.响应处理流程: 对于 route.fetch() 操作: Playwright 会实际发送请求到服务器 获取响应后允许修改再返回给页面 对于 route.fulfill() 操作: 直接构造响应返回给浏览器 不接触真实服务器
- 1.请求匹配系统: 使用 glob 模式或正则表达式匹配 URL 可以按资源类型 (XHR, stylesheet, image等) 过滤 2.内存管理: 每个路由处理器都保持引用直到手动取消 需要调用 page.unroute() 避免内存泄漏 3.安全机制: HTTPS 拦截需要 Playwright 安装自己的 CA 证书 修改后的请求仍保持安全特性 遵循同源策略和 CORS 规则 4.性能优化: 拦截器运行在浏览器进程中 最小化 Python 和浏览器之间的通信 批量处理多个请求
- 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请求会自动带上认证头
目录
- 一、基本网络监控
- 1. 记录所有请求
- 2. 获取请求详细信息
- 二、网络请求修改
- 1. 修改请求头
- 2. 修改请求体
- 3. 拦截并返回模拟响应
- 三、高级网络控制
- 1. 延迟响应
- 2. 阻止特定请求
- 3. 修改响应
- 四、底层运行原理
- 1. 整体架构
- 2. 请求生命周期
- 3. 关键技术点
- 五、实际应用示例
- 1. API 测试模拟
- 2. 性能分析
- 3. 认证处理
- 六、最佳实践
Playwright 提供了强大的网络流量控制能力,可以拦截、修改和分析所有 HTTP/HTTPS 请求。下面我将详细介绍各种使用方法和底层原理。
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)
+——————-+ +——————-+ +——————-+
| Playwright | | Browser | | Remote Server |
| Python Client |<—>| (Chromium/etc) |<—>| (example.com) |
+——————-+ +——————-+ +——————-+
| ^ ^
| CDP (Chrome DevTools) | |
| WebSocket Connection | | Actual Network Traffic
v | |
+——————-+ | |
| Network Proxy |———–+ |
| Layer |—————-+
+——————-+
+——————-+ +——————-+ +——————-+
| Playwright | | Browser | | Remote Server |
| Python Client |<—>| (Chromium/etc) |<—>| (example.com) |
+——————-+ +——————-+ +——————-+
| ^ ^
| CDP (Chrome DevTools) | |
| WebSocket Connection | | Actual Network Traffic
v | |
+——————-+ | |
| Network Proxy |———–+ |
| Layer |—————-+
+——————-+
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->>Playwright: route.abort()
Playwright->>Browser: 中止请求
end
3.修改请求流程:
Playwright 使用中间人技术拦截请求
可以修改的部分包括:
- URL
- 方法 (GET/POST等)
- 请求头
- 请求体
- 重定向行为
4.响应处理流程:
对于 route.fetch() 操作:
- Playwright 会实际发送请求到服务器
- 获取响应后允许修改再返回给页面
对于 route.fulfill() 操作:
- 直接构造响应返回给浏览器
- 不接触真实服务器
1.请求匹配系统:
- 使用 glob 模式或正则表达式匹配 URL
- 可以按资源类型 (XHR, stylesheet, image等) 过滤
2.内存管理:
- 每个路由处理器都保持引用直到手动取消
- 需要调用
page.unroute()避免内存泄漏
3.安全机制:
- HTTPS 拦截需要 Playwright 安装自己的 CA 证书
- 修改后的请求仍保持安全特性
- 遵循同源策略和 CORS 规则
4.性能优化:
- 拦截器运行在浏览器进程中
- 最小化 Python 和浏览器之间的通信
- 批量处理多个请求
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请求会自动带上认证头
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.性能敏感操作:
- 对于大量请求,考虑在 Python 端实现缓存
- 避免在路由处理器中进行复杂计算
5.调试技巧:
# 打印未处理的请求
page.on("request", lambda r: print("Unhandled:", r.url))
Playwright 的网络拦截 API 提供了对浏览器网络活动的完全控制,使得测试复杂 web 应用变得更加容易,特别是在需要模拟各种网络条件或测试边缘情况时。理解其底层原理有助于更有效地使用这些功能。
到此这篇关于Playwright实现网络流量监控与修改指南的文章就介绍到这了,更多相关Playwright网络流量监控内容请搜索风君子博客以前的文章或继续浏览下面的相关文章希望大家以后多多支持风君子博客!
您可能感兴趣的文章:
- Python系统公网私网流量监控实现流程
- Python脚本实现网卡流量监控
- python监控网卡流量并使用graphite绘图的示例
- 使用Python统计网站访问流量
- Python实现端口流量转发的示例代码
- python如何查看系统网络流量的信息