被动监听网络事件，不阻塞也不修改请求。五种事件覆盖请求的完整生命周期，适合做请求审计、链路诊断和认证检测。
| 事件名 | 触发时机 | 典型用途 |
|---|---|---|
| network.beforeRequestSent | 请求发出前 | 记录所有发出的请求 URL 和方法 |
| network.responseStarted | 响应头到达 | 检查响应状态码 |
| network.responseCompleted | 响应体传输完成 | 确认请求是否正常完成 |
| network.fetchError | 请求失败 | 检测网络异常或被拦截的请求 |
| network.authRequired | 需要认证 | 检测需要 HTTP 认证的资源 |
from ruyipage import FirefoxPage
page = FirefoxPage()
page.get("about:blank")
# 订阅网络事件
page.events.start([
"network.beforeRequestSent",
"network.responseStarted",
"network.responseCompleted",
"network.fetchError",
], contexts=[page.tab_id])
# 触发一个请求
page.run_js('fetch("https://httpbin.org/get").catch(()=>null); return true;', as_expr=False)
# 依次等待事件
before = page.events.wait("network.beforeRequestSent", timeout=5)
if before:
url = (before.request or {}).get("url", "")
print(f"请求发出: {url}")
started = page.events.wait("network.responseStarted", timeout=5)
if started:
status = (started.response or {}).get("status", "")
print(f"响应开始: status={status}")
completed = page.events.wait("network.responseCompleted", timeout=5)
if completed:
status = (completed.response or {}).get("status", "")
print(f"响应完成: status={status}")
page.events.stop()
page.quit()page.events.start(["network.fetchError"], contexts=[page.tab_id])
page.events.clear()
# 请求一个不存在的服务
page.run_js('fetch("http://127.0.0.1:9/").catch(()=>null); return true;', as_expr=False)
error = page.events.wait("network.fetchError", timeout=5)
if error:
print(f"请求失败: {error.error_text}")
page.events.stop()page.events.start(events, contexts=None)开始订阅指定的 BiDi 事件列表。
| 参数 | 说明 |
|---|---|
| events | 事件名称列表，如 ['network.beforeRequestSent', 'network.responseCompleted']。 |
| contexts | 可选，限定监听的标签页范围，如 [page.tab_id]。 |
page.events.wait(event, timeout=10)阻塞等待下一个匹配的事件触发。
| 参数 | 说明 |
|---|---|
| event | 事件名称字符串。 |
| timeout | 超时时间（秒），超时返回 None。 |
是 network-events 页最直接的主示例。
源码文件：app/content/ruyipage_examples/31_network_events.py · 仓库：31_network_events.py
#!/usr/bin/env python
# =============================
# RuyiPage 示例注释版
# 标题: 网络事件：请求开始、响应开始、响应完成、错误与认证
# 说明: 是 network-events 页最直接的主示例。
# 你会学到:
# 1. 观察 beforeRequestSent、responseStarted、responseCompleted
# 2. 监听 fetchError
# 3. 处理 authRequired 并补充用户名密码
# 相关 API: page.events.start, page.events.wait, page.intercept.start, page.intercept.wait, continue_with_auth
# =============================
# -*- coding: utf-8 -*-
"""示例31: Network Events（严格结果版）"""
import io
import sys
from typing import Dict, List
if sys.platform == "win32":
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8")
from ruyipage import FirefoxPage, InterceptedRequest
from ruyipage._functions.tools import find_free_port
from test_server import TestServer
def add_result(
results: List[Dict[str, str]], item: str, status: str, note: str
) -> None:
results.append({"item": item, "status": status, "note": note})
def print_results(results: List[Dict[str, str]]) -> None:
print("\n| 项目 | 状态 | 说明 |")
print("| --- | --- | --- |")
for row in results:
print(f"| {row['item']} | {row['status']} | {row['note']} |")
def main() -> None:
print("=" * 70)
print("测试 31: Network Events")
print("=" * 70)
page = FirefoxPage()
server = None
results: List[Dict[str, str]] = []
try:
server = TestServer(port=find_free_port(9330, 9430)).start()
base_url = server.get_url("")[:-1]
page.get("about:blank")
add_result(
results, "页面加载", "成功", f"about:blank 已加载，服务地址: {base_url}"
)
page.events.start(
[
"network.beforeRequestSent",
"network.responseStarted",
"network.responseCompleted",
"network.fetchError",
"network.authRequired",
],
contexts=[page.tab_id],
)
# 1. 普通请求事件链
page.events.clear()
page.run_js(
f"fetch('{base_url}/api/data').catch(()=>null); return true;",
as_expr=False,
)
before_event = page.events.wait("network.beforeRequestSent", timeout=5)
started_event = page.events.wait("network.responseStarted", timeout=5)
completed_event = page.events.wait("network.responseCompleted", timeout=5)
if before_event and str((before_event.request or {}).get("url", "")).endswith(
"/api/data"
):
add_result(
results,
"network.beforeRequestSent",
"成功",
(before_event.request or {}).get("url", ""),
)
else:
add_result(
results,
"network.beforeRequestSent",
"失败",
"未观察到 /api/data 请求开始事件",
)
if (
started_event
and int((started_event.response or {}).get("status", 0)) == 200
):
add_result(
results,
"network.responseStarted",
"成功",
f"status={(started_event.response or {}).get('status')}",
)
else:
add_result(
results, "network.responseStarted", "失败", "未观察到 200 响应开始事件"
)
if (
completed_event
and int((completed_event.response or {}).get("status", 0)) == 200
):
add_result(
results,
"network.responseCompleted",
"成功",
f"status={(completed_event.response or {}).get('status')}",
)
else:
add_result(
results,
"network.responseCompleted",
"失败",
"未观察到 200 响应完成事件",
)
# 2. fetchError
page.events.clear()
page.run_js(
"fetch('http://127.0.0.1:9/').catch(()=>null); return true;",
as_expr=False,
)
error_event = page.events.wait("network.fetchError", timeout=5)
if error_event:
add_result(
results, "network.fetchError", "成功", error_event.error_text or ""
)
else:
add_result(
results, "network.fetchError", "跳过", "当前环境未稳定观察到 fetchError"
)
# 3. authRequired
page.events.clear()
page.intercept.start(
url_patterns=[{"type": "string", "pattern": base_url + "/api/auth"}],
phases=["authRequired"],
)
try:
page.get(base_url + "/api/auth")
except Exception:
pass
auth_req: InterceptedRequest | None = page.intercept.wait(timeout=5)
auth_event = page.events.wait("network.authRequired", timeout=5)
if auth_event and auth_req:
add_result(results, "network.authRequired", "成功", auth_req.url)
auth_req.continue_with_auth(
action="provideCredentials",
username="user",
password="pass",
)
auth_done = page.events.wait("network.responseCompleted", timeout=5)
if auth_done and int((auth_done.response or {}).get("status", 0)) == 200:
add_result(
results,
"network.authRequired credentials",
"成功",
"提供凭证后认证通过",
)
else:
add_result(
results,
"network.authRequired credentials",
"失败",
"提供凭证后未观察到 200 完成事件",
)
else:
add_result(
results,
"network.authRequired",
"跳过",
"当前环境未稳定观察到 authRequired",
)
add_result(
results,
"network.authRequired credentials",
"跳过",
"未观察到 authRequired，跳过凭证验证",
)
print_results(results)
except Exception as e:
add_result(results, "示例执行", "失败", str(e)[:120])
print_results(results)
raise
finally:
try:
page.intercept.stop()
except Exception:
pass
try:
page.events.stop()
except Exception:
pass
try:
if server is not None:
server.stop()
except Exception:
pass
try:
page.quit()
except Exception:
pass
if __name__ == "__main__":
main()
展示 intercept 与 events 组合后的完整网络控制能力。
源码文件：app/content/ruyipage_examples/18_advanced_network.py · 仓库：18_advanced_network.py
# =============================
# RuyiPage 示例注释版
# 标题: 高级网络功能：请求/响应拦截、mock、阻止、fetchError
# 说明: 展示 intercept 与 events 组合后的完整网络控制能力。
# 你会学到:
# 1. 在 beforeRequestSent 阶段继续请求
# 2. 修改请求头
# 3. mock JSON 响应
# 4. 阻止请求
# 5. 在 responseStarted 阶段修改响应状态
# 相关 API: page.intercept.start_requests, page.intercept.start_responses, req.continue_request, req.mock, req.fail, req.continue_response, page.events.start, page.events.wait
# =============================
# -*- coding: utf-8 -*-
"""示例18: 高级网络功能（完整场景）
覆盖：
1) beforeRequestSent 拦截 + 继续请求
2) 修改请求头
3) mock 响应
4) 阻止请求
5) responseStarted 阶段修改响应状态
6) 队列模式 wait() 手动处理
7) fetchError 事件监听
"""
import io
import os
import sys
if sys.platform == "win32":
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8")
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
sys.path.insert(0, os.path.dirname(__file__))
from ruyipage import FirefoxOptions, FirefoxPage
from ruyipage._functions.tools import find_free_port
from test_server import TestServer
def test_advanced_network():
print("=" * 60)
print("测试18: 高级网络功能")
print("=" * 60)
server = TestServer(port=find_free_port(8889, 8989))
server.start()
opts = FirefoxOptions()
opts.headless(False)
page = FirefoxPage(opts)
fetch_error_list = []
try:
page.get("about:blank")
page.wait(0.6)
# 1) 请求拦截
print("\n1. beforeRequestSent 拦截 + 继续请求:")
intercepted = []
def req_handler(req):
if "/api/data" in req.url:
intercepted.append(req.url)
print(f" 拦截到请求: {req.method} {req.url}")
req.continue_request()
page.intercept.start_requests(req_handler)
data = page.run_js(
"""
return fetch(arguments[0]).then(r => r.json()).then(d => d.status).catch(e => String(e));
""",
server.get_url("/api/data"),
as_expr=False,
)
page.wait(0.5)
print(f" 请求结果: {data}")
print(f" 拦截计数: {len(intercepted)}")
page.intercept.stop()
# 2) 修改请求头
print("\n2. 修改请求头:")
def header_handler(req):
if "/api/headers" in req.url:
req.continue_request(
headers=[
{
"name": "X-Ruyi-Demo",
"value": {"type": "string", "value": "yes"},
},
{
"name": "User-Agent",
"value": {"type": "string", "value": "RuyiPage/Example18"},
},
]
)
print(" ✓ 已注入请求头")
else:
req.continue_request()
page.intercept.start_requests(header_handler)
headers = page.run_js(
"""
return fetch(arguments[0]).then(r => r.json()).catch(e => ({error:String(e)}));
""",
server.get_url("/api/headers"),
as_expr=False,
)
print(
f" X-Ruyi-Demo: {headers.get('X-Ruyi-Demo') or headers.get('x-ruyi-demo')}"
)
page.intercept.stop()
# 3) mock 响应
print("\n3. mock 响应:")
def mock_handler(req):
if "/api/data" in req.url:
req.mock(
'{"status":"mocked","data":{"message":"这是Mock数据"}}',
status_code=200,
headers=[
{
"name": "content-type",
"value": {"type": "string", "value": "application/json"},
},
{
"name": "access-control-allow-origin",
"value": {"type": "string", "value": "*"},
},
],
)
print(" ✓ 已返回Mock响应")
else:
req.continue_request()
page.intercept.start_requests(mock_handler)
mock_msg = page.run_js(
"""
return fetch(arguments[0]).then(r => r.json()).then(d => d.status + ':' + d.data.message).catch(e => String(e));
""",
server.get_url("/api/data"),
as_expr=False,
)
print(f" Mock结果: {mock_msg}")
page.intercept.stop()
# 4) 阻止请求
print("\n4. 阻止请求:")
def block_handler(req):
if "/api/error" in req.url:
req.fail()
print(" ✓ 请求已阻止")
else:
req.continue_request()
page.intercept.start_requests(block_handler)
blocked = page.run_js(
"""
return fetch(arguments[0]).then(() => 'unexpected-success').catch(e => 'blocked:' + e.name);
""",
server.get_url("/api/error"),
as_expr=False,
)
print(f" 阻止结果: {blocked}")
page.intercept.stop()
# 5) 响应阶段修改状态码
print("\n5. responseStarted 阶段修改响应状态码:")
def response_handler(req):
if "/api/data" in req.url:
req.continue_response(status_code=299, reason_phrase="RuyiModified")
print(" ✓ 响应状态码已改为 299")
else:
req.continue_response()
page.intercept.start_responses(response_handler)
modified_status = page.run_js(
"""
return fetch(arguments[0]).then(r => r.status + ':' + r.statusText).catch(e => String(e));
""",
server.get_url("/api/data"),
as_expr=False,
)
print(f" 响应状态: {modified_status}")
page.intercept.stop()
# 6) 队列模式
print("\n6. 队列模式 wait() 手动处理:")
page.intercept.start(handler=None, phases=["beforeRequestSent"])
page.run_js(
"""
fetch(arguments[0]).catch(() => null);
return 'sent';
""",
server.get_url("/api/data"),
as_expr=False,
)
queued = page.intercept.wait(timeout=5)
if queued:
print(f" wait捕获: {queued.method} {queued.url}")
queued.continue_request()
else:
print(" ⚠ 未捕获到请求")
page.intercept.stop()
# 7) fetchError 事件监听
print("\n7. fetchError 事件监听:")
page.events.start(["network.fetchError"], contexts=[page.tab_id])
page.events.clear()
page.intercept.start_requests(block_handler)
page.run_js(
"""
fetch(arguments[0]).catch(() => null);
return 'sent';
""",
server.get_url("/api/error"),
as_expr=False,
)
page.wait(1)
page.intercept.stop()
while True:
evt = page.events.wait("network.fetchError", timeout=0.2)
if not evt:
break
fetch_error_list.append(evt)
print(f" fetchError 事件数量: {len(fetch_error_list)}")
print("\n" + "=" * 60)
print("✓ 所有高级网络功能测试通过！")
print("=" * 60)
except Exception as e:
print(f"\n✗ 测试失败: {e}")
import traceback
traceback.print_exc()
finally:
try:
page.intercept.stop()
except Exception:
pass
try:
page.events.stop()
except Exception:
pass
page.wait(1)
page.quit()
server.stop()
if __name__ == "__main__":
test_advanced_network()
用于 scraper 与接口采集场景的抓包示例，重点是先观察请求链路，再决定后续是监听、拦截还是 collector。