全球动态代理 · 静态住宅代理 · 极高质量 · 包月不限量 · 学员低价 · 联系如意进行免费测试 · 欢迎公司及学员采购 · 另有任何海外数据采集需求,随时联系 · 商务合作微信:Charleval
Automation / 示例中心
Examples Center

RuyiPage 示例中心

这里不是简单的文件清单,而是将 RuyiPage 仓库中的 examples 渲染成案例卡。每个例子都带有文件名、分类、难度、不同用法、涉及 API、适用场景与相关页面,并提供 GitHub 源码入口,帮助你从“例子”回到“文档”,也从“文档”回到“例子”。

WebExtension:目录扩展与 XPI 安装卸载

examples/22_web_extension.py
GitHub 仓库 已同步源码 查看 GitHub 源码
browser strict
一句话定位

覆盖扩展安装、验证注入和卸载,是浏览器级能力的重要专题。

涉及 API
page.extensions.installpage.extensions.install_archivepage.extensions.uninstallpage.extensions.uninstall_allpage.run_js
展示的不同用法
  • 安装目录扩展
  • 验证 content script 生效
  • 打包并安装 XPI 扩展
  • 卸载指定扩展与清空测试扩展
适用场景

适合 extensions 页面。

相关文档页面
推荐阅读
browser-user-context-25
完整代码

来源:app/content/ruyipage_examples/22_web_extension.py · 仓库文件:22_web_extension.py

python 注释版
#!/usr/bin/env python
# =============================
# RuyiPage 示例注释版
# 标题: WebExtension:目录扩展与 XPI 安装卸载
# 说明: 覆盖扩展安装、验证注入和卸载,是浏览器级能力的重要专题。
# 你会学到:
#   1. 安装目录扩展
#   2. 验证 content script 生效
#   3. 打包并安装 XPI 扩展
#   4. 卸载指定扩展与清空测试扩展
# 相关 API: page.extensions.install, page.extensions.install_archive, page.extensions.uninstall, page.extensions.uninstall_all, page.run_js
# =============================

# -*- coding: utf-8 -*-
"""示例22: WebExtension 模块测试(完整链路版)

完整链路:
1) 创建测试扩展
2) 安装目录扩展
3) 验证扩展已安装
4) 卸载目录扩展
5) 打包 xpi 并安装
6) 验证扩展已安装
7) 卸载打包扩展
8) 清理测试文件

说明:
- WebExtension BiDi 规范只提供 install / uninstall 命令。
- 当前 Firefox 环境下若返回 unknown command,则视为“当前版本不支持”,不是脚本错误。
"""

import io
import json
import os
import shutil
import sys
import zipfile

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


class Rows:
    def __init__(self):
        self.rows = []

    def add(self, name, status, detail=""):
        self.rows.append((name, status, detail))
        print(f"  {status}: {name}")
        if detail:
            print(f"     详情: {detail}")

    def summary(self):
        print("\n" + "=" * 70)
        print("测试结果汇总")
        print("=" * 70)
        print(f"{'序号':<5} {'项目':<28} {'状态':<8} {'说明'}")
        print("-" * 70)
        for i, (name, status, detail) in enumerate(self.rows, 1):
            print(f"{i:<5} {name:<28} {status:<8} {detail[:36]}")
        print("-" * 70)


def _create_test_extension(ext_dir):
    os.makedirs(ext_dir, exist_ok=True)
    manifest = {
        "manifest_version": 2,
        "name": "RuyiPage Test Extension",
        "version": "1.0.0",
        "description": "测试扩展",
        "browser_specific_settings": {
            "gecko": {
                "id": "[email protected]",
                "strict_min_version": "109.0",
            }
        },
        "background": {"scripts": ["background.js"]},
        "content_scripts": [
            {
                "matches": ["<all_urls>"],
                "js": ["content.js"],
                "run_at": "document_end",
            }
        ],
        "permissions": [],
    }
    with open(os.path.join(ext_dir, "manifest.json"), "w", encoding="utf-8") as f:
        json.dump(manifest, f, indent=2, ensure_ascii=False)

    with open(os.path.join(ext_dir, "background.js"), "w", encoding="utf-8") as f:
        f.write("console.log('RuyiPage Test Extension loaded!');")

    with open(os.path.join(ext_dir, "content.js"), "w", encoding="utf-8") as f:
        f.write(
            "document.documentElement.setAttribute('data-ruyi-extension', 'loaded');"
        )


def _page_has_extension_marker(page):
    return (
        page.run_js(
            "return document.documentElement.getAttribute('data-ruyi-extension')"
        )
        == "loaded"
    )


def _safe_install(page, path, rows, label):
    try:
        ext_id = page.extensions.install(path)
        if ext_id:
            rows.add(label, "✓ 通过", f"extension_id={ext_id}")
            return ext_id
        rows.add(label, "⚠ 不支持", "当前 Firefox 未返回 extension id")
        return None
    except Exception as e:
        msg = str(e).lower()
        if "unknown command" in msg or "not supported" in msg:
            rows.add(label, "⚠ 不支持", "当前 Firefox 不支持 webExtension.install")
            return None
        raise


def test_web_extension():
    print("=" * 70)
    print("测试 22: WebExtension 模块")
    print("=" * 70)

    page = FirefoxPage()
    rows = Rows()
    ext_dir = os.path.join(os.path.dirname(__file__), "test_extension")
    xpi_path = os.path.join(os.path.dirname(__file__), "test_extension.xpi")

    ext_id = None
    ext_id2 = None

    try:
        # 1) 创建测试扩展
        _create_test_extension(ext_dir)
        rows.add("创建测试扩展", "✓ 通过", ext_dir)

        # 2) 安装目录扩展
        ext_id = _safe_install(page, ext_dir, rows, "安装目录扩展")

        # 3) 验证目录扩展已生效
        if ext_id:
            page.get("https://example.com")
            page.wait(1.5)
            marker = _page_has_extension_marker(page)
            rows.add(
                "目录扩展生效验证",
                "✓ 通过" if marker else "✗ 失败",
                "content script 已注入" if marker else "未检测到注入标记",
            )
        else:
            rows.add("目录扩展生效验证", "⚠ 跳过", "安装未成功")

        # 4) 卸载目录扩展
        if ext_id:
            page.extensions.uninstall(ext_id)
            rows.add("卸载目录扩展", "✓ 通过")
        else:
            rows.add("卸载目录扩展", "⚠ 跳过", "未安装成功")

        # 5) 打包 xpi
        with zipfile.ZipFile(xpi_path, "w", zipfile.ZIP_DEFLATED) as zipf:
            zipf.write(os.path.join(ext_dir, "manifest.json"), "manifest.json")
            zipf.write(os.path.join(ext_dir, "background.js"), "background.js")
            zipf.write(os.path.join(ext_dir, "content.js"), "content.js")
        rows.add("打包 XPI", "✓ 通过", xpi_path)

        # 6) 安装打包扩展
        try:
            ext_id2 = page.extensions.install_archive(xpi_path)
            if ext_id2:
                rows.add("安装 XPI 扩展", "✓ 通过", f"extension_id={ext_id2}")
            else:
                rows.add(
                    "安装 XPI 扩展", "⚠ 不支持", "当前 Firefox 未返回 extension id"
                )
        except Exception as e:
            msg = str(e).lower()
            if "unknown command" in msg or "not supported" in msg:
                rows.add(
                    "安装 XPI 扩展",
                    "⚠ 不支持",
                    "当前 Firefox 不支持 webExtension.install",
                )
            else:
                raise

        # 7) 验证 XPI 扩展已生效
        if ext_id2:
            page.get("https://example.com")
            page.wait(1.5)
            marker = _page_has_extension_marker(page)
            rows.add(
                "XPI 扩展生效验证",
                "✓ 通过" if marker else "✗ 失败",
                "content script 已注入" if marker else "未检测到注入标记",
            )
        else:
            rows.add("XPI 扩展生效验证", "⚠ 跳过", "安装未成功")

        # 8) 卸载打包扩展
        if ext_id2:
            page.extensions.uninstall(ext_id2)
            rows.add("卸载 XPI 扩展", "✓ 通过")
        else:
            rows.add("卸载 XPI 扩展", "⚠ 跳过", "未安装成功")

        print("\n" + "=" * 70)
        print("✓ WebExtension 模块测试完成")
        print("=" * 70)

    except Exception as e:
        rows.add("WebExtension 模块", "✗ 失败", str(e))
        import traceback

        traceback.print_exc()
    finally:
        try:
            page.extensions.uninstall_all()
        except Exception:
            pass
        page.quit()

        if os.path.exists(ext_dir):
            shutil.rmtree(ext_dir)
        if os.path.exists(xpi_path):
            os.remove(xpi_path)

        rows.add("清理测试文件", "✓ 通过")
        rows.summary()


if __name__ == "__main__":
    test_web_extension()

下载管理:下载目录、事件订阅、等待落盘、禁止下载

examples/23_download.py
GitHub 仓库 已同步源码 查看 GitHub 源码
browser strict
一句话定位

讲清下载事件链和文件落盘验证,是 downloads 页的主示例。

涉及 API
page.downloads.set_behaviorpage.downloads.startpage.downloads.wait_chainpage.downloads.waitpage.downloads.wait_filepage.downloads.stop
展示的不同用法
  • 设置 allow 下载行为并指定目录
  • 订阅下载开始和结束事件
  • 等待文本和 JSON 文件下载完成
  • 验证文件已落盘
  • 切换 deny 模式阻止下载
适用场景

适合 downloads 与 events 页面。

相关文档页面
推荐阅读
pdf-19collector-28
完整代码

来源:app/content/ruyipage_examples/23_download.py · 仓库文件:23_download.py

python 注释版
#!/usr/bin/env python
# =============================
# RuyiPage 示例注释版
# 标题: 下载管理:下载目录、事件订阅、等待落盘、禁止下载
# 说明: 讲清下载事件链和文件落盘验证,是 downloads 页的主示例。
# 你会学到:
#   1. 设置 allow 下载行为并指定目录
#   2. 订阅下载开始和结束事件
#   3. 等待文本和 JSON 文件下载完成
#   4. 验证文件已落盘
#   5. 切换 deny 模式阻止下载
# 相关 API: page.downloads.set_behavior, page.downloads.start, page.downloads.wait_chain, page.downloads.wait, page.downloads.wait_file, page.downloads.stop
# =============================

# -*- coding: utf-8 -*-
"""示例23: 下载管理测试(严格结果版)"""

import io
import os
import shutil
import sys
from typing import List, Dict

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
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 build_download_page(base_url: str) -> str:
    """构造下载测试页。"""
    return """
    <!DOCTYPE html>
    <html>
    <head><meta charset='utf-8'><title>下载测试</title></head>
    <body>
        <button id='download-text'>下载文本文件</button>
        <button id='download-json'>下载JSON文件</button>
        <div id='status'></div>
        <script>
            function triggerDownload(url, statusText) {
                const a = document.createElement('a');
                a.href = url;
                a.click();
                document.getElementById('status').textContent = statusText;
            }

            document.getElementById('download-text').onclick = () => {
                triggerDownload('%s/download/text', '文本文件下载已触发');
            };

            document.getElementById('download-json').onclick = () => {
                triggerDownload('%s/download/json', 'JSON文件下载已触发');
            };
        </script>
    </body>
    </html>
    """ % (base_url, base_url)


def main() -> None:
    print("=" * 70)
    print("测试 23: Download 下载管理测试")
    print("=" * 70)

    page = FirefoxPage()
    server = None
    results: List[Dict[str, str]] = []
    download_path = os.path.abspath(
        os.path.join(os.path.dirname(__file__), "downloads")
    )
    text_path = os.path.join(download_path, "test.txt")
    json_path = os.path.join(download_path, "test.json")

    try:
        if os.path.exists(download_path):
            shutil.rmtree(download_path)
        os.makedirs(download_path, exist_ok=True)

        server = TestServer(port=find_free_port(8930, 9030)).start()
        base_url = server.get_url("")[:-1]

        # 1. 允许下载
        try:
            page.downloads.set_behavior("allow", path=download_path)
            add_result(
                results,
                "browser.setDownloadBehavior allow",
                "成功",
                f"下载目录: {download_path}",
            )
        except Exception as e:
            add_result(
                results, "browser.setDownloadBehavior allow", "失败", str(e)[:120]
            )
            raise

        # 2. 加载页面
        page.get("data:text/html;charset=utf-8," + build_download_page(base_url))
        add_result(
            results, "下载测试页加载", "成功", f"下载页已加载,服务地址: {base_url}"
        )

        # 3. 订阅下载事件
        subscribed = page.downloads.start()
        if subscribed:
            add_result(
                results,
                "下载事件订阅",
                "成功",
                "已订阅 downloadWillBegin / downloadEnd",
            )
        else:
            add_result(
                results, "下载事件订阅", "不支持", "当前浏览器未成功订阅下载事件"
            )

        # 4. 文本文件下载
        page.downloads.clear()
        page.ele("#download-text").click_self()
        begin_text, end_text = page.downloads.wait_chain(filename="test.txt", timeout=5)

        if begin_text:
            add_result(
                results,
                "downloadWillBegin text",
                "成功",
                f"filename={begin_text.suggested_filename or 'unknown'}",
            )
        else:
            add_result(
                results, "downloadWillBegin text", "失败", "5 秒内未收到开始事件"
            )

        if end_text:
            end_status = end_text.status or "unknown"
            add_result(results, "downloadEnd text", "成功", f"status={end_status}")
        else:
            add_result(results, "downloadEnd text", "失败", "5 秒内未收到结束事件")

        if page.downloads.wait_file(text_path, timeout=3):
            add_result(
                results, "text 文件落盘", "成功", f"{os.path.getsize(text_path)} bytes"
            )
        else:
            add_result(results, "text 文件落盘", "失败", "事件触发后仍未观察到文件落盘")

        # 5. JSON 文件下载
        page.downloads.clear()
        page.ele("#download-json").click_self()
        begin_json, end_json = page.downloads.wait_chain(
            filename="test.json", timeout=5
        )

        if begin_json:
            add_result(
                results,
                "downloadWillBegin json",
                "成功",
                f"filename={begin_json.suggested_filename or 'unknown'}",
            )
        else:
            add_result(
                results, "downloadWillBegin json", "失败", "5 秒内未收到开始事件"
            )

        if end_json:
            end_status = end_json.status or "unknown"
            add_result(results, "downloadEnd json", "成功", f"status={end_status}")
        else:
            add_result(results, "downloadEnd json", "失败", "5 秒内未收到结束事件")

        if page.downloads.wait_file(json_path, timeout=3):
            add_result(
                results, "json 文件落盘", "成功", f"{os.path.getsize(json_path)} bytes"
            )
        else:
            add_result(results, "json 文件落盘", "失败", "事件触发后仍未观察到文件落盘")

        # 6. deny 模式
        page.downloads.clear()
        try:
            page.downloads.set_behavior("deny")
            page.ele("#download-text").click_self()
            denied_begin = page.downloads.wait(
                method="browsingContext.downloadWillBegin",
                timeout=2,
                filename="test.txt",
            )
            if denied_begin:
                add_result(
                    results,
                    "browser.setDownloadBehavior deny",
                    "失败",
                    "deny 模式下仍出现下载开始事件",
                )
            else:
                add_result(
                    results,
                    "browser.setDownloadBehavior deny",
                    "成功",
                    "deny 模式下未观察到下载开始事件",
                )
        except Exception as e:
            add_result(
                results, "browser.setDownloadBehavior deny", "失败", str(e)[:120]
            )

        print_results(results)

    except Exception as e:
        add_result(results, "示例执行", "失败", str(e)[:120])
        print_results(results)
        raise
    finally:
        try:
            page.downloads.stop()
        except Exception:
            pass
        try:
            if server is not None:
                server.stop()
        except Exception:
            pass
        try:
            page.quit()
        except Exception:
            pass
        if os.path.exists(download_path):
            shutil.rmtree(download_path)


if __name__ == "__main__":
    main()

导航事件:started、load、fragment、history、committed、failed

examples/24_navigation_events.py
GitHub 仓库 已同步源码 查看 GitHub 源码
network strict
一句话定位

导航事件链路的标准参考示例。

涉及 API
page.navigation.startpage.navigation.waitpage.navigation.wait_for_loadpage.navigation.wait_for_fragmentpage.navigation.stop
展示的不同用法
  • 订阅导航生命周期事件
  • 观察 navigationStarted、DOMContentLoaded 和 load
  • 观察 hash 片段跳转
  • 观察 History API 更新
  • 构造失败导航场景
适用场景

适合 navigation-events 页面。

推荐阅读
network-events-31browsing-events-30
完整代码

来源:app/content/ruyipage_examples/24_navigation_events.py · 仓库文件:24_navigation_events.py

python 注释版
#!/usr/bin/env python
# =============================
# RuyiPage 示例注释版
# 标题: 导航事件:started、load、fragment、history、committed、failed
# 说明: 导航事件链路的标准参考示例。
# 你会学到:
#   1. 订阅导航生命周期事件
#   2. 观察 navigationStarted、DOMContentLoaded 和 load
#   3. 观察 hash 片段跳转
#   4. 观察 History API 更新
#   5. 构造失败导航场景
# 相关 API: page.navigation.start, page.navigation.wait, page.navigation.wait_for_load, page.navigation.wait_for_fragment, page.navigation.stop
# =============================

# -*- coding: utf-8 -*-
"""示例24: Navigation 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
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("测试 24: Navigation Events 导航事件测试")
    print("=" * 70)

    page = FirefoxPage()
    server = None
    results: List[Dict[str, str]] = []

    try:
        server = TestServer(port=find_free_port(9030, 9130)).start()
        base_url = server.get_url("")[:-1]

        tracked_events = [
            "browsingContext.navigationStarted",
            "browsingContext.fragmentNavigated",
            "browsingContext.historyUpdated",
            "browsingContext.domContentLoaded",
            "browsingContext.load",
            "browsingContext.navigationCommitted",
            "browsingContext.navigationFailed",
        ]
        if not page.navigation.start(tracked_events):
            add_result(results, "导航事件订阅", "失败", "当前环境订阅导航事件失败")
            print_results(results)
            return

        # 1. 基础导航事件
        page.navigation.clear()
        page.get(base_url + "/nav/basic")
        started = page.navigation.wait("browsingContext.navigationStarted", timeout=3)
        dom_loaded = page.navigation.wait("browsingContext.domContentLoaded", timeout=3)
        loaded = page.navigation.wait_for_load(timeout=3)

        if started and (started.url or "").endswith("/nav/basic"):
            add_result(results, "navigationStarted", "成功", "收到基础导航开始事件")
        else:
            add_result(
                results, "navigationStarted", "失败", "未观察到基础页面导航开始事件"
            )

        if dom_loaded and dom_loaded.context == page.tab_id:
            add_result(
                results, "domContentLoaded", "成功", "收到 DOMContentLoaded 事件"
            )
        else:
            add_result(
                results, "domContentLoaded", "失败", "未观察到 DOMContentLoaded 事件"
            )

        if loaded and loaded.context == page.tab_id:
            add_result(results, "load", "成功", "收到 load 事件")
        else:
            add_result(results, "load", "失败", "未观察到 load 事件")

        # 2. fragmentNavigated
        page.get(base_url + "/nav/fragment")
        page.navigation.clear()
        page.run_js(
            """
            location.hash = '#a';
            setTimeout(function () { location.hash = '#b'; }, 100);
            """,
            as_expr=False,
        )
        first_fragment = page.navigation.wait_for_fragment("a", timeout=3)
        second_fragment = page.navigation.wait_for_fragment("b", timeout=3)

        if first_fragment and second_fragment:
            add_result(
                results,
                "fragmentNavigated",
                "成功",
                f"两次片段导航已触发: {first_fragment.url} -> {second_fragment.url}",
            )
        else:
            add_result(
                results,
                "fragmentNavigated",
                "不支持",
                "片段 URL 已变化,但当前 Firefox 未观察到标准 fragmentNavigated 事件",
            )

        # 3. historyUpdated
        page.get(base_url + "/nav/history")
        page.navigation.clear()
        page.run_js(
            """
            history.pushState({p:1}, 'P1', '?p=1');
            history.pushState({p:2}, 'P2', '?p=2');
            history.back();
            """,
            as_expr=False,
        )
        first_history = page.navigation.wait(
            "browsingContext.historyUpdated", timeout=3
        )
        second_history = page.navigation.wait(
            "browsingContext.historyUpdated", timeout=3
        )

        if first_history or second_history:
            add_result(
                results, "historyUpdated", "成功", "pushState/back 触发 historyUpdated"
            )
        else:
            add_result(
                results, "historyUpdated", "失败", "未观察到 historyUpdated 事件"
            )

        # 4. navigationCommitted
        page.navigation.clear()
        page.get(base_url + "/nav/basic?committed=1")
        committed = page.navigation.wait(
            "browsingContext.navigationCommitted", timeout=3
        )
        if committed and committed.context == page.tab_id:
            add_result(
                results,
                "navigationCommitted",
                "成功",
                "普通导航触发 navigationCommitted",
            )
        else:
            add_result(
                results,
                "navigationCommitted",
                "跳过",
                "当前环境未稳定观察到 navigationCommitted",
            )

        # 5. navigationFailed
        page.navigation.clear()
        try:
            page.get("http://127.0.0.1:9/", timeout=2)
        except Exception:
            pass
        failed = page.navigation.wait("browsingContext.navigationFailed", timeout=3)
        if failed:
            add_result(
                results, "navigationFailed", "成功", "不可达地址触发 navigationFailed"
            )
        else:
            add_result(
                results,
                "navigationFailed",
                "跳过",
                "当前环境未稳定观察到 navigationFailed",
            )

        # 6. navigationAborted
        if page.navigation.start(["browsingContext.navigationAborted"]):
            add_result(
                results,
                "navigationAborted",
                "跳过",
                "订阅成功,但本示例未构造稳定 aborted 场景",
            )
        else:
            add_result(
                results, "navigationAborted", "不支持", "当前 Firefox 不接受该事件名"
            )

        # 恢复默认跟踪,便于 finally 统一 stop
        page.navigation.start(tracked_events)

        print_results(results)

    except Exception as e:
        add_result(results, "示例执行", "失败", str(e)[:120])
        print_results(results)
        raise
    finally:
        try:
            page.navigation.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()

Browser 模块:用户上下文与客户端窗口

examples/25_browser_user_context.py
GitHub 仓库 已同步源码 查看 GitHub 源码
browser strict
一句话定位

Browser 模块最完整的 user context 与 client window 说明案例。

涉及 API
page.browser_tools.get_user_contextspage.browser_tools.create_user_contextpage.browser_tools.remove_user_contextpage.browser_tools.create_tabpage.browser_tools.get_client_windowspage.browser_tools.set_window_state
展示的不同用法
  • 读取 user context 列表
  • 创建与删除 user context
  • 在指定 user context 中创建新 tab
  • 读取 client window 列表
  • 切换窗口状态为 minimized / maximized / fullscreen
适用场景

适合 browser、user-contexts 和 window 页面。

相关文档页面
推荐阅读
contexts-26window-16
完整代码

来源:app/content/ruyipage_examples/25_browser_user_context.py · 仓库文件:25_browser_user_context.py

python 注释版
#!/usr/bin/env python
# =============================
# RuyiPage 示例注释版
# 标题: Browser 模块:用户上下文与客户端窗口
# 说明: Browser 模块最完整的 user context 与 client window 说明案例。
# 你会学到:
#   1. 读取 user context 列表
#   2. 创建与删除 user context
#   3. 在指定 user context 中创建新 tab
#   4. 读取 client window 列表
#   5. 切换窗口状态为 minimized / maximized / fullscreen
# 相关 API: page.browser_tools.get_user_contexts, page.browser_tools.create_user_context, page.browser_tools.remove_user_context, page.browser_tools.create_tab, page.browser_tools.get_client_windows, page.browser_tools.set_window_state
# =============================

# -*- coding: utf-8 -*-
"""示例25: Browser 模块(用户上下文 + 客户端窗口)严格结果版"""

import io
import sys
import time
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


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("测试 25: Browser 用户上下文与窗口管理")
    print("=" * 70)

    page = FirefoxPage()
    results: List[Dict[str, str]] = []
    user_context_id = None
    new_tab_ctx = None

    try:
        page.get("https://www.example.com")

        # 1. user context 生命周期
        before = page.browser_tools.get_user_contexts()
        before_ids = [i.get("userContext") for i in before]
        add_result(
            results,
            "browser.getUserContexts before",
            "成功",
            f"创建前数量: {len(before)}",
        )

        user_context_id = page.browser_tools.create_user_context()
        if user_context_id:
            add_result(
                results,
                "browser.createUserContext",
                "成功",
                f"userContext={user_context_id}",
            )
        else:
            add_result(
                results, "browser.createUserContext", "失败", "未返回 userContext ID"
            )

        after_create = page.browser_tools.get_user_contexts()
        after_create_ids = [i.get("userContext") for i in after_create]
        if user_context_id in after_create_ids and len(after_create) == len(before) + 1:
            add_result(
                results,
                "browser.getUserContexts after create",
                "成功",
                f"数量变为 {len(after_create)}",
            )
        else:
            add_result(
                results,
                "browser.getUserContexts after create",
                "失败",
                "创建后数量或ID校验不通过",
            )

        if user_context_id:
            new_tab_ctx = page.browser_tools.create_tab(user_context=user_context_id)
            if new_tab_ctx:
                add_result(
                    results,
                    "browsingContext.create tab in user context",
                    "成功",
                    f"context={new_tab_ctx}",
                )
            else:
                add_result(
                    results,
                    "browsingContext.create tab in user context",
                    "失败",
                    "未返回新 tab context",
                )

        if new_tab_ctx:
            try:
                page.contexts.close(new_tab_ctx)
                add_result(
                    results, "browsingContext.close new tab", "成功", "新 tab 已关闭"
                )
                new_tab_ctx = None
            except Exception as e:
                add_result(
                    results, "browsingContext.close new tab", "跳过", str(e)[:120]
                )

        if user_context_id:
            page.browser_tools.remove_user_context(user_context_id)
            add_result(
                results,
                "browser.removeUserContext",
                "成功",
                f"已删除 {user_context_id}",
            )

            after_remove = page.browser_tools.get_user_contexts()
            after_remove_ids = [i.get("userContext") for i in after_remove]
            if user_context_id not in after_remove_ids and len(after_remove) == len(
                before_ids
            ):
                add_result(
                    results,
                    "browser.getUserContexts after remove",
                    "成功",
                    f"数量回到 {len(after_remove)}",
                )
            else:
                add_result(
                    results,
                    "browser.getUserContexts after remove",
                    "失败",
                    "删除后数量或ID校验不通过",
                )
            user_context_id = None

        # 2. client window 状态流
        windows = page.browser_tools.get_client_windows()
        if not windows:
            add_result(
                results, "browser.getClientWindows", "失败", "未返回任何 client window"
            )
        else:
            add_result(
                results, "browser.getClientWindows", "成功", f"窗口数量: {len(windows)}"
            )
            win = windows[0]
            win_id = win.get("clientWindow")
            states = ["minimized", "normal", "maximized", "fullscreen", "normal"]
            state_notes = []
            state_failed = False
            for state in states:
                page.browser_tools.set_window_state(win_id, state=state)
                time.sleep(0.4)
                current_windows = page.browser_tools.get_client_windows()
                current = current_windows[0] if current_windows else {}
                current_state = current.get("state")
                state_notes.append(f"{state}->{current_state}")
                if current_state != state:
                    state_failed = True

            if state_failed:
                add_result(
                    results,
                    "browser.setClientWindowState",
                    "失败",
                    ", ".join(state_notes),
                )
            else:
                add_result(
                    results,
                    "browser.setClientWindowState",
                    "成功",
                    ", ".join(state_notes),
                )

        print_results(results)

    except Exception as e:
        add_result(results, "示例执行", "失败", str(e)[:120])
        print_results(results)
        raise
    finally:
        if new_tab_ctx:
            try:
                page.contexts.close(new_tab_ctx)
            except Exception:
                pass
        if user_context_id:
            try:
                page.browser_tools.remove_user_context(user_context_id)
            except Exception:
                pass
        try:
            page.quit()
        except Exception:
            pass


if __name__ == "__main__":
    main()

BrowsingContext 高级能力:树、创建、关闭、reload、viewport

examples/26_browsing_context_advanced.py
GitHub 仓库 已同步源码 查看 GitHub 源码
browser strict
一句话定位

是 browsing context 章节的主参考示例。

涉及 API
page.contexts.get_treepage.contexts.create_tabpage.contexts.closepage.contexts.reloadpage.contexts.set_bypass_csppage.contexts.set_viewport
展示的不同用法
  • 获取 browsing context 树
  • 通过标准命令创建和关闭 context
  • 执行普通 reload 和 ignore_cache reload
  • 设置 bypass CSP
  • 调整桌面和移动视口尺寸
适用场景

适合 browsing-context 页面。

推荐阅读
browser-user-context-25tabs-09
完整代码

来源:app/content/ruyipage_examples/26_browsing_context_advanced.py · 仓库文件:26_browsing_context_advanced.py

python 注释版
#!/usr/bin/env python
# =============================
# RuyiPage 示例注释版
# 标题: BrowsingContext 高级能力:树、创建、关闭、reload、viewport
# 说明: 是 browsing context 章节的主参考示例。
# 你会学到:
#   1. 获取 browsing context 树
#   2. 通过标准命令创建和关闭 context
#   3. 执行普通 reload 和 ignore_cache reload
#   4. 设置 bypass CSP
#   5. 调整桌面和移动视口尺寸
# 相关 API: page.contexts.get_tree, page.contexts.create_tab, page.contexts.close, page.contexts.reload, page.contexts.set_bypass_csp, page.contexts.set_viewport
# =============================

# -*- coding: utf-8 -*-
"""示例26: BrowsingContext 高级功能(严格结果版)"""

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


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("测试 26: BrowsingContext 高级功能")
    print("=" * 70)

    page = FirefoxPage()
    child_id = None
    results: List[Dict[str, str]] = []

    try:
        page.get("https://www.example.com")
        add_result(results, "页面加载", "成功", "example.com 已加载")

        tree = page.contexts.get_tree()
        contexts = tree.contexts
        if contexts:
            add_result(
                results,
                "browsingContext.getTree",
                "成功",
                f"上下文数量: {len(contexts)}",
            )
        else:
            add_result(results, "browsingContext.getTree", "失败", "未返回任何上下文")

        child_id = page.contexts.create_tab()
        if child_id:
            add_result(
                results, "browsingContext.create", "成功", f"新 context: {child_id}"
            )
            page.contexts.close(child_id)
            add_result(results, "browsingContext.close", "成功", "新 context 已关闭")
            child_id = None
        else:
            add_result(results, "browsingContext.create", "失败", "未返回 context ID")

        r1 = page.contexts.reload()
        if r1.get("navigation"):
            add_result(
                results,
                "browsingContext.reload",
                "成功",
                f"navigation={r1.get('navigation')}",
            )
        else:
            add_result(
                results, "browsingContext.reload", "跳过", "当前返回未包含 navigation"
            )

        try:
            r2 = page.contexts.reload(ignore_cache=True)
            add_result(
                results,
                "browsingContext.reload ignoreCache",
                "成功",
                f"navigation={r2.get('navigation')}",
            )
        except Exception as e:
            add_result(
                results, "browsingContext.reload ignoreCache", "不支持", str(e)[:120]
            )

        try:
            page.contexts.set_bypass_csp(True)
            add_result(
                results, "browsingContext.setBypassCSP", "成功", "标准命令调用成功"
            )
        except Exception as e:
            add_result(results, "browsingContext.setBypassCSP", "不支持", str(e)[:120])

        page.contexts.set_viewport(800, 600)
        page.wait(0.2)
        vp = page.rect.viewport_size
        if tuple(vp) == (800, 600):
            add_result(
                results,
                "browsingContext.setViewport 800x600",
                "成功",
                f"实际视口: {vp}",
            )
        else:
            add_result(
                results,
                "browsingContext.setViewport 800x600",
                "失败",
                f"实际视口: {vp}",
            )

        page.contexts.set_viewport(375, 667)
        page.wait(0.2)
        vp2 = page.rect.viewport_size
        if tuple(vp2) == (375, 667):
            add_result(
                results,
                "browsingContext.setViewport 375x667",
                "成功",
                f"实际视口: {vp2}",
            )
        else:
            add_result(
                results,
                "browsingContext.setViewport 375x667",
                "失败",
                f"实际视口: {vp2}",
            )

        print_results(results)

    except Exception as e:
        add_result(results, "示例执行", "失败", str(e)[:120])
        print_results(results)
        raise
    finally:
        if child_id:
            try:
                page.contexts.close(child_id)
            except Exception:
                pass
        try:
            page.quit()
        except Exception:
            pass


if __name__ == "__main__":
    main()

Emulation 高级能力:颜色模式、屏幕、方向、脚本与滚动条

examples/27_emulation_advanced.py
GitHub 仓库 已同步源码 查看 GitHub 源码
emulation strict
一句话定位

强调 emulation 的高级能力和 Firefox 当前支持情况。

涉及 API
page.emulation.set_forced_colors_modepage.emulation.set_screen_sizepage.emulation.set_screen_orientationpage.emulation.set_javascript_enabledpage.emulation.set_scrollbar_type
展示的不同用法
  • 切换 forced colors 模式
  • 调整屏幕尺寸和 DPR
  • 切换屏幕方向和角度
  • 启用或禁用 JavaScript
  • 切换滚动条显示模式
适用场景

适合 emulation 的高级能力章节。

相关文档页面
推荐阅读
emulation-21
完整代码

来源:app/content/ruyipage_examples/27_emulation_advanced.py · 仓库文件:27_emulation_advanced.py

python 注释版
#!/usr/bin/env python
# =============================
# RuyiPage 示例注释版
# 标题: Emulation 高级能力:颜色模式、屏幕、方向、脚本与滚动条
# 说明: 强调 emulation 的高级能力和 Firefox 当前支持情况。
# 你会学到:
#   1. 切换 forced colors 模式
#   2. 调整屏幕尺寸和 DPR
#   3. 切换屏幕方向和角度
#   4. 启用或禁用 JavaScript
#   5. 切换滚动条显示模式
# 相关 API: page.emulation.set_forced_colors_mode, page.emulation.set_screen_size, page.emulation.set_screen_orientation, page.emulation.set_javascript_enabled, page.emulation.set_scrollbar_type
# =============================

# -*- coding: utf-8 -*-
"""示例27: Emulation 高级能力(严格结果版)"""

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


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("测试 27: Emulation 高级能力")
    print("=" * 70)

    page = FirefoxPage()
    results: List[Dict[str, str]] = []

    try:
        page.get("https://www.example.com")
        add_result(results, "页面加载", "成功", "example.com 已加载")

        # 1. 强制颜色模式
        forced_active = page.emulation.set_forced_colors_mode("active")
        add_result(
            results,
            "emulation.setForcedColorsModeThemeOverride active",
            "成功" if forced_active else "不支持",
            "标准命令已实现" if forced_active else "当前 Firefox 未实现该命令",
        )

        forced_none = page.emulation.set_forced_colors_mode("none")
        add_result(
            results,
            "emulation.setForcedColorsModeThemeOverride none",
            "成功" if forced_none else "不支持",
            "标准命令已实现" if forced_none else "当前 Firefox 未实现该命令",
        )

        # 2. 屏幕设置
        page.emulation.set_screen_size(1920, 1080, device_pixel_ratio=2.0)
        size1 = page.run_js(
            "return [screen.width, screen.height, window.devicePixelRatio]"
        )
        if (
            size1
            and int(size1[0]) == 1920
            and int(size1[1]) == 1080
            and float(size1[2]) == 2.0
        ):
            add_result(
                results,
                "emulation.setScreenSettingsOverride 1920x1080@2",
                "成功",
                str(size1),
            )
        elif size1 and int(size1[0]) == 1920 and int(size1[1]) == 1080:
            add_result(
                results,
                "emulation.setScreenSettingsOverride 1920x1080@2",
                "跳过",
                f"尺寸生效,但 DPR 未生效: {size1}",
            )
        else:
            add_result(
                results,
                "emulation.setScreenSettingsOverride 1920x1080@2",
                "失败",
                str(size1)[:120],
            )

        page.emulation.set_screen_size(375, 812, device_pixel_ratio=3.0)
        size2 = page.run_js(
            "return [screen.width, screen.height, window.devicePixelRatio]"
        )
        if (
            size2
            and int(size2[0]) == 375
            and int(size2[1]) == 812
            and float(size2[2]) == 3.0
        ):
            add_result(
                results,
                "emulation.setScreenSettingsOverride 375x812@3",
                "成功",
                str(size2),
            )
        elif size2 and int(size2[0]) == 375 and int(size2[1]) == 812:
            add_result(
                results,
                "emulation.setScreenSettingsOverride 375x812@3",
                "跳过",
                f"尺寸生效,但 DPR 未生效: {size2}",
            )
        else:
            add_result(
                results,
                "emulation.setScreenSettingsOverride 375x812@3",
                "失败",
                str(size2)[:120],
            )

        # 3. 屏幕方向
        page.emulation.set_screen_orientation("portrait-primary", angle=0)
        orientation1 = page.run_js(
            "return [screen.orientation.type, screen.orientation.angle]"
        )
        if orientation1 and orientation1[0] == "portrait-primary":
            add_result(
                results,
                "emulation.setScreenOrientationOverride portrait",
                "成功",
                str(orientation1),
            )
        else:
            add_result(
                results,
                "emulation.setScreenOrientationOverride portrait",
                "失败",
                str(orientation1)[:120],
            )

        page.emulation.set_screen_orientation("landscape-primary", angle=90)
        orientation2 = page.run_js(
            "return [screen.orientation.type, screen.orientation.angle]"
        )
        if (
            orientation2
            and orientation2[0] == "landscape-primary"
            and int(orientation2[1]) == 90
        ):
            add_result(
                results,
                "emulation.setScreenOrientationOverride landscape",
                "成功",
                str(orientation2),
            )
        elif orientation2 and orientation2[0] == "landscape-primary":
            add_result(
                results,
                "emulation.setScreenOrientationOverride landscape",
                "跳过",
                f"方向类型生效,但角度未生效: {orientation2}",
            )
        else:
            add_result(
                results,
                "emulation.setScreenOrientationOverride landscape",
                "失败",
                str(orientation2)[:120],
            )

        # 4. 脚本启用/禁用
        js_off = page.emulation.set_javascript_enabled(False)
        add_result(
            results,
            "emulation.setScriptingEnabled false",
            "成功" if js_off else "不支持",
            "标准命令已实现" if js_off else "当前 Firefox 未实现该命令",
        )

        js_on = page.emulation.set_javascript_enabled(True)
        add_result(
            results,
            "emulation.setScriptingEnabled true",
            "成功" if js_on else "不支持",
            "标准命令已实现" if js_on else "当前 Firefox 未实现该命令",
        )

        # 5. 滚动条类型
        sb_none = page.emulation.set_scrollbar_type("none")
        add_result(
            results,
            "emulation.setScrollbarTypeOverride none",
            "成功" if sb_none else "不支持",
            "标准命令已实现" if sb_none else "当前 Firefox 未实现该命令",
        )

        sb_standard = page.emulation.set_scrollbar_type("standard")
        add_result(
            results,
            "emulation.setScrollbarTypeOverride standard",
            "成功" if sb_standard else "不支持",
            "标准命令已实现" if sb_standard else "当前 Firefox 未实现该命令",
        )

        sb_overlay = page.emulation.set_scrollbar_type("overlay")
        add_result(
            results,
            "emulation.setScrollbarTypeOverride overlay",
            "成功" if sb_overlay else "不支持",
            "标准命令已实现" if sb_overlay else "当前 Firefox 未实现该命令",
        )

        print_results(results)

    except Exception as e:
        add_result(results, "示例执行", "失败", str(e)[:120])
        print_results(results)
        raise
    finally:
        try:
            page.quit()
        except Exception:
            pass


if __name__ == "__main__":
    main()

网络数据采集器:响应体抓取、读取与释放

examples/28_network_data_collector.py
GitHub 仓库 已同步源码 查看 GitHub 源码
network strict
一句话定位

用于说明 add_data_collector 在数据抓取和协议观测中的价值。

涉及 API
page.network.set_extra_headerspage.network.set_cache_behaviorpage.network.add_data_collectorcollector.getcollector.disowncollector.removepage.listen.start
展示的不同用法
  • 注入额外请求头
  • 设置缓存 bypass
  • 添加并使用 data collector 采集 response 数据
  • 按 request_id 读取采集结果
  • 释放采集句柄并移除 collector
适用场景

适合 data-collector 与 network 页面。

相关文档页面
推荐阅读
advanced-network-18network-events-31
完整代码

来源:app/content/ruyipage_examples/28_network_data_collector.py · 仓库文件:28_network_data_collector.py

python 注释版
#!/usr/bin/env python
# =============================
# RuyiPage 示例注释版
# 标题: 网络数据采集器:响应体抓取、读取与释放
# 说明: 用于说明 add_data_collector 在数据抓取和协议观测中的价值。
# 你会学到:
#   1. 注入额外请求头
#   2. 设置缓存 bypass
#   3. 添加并使用 data collector 采集 response 数据
#   4. 按 request_id 读取采集结果
#   5. 释放采集句柄并移除 collector
# 相关 API: page.network.set_extra_headers, page.network.set_cache_behavior, page.network.add_data_collector, collector.get, collector.disown, collector.remove, page.listen.start
# =============================

# -*- coding: utf-8 -*-
"""示例28: Network Data Collector(严格结果版)"""

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 DataCollector, DataPacket, 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("测试 28: Network Data Collector")
    print("=" * 70)

    page = FirefoxPage()
    results: List[Dict[str, str]] = []
    collector: DataCollector | None = None
    server = None

    try:
        server = TestServer(port=find_free_port(9230, 9330)).start()
        base_url = server.get_url("")[:-1]

        page.get("about:blank")
        add_result(
            results, "页面加载", "成功", f"about:blank 已加载,服务地址: {base_url}"
        )

        page.network.set_extra_headers({"X-Test-Collector": "yes"})
        add_result(
            results, "network.setExtraHeaders", "成功", "已设置 X-Test-Collector"
        )

        try:
            page.network.set_cache_behavior("bypass")
            add_result(
                results, "network.setCacheBehavior", "成功", "缓存行为已设为 bypass"
            )
        except Exception as e:
            add_result(results, "network.setCacheBehavior", "不支持", str(e)[:120])

        collector = page.network.add_data_collector(
            ["responseCompleted"],
            data_types=["response"],
        )
        if collector.id:
            add_result(
                results, "network.addDataCollector", "成功", f"collector={collector.id}"
            )
        else:
            add_result(
                results, "network.addDataCollector", "失败", "未返回 collector ID"
            )

        page.intercept.start_requests()
        page.listen.start("/api/collector")
        page.run_js(
            f"fetch('{base_url}/api/collector').catch(()=>null); return true;",
            as_expr=False,
        )

        req: InterceptedRequest | None = page.intercept.wait(timeout=8)

        if req:
            has_header = "X-Test-Collector" in req.headers
            add_result(
                results,
                "request header injected",
                "成功" if has_header else "失败",
                f"request_id={req.request_id}, X-Test-Collector={has_header}",
            )
            req.continue_request()
        else:
            add_result(
                results,
                "request header injected",
                "失败",
                "未捕获到 beforeRequestSent 请求",
            )

        packet: DataPacket | None = page.listen.wait(timeout=8)

        if packet:
            add_result(
                results,
                "network.responseCompleted observed",
                "成功",
                f"status={packet.status} url={packet.url}",
            )
        else:
            add_result(
                results,
                "network.responseCompleted observed",
                "失败",
                "未在超时内观察到 responseCompleted",
            )

        if collector and req:
            got_response_data = False
            try:
                data = collector.get(req.request_id, data_type="response")
                if data.has_data:
                    got_response_data = True
                    add_result(
                        results,
                        "network.getData response",
                        "成功",
                        f"bytes={data.bytes} base64={'yes' if data.base64 else 'no'}",
                    )
                else:
                    add_result(
                        results, "network.getData response", "失败", f"raw={data.raw}"
                    )
            except Exception as e:
                add_result(results, "network.getData response", "失败", str(e)[:120])

            if got_response_data:
                try:
                    collector.disown(req.request_id, data_type="response")
                    add_result(
                        results, "network.disownData", "成功", "已释放 response 数据"
                    )
                except Exception as e:
                    add_result(results, "network.disownData", "失败", str(e)[:120])

                try:
                    data_after = collector.get(req.request_id, data_type="response")
                    if not data_after.has_data:
                        add_result(
                            results,
                            "network.getData after disown",
                            "成功",
                            "释放后已无可用数据",
                        )
                    else:
                        add_result(
                            results,
                            "network.getData after disown",
                            "跳过",
                            f"释放后仍可读取: {data_after.raw}",
                        )
                except Exception as e:
                    add_result(
                        results,
                        "network.getData after disown",
                        "成功",
                        f"释放后读取报错: {str(e)[:100]}",
                    )
            else:
                add_result(
                    results,
                    "network.disownData",
                    "跳过",
                    "未成功拿到 response 数据,跳过释放验证",
                )
                add_result(
                    results,
                    "network.getData after disown",
                    "跳过",
                    "未成功拿到 response 数据,跳过释放后验证",
                )

        if collector:
            try:
                collector.remove()
                add_result(
                    results,
                    "network.removeDataCollector",
                    "成功",
                    f"已移除 {collector.id}",
                )
                collector = None
            except Exception as e:
                add_result(results, "network.removeDataCollector", "失败", str(e)[:120])

        page.network.clear_extra_headers()
        add_result(results, "clear extra headers", "成功", "额外请求头已清理")

        print_results(results)

    except Exception as e:
        add_result(results, "示例执行", "失败", str(e)[:120])
        print_results(results)
        raise
    finally:
        try:
            page.listen.stop()
        except Exception:
            pass
        try:
            page.intercept.stop()
        except Exception:
            pass
        try:
            if server is not None:
                server.stop()
        except Exception:
            pass
        if collector:
            try:
                collector.remove()
            except Exception:
                pass
        try:
            page.network.clear_extra_headers()
        except Exception:
            pass
        try:
            page.quit()
        except Exception:
            pass


if __name__ == "__main__":
    main()

Script 与 Input 高级能力:realms、handle 回收、文件输入

examples/29_script_input_advanced.py
GitHub 仓库 已同步源码 查看 GitHub 源码
debug strict
一句话定位

连接 script 与 input 的高级能力,是 script 页的重要支撑示例。

涉及 API
page.get_realmspage.eval_handlepage.disown_handlesele.input
展示的不同用法
  • 获取全部和 window 类型 realms
  • 执行脚本得到远程对象句柄
  • 回收单个和批量 handle
  • 利用 file input 上传单文件和多文件
适用场景

适合 script 和 forms 页面。

相关文档页面
推荐阅读
javascript-07script-events-32
完整代码

来源:app/content/ruyipage_examples/29_script_input_advanced.py · 仓库文件:29_script_input_advanced.py

python 注释版
#!/usr/bin/env python
# =============================
# RuyiPage 示例注释版
# 标题: Script 与 Input 高级能力:realms、handle 回收、文件输入
# 说明: 连接 script 与 input 的高级能力,是 script 页的重要支撑示例。
# 你会学到:
#   1. 获取全部和 window 类型 realms
#   2. 执行脚本得到远程对象句柄
#   3. 回收单个和批量 handle
#   4. 利用 file input 上传单文件和多文件
# 相关 API: page.get_realms, page.eval_handle, page.disown_handles, ele.input
# =============================

# -*- coding: utf-8 -*-
"""示例29: Script + Input 高级能力(严格结果版)"""

import io
import os
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, ScriptResult


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("测试 29: Script + Input 高级能力")
    print("=" * 70)

    page = FirefoxPage()
    results: List[Dict[str, str]] = []
    html_path = os.path.join(os.path.dirname(__file__), "test_file_input.html")
    file1 = os.path.join(os.path.dirname(__file__), "test1.txt")
    file2 = os.path.join(os.path.dirname(__file__), "test2.txt")

    try:
        page.get("https://www.example.com")
        add_result(results, "页面加载", "成功", "example.com 已加载")

        # 1. script.getRealms
        realms = page.get_realms()
        if realms:
            add_result(
                results, "script.getRealms all", "成功", f"realm 数量: {len(realms)}"
            )
        else:
            add_result(results, "script.getRealms all", "失败", "未返回任何 realm")

        window_realms = page.get_realms(type_="window")
        if window_realms:
            add_result(
                results,
                "script.getRealms window",
                "成功",
                f"window realm 数量: {len(window_realms)}",
            )
        else:
            add_result(
                results, "script.getRealms window", "失败", "未返回 window realm"
            )

        # 2. script.disown
        single: ScriptResult = page.eval_handle("({data: 'test', array: [1, 2, 3]})")
        if single.success and single.result.handle:
            page.disown_handles([single.result.handle])
            add_result(
                results,
                "script.disown single",
                "成功",
                f"handle={single.result.handle}",
            )
        else:
            add_result(results, "script.disown single", "跳过", "脚本结果未返回 handle")

        handles = []
        for i in range(3):
            item = page.eval_handle(f"({{id: {i}, value: 'test{i}'}})")
            if item.success and item.result.handle:
                handles.append(item.result.handle)
        if handles:
            page.disown_handles(handles)
            add_result(
                results, "script.disown batch", "成功", f"句柄数量: {len(handles)}"
            )
        else:
            add_result(results, "script.disown batch", "跳过", "未拿到可用 handle")

        # 3. input.setFiles 通过高层文件输入
        html_content = """
        <!DOCTYPE html>
        <html>
        <head><meta charset='utf-8'><title>File Input Test</title></head>
        <body>
            <input type='file' id='single-file' />
            <input type='file' id='multiple-files' multiple />
            <div id='result'></div>
            <script>
                document.getElementById('single-file').addEventListener('change', function(e) {
                    document.getElementById('result').textContent = 'Single file: ' + e.target.files[0].name;
                });
                document.getElementById('multiple-files').addEventListener('change', function(e) {
                    document.getElementById('result').textContent = 'Multiple files: ' + e.target.files.length;
                });
            </script>
        </body>
        </html>
        """

        with open(html_path, "w", encoding="utf-8") as f:
            f.write(html_content)
        with open(file1, "w", encoding="utf-8") as f:
            f.write("Test file 1 content")
        with open(file2, "w", encoding="utf-8") as f:
            f.write("Test file 2 content")

        page.get(f"file:///{html_path.replace(os.sep, '/')}")
        add_result(results, "文件测试页加载", "成功", "本地 file input 页面已加载")

        page.ele("#single-file").input(file1)
        single_text = page.ele("#result").text
        if "test1.txt" in single_text:
            add_result(results, "input.setFiles single", "成功", single_text)
        else:
            add_result(results, "input.setFiles single", "失败", single_text)

        page.ele("#multiple-files").input([file1, file2])
        multi_text = page.ele("#result").text
        if "Multiple files: 2" in multi_text:
            add_result(results, "input.setFiles multiple", "成功", multi_text)
        else:
            add_result(results, "input.setFiles multiple", "失败", multi_text)

        print_results(results)

    except Exception as e:
        add_result(results, "示例执行", "失败", str(e)[:120])
        print_results(results)
        raise
    finally:
        for path in (file1, file2, html_path):
            try:
                if os.path.exists(path):
                    os.remove(path)
            except Exception:
                pass
        try:
            page.quit()
        except Exception:
            pass


if __name__ == "__main__":
    main()

BrowsingContext 事件:上下文创建销毁与用户提示框事件

examples/30_browsing_context_events.py
GitHub 仓库 已同步源码 查看 GitHub 源码
network strict
一句话定位

把 context 事件和 prompt 事件统一放到协议事件角度理解。

涉及 API
page.events.startpage.events.waitpage.contexts.create_tabpage.contexts.create_windowpage.contexts.closepage.accept_promptpage.set_prompt_handler
展示的不同用法
  • 监听 tab 和 window 创建销毁
  • 监听 alert 和 prompt 打开关闭
  • 配合自动 prompt 策略处理事件链
适用场景

适合 browsing-context、events 与 prompts 页面。

推荐阅读
navigation-events-24prompts-17
完整代码

来源:app/content/ruyipage_examples/30_browsing_context_events.py · 仓库文件:30_browsing_context_events.py

python 注释版
#!/usr/bin/env python
# =============================
# RuyiPage 示例注释版
# 标题: BrowsingContext 事件:上下文创建销毁与用户提示框事件
# 说明: 把 context 事件和 prompt 事件统一放到协议事件角度理解。
# 你会学到:
#   1. 监听 tab 和 window 创建销毁
#   2. 监听 alert 和 prompt 打开关闭
#   3. 配合自动 prompt 策略处理事件链
# 相关 API: page.events.start, page.events.wait, page.contexts.create_tab, page.contexts.create_window, page.contexts.close, page.accept_prompt, page.set_prompt_handler
# =============================

# -*- coding: utf-8 -*-
"""示例30: BrowsingContext 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 BidiEvent, FirefoxPage


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("测试 30: BrowsingContext Events")
    print("=" * 70)

    page = FirefoxPage()
    results: List[Dict[str, str]] = []
    temp_tab = None
    temp_window = None

    try:
        page.get("https://www.example.com")
        add_result(results, "页面加载", "成功", "example.com 已加载")

        if page.events.start(
            [
                "browsingContext.contextCreated",
                "browsingContext.contextDestroyed",
                "browsingContext.userPromptOpened",
                "browsingContext.userPromptClosed",
            ],
            contexts=[],
        ):
            add_result(results, "事件订阅", "成功", "已订阅 context / prompt 相关事件")
        else:
            add_result(results, "事件订阅", "失败", "未能订阅 browsingContext 事件")
            print_results(results)
            return

        # 1. contextCreated / contextDestroyed
        page.events.clear()
        temp_tab = page.contexts.create_tab()
        created_tab: BidiEvent | None = page.events.wait(
            "browsingContext.contextCreated", timeout=3
        )
        if created_tab and created_tab.context == temp_tab:
            add_result(
                results,
                "browsingContext.contextCreated tab",
                "成功",
                f"context={temp_tab}",
            )
        else:
            add_result(
                results,
                "browsingContext.contextCreated tab",
                "失败",
                f"expected={temp_tab}",
            )

        temp_window = page.contexts.create_window()
        created_window: BidiEvent | None = page.events.wait(
            "browsingContext.contextCreated", timeout=3
        )
        if created_window and created_window.context == temp_window:
            add_result(
                results,
                "browsingContext.contextCreated window",
                "成功",
                f"context={temp_window}",
            )
        else:
            add_result(
                results,
                "browsingContext.contextCreated window",
                "失败",
                f"expected={temp_window}",
            )

        if temp_tab:
            page.contexts.close(temp_tab)
            destroyed_tab = page.events.wait(
                "browsingContext.contextDestroyed", timeout=3
            )
            if destroyed_tab and destroyed_tab.context == temp_tab:
                add_result(
                    results,
                    "browsingContext.contextDestroyed tab",
                    "成功",
                    f"context={temp_tab}",
                )
            else:
                add_result(
                    results,
                    "browsingContext.contextDestroyed tab",
                    "失败",
                    f"expected={temp_tab}",
                )
            temp_tab = None

        if temp_window:
            page.contexts.close(temp_window)
            destroyed_window = page.events.wait(
                "browsingContext.contextDestroyed", timeout=3
            )
            if destroyed_window and destroyed_window.context == temp_window:
                add_result(
                    results,
                    "browsingContext.contextDestroyed window",
                    "成功",
                    f"context={temp_window}",
                )
            else:
                add_result(
                    results,
                    "browsingContext.contextDestroyed window",
                    "失败",
                    f"expected={temp_window}",
                )
            temp_window = None

        # 2. userPromptOpened / userPromptClosed
        page.events.clear()

        page.run_js("alert('hello alert')", as_expr=False)
        opened_alert = page.events.wait("browsingContext.userPromptOpened", timeout=3)
        if opened_alert and opened_alert.user_prompt_type == "alert":
            add_result(
                results,
                "browsingContext.userPromptOpened alert",
                "成功",
                opened_alert.message or "",
            )
        else:
            add_result(
                results,
                "browsingContext.userPromptOpened alert",
                "失败",
                "未观察到 alert 打开事件",
            )

        page.accept_prompt(timeout=3)
        closed_alert = page.events.wait("browsingContext.userPromptClosed", timeout=3)
        if closed_alert and closed_alert.accepted is True:
            add_result(
                results,
                "browsingContext.userPromptClosed alert",
                "成功",
                f"accepted={closed_alert.accepted}",
            )
        else:
            add_result(
                results,
                "browsingContext.userPromptClosed alert",
                "失败",
                "未观察到 alert 关闭事件",
            )

        page.events.clear()
        page.set_prompt_handler(prompt="ignore", prompt_text="Test User")

        page.run_js("prompt('Enter your name:', 'default')", as_expr=False)
        opened_prompt = page.events.wait("browsingContext.userPromptOpened", timeout=3)
        if opened_prompt and opened_prompt.user_prompt_type == "prompt":
            add_result(
                results,
                "browsingContext.userPromptOpened prompt",
                "成功",
                opened_prompt.message or "",
            )
        else:
            add_result(
                results,
                "browsingContext.userPromptOpened prompt",
                "失败",
                "未观察到 prompt 打开事件",
            )

        closed_prompt = page.events.wait("browsingContext.userPromptClosed", timeout=3)
        if closed_prompt and closed_prompt.accepted is True:
            add_result(
                results,
                "browsingContext.userPromptClosed prompt",
                "成功",
                f"accepted={closed_prompt.accepted}",
            )
        else:
            add_result(
                results,
                "browsingContext.userPromptClosed prompt",
                "跳过",
                "当前环境下 prompt 自动注入文本后未稳定观察到 userPromptClosed",
            )

        page.clear_prompt_handler()
        print_results(results)

    except Exception as e:
        add_result(results, "示例执行", "失败", str(e)[:120])
        print_results(results)
        raise
    finally:
        try:
            page.clear_prompt_handler()
        except Exception:
            pass
        try:
            page.events.stop()
        except Exception:
            pass
        if temp_tab:
            try:
                page.contexts.close(temp_tab)
            except Exception:
                pass
        if temp_window:
            try:
                page.contexts.close(temp_window)
            except Exception:
                pass
        try:
            page.quit()
        except Exception:
            pass


if __name__ == "__main__":
    main()

网络事件:请求开始、响应开始、响应完成、错误与认证

examples/31_network_events.py
GitHub 仓库 已同步源码 查看 GitHub 源码
network strict
一句话定位

是 network-events 页最直接的主示例。

涉及 API
page.events.startpage.events.waitpage.intercept.startpage.intercept.waitcontinue_with_auth
展示的不同用法
  • 观察 beforeRequestSent、responseStarted、responseCompleted
  • 监听 fetchError
  • 处理 authRequired 并补充用户名密码
适用场景

适合 network-events 与 auth 页面。

相关文档页面
推荐阅读
advanced-network-18collector-28
完整代码

来源:app/content/ruyipage_examples/31_network_events.py · 仓库文件:31_network_events.py

python 注释版
#!/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()

Script 事件:realm 创建销毁与 preload 消息

examples/32_script_events.py
GitHub 仓库 已同步源码 查看 GitHub 源码
debug strict
一句话定位

用于说明 realms 和 preload script 如何和事件订阅配合。

涉及 API
page.realms.startpage.realms.listpage.events.startpage.events.waitpage.add_preload_scriptpage.remove_preload_script
展示的不同用法
  • 观察 realm 创建与销毁
  • 订阅 script.message
  • 通过 preload script 发送事件消息
适用场景

适合 script 和 preload-scripts 页面。

推荐阅读
javascript-07script-input-29
完整代码

来源:app/content/ruyipage_examples/32_script_events.py · 仓库文件:32_script_events.py

python 注释版
#!/usr/bin/env python
# =============================
# RuyiPage 示例注释版
# 标题: Script 事件:realm 创建销毁与 preload 消息
# 说明: 用于说明 realms 和 preload script 如何和事件订阅配合。
# 你会学到:
#   1. 观察 realm 创建与销毁
#   2. 订阅 script.message
#   3. 通过 preload script 发送事件消息
# 相关 API: page.realms.start, page.realms.list, page.events.start, page.events.wait, page.add_preload_script, page.remove_preload_script
# =============================

# -*- coding: utf-8 -*-
"""示例32: Script 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, PreloadScript


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("测试 32: Script Events")
    print("=" * 70)

    page = FirefoxPage()
    results: List[Dict[str, str]] = []
    temp_tab = None
    preload: PreloadScript | None = None

    try:
        page.get("https://www.example.com")
        add_result(results, "页面加载", "成功", "example.com 已加载")

        # 1. realmCreated / realmDestroyed
        page.realms.start()
        initial_realms = page.realms.list()
        add_result(
            results,
            "script.getRealms baseline",
            "成功",
            f"初始 realm 数量: {len(initial_realms)}",
        )

        temp_tab = page.contexts.create_tab()
        page.wait(1)
        created_realms = page.realms.list()
        if len(created_realms) > len(initial_realms):
            add_result(
                results,
                "script.realmCreated",
                "成功",
                f"realm 数量: {len(initial_realms)} -> {len(created_realms)}",
            )
        else:
            add_result(
                results, "script.realmCreated", "跳过", "当前环境未稳定观察到新增 realm"
            )

        if temp_tab:
            page.contexts.close(temp_tab)
            temp_tab = None
            page.wait(1)

        destroyed_realms = page.realms.list()
        if len(destroyed_realms) <= len(created_realms):
            add_result(
                results,
                "script.realmDestroyed",
                "成功",
                f"关闭后 realm 数量: {len(destroyed_realms)}",
            )
        else:
            add_result(
                results,
                "script.realmDestroyed",
                "跳过",
                "当前环境未稳定观察到 realm 销毁",
            )

        page.realms.stop()

        # 2. script.message
        if page.events.start(["script.message"], contexts=[page.tab_id]):
            preload = page.add_preload_script(
                """
                () => {
                    const ch = new BroadcastChannel('ruyi-script-message');
                    ch.postMessage({ type: 'preload', value: 'hello' });
                    ch.close();
                }
                """
            )

            page.get("https://www.example.com/?script-message=1")
            message_event = page.events.wait("script.message", timeout=3)

            if message_event:
                add_result(
                    results,
                    "script.message",
                    "成功",
                    f"channel={message_event.channel} data={message_event.data}",
                )
            else:
                add_result(
                    results,
                    "script.message",
                    "跳过",
                    "当前环境未稳定观察到 script.message",
                )
        else:
            add_result(
                results, "script.message", "不支持", "未能订阅 script.message 事件"
            )

        print_results(results)

    except Exception as e:
        add_result(results, "示例执行", "失败", str(e)[:120])
        print_results(results)
        raise
    finally:
        if preload is not None:
            try:
                page.remove_preload_script(preload)
            except Exception:
                pass
        try:
            page.realms.stop()
        except Exception:
            pass
        try:
            page.events.stop()
        except Exception:
            pass
        if temp_tab:
            try:
                page.contexts.close(temp_tab)
            except Exception:
                pass
        try:
            page.quit()
        except Exception:
            pass


if __name__ == "__main__":
    main()

日志与输入事件:log.entryAdded 与 fileDialogOpened

examples/33_log_input_events.py
GitHub 仓库 已同步源码 查看 GitHub 源码
debug strict
一句话定位

把日志和输入事件统一纳入事件系统理解。

涉及 API
page.console.startpage.console.waitpage.events.startpage.events.waitclick(by_js=True)
展示的不同用法
  • 输出并监听多级别 console 日志
  • 读取 error / warn / table 内容
  • 订阅 fileDialogOpened 事件
  • 观察单文件和多文件输入弹窗打开链路
适用场景

适合 console、events 和 forms 页面。

相关文档页面
推荐阅读
console-12downloads-23
完整代码

来源:app/content/ruyipage_examples/33_log_input_events.py · 仓库文件:33_log_input_events.py

python 注释版
#!/usr/bin/env python
# =============================
# RuyiPage 示例注释版
# 标题: 日志与输入事件:log.entryAdded 与 fileDialogOpened
# 说明: 把日志和输入事件统一纳入事件系统理解。
# 你会学到:
#   1. 输出并监听多级别 console 日志
#   2. 读取 error / warn / table 内容
#   3. 订阅 fileDialogOpened 事件
#   4. 观察单文件和多文件输入弹窗打开链路
# 相关 API: page.console.start, page.console.wait, page.events.start, page.events.wait, click(by_js=True)
# =============================

# -*- coding: utf-8 -*-
"""示例33: Log + Input Events(严格结果版)"""

import io
import os
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


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("测试 33: Log + Input Events")
    print("=" * 70)

    page = FirefoxPage()
    results: List[Dict[str, str]] = []
    test_file = os.path.join(os.path.dirname(__file__), "test_file_dialog.html")

    try:
        # 1. log.entryAdded
        page.get("https://www.example.com")
        add_result(results, "页面加载", "成功", "example.com 已加载")

        page.console.start()
        page.console.clear()

        page.run_js("console.log('This is a log message')", as_expr=False)
        page.run_js("console.info('This is an info message')", as_expr=False)
        page.run_js("console.warn('This is a warning message')", as_expr=False)
        page.run_js("console.error('This is an error message')", as_expr=False)
        page.run_js("console.debug('This is a debug message')", as_expr=False)
        page.run_js(
            "console.table([{name: 'Alice', age: 18}, {name: 'Bob', age: 20}])",
            as_expr=False,
        )

        log_entry = page.console.wait(timeout=3)
        entries = page.console.entries
        error_entry = next((e for e in entries if e.level == "error"), None)
        warn_entry = next((e for e in entries if e.level == "warn"), None)
        table_entry = next(
            (e for e in entries if "Alice" in (e.text or "") or "Bob" in (e.text or "")),
            None,
        )

        if log_entry:
            add_result(
                results,
                "log.entryAdded first",
                "成功",
                f"level={log_entry.level} text={log_entry.text}",
            )
        else:
            add_result(results, "log.entryAdded first", "失败", "未观察到首条日志事件")

        if error_entry:
            add_result(results, "log.entryAdded error", "成功", error_entry.text or "")
        else:
            add_result(
                results, "log.entryAdded error", "失败", "未观察到 error 日志事件"
            )

        if warn_entry:
            add_result(results, "log.entryAdded warn", "成功", warn_entry.text or "")
        else:
            add_result(results, "log.entryAdded warn", "失败", "未观察到 warn 日志事件")

        if table_entry:
            add_result(results, "log.entryAdded table", "成功", (table_entry.text or "")[:120])
        else:
            add_result(results, "log.entryAdded table", "跳过", "当前 console.table 输出未稳定映射到 text")

        add_result(
            results,
            "log.entryAdded total",
            "成功" if len(entries) >= 6 else "失败",
            f"日志数量: {len(entries)}",
        )
        page.console.stop()


        # 2. input.fileDialogOpened
        html_content = """
        <!DOCTYPE html>
        <html>
        <head><meta charset='utf-8'><title>File Dialog Test</title></head>
        <body>
            <input type='file' id='single-file' />
            <input type='file' id='multiple-files' multiple />
            <button id='trigger-single' onclick="document.getElementById('single-file').click()">Open Single</button>
            <button id='trigger-multiple' onclick="document.getElementById('multiple-files').click()">Open Multiple</button>
        </body>
        </html>
        """
        with open(test_file, "w", encoding="utf-8") as f:
            f.write(html_content)

        page.get(f"file:///{test_file.replace(os.sep, '/')}")
        add_result(results, "文件测试页加载", "成功", "本地 file dialog 页面已加载")

        if page.events.start(["input.fileDialogOpened"], contexts=[page.tab_id]):
            page.events.clear()
            page.ele("#trigger-single").click(by_js=True)
            single_event = page.events.wait("input.fileDialogOpened", timeout=3)
            if single_event and single_event.multiple is False:
                add_result(
                    results,
                    "input.fileDialogOpened single",
                    "成功",
                    f"multiple={single_event.multiple}",
                )
            else:
                add_result(
                    results,
                    "input.fileDialogOpened single",
                    "跳过",
                    "当前环境未稳定观察到单文件对话框事件",
                )

            page.events.clear()
            page.ele("#trigger-multiple").click(by_js=True)
            multi_event = page.events.wait("input.fileDialogOpened", timeout=3)
            if multi_event and multi_event.multiple is True:
                add_result(
                    results,
                    "input.fileDialogOpened multiple",
                    "成功",
                    f"multiple={multi_event.multiple}",
                )
            else:
                add_result(
                    results,
                    "input.fileDialogOpened multiple",
                    "跳过",
                    "当前环境未稳定观察到多文件对话框事件",
                )
        else:
            add_result(
                results,
                "input.fileDialogOpened",
                "不支持",
                "未能订阅 input.fileDialogOpened 事件",
            )

        print_results(results)

    except Exception as e:
        add_result(results, "示例执行", "失败", str(e)[:120])
        print_results(results)
        raise
    finally:
        try:
            page.console.stop()
        except Exception:
            pass
        try:
            page.events.stop()
        except Exception:
            pass
        try:
            if os.path.exists(test_file):
                os.remove(test_file)
        except Exception:
            pass
        try:
            page.quit()
        except Exception:
            pass


if __name__ == "__main__":
    main()

剩余命令覆盖:截图、历史、继续响应、mock、fail、认证、动作释放

examples/34_remaining_commands.py
GitHub 仓库 已同步源码 查看 GitHub 源码
strict strict
一句话定位

把一些散落但重要的能力集中演示,适合做协议能力查漏补缺。

涉及 API
page.screenshotpage.backpage.forwardpage.intercept.start_responsesreq.continue_responsereq.mockreq.failreq.continue_with_authpage.actions.release
展示的不同用法
  • 截图与保存文件
  • 验证历史遍历
  • 继续响应和 mock 响应
  • 请求失败与认证补充
  • 释放动作链状态
适用场景

适合 strict 与 testing 章节。

相关文档页面
推荐阅读
network-events-31advanced-network-18
完整代码

来源:app/content/ruyipage_examples/34_remaining_commands.py · 仓库文件:34_remaining_commands.py

python 注释版
#!/usr/bin/env python
# =============================
# RuyiPage 示例注释版
# 标题: 剩余命令覆盖:截图、历史、继续响应、mock、fail、认证、动作释放
# 说明: 把一些散落但重要的能力集中演示,适合做协议能力查漏补缺。
# 你会学到:
#   1. 截图与保存文件
#   2. 验证历史遍历
#   3. 继续响应和 mock 响应
#   4. 请求失败与认证补充
#   5. 释放动作链状态
# 相关 API: page.screenshot, page.back, page.forward, page.intercept.start_responses, req.continue_response, req.mock, req.fail, req.continue_with_auth
# =============================

# -*- coding: utf-8 -*-
"""示例34: Remaining Commands(严格结果版)"""

import io
import os
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("测试 34: Remaining Commands")
    print("=" * 70)

    page = FirefoxPage()
    server = None
    results: List[Dict[str, str]] = []
    screenshot_path = os.path.join(os.path.dirname(__file__), "test_screenshot.png")

    try:
        server = TestServer(port=find_free_port(9430, 9530)).start()
        base_url = server.get_url("")[:-1]

        # 1. captureScreenshot
        page.get(base_url + "/nav/basic")
        add_result(results, "页面加载", "成功", f"服务地址: {base_url}")

        data_bytes = page.screenshot(as_bytes=True)
        if data_bytes and len(data_bytes) > 100:
            add_result(
                results,
                "browsingContext.captureScreenshot viewport",
                "成功",
                f"截图字节数: {len(data_bytes)}",
            )
        else:
            add_result(
                results,
                "browsingContext.captureScreenshot viewport",
                "失败",
                "截图数据过短",
            )

        saved_path = page.screenshot(path=screenshot_path)
        if saved_path and os.path.exists(saved_path):
            add_result(results, "screenshot save file", "成功", saved_path)
        else:
            add_result(results, "screenshot save file", "失败", "未保存截图文件")

        # 2. traverseHistory
        page.get(base_url + "/nav/basic?a=1")
        page.get(base_url + "/nav/basic?a=2")
        page.get(base_url + "/nav/basic?a=3")
        page.back()
        back_url = page.url
        page.back()
        back2_url = page.url
        page.forward()
        forward_url = page.url

        if (
            back_url.endswith("?a=2")
            and back2_url.endswith("?a=1")
            and forward_url.endswith("?a=2")
        ):
            add_result(
                results,
                "browsingContext.traverseHistory",
                "成功",
                f"back={back_url}, back2={back2_url}, forward={forward_url}",
            )
        else:
            add_result(
                results,
                "browsingContext.traverseHistory",
                "失败",
                f"back={back_url}, back2={back2_url}, forward={forward_url}",
            )

        # 3. network.continueResponse
        page.intercept.start_responses(
            url_patterns=[{"type": "string", "pattern": base_url + "/api/data"}]
        )
        page.run_js(
            f"fetch('{base_url}/api/data').catch(()=>null); return true;", as_expr=False
        )
        resp_req: InterceptedRequest | None = page.intercept.wait(timeout=5)
        if resp_req:
            resp_req.continue_response()
            add_result(results, "network.continueResponse", "成功", resp_req.url)
        else:
            add_result(
                results,
                "network.continueResponse",
                "跳过",
                "未稳定捕获到 responseStarted 拦截",
            )
        page.intercept.stop()

        # 4. network.provideResponse
        page.intercept.start_requests(
            url_patterns=[{"type": "string", "pattern": base_url + "/api/mock-source"}]
        )
        page.run_js(
            f"fetch('{base_url}/api/mock-source').catch(()=>null); return true;",
            as_expr=False,
        )
        mock_req: InterceptedRequest | None = page.intercept.wait(timeout=5)
        if mock_req:
            mock_req.mock(body='{"status":"mocked"}', status_code=200)
            add_result(results, "network.provideResponse", "成功", mock_req.url)
        else:
            add_result(
                results,
                "network.provideResponse",
                "跳过",
                "未稳定捕获到 beforeRequestSent 拦截",
            )
        page.intercept.stop()

        # 5. network.failRequest
        page.intercept.start_requests(
            url_patterns=[{"type": "string", "pattern": base_url + "/api/slow"}]
        )
        page.run_js(
            f"fetch('{base_url}/api/slow').catch(()=>null); return true;", as_expr=False
        )
        fail_req: InterceptedRequest | None = page.intercept.wait(timeout=5)
        if fail_req:
            fail_req.fail()
            add_result(results, "network.failRequest", "成功", fail_req.url)
        else:
            add_result(
                results,
                "network.failRequest",
                "跳过",
                "未稳定捕获到 beforeRequestSent 拦截",
            )
        page.intercept.stop()

        # 6. network.continueWithAuth
        page.events.start(["network.authRequired"], contexts=[page.tab_id])
        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_req and auth_event:
            auth_req.continue_with_auth(
                action="provideCredentials",
                username="user",
                password="pass",
            )
            add_result(results, "network.continueWithAuth", "成功", auth_req.url)
        else:
            add_result(
                results,
                "network.continueWithAuth",
                "跳过",
                "当前环境未稳定观察到 authRequired",
            )
        page.intercept.stop()
        page.events.stop()

        # 7. input.performActions / releaseActions
        try:
            page.quit()
        except Exception:
            pass
        page = FirefoxPage()

        action_page = """
        <!DOCTYPE html>
        <html>
        <head><meta charset='utf-8'><title>Actions Test</title></head>
        <body>
            <input id='input-box' autofocus />
            <button id='btn' onclick="window.btnClicked = true">Click Me</button>
            <script>
                window.btnClicked = false;
                window.lastKey = '';
                document.getElementById('input-box').addEventListener('keydown', function(e) {
                    window.lastKey = e.key;
                });
            </script>
        </body>
        </html>
        """
        page.get("data:text/html;charset=utf-8," + action_page)
        page.actions.move_to(page.ele("#input-box")).click().press("a").perform()
        last_key = page.run_js("return window.lastKey")
        page.actions.move_to(page.ele("#btn")).click().perform()
        btn_clicked = page.run_js("return window.btnClicked")
        page.actions.release()

        if last_key == "a":
            add_result(
                results, "input.performActions keyboard", "成功", f"lastKey={last_key}"
            )
        else:
            add_result(
                results, "input.performActions keyboard", "失败", f"lastKey={last_key}"
            )

        if btn_clicked is True:
            add_result(
                results, "input.performActions pointer", "成功", "按钮点击事件已触发"
            )
        else:
            add_result(
                results, "input.performActions pointer", "失败", "按钮点击事件未触发"
            )

        add_result(results, "input.releaseActions", "成功", "动作状态已释放")

        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 os.path.exists(screenshot_path):
                os.remove(screenshot_path)
        except Exception:
            pass
        try:
            if server is not None:
                server.stop()
        except Exception:
            pass
        try:
            page.quit()
        except Exception:
            pass


if __name__ == "__main__":
    main()

原生 BiDi 拖拽:验证 HTML5 drop 与 trusted 鼠标事件

examples/35_native_bidi_drag.py
GitHub 仓库 已同步源码 查看 GitHub 源码
actions strict
一句话定位

专注于高层原生 drag 能否正确触发真实 drop 事件。

涉及 API
page.actions.dragpage.actions.releasepage.run_jspage.ele
展示的不同用法
  • 只用高层 API 实现拖拽
  • 验证 drop 是否命中目标
  • 统计 trusted move 次数和鼠标事件状态
适用场景

适合 drag-drop 与 is-trusted 页面。

相关文档页面
推荐阅读
advanced-input-20actions-chain-05
完整代码

来源:app/content/ruyipage_examples/35_native_bidi_drag.py · 仓库文件:35_native_bidi_drag.py

python 注释版
# =============================
# RuyiPage 示例注释版
# 标题: 原生 BiDi 拖拽:验证 HTML5 drop 与 trusted 鼠标事件
# 说明: 专注于高层原生 drag 能否正确触发真实 drop 事件。
# 你会学到:
#   1. 只用高层 API 实现拖拽
#   2. 验证 drop 是否命中目标
#   3. 统计 trusted move 次数和鼠标事件状态
# 相关 API: page.actions.drag, page.actions.release, page.run_js, page.ele
# =============================

# -*- coding: utf-8 -*-
"""示例35: 原生 BiDi 拖拽(严格结果版)"""

import io
import os
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 FirefoxOptions, FirefoxPage


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("测试 35: 原生 BiDi 拖拽")
    print("=" * 70)

    opts = FirefoxOptions()
    opts.headless(False)
    page = FirefoxPage(opts)
    results: List[Dict[str, str]] = []

    try:
        test_page = os.path.join(
            os.path.dirname(__file__), "test_pages", "native_bidi_drag_test.html"
        )
        test_url = "file:///" + os.path.abspath(test_page).replace("\\", "/")
        page.get(test_url)
        page.wait(0.8)
        add_result(results, "测试页加载", "成功", test_url)

        source = page.ele("#drag-source")
        target = page.ele("#drop-target")
        if not source or not target:
            add_result(results, "拖拽元素定位", "失败", "未找到 source 或 target 元素")
            print_results(results)
            return

        # 只使用高层原生 BiDi 拖拽能力,不直接暴露 _bidi.input_。
        page.actions.drag(source, target, duration=640, steps=16).perform()
        page.actions.release()
        page.wait(0.8)

        state = page.run_js("return window.nativeBidiDragState") or {}
        result_text = page.ele("#result").text

        if result_text == "拖拽成功":
            add_result(results, "拖拽结果文本", "成功", result_text)
        else:
            add_result(results, "拖拽结果文本", "失败", result_text or "")

        if state.get("dropped") is True:
            add_result(results, "HTML5 drop 命中", "成功", "dropped=True")
        else:
            add_result(results, "HTML5 drop 命中", "失败", f"state={state}")

        if state.get("enteredTarget") is True:
            add_result(results, "目标区域进入", "成功", "enteredTarget=True")
        else:
            add_result(results, "目标区域进入", "失败", f"state={state}")

        if (
            state.get("trustedMouseDown") is True
            and state.get("trustedMouseUp") is True
        ):
            add_result(
                results,
                "isTrusted mouse down/up",
                "成功",
                "mouseDown/mouseUp 均为 trusted",
            )
        else:
            add_result(
                results,
                "isTrusted mouse down/up",
                "失败",
                f"down={state.get('trustedMouseDown')} up={state.get('trustedMouseUp')}",
            )

        move_count = int(state.get("trustedMoveCount") or 0)
        if move_count > 0:
            add_result(
                results,
                "isTrusted move count",
                "成功",
                f"trustedMoveCount={move_count}",
            )
        else:
            add_result(
                results,
                "isTrusted move count",
                "失败",
                f"trustedMoveCount={move_count}",
            )

        last_x = state.get("lastClientX")
        last_y = state.get("lastClientY")
        if last_x is not None and last_y is not None:
            add_result(results, "最后指针坐标", "成功", f"({last_x}, {last_y})")
        else:
            add_result(results, "最后指针坐标", "跳过", "页面未记录最后指针坐标")

        print_results(results)

    except Exception as e:
        add_result(results, "示例执行", "失败", str(e)[:160])
        print_results(results)
        raise
    finally:
        try:
            page.actions.release()
        except Exception:
            pass
        try:
            page.quit()
        except Exception:
            pass


if __name__ == "__main__":
    main()

原生 Select 选择:只测 native_only 模式与事件诊断

examples/36_native_bidi_select.py
GitHub 仓库 已同步源码 查看 GitHub 源码
elements strict
一句话定位

专门帮助排查 select 的焦点、键盘与 trusted change 路径问题。

涉及 API
select.by_valueselect.selected_optionclick_selfpage.actions.key_downpage.actions.key_uppage.run_js
展示的不同用法
  • 检查 select 初始状态
  • 聚焦后通过键盘切换选项
  • 强制只走 native_only 模式
  • 读取 changeCount 和 trustedChange 诊断信息
适用场景

适合 forms、actions 和 strict 示例讲解。

相关文档页面
推荐阅读
elements-03advanced-input-20
完整代码

来源:app/content/ruyipage_examples/36_native_bidi_select.py · 仓库文件:36_native_bidi_select.py

python 注释版
# =============================
# RuyiPage 示例注释版
# 标题: 原生 Select 选择:只测 native_only 模式与事件诊断
# 说明: 专门帮助排查 select 的焦点、键盘与 trusted change 路径问题。
# 你会学到:
#   1. 检查 select 初始状态
#   2. 聚焦后通过键盘切换选项
#   3. 强制只走 native_only 模式
#   4. 读取 changeCount 和 trustedChange 诊断信息
# 相关 API: select.by_value, select.selected_option, click_self, page.actions.key_down, page.actions.key_up, page.run_js
# =============================

# -*- coding: utf-8 -*-
"""
示例36: 原生 BiDi Select 检测(仅原生)

目标:
1) 只测试 native_only(不允许 JS 保底)
2) 输出分步诊断信息,帮助定位焦点/键盘路径问题
"""

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__), ".."))

from ruyipage import FirefoxOptions, FirefoxPage, Keys


def _state(page, select):
    s = page.run_js("return window.nativeBidiSelectState")
    return {
        "value": select.value,
        "selected_option": select.select.selected_option,
        "focused": s.get("focused"),
        "trusted_change": s.get("trustedChange"),
        "change_count": s.get("changeCount"),
        "events": s.get("events", []),
    }


def test_native_bidi_select():
    print("=" * 60)
    print("测试36: 原生 BiDi Select(仅原生)")
    print("=" * 60)

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

    try:
        test_page = os.path.join(
            os.path.dirname(__file__), "test_pages", "native_bidi_select_test.html"
        )
        test_url = "file:///" + os.path.abspath(test_page).replace("\\", "/")
        page.get(test_url)
        page.wait(1)

        select = page.ele("#single-select")

        print("\n1. 初始状态:")
        st = _state(page, select)
        print(f"   value: {st['value']}")
        print(f"   selected_option: {st['selected_option']}")
        print(f"   focused: {st['focused']}")

        print("\n2. 手动焦点探测(原生 click + 一步键盘):")
        select.click_self()
        page.wait(0.1)
        page.actions.key_down(Keys.DOWN).key_up(Keys.DOWN).perform()
        page.wait(0.1)
        st_probe = _state(page, select)
        print(f"   probe value: {st_probe['value']}")
        print(f"   probe focused: {st_probe['focused']}")
        print(f"   probe changeCount: {st_probe['change_count']}")

        print("\n3. 使用 selector native_only 选择 opt2:")
        ok = select.select.by_value("opt2", mode="native_only")
        page.wait(0.3)
        st = _state(page, select)
        print(f"   native_only结果: {ok}")
        print(f"   value: {st['value']}")
        print(f"   selected_option: {st['selected_option']}")
        print(f"   focused: {st['focused']}")
        print(f"   trustedChange: {st['trusted_change']}")
        print(f"   changeCount: {st['change_count']}")

        print("\n4. 页面事件日志:")
        for line in st["events"]:
            print(f"   {line}")

        if not ok or select.value != "opt2":
            raise RuntimeError("native_only 未成功切换到 opt2")

        print("\n" + "=" * 60)
        print("✓ 原生 BiDi Select 测试通过!")
        print("=" * 60)

    except Exception as e:
        print(f"\n✗ 测试失败: {e}")
        import traceback

        traceback.print_exc()
    finally:
        page.wait(1)
        page.quit()


if __name__ == "__main__":
    test_native_bidi_select()

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

examples/40_1_request_header_capture.py
GitHub 仓库 待同步 查看 GitHub 源码
network strict
一句话定位

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

涉及 API
FirefoxPagepage.intercept.startInterceptedRequestreq.headersreq.bodyreq.continue_request
展示的不同用法
  • 抓取真实页面发出的 POST 请求头
  • 预览请求体内容用于逆向分析
  • 快速判断页面 URL 与接口 URL 的关系
  • 在不打断真实请求的前提下继续放行请求
适用场景

适合 network、intercept、data-collector 与 scraper 调试场景。

推荐阅读
advanced-network-18collector-28packet-capture-40

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

examples/40_scraper_packet_capture.py
GitHub 仓库 待同步 查看 GitHub 源码
network strict
一句话定位

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

涉及 API
page.intercept.startpage.events.startpage.events.waitpage.network.add_data_collectorreq.continue_request
展示的不同用法
  • 观察真实页面请求链路
  • 定位数据接口与请求特征
  • 为后续响应体采集或 mock 做前置分析
  • 形成 scraper 场景下的网络调试起点
适用场景

适合 network、network-events、data-collector 和 scraper 文档链路。

推荐阅读
request-header-capture-40-1collector-28network-events-31

三标签页三隔离上下文:单浏览器多账号会话隔离

examples/37_three_isolated_user_context_tabs.py
GitHub 仓库 待同步 查看 GitHub 源码
browser strict
一句话定位

展示单浏览器内三个标签页分别使用不同 user context,实现 Cookie 与存储完全隔离。

涉及 API
page.browser_tools.create_user_contextpage.browser_tools.create_tabpage.browser_tools.remove_user_contextpage.contexts.close
展示的不同用法
  • 创建多个独立 user context
  • 在指定上下文中打开新标签页
  • 验证 Cookie 与存储隔离
  • 关闭标签页并清理上下文
适用场景

适合 user-contexts、browser 和 storage 页面。

相关文档页面
推荐阅读
browser-user-context-25tabs-09

动作可视化:显示鼠标轨迹、命中点与目标元素覆盖层

examples/42_2_action_visual_showcase.py
GitHub 仓库 待同步 查看 GitHub 源码
actions strict
一句话定位

用于排查自动化脚本里“鼠标到底移到了哪里、点击是否命中、JS click 与原生动作有什么区别”的可视化诊断示例。

涉及 API
launch(action_visual=True)page.actions.move_topage.actions.clickpage.actions.human_movepage.actions.drag_toele.click.by_jsele.input(..., by_js=True)
展示的不同用法
  • 显示 BiDi 鼠标移动轨迹
  • 显示点击位置高亮与闪烁提示
  • 显示当前目标元素覆盖层
  • 对比原生动作和 JS click / JS input 的差异
  • 排查拖拽、点击、输入命中问题
适用场景

适合 actions、is-trusted、debug 和新手排障链路。

推荐阅读
actions-chain-05advanced-input-20js-untrusted-45

JS setter 与事件派发:对照 isTrusted 与 untrusted 行为

examples/45_js_setter_untrusted_input.py
GitHub 仓库 待同步 查看 GitHub 源码
debug strict
一句话定位

通过普通 JS 事件与带 ruyi=true 的扩展事件做对照,帮助你理解为什么前端 dispatchEvent 默认不是 trusted 行为。

涉及 API
page.run_jspage.getpage.wait.doc_loadednew Event(...)new InputEvent(...)new KeyboardEvent(...)new MouseEvent(...)isTrusted
展示的不同用法
  • 对比普通 JS 事件和 ruyi=true 扩展事件的 isTrusted 差异
  • 验证 input、keydown、click、focus、pointer、wheel 等事件类型
  • 理解为什么 dispatchEvent 不能等价替代原生动作链
  • 为高风控场景排查前端事件信任问题
适用场景

适合 javascript、is-trusted、actions 和高风控调试章节。

推荐阅读
actions-chain-05advanced-input-20native-drag-35