RuyiPage Docs

RuyiPage 拦截响应

拦截响应是抓包采集和接口数据提取的核心能力。在 responseStarted 阶段，响应头已到达但 body 尚未传输，你可以读取状态码和响应头、修改它们，还能配合 collect_response=True 一步读取完整响应体。

什么时候用拦截响应

读取响应头和状态码

用 start_responses() 拦截后，通过 req.response_status 和 req.response_headers 直接读取。

修改响应

用 continue_response(status_code=..., headers=...) 修改状态码、响应头后放行。

采集响应体

加上 collect_response=True，放行后通过 req.response_body 一步读取完整响应内容。

请求体 + 响应体双向采集

用 start_requests(handler, collect_response=True) 在请求阶段拦截，同时拿到 req.body 和 req.response_body。

拦截响应 vs 被动监听

对比项	拦截响应（intercept）	被动监听（events/listen）
是否阻塞	是，必须 continue_response() 放行	否，只观察不干预
能否修改	可以改状态码、响应头	只读，不能修改
响应体读取	collect_response=True 自动采集	需手动配合 DataCollector
适合场景	抓包采集、响应改写、接口测试	链路审计、请求统计

示例 1：读取响应头和状态码

python start_responses 查看响应信息

from ruyipage import FirefoxPage

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

def response_handler(req):
    """responseStarted 阶段：响应头已到达。"""
    print(f"URL: {req.url}")
    print(f"状态码: {req.response_status}")       # 如 200
    print(f"响应头: {req.response_headers}")       # dict 格式
    print(f"是响应阶段: {req.is_response_phase}")   # True
    req.continue_response()  # 必须放行

# start_responses 是 start(phases=["responseStarted"]) 的快捷方式
page.intercept.start_responses(response_handler)

page.run_js(
    'fetch("https://httpbin.org/get").catch(()=>null); return true;',
    as_expr=False
)
page.wait(2)
page.intercept.stop()
page.quit()

示例 2：修改响应状态码和响应头

python continue_response 修改响应

from ruyipage import FirefoxPage

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

def modify_response(req):
    if "/api/data" in req.url:
        # 修改状态码和响应头后放行
        req.continue_response(
            status_code=299,
            reason_phrase="RuyiModified",
            headers={"X-Modified-By": "RuyiPage"}
        )
        print(f"已修改: {req.url} -> 299")
    else:
        req.continue_response()  # 其他响应原样放行

page.intercept.start_responses(modify_response)

# 验证修改效果
result = page.run_js("""
    return fetch("https://httpbin.org/get")
        .then(r => r.status + ":" + r.statusText)
""", as_expr=False)
print(f"浏览器看到的状态: {result}")

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

示例 3：拦截响应并读取响应体

这是最常用的场景：在响应阶段拦截，读取状态码和 Content-Type，然后通过 req.response_body 获取完整响应体。

关键参数是 collect_response=True，它会自动创建 DataCollector，在 continue_response() 放行后等待 body 传输完成并自动解码。

python start_responses + collect_response（推荐写法）

from ruyipage import FirefoxPage
import json

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

captured = []

def capture_response(req):
    """在响应阶段拦截，读状态码 + 自动采集响应体。"""
    print(f"\n[响应] {req.url}")
    print(f"  状态码: {req.response_status}")
    print(f"  Content-Type: {req.response_headers.get('Content-Type', '未知')}")

    # 先放行响应（必须先放行才能读 body）
    req.continue_response()

    # response_body 自动等待 body 传输完成并解码
    body = req.response_body
    if body:
        print(f"  响应体长度: {len(body)} 字符")
        print(f"  预览: {body[:200]}")
        captured.append({
            "url": req.url,
            "status": req.response_status,
            "body": body
        })

# collect_response=True 是关键
page.intercept.start_responses(capture_response, collect_response=True)

page.get("https://httpbin.org/get", wait="none")
page.wait(5)
page.intercept.stop()

# 查看采集结果
print(f"\n共采集到 {len(captured)} 个响应")
for item in captured:
    print(f"  {item['status']} {item['url']}")
    try:
        data = json.loads(item["body"])
        print(f"    JSON keys: {list(data.keys())}")
    except Exception:
        pass

page.quit()

示例 4：队列模式拦截响应

python handler=None + wait() 手动处理

from ruyipage import FirefoxPage

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

# 队列模式：handler=None
page.intercept.start(
    handler=None,
    phases=["responseStarted"],
    collect_response=True
)

# 触发请求
page.run_js(
    'fetch("https://httpbin.org/get").catch(()=>null); return true;',
    as_expr=False
)

# 手动取出并处理
req = page.intercept.wait(timeout=5)
if req:
    print(f"状态码: {req.response_status}")
    print(f"Content-Type: {req.response_headers.get('Content-Type')}")
    req.continue_response()  # 放行

    body = req.response_body
    print(f"响应体: {body[:300]}")

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

示例 5：POST 请求体 + 响应体双向采集

实际抓包场景中，往往需要同时拿到请求体（POST payload）和响应体（接口返回数据）。

方法是用 start_requests(handler, collect_response=True) 在请求阶段拦截：req.body 拿请求体，放行后 req.response_body 拿响应体。

python beforeRequestSent + collect_response 双向采集

from ruyipage import FirefoxPage, InterceptedRequest

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

results = []

def full_capture(req: InterceptedRequest):
    """请求阶段拦截：先读请求体，放行后再读响应体。"""
    if "/api/" not in req.url and "httpbin" not in req.url:
        req.continue_request()
        return

    # 1. 读取请求信息
    print(f"\n[请求] {req.method} {req.url}")
    print(f"  请求头: {req.headers}")
    if req.body:
        print(f"  请求体: {req.body[:300]}")

    # 2. 放行请求
    req.continue_request()

    # 3. 自动等待并读取响应体
    resp_body = req.response_body
    print(f"  响应体: {resp_body[:300] if resp_body else '(空)'}")

    results.append({
        "method": req.method,
        "url": req.url,
        "request_body": req.body,
        "response_body": resp_body,
    })

# beforeRequestSent 阶段 + 自动采集响应体
page.intercept.start_requests(full_capture, collect_response=True)

# 触发 POST 请求
page.run_js("""
    fetch("https://httpbin.org/post", {
        method: "POST",
        headers: {"Content-Type": "application/json"},
        body: JSON.stringify({keyword: "ruyipage", page: 1})
    }).catch(() => null);
    return true;
""", as_expr=False)
page.wait(3)
page.intercept.stop()

print(f"\n共捕获 {len(results)} 个请求")
for r in results:
    print(f"  {r['method']} {r['url']}")
    if r["request_body"]:
        print(f"    请求体: {r['request_body'][:100]}")
    if r["response_body"]:
        print(f"    响应体: {r['response_body'][:100]}")

page.quit()

InterceptedRequest 响应阶段属性

属性	类型	说明	可用条件
req.response_status	int	HTTP 状态码（如 200、404、500）	responseStarted 阶段
req.response_headers	dict	响应头字典（自动转为 key-value 格式）	responseStarted 阶段
req.is_response_phase	bool	是否处于响应阶段	responseStarted 阶段
req.response_body	str \| None	完整响应体（自动解码）	需要 collect_response=True
req.request_id	str	请求唯一标识，关联请求与响应	所有阶段
req.url	str	请求的 URL（响应阶段也可读取原始请求 URL）	所有阶段

注意事项

responseStarted 阶段必须调用 continue_response() 放行，否则响应会一直卡住。
req.response_body 必须在 continue_response() 之后读取，因为需要等待 body 传输完成。
collect_response=True 会自动创建和清理 DataCollector，无需手动管理。
start_responses() 默认 collect_response=True，start_requests() 默认 False 需要手动打开。
如果只需要读响应头和状态码（不需要 body），可以不传 collect_response。

重点 API 参考

page.intercept.start_responses

返回: None

page.intercept.start_responses(handler=None, url_patterns=None, collect_response=True)

拦截 responseStarted 阶段。默认自动采集响应体。

参数说明

参数	说明
handler	回调函数或 None（队列模式）。
url_patterns	可选，URL 匹配模式。
collect_response	默认 True，自动创建 DataCollector 采集响应体。

常见坑

handler 中必须调用 continue_response()，否则响应卡住。

req.continue_response

返回: None

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

放行响应，可选修改响应头、状态码和原因短语。

参数说明

参数	说明
headers	可选，替换响应头（dict 格式自动转换）。
status_code	可选，替换 HTTP 状态码。
reason_phrase	可选，替换原因短语（如 'OK'、'Not Found'）。

常见坑

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

req.response_body

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

req.response_body

读取完整响应体。自动等待传输完成并解码，需要 collect_response=True。

常见坑

必须在 continue_response() 或 continue_request() 后读取。
未使用 collect_response=True 时不可用。

相关示例源码

4 个示例 ▶ 点击展开

高级网络功能：请求/响应拦截、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()

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

examples/40_scraper_packet_capture.py

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

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

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

网络拦截入门：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()

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

examples/40_1_request_header_capture.py

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

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

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