RuyiPage 测试工具

# =============================
# RuyiPage 示例注释版
# 标题: 示例测试服务器：为 network、downloads、cookies、navigation 提供本地接口
# 说明: 大量示例依赖的本地稳定环境，是测试型章节的重要配套工具。
# 你会学到:
#   1. 为网络拦截、下载、Cookie 等示例提供稳定服务端
#   2. 避免外网波动影响案例复现
# 相关 API: TestServer.start, TestServer.stop, TestServer.get_url
# =============================

# -*- coding: utf-8 -*-
"""示例用本地测试服务器。"""

import base64
import json
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse


class TestServer(object):
    """用于 examples 的轻量 HTTP 测试服务器。"""

    def __init__(self, host="127.0.0.1", port=8888):
        self.host = host
        self.port = port
        self._server = None
        self._thread = None

    def start(self):
        if self._server:
            return self

        server = self

        class Handler(BaseHTTPRequestHandler):
            def log_message(self, format, *args):
                return

            def _cors_headers(self):
                return [
                    ("Access-Control-Allow-Origin", "*"),
                    ("Access-Control-Allow-Methods", "GET, POST, OPTIONS"),
                    (
                        "Access-Control-Allow-Headers",
                        "Content-Type, X-Ruyi-Demo, User-Agent",
                    ),
                ]

            def _write_text(self, status, body, headers=None):
                body_bytes = body.encode("utf-8")
                self.send_response(status)
                self.send_header("Content-Type", "text/plain; charset=utf-8")
                self.send_header("Content-Length", str(len(body_bytes)))
                if headers:
                    for name, value in headers:
                        self.send_header(name, value)
                self.end_headers()
                self.wfile.write(body_bytes)

            def _write_json(self, status, payload, headers=None):
                body = json.dumps(payload, ensure_ascii=False)
                body_bytes = body.encode("utf-8")
                self.send_response(status)
                self.send_header("Content-Type", "application/json; charset=utf-8")
                self.send_header("Content-Length", str(len(body_bytes)))
                if headers:
                    for name, value in headers:
                        self.send_header(name, value)
                self.end_headers()
                self.wfile.write(body_bytes)

            def do_GET(self):
                parsed = urlparse(self.path)
                path = parsed.path

                if path == "/set-cookie":
                    self._write_text(
                        200,
                        "cookie set ok",
                        headers=[
                            ("Set-Cookie", "server_cookie=server_value; Path=/"),
                            ("Set-Cookie", "session_id=abc123; Path=/; HttpOnly"),
                        ],
                    )
                    return

                if path == "/get-cookie":
                    cookie_header = self.headers.get("Cookie", "")
                    self._write_text(200, cookie_header or "")
                    return

                if path == "/api/data":
                    self._write_json(
                        200,
                        {
                            "status": "ok",
                            "data": {"message": "来自测试服务器的数据"},
                        },
                        headers=self._cors_headers(),
                    )
                    return

                if path == "/api/headers":
                    headers = {key: value for key, value in self.headers.items()}
                    self._write_json(200, headers, headers=self._cors_headers())
                    return

                if path == "/api/collector":
                    headers = {key: value for key, value in self.headers.items()}
                    self._write_json(
                        200,
                        {
                            "status": "ok",
                            "source": "collector",
                            "message": "stable response body for data collector",
                            "headers": headers,
                        },
                        headers=self._cors_headers(),
                    )
                    return

                if path == "/api/error":
                    self._write_json(
                        500,
                        {"status": "error", "message": "server error"},
                        headers=self._cors_headers(),
                    )
                    return

                if path == "/api/mock-source":
                    self._write_json(
                        200,
                        {"status": "ok", "source": "real-server"},
                        headers=self._cors_headers(),
                    )
                    return

                if path == "/api/slow":
                    self._write_json(
                        200,
                        {"status": "ok", "source": "slow-server"},
                        headers=self._cors_headers(),
                    )
                    return

                if path == "/api/auth":
                    auth_header = self.headers.get("Authorization", "")
                    expected = "Basic " + base64.b64encode(b"user:pass").decode("ascii")
                    if auth_header != expected:
                        self.send_response(401)
                        self.send_header("WWW-Authenticate", 'Basic realm="RuyiTest"')
                        self.send_header("Content-Length", "0")
                        self.end_headers()
                        return

                    self._write_json(
                        200,
                        {"status": "ok", "auth": True, "user": "user"},
                        headers=self._cors_headers(),
                    )
                    return

                if path == "/download/text":
                    self._write_text(
                        200,
                        "hello download",
                        headers=[
                            ("Content-Disposition", 'attachment; filename="test.txt"'),
                            ("Cache-Control", "no-store"),
                        ],
                    )
                    return

                if path == "/download/json":
                    self._write_json(
                        200,
                        {"ok": True},
                        headers=[
                            ("Content-Disposition", 'attachment; filename="test.json"'),
                            ("Cache-Control", "no-store"),
                        ],
                    )
                    return

                if path == "/nav/basic":
                    body = """<!DOCTYPE html>
<html><head><meta charset='utf-8'><title>Nav Basic</title></head>
<body><h1>Nav Basic</h1></body></html>"""
                    body_bytes = body.encode("utf-8")
                    self.send_response(200)
                    self.send_header("Content-Type", "text/html; charset=utf-8")
                    self.send_header("Content-Length", str(len(body_bytes)))
                    self.end_headers()
                    self.wfile.write(body_bytes)
                    return

                if path == "/nav/fragment":
                    body = """<!DOCTYPE html>
<html><head><meta charset='utf-8'><title>Nav Fragment</title></head>
<body>
<h1 id='a'>A</h1>
<div style='height:1200px'></div>
<h1 id='b'>B</h1>
</body></html>"""
                    body_bytes = body.encode("utf-8")
                    self.send_response(200)
                    self.send_header("Content-Type", "text/html; charset=utf-8")
                    self.send_header("Content-Length", str(len(body_bytes)))
                    self.end_headers()
                    self.wfile.write(body_bytes)
                    return

                if path == "/nav/history":
                    body = """<!DOCTYPE html>
<html><head><meta charset='utf-8'><title>Nav History</title></head>
<body><h1>Nav History</h1></body></html>"""
                    body_bytes = body.encode("utf-8")
                    self.send_response(200)
                    self.send_header("Content-Type", "text/html; charset=utf-8")
                    self.send_header("Content-Length", str(len(body_bytes)))
                    self.end_headers()
                    self.wfile.write(body_bytes)
                    return

                self._write_text(404, "not found")

            def do_OPTIONS(self):
                self.send_response(204)
                for name, value in self._cors_headers():
                    self.send_header(name, value)
                self.send_header("Content-Length", "0")
                self.end_headers()

        self._server = ThreadingHTTPServer((self.host, self.port), Handler)
        self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)
        self._thread.start()
        return self

    def stop(self):
        if not self._server:
            return
        self._server.shutdown()
        self._server.server_close()
        self._server = None
        self._thread = None

    def get_url(self, path="/"):
        if not path.startswith("/"):
            path = "/" + path
        return f"http://{self.host}:{self.port}{path}"

# -*- coding: utf-8 -*-
"""示例用本地测试服务器。"""

import base64
import json
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse


class TestServer(object):
    """用于 examples 的轻量 HTTP 测试服务器。"""

    def __init__(self, host="127.0.0.1", port=8888):
        self.host = host
        self.port = port
        self._server = None
        self._thread = None

    def start(self):
        if self._server:
            return self

        server = self

        class Handler(BaseHTTPRequestHandler):
            def log_message(self, format, *args):
                return

            def _cors_headers(self):
                return [
                    ("Access-Control-Allow-Origin", "*"),
                    ("Access-Control-Allow-Methods", "GET, POST, OPTIONS"),
                    (
                        "Access-Control-Allow-Headers",
                        "Content-Type, X-Ruyi-Demo, User-Agent",
                    ),
                ]

            def _write_text(self, status, body, headers=None):
                body_bytes = body.encode("utf-8")
                self.send_response(status)
                self.send_header("Content-Type", "text/plain; charset=utf-8")
                self.send_header("Content-Length", str(len(body_bytes)))
                if headers:
                    for name, value in headers:
                        self.send_header(name, value)
                self.end_headers()
                self.wfile.write(body_bytes)

            def _write_json(self, status, payload, headers=None):
                body = json.dumps(payload, ensure_ascii=False)
                body_bytes = body.encode("utf-8")
                self.send_response(status)
                self.send_header("Content-Type", "application/json; charset=utf-8")
                self.send_header("Content-Length", str(len(body_bytes)))
                if headers:
                    for name, value in headers:
                        self.send_header(name, value)
                self.end_headers()
                self.wfile.write(body_bytes)

            def do_GET(self):
                parsed = urlparse(self.path)
                path = parsed.path

                if path == "/set-cookie":
                    self._write_text(
                        200,
                        "cookie set ok",
                        headers=[
                            ("Set-Cookie", "server_cookie=server_value; Path=/"),
                            ("Set-Cookie", "session_id=abc123; Path=/; HttpOnly"),
                        ],
                    )
                    return

                if path == "/get-cookie":
                    cookie_header = self.headers.get("Cookie", "")
                    self._write_text(200, cookie_header or "")
                    return

                if path == "/api/data":
                    self._write_json(
                        200,
                        {
                            "status": "ok",
                            "data": {"message": "来自测试服务器的数据"},
                        },
                        headers=self._cors_headers(),
                    )
                    return

                if path == "/api/headers":
                    headers = {key: value for key, value in self.headers.items()}
                    self._write_json(200, headers, headers=self._cors_headers())
                    return

                if path == "/api/collector":
                    headers = {key: value for key, value in self.headers.items()}
                    self._write_json(
                        200,
                        {
                            "status": "ok",
                            "source": "collector",
                            "message": "stable response body for data collector",
                            "headers": headers,
                        },
                        headers=self._cors_headers(),
                    )
                    return

                if path == "/api/error":
                    self._write_json(
                        500,
                        {"status": "error", "message": "server error"},
                        headers=self._cors_headers(),
                    )
                    return

                if path == "/api/mock-source":
                    self._write_json(
                        200,
                        {"status": "ok", "source": "real-server"},
                        headers=self._cors_headers(),
                    )
                    return

                if path == "/api/slow":
                    self._write_json(
                        200,
                        {"status": "ok", "source": "slow-server"},
                        headers=self._cors_headers(),
                    )
                    return

                if path == "/api/auth":
                    auth_header = self.headers.get("Authorization", "")
                    expected = "Basic " + base64.b64encode(b"user:pass").decode("ascii")
                    if auth_header != expected:
                        self.send_response(401)
                        self.send_header("WWW-Authenticate", 'Basic realm="RuyiTest"')
                        self.send_header("Content-Length", "0")
                        self.end_headers()
                        return

                    self._write_json(
                        200,
                        {"status": "ok", "auth": True, "user": "user"},
                        headers=self._cors_headers(),
                    )
                    return

                if path == "/download/text":
                    self._write_text(
                        200,
                        "hello download",
                        headers=[
                            ("Content-Disposition", 'attachment; filename="test.txt"'),
                            ("Cache-Control", "no-store"),
                        ],
                    )
                    return

                if path == "/download/json":
                    self._write_json(
                        200,
                        {"ok": True},
                        headers=[
                            ("Content-Disposition", 'attachment; filename="test.json"'),
                            ("Cache-Control", "no-store"),
                        ],
                    )
                    return

                if path == "/nav/basic":
                    body = """<!DOCTYPE html>
<html><head><meta charset='utf-8'><title>Nav Basic</title></head>
<body><h1>Nav Basic</h1></body></html>"""
                    body_bytes = body.encode("utf-8")
                    self.send_response(200)
                    self.send_header("Content-Type", "text/html; charset=utf-8")
                    self.send_header("Content-Length", str(len(body_bytes)))
                    self.end_headers()
                    self.wfile.write(body_bytes)
                    return

                if path == "/nav/fragment":
                    body = """<!DOCTYPE html>
<html><head><meta charset='utf-8'><title>Nav Fragment</title></head>
<body>
<h1 id='a'>A</h1>
<div style='height:1200px'></div>
<h1 id='b'>B</h1>
</body></html>"""
                    body_bytes = body.encode("utf-8")
                    self.send_response(200)
                    self.send_header("Content-Type", "text/html; charset=utf-8")
                    self.send_header("Content-Length", str(len(body_bytes)))
                    self.end_headers()
                    self.wfile.write(body_bytes)
                    return

                if path == "/nav/history":
                    body = """<!DOCTYPE html>
<html><head><meta charset='utf-8'><title>Nav History</title></head>
<body><h1>Nav History</h1></body></html>"""
                    body_bytes = body.encode("utf-8")
                    self.send_response(200)
                    self.send_header("Content-Type", "text/html; charset=utf-8")
                    self.send_header("Content-Length", str(len(body_bytes)))
                    self.end_headers()
                    self.wfile.write(body_bytes)
                    return

                self._write_text(404, "not found")

            def do_OPTIONS(self):
                self.send_response(204)
                for name, value in self._cors_headers():
                    self.send_header(name, value)
                self.send_header("Content-Length", "0")
                self.end_headers()

        self._server = ThreadingHTTPServer((self.host, self.port), Handler)
        self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)
        self._thread.start()
        return self

    def stop(self):
        if not self._server:
            return
        self._server.shutdown()
        self._server.server_close()
        self._server = None
        self._thread = None

    def get_url(self, path="/"):
        if not path.startswith("/"):
            path = "/" + path
        return f"http://{self.host}:{self.port}{path}"

#!/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()

#!/usr/bin/env python
# -*- 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()

#!/usr/bin/env python
# =============================
# RuyiPage 示例注释版
# 标题: 提取 W3C WebDriver BiDi 规范 API 清单
# 说明: 适合放在 BiDi 对照页，强调项目对协议演进的关注。
# 你会学到:
#   1. 抓取 W3C BiDi 规范目录和 API 清单
#   2. 为对照表和文档生成提供原始数据
# 相关 API: FirefoxPage, page.get, page.run_js
# =============================

# -*- coding: utf-8 -*-
"""从W3C WebDriver BiDi规范网页提取所有API（使用正确的选择器）"""

import sys
import io

if sys.platform == "win32":
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")

from ruyipage import *
import os
import re

# 创建浏览器
page = FirefoxPage()

try:
    print("正在访问 W3C WebDriver BiDi 规范...")
    page.get("https://w3c.github.io/webdriver-bidi/")

    print("等待页面加载...")
    page.wait(3)

    print("\n从目录 (ol.toc li a) 提取所有API...")

    # 提取目录中的所有链接
    toc_script = """
    (() => {
        const items = [];
        document.querySelectorAll('ol.toc li a').forEach(link => {
            const href = link.getAttribute('href');
            const text = link.textContent.trim();
            if (href && text) {
                items.push({
                    href: href,
                    text: text,
                    id: href.replace('#', '')
                });
            }
        });
        return items;
    })()
    """

    toc_items = page.run_js(toc_script)

    print(f"\n找到 {len(toc_items)} 个目录项")

    # 分析提取命令和事件
    commands = {}
    events = {}
    modules = set()

    for item in toc_items:
        text = item["text"]
        id_str = item["id"]

        # 提取命令（格式：The module.command Command）
        cmd_match = re.search(r"The\s+(\w+)\.(\w+)\s+Command", text)
        if cmd_match:
            module = cmd_match.group(1)
            cmd_name = f"{module}.{cmd_match.group(2)}"
            modules.add(module)
            if module not in commands:
                commands[module] = []
            commands[module].append(cmd_name)
            print(f"  命令: {cmd_name}")

        # 提取事件（格式：The module.event Event）
        evt_match = re.search(r"The\s+(\w+)\.(\w+)\s+Event", text)
        if evt_match:
            module = evt_match.group(1)
            evt_name = f"{module}.{evt_match.group(2)}"
            modules.add(module)
            if module not in events:
                events[module] = []
            events[module].append(evt_name)
            print(f"  事件: {evt_name}")

    print(f"\n\n=== 统计结果 ===")
    print(f"模块数: {len(modules)}")
    print(f"命令数: {sum(len(v) for v in commands.values())}")
    print(f"事件数: {sum(len(v) for v in events.values())}")

    print("\n\n=== 按模块分类的命令 ===")
    for module in sorted(commands.keys()):
        print(f"\n{module} 模块 ({len(commands[module])} 个命令):")
        for cmd in sorted(commands[module]):
            print(f"  - {cmd}")

    print("\n\n=== 按模块分类的事件 ===")
    for module in sorted(events.keys()):
        print(f"\n{module} 模块 ({len(events[module])} 个事件):")
        for evt in sorted(events[module]):
            print(f"  - {evt}")

    # 保存结构化数据
    import json

    result = {
        "modules": sorted(list(modules)),
        "commands": commands,
        "events": events,
        "total_commands": sum(len(v) for v in commands.values()),
        "total_events": sum(len(v) for v in events.values()),
    }

    _out_path = os.path.join(os.path.dirname(__file__), "..", "w3c_bidi_apis.json")
    with open(_out_path, "w", encoding="utf-8") as f:
        json.dump(result, f, indent=2, ensure_ascii=False)

    print(f"\n\n结构化数据已保存到: {os.path.abspath(_out_path)}")

finally:
    page.quit()

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""从W3C WebDriver BiDi规范网页提取所有API（使用正确的选择器）"""

import sys
import io

if sys.platform == "win32":
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")

from ruyipage import *
import os
import re

# 创建浏览器
page = FirefoxPage()

try:
    print("正在访问 W3C WebDriver BiDi 规范...")
    page.get("https://w3c.github.io/webdriver-bidi/")

    print("等待页面加载...")
    page.wait(3)

    print("\n从目录 (ol.toc li a) 提取所有API...")

    # 提取目录中的所有链接
    toc_script = """
    (() => {
        const items = [];
        document.querySelectorAll('ol.toc li a').forEach(link => {
            const href = link.getAttribute('href');
            const text = link.textContent.trim();
            if (href && text) {
                items.push({
                    href: href,
                    text: text,
                    id: href.replace('#', '')
                });
            }
        });
        return items;
    })()
    """

    toc_items = page.run_js(toc_script)

    print(f"\n找到 {len(toc_items)} 个目录项")

    # 分析提取命令和事件
    commands = {}
    events = {}
    modules = set()

    for item in toc_items:
        text = item["text"]
        id_str = item["id"]

        # 提取命令（格式：The module.command Command）
        cmd_match = re.search(r"The\s+(\w+)\.(\w+)\s+Command", text)
        if cmd_match:
            module = cmd_match.group(1)
            cmd_name = f"{module}.{cmd_match.group(2)}"
            modules.add(module)
            if module not in commands:
                commands[module] = []
            commands[module].append(cmd_name)
            print(f"  命令: {cmd_name}")

        # 提取事件（格式：The module.event Event）
        evt_match = re.search(r"The\s+(\w+)\.(\w+)\s+Event", text)
        if evt_match:
            module = evt_match.group(1)
            evt_name = f"{module}.{evt_match.group(2)}"
            modules.add(module)
            if module not in events:
                events[module] = []
            events[module].append(evt_name)
            print(f"  事件: {evt_name}")

    print(f"\n\n=== 统计结果 ===")
    print(f"模块数: {len(modules)}")
    print(f"命令数: {sum(len(v) for v in commands.values())}")
    print(f"事件数: {sum(len(v) for v in events.values())}")

    print("\n\n=== 按模块分类的命令 ===")
    for module in sorted(commands.keys()):
        print(f"\n{module} 模块 ({len(commands[module])} 个命令):")
        for cmd in sorted(commands[module]):
            print(f"  - {cmd}")

    print("\n\n=== 按模块分类的事件 ===")
    for module in sorted(events.keys()):
        print(f"\n{module} 模块 ({len(events[module])} 个事件):")
        for evt in sorted(events[module]):
            print(f"  - {evt}")

    # 保存结构化数据
    import json

    result = {
        "modules": sorted(list(modules)),
        "commands": commands,
        "events": events,
        "total_commands": sum(len(v) for v in commands.values()),
        "total_events": sum(len(v) for v in events.values()),
    }

    _out_path = os.path.join(os.path.dirname(__file__), "..", "w3c_bidi_apis.json")
    with open(_out_path, "w", encoding="utf-8") as f:
        json.dump(result, f, indent=2, ensure_ascii=False)

    print(f"\n\n结构化数据已保存到: {os.path.abspath(_out_path)}")

finally:
    page.quit()