RuyiPage Docs

RuyiPage 请求拦截

请求拦截是 RuyiPage 最强的网络控制手段。三个拦截阶段 + 两种使用模式，覆盖从请求修改到响应伪造的全部场景。

三个拦截阶段

阶段	触发时机	可做的操作
beforeRequestSent	请求发出前	修改 URL/headers/body、mock 响应、fail 请求
responseStarted	响应头到达后、body 传输前	修改 status_code、headers、reason_phrase
authRequired	服务器返回 401 认证质询	提供用户名密码、取消认证、使用浏览器默认

两种使用模式

回调模式

传入 handler 函数，每个拦截请求自动触发回调。适合简单的条件判断和批量处理。

队列模式

handler=None，拦截请求进入队列，用 wait() 手动逐个取出处理。适合需要精确控制处理顺序的场景。

示例 1：拦截并 mock 响应

python mock 响应

from ruyipage import FirefoxPage

page = FirefoxPage()
page.get("about:blank")

def my_handler(req):
    if "/api/data" in req.url:
        # 不发真实请求，直接返回自定义 JSON
        req.mock(
            '{"status": "mocked", "message": "来自拦截器"}',
            status_code=200,
            headers={"content-type": "application/json"}
        )
    else:
        req.continue_request()  # 其他请求正常放行

page.intercept.start(my_handler, phases=["beforeRequestSent"])
result = page.run_js("""
    return fetch("https://api.example.com/api/data")
        .then(r => r.json())
        .then(d => d.message)
""", as_expr=False)
print(result)  # 输出: 来自拦截器

page.intercept.stop()
page.quit()

示例 2：阻止请求

python fail 请求

def block_handler(req):
    if "/ads/" in req.url or "/tracking/" in req.url:
        req.fail()  # 阻止广告和追踪请求
    else:
        req.continue_request()

page.intercept.start_requests(block_handler)

示例 3：修改请求头

python 修改请求头

def header_handler(req):
    if "/api/" in req.url:
        req.continue_request(
            headers={"User-Agent": "RuyiPage/1.0", "X-Custom": "hello"}
        )
    else:
        req.continue_request()

page.intercept.start_requests(header_handler)

示例 4：队列模式

python wait() 手动处理

# handler=None 进入队列模式
page.intercept.start(handler=None, phases=["beforeRequestSent"])

# 触发一个请求
page.run_js('fetch("https://api.example.com/data").catch(()=>null)')

# 手动取出并处理
req = page.intercept.wait(timeout=5)
if req:
    print(f"捕获: {req.method} {req.url}")
    req.continue_request()  # 记得放行，否则请求会卡住

page.intercept.stop()

响应拦截与响应体采集

拦截响应（responseStarted 阶段）可以读取和修改响应头、状态码，配合 collect_response=True 还能一步读取完整响应体。

详细说明、完整代码示例和 API 参考请查看「拦截响应」专题页。

python 响应拦截最小示例

def response_handler(req):
    print(f"状态码: {req.response_status}")      # 读取状态码
    print(f"响应头: {req.response_headers}")      # 读取响应头
    req.continue_response()                      # 放行
    body = req.response_body                     # 读取响应体
    print(f"响应体: {body[:200]}")

# collect_response=True 自动采集响应体
page.intercept.start_responses(response_handler, collect_response=True)

重点 API 参考

page.intercept.start

返回: None

page.intercept.start(handler, url_patterns=None, phases=['beforeRequestSent'], collect_response=False)

启动拦截。handler 为回调函数或 None（队列模式）。

参数说明

参数	说明
handler	回调函数（接收 InterceptedRequest）或 None（队列模式）。
url_patterns	URL 匹配模式列表，不传则拦截所有请求。
phases	拦截阶段：beforeRequestSent / responseStarted / authRequired。
collect_response	True 时自动创建 DataCollector。

常见坑

handler 内不处理的请求必须 continue_request()，否则挂起。
每次只能有一个 intercept 实例运行，start 前需 stop 上一个。

req.mock

返回: None

req.mock(body, status_code=200, headers=None, reason_phrase='OK')

不发送真实请求，直接返回自定义响应体。

参数说明

参数	说明
body	响应体字符串（JSON、HTML 等）。
status_code	HTTP 状态码，默认 200。
headers	响应头，支持 dict 格式或 BiDi 原始格式。

常见坑

跨域请求 mock 时需要带上 access-control-allow-origin 头。

req.continue_request

返回: None

req.continue_request(url=None, method=None, headers=None, body=None)

放行请求，可选参数修改请求的任意部分。

参数说明

参数	说明
url	可选，替换请求 URL。
method	可选，替换 HTTP 方法。
headers	可选，替换请求头（dict 或 BiDi 格式）。
body	可选，替换请求体。

常见坑

不传参数等同于原样放行。

req.continue_response

返回: None

req.continue_response(headers=None, status_code=None, reason_phrase=None)

在 responseStarted 阶段放行响应，可选参数修改响应头、状态码和原因短语。

参数说明

参数	说明
headers	可选，替换响应头（dict 或 BiDi 格式）。
status_code	可选，替换 HTTP 状态码。
reason_phrase	可选，替换 HTTP 原因短语。

常见坑

仅在 responseStarted 阶段可调用。
不传参数等同于原样放行响应。

req.response_body

返回: str（响应体文本）或 None

req.response_body

读取完整响应体（自动等待传输完成并解码）。需要启动拦截时传入 collect_response=True。

常见坑

必须先 continue_request() 或 continue_response() 放行后才能读取。
未使用 collect_response=True 时该属性不可用。

相关示例

抓包采集：从真实页面请求链路定位接口数据

相关示例源码

4 个示例 ▶ 点击展开

网络拦截入门：mock、fail、改请求头、队列模式

examples/11_network_intercept.py

通过本地 TestServer 让请求拦截、mock 和 fail 行为可重复验证。

GitHub 仓库查看仓库源码 page.intercept.startpage.intercept.waitreq.mockreq.failreq.continue_requestpage.run_js

拦截请求并直接 mock 响应
主动 fail 请求
修改请求头后继续请求
无 handler 队列模式手动处理请求

源码文件：app/content/ruyipage_examples/11_network_intercept.py · 仓库：11_network_intercept.py

python 注释版

# =============================
# RuyiPage 示例注释版
# 标题: 网络拦截入门：mock、fail、改请求头、队列模式
# 说明: 通过本地 TestServer 让请求拦截、mock 和 fail 行为可重复验证。
# 你会学到:
#   1. 拦截请求并直接 mock 响应
#   2. 主动 fail 请求
#   3. 修改请求头后继续请求
#   4. 无 handler 队列模式手动处理请求
# 相关 API: page.intercept.start, page.intercept.wait, req.mock, req.fail, req.continue_request, page.run_js
# =============================

# -*- coding: utf-8 -*-
"""
示例11: 网络拦截（本地可重复版）

测试覆盖：
1) beforeRequestSent + mock 响应
2) beforeRequestSent + fail 请求
3) beforeRequestSent + 修改请求头并继续
4) 无 handler 队列模式 wait()

说明：
- 本示例全部使用本地 TestServer，避免外网波动导致结果不稳定。
- 所有拦截都在当前页面 context 内触发，便于复现与调试。
"""

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 test_server import TestServer


def test_network_intercept():
    """网络拦截演示主流程（适合初学者逐步阅读）。"""
    print("=" * 60)
    print("测试11: 网络拦截")
    print("=" * 60)

    server = TestServer(port=8888)
    server.start()

    opts = FirefoxOptions()
    opts.headless(False)
    page = FirefoxPage(opts)

    try:
        # 准备一个干净页面，确保后续 fetch 都在当前 context 内触发。
        page.get("about:blank")
        page.wait(0.5)

        # ==========================================================
        # 1) mock 响应
        # ==========================================================
        print("\n1. 拦截并 Mock 本地 API:")

        def mock_handler(req):
            if "/api/data" in req.url:
                print(f"   拦截到请求: {req.method} {req.url}")
                # 注意：mock 需要带上 CORS 头，不然浏览器会把跨域响应拦掉，
                # fetch 侧会看到 NetworkError。
                req.mock(
                    '{"status":"ok","data":{"message":"mocked-by-interceptor"}}',
                    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(mock_handler, phases=["beforeRequestSent"])
        mocked = page.run_js(
            """
            return fetch(arguments[0]).then(r => r.json()).then(d => d.data.message).catch(e => String(e));
            """,
            server.get_url("/api/data"),
            as_expr=False,
        )
        print(f"   mock结果: {mocked}")
        page.intercept.stop()

        # ==========================================================
        # 2) fail 请求
        # ==========================================================
        print("\n2. 阻止本地 API 请求:")

        def fail_handler(req):
            if "/api/data" in req.url:
                print(f"   阻止请求: {req.url}")
                req.fail()
            else:
                req.continue_request()

        page.intercept.start(fail_handler, phases=["beforeRequestSent"])
        blocked = page.run_js(
            """
            return fetch(arguments[0]).then(() => 'unexpected-success').catch(e => 'blocked:' + e.name);
            """,
            server.get_url("/api/data"),
            as_expr=False,
        )
        print(f"   fail结果: {blocked}")
        page.intercept.stop()

        # ==========================================================
        # 3) 修改请求头并继续
        # ==========================================================
        print("\n3. 修改请求头并继续:")

        def header_handler(req):
            if "/api/headers" in req.url:
                # continue_request(headers=...) 会覆盖本次请求头。
                req.continue_request(
                    headers=[
                        {
                            "name": "X-Ruyi-Demo",
                            "value": {"type": "string", "value": "yes"},
                        },
                        {
                            "name": "User-Agent",
                            "value": {"type": "string", "value": "RuyiPage/Example11"},
                        },
                    ]
                )
                print("   ✓ 已注入自定义请求头")
            else:
                req.continue_request()

        page.intercept.start(header_handler, phases=["beforeRequestSent"])
        headers_json = page.run_js(
            """
            return fetch(arguments[0]).then(r => r.json()).catch(e => ({error:String(e)}));
            """,
            server.get_url("/api/headers"),
            as_expr=False,
        )
        x_header = ""
        if isinstance(headers_json, dict):
            x_header = headers_json.get("X-Ruyi-Demo") or headers_json.get(
                "x-ruyi-demo"
            )
        print(f"   服务器看到的 X-Ruyi-Demo: {x_header}")
        page.intercept.stop()

        # ==========================================================
        # 4) 无 handler 队列模式
        # ==========================================================
        print("\n4. 队列模式 wait() 捕获拦截请求:")
        # handler=None 表示不在回调里处理，而是进入队列，供 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,
        )
        req = page.intercept.wait(timeout=5)
        if req:
            print(f"   wait捕获: {req.method} {req.url}")
            # 队列模式下记得手动继续/中止请求，否则请求会悬挂。
            req.continue_request()
        else:
            print("   ⚠ 未在超时内捕获请求")
        page.intercept.stop()

        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
        page.wait(1)
        page.quit()
        server.stop()


if __name__ == "__main__":
    test_network_intercept()

# -*- coding: utf-8 -*-
"""
示例11: 网络拦截（本地可重复版）

测试覆盖：
1) beforeRequestSent + mock 响应
2) beforeRequestSent + fail 请求
3) beforeRequestSent + 修改请求头并继续
4) 无 handler 队列模式 wait()

说明：
- 本示例全部使用本地 TestServer，避免外网波动导致结果不稳定。
- 所有拦截都在当前页面 context 内触发，便于复现与调试。
"""

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 test_server import TestServer


def test_network_intercept():
    """网络拦截演示主流程（适合初学者逐步阅读）。"""
    print("=" * 60)
    print("测试11: 网络拦截")
    print("=" * 60)

    server = TestServer(port=8888)
    server.start()

    opts = FirefoxOptions()
    opts.headless(False)
    page = FirefoxPage(opts)

    try:
        # 准备一个干净页面，确保后续 fetch 都在当前 context 内触发。
        page.get("about:blank")
        page.wait(0.5)

        # ==========================================================
        # 1) mock 响应
        # ==========================================================
        print("\n1. 拦截并 Mock 本地 API:")

        def mock_handler(req):
            if "/api/data" in req.url:
                print(f"   拦截到请求: {req.method} {req.url}")
                # 注意：mock 需要带上 CORS 头，不然浏览器会把跨域响应拦掉，
                # fetch 侧会看到 NetworkError。
                req.mock(
                    '{"status":"ok","data":{"message":"mocked-by-interceptor"}}',
                    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(mock_handler, phases=["beforeRequestSent"])
        mocked = page.run_js(
            """
            return fetch(arguments[0]).then(r => r.json()).then(d => d.data.message).catch(e => String(e));
            """,
            server.get_url("/api/data"),
            as_expr=False,
        )
        print(f"   mock结果: {mocked}")
        page.intercept.stop()

        # ==========================================================
        # 2) fail 请求
        # ==========================================================
        print("\n2. 阻止本地 API 请求:")

        def fail_handler(req):
            if "/api/data" in req.url:
                print(f"   阻止请求: {req.url}")
                req.fail()
            else:
                req.continue_request()

        page.intercept.start(fail_handler, phases=["beforeRequestSent"])
        blocked = page.run_js(
            """
            return fetch(arguments[0]).then(() => 'unexpected-success').catch(e => 'blocked:' + e.name);
            """,
            server.get_url("/api/data"),
            as_expr=False,
        )
        print(f"   fail结果: {blocked}")
        page.intercept.stop()

        # ==========================================================
        # 3) 修改请求头并继续
        # ==========================================================
        print("\n3. 修改请求头并继续:")

        def header_handler(req):
            if "/api/headers" in req.url:
                # continue_request(headers=...) 会覆盖本次请求头。
                req.continue_request(
                    headers=[
                        {
                            "name": "X-Ruyi-Demo",
                            "value": {"type": "string", "value": "yes"},
                        },
                        {
                            "name": "User-Agent",
                            "value": {"type": "string", "value": "RuyiPage/Example11"},
                        },
                    ]
                )
                print("   ✓ 已注入自定义请求头")
            else:
                req.continue_request()

        page.intercept.start(header_handler, phases=["beforeRequestSent"])
        headers_json = page.run_js(
            """
            return fetch(arguments[0]).then(r => r.json()).catch(e => ({error:String(e)}));
            """,
            server.get_url("/api/headers"),
            as_expr=False,
        )
        x_header = ""
        if isinstance(headers_json, dict):
            x_header = headers_json.get("X-Ruyi-Demo") or headers_json.get(
                "x-ruyi-demo"
            )
        print(f"   服务器看到的 X-Ruyi-Demo: {x_header}")
        page.intercept.stop()

        # ==========================================================
        # 4) 无 handler 队列模式
        # ==========================================================
        print("\n4. 队列模式 wait() 捕获拦截请求:")
        # handler=None 表示不在回调里处理，而是进入队列，供 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,
        )
        req = page.intercept.wait(timeout=5)
        if req:
            print(f"   wait捕获: {req.method} {req.url}")
            # 队列模式下记得手动继续/中止请求，否则请求会悬挂。
            req.continue_request()
        else:
            print("   ⚠ 未在超时内捕获请求")
        page.intercept.stop()

        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
        page.wait(1)
        page.quit()
        server.stop()


if __name__ == "__main__":
    test_network_intercept()

高级网络功能：请求/响应拦截、mock、阻止、fetchError

examples/18_advanced_network.py

展示 intercept 与 events 组合后的完整网络控制能力。

GitHub 仓库查看仓库源码 page.intercept.start_requestspage.intercept.start_responsesreq.continue_requestreq.mockreq.failreq.continue_responsepage.events.startpage.events.wait

在 beforeRequestSent 阶段继续请求
修改请求头
mock JSON 响应
阻止请求
在 responseStarted 阶段修改响应状态
监听 fetchError

源码文件：app/content/ruyipage_examples/18_advanced_network.py · 仓库：18_advanced_network.py

python 注释版

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

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

请求头抓包：真实跟踪页 POST 请求头与请求体诊断

examples/40_1_request_header_capture.py

面向抓包与数据采集诊断的示例，重点不是改包，而是先看清真实页面发出的 POST 请求头和请求体形状。

GitHub 仓库查看仓库源码 FirefoxPagepage.intercept.startInterceptedRequestreq.headersreq.bodyreq.continue_request

抓取真实页面发出的 POST 请求头
预览请求体内容用于逆向分析
快速判断页面 URL 与接口 URL 的关系
在不打断真实请求的前提下继续放行请求

抓包采集：从真实页面请求链路定位接口数据

examples/40_scraper_packet_capture.py

用于 scraper 与接口采集场景的抓包示例，重点是先观察请求链路，再决定后续是监听、拦截还是 collector。

GitHub 仓库查看仓库源码 page.intercept.startpage.events.startpage.events.waitpage.network.add_data_collectorreq.continue_request

观察真实页面请求链路
定位数据接口与请求特征
为后续响应体采集或 mock 做前置分析
形成 scraper 场景下的网络调试起点