browser.py

# 修改历史 (Revision History)
# ==================================
# 版本: v1.16
# 日期: 2026-06-25
# 修改说明: 在 BrowserCopilot 初始化的代理检测中增加对全局环境变量 ALL_PROXY / all_proxy 的读取支持，确保在仅配置 ALL_PROXY 时，无头及有头浏览器能正常路由代理，解决后台 silent token refresh 超时与人机验证被阻。
#
# 版本: v1.15
# 日期: 2026-06-24
# 修改说明: 在 login 方法中提取 Token 时，针对 page.evaluate 增加 try-except 捕获，容错并重试处理页面因重定向/刷新导致的 Execution context destroyed 崩溃问题。
#
# 版本: v1.14
# 日期: 2026-06-24
# 修改说明: 将交互式登录凭证获取机制从自动轮询改成回车手动确认模式，允许用户调试完成后再进行一次性提取，规避页面加载时因已有缓存 Token 触发程序提前退出的缺陷。
#
# 版本: v1.13
# 日期: 2026-06-24
# 修改说明: 在 ALLOWED_COOKIE_KEYS 核心白名单中添加 cf_clearance、__cf_bm 和 __Host-copilot-anon，防止登录凭证截取时丢弃 Cloudflare Bot 校验与会话 Cookie 造成 WebSocket 握手报 invalid-event 错误。
#
# 版本: v1.12
# 日期: 2026-06-24
# 修改说明: 实现源头 Cookie 精细过滤机制，新增必要 Cookie 白名单 ALLOWED_COOKIE_KEYS，并在 login 与 cookies 中自动剥离 OParams 等超大 SSO 冗余临时 Cookie，彻底防止 WebSocket 升级 403 阻断。

# 版本: v1.11
# 日期: 2026-06-24
# 修改说明: 修复 v1.10 中遗漏 cookies_list = self._context.cookies() 变量定义导致的 NameError 致命缺陷，确保交互登录流程正常捕获并自动退出。
#
# 版本: v1.10
# 日期: 2026-06-24
# 修改说明: 拓宽 Cookie 过滤域名，支持捕获 live.com、bing.com 与 microsoftonline.com 相关的 MSA 账户认证 Cookie，解决接口由于关键凭证丢失沦为匿名会话导致多轮对话连接断连/invalid-event 的缺陷。
#
# 版本: v1.9
# 日期: 2026-06-24
# 修改说明: 在 login 方法启动调试 Chrome 前，新增调试端口 9527 占用检测与强杀进程树逻辑，彻底解决后台残留僵尸进程占用调试端口导致 Playwright CDP 控制失效、无法自动导航跳转的缺陷；同时剔除 _FIND_TOKEN_JS 中的 fallback 备选逻辑，只允许提取真正的 ChatAI 聊天 Token。
#
# 版本: v1.8
# 日期: 2026-06-24
# 修改说明: 将所有 print 输出中的 emoji 字符（如 🎉 和 ⚠️）替换为文本前缀（如 [Success] 和 [Warning]），解决在 Windows GBK 终端中可能触发 UnicodeEncodeError 崩溃的隐患。


"""Browser-backed Copilot driver.

A Playwright fallback for the pure-HTTP :class:`copilot.client.Copilot`: it runs
the *exact same protocol* inside a real browser that already holds Cloudflare
clearance and (optionally) a signed-in Microsoft session. Useful if Microsoft
ever escalates the challenge to a Cloudflare Turnstile CAPTCHA, which needs a
browser-solved token.

``BrowserCopilot`` launches a **persistent** Playwright Chromium profile so that
Cloudflare clearance and any sign-in survive restarts. The chat protocol
(``POST /c/api/conversations`` then a ``wss://.../c/api/chat`` WebSocket speaking
``send`` -> ``appendText``* -> ``done``) is executed *in the page* via
``page.evaluate`` so the browser's own ``fetch``/``WebSocket`` carry the cookies,
Cloudflare token, and auth headers.

It exposes the same ``create_completion(prompt, stream=...)`` generator API as
:class:`copilot.client.Copilot`, so it is a drop-in replacement.

PROTOCOL ASSUMPTIONS (verify at runtime against a live session):
  * Conversation create:  POST /c/api/conversations  -> {"id": "..."}
  * Chat socket:          wss://copilot.microsoft.com/c/api/chat?api-version=2
                          (with &accessToken=<token> when signed in)
  * Send frame:           {"event":"send","conversationId":...,
                           "content":[{"type":"text","text":...}],"mode":"chat"}
  * Stream frames:        {"event":"appendText","text":...}, then {"event":"done"}
These mirror the captured protocol in ``client.py``. If Microsoft changes them,
adjust the JS templates below.
"""

from __future__ import annotations

import json
import time
from pathlib import Path
from typing import Dict, Generator, Optional

from playwright.sync_api import sync_playwright, Error as PlaywrightError

from .auth import DEFAULT_AUTH_FILE, DEFAULT_PROFILE_DIR

COPILOT_URL = "https://copilot.microsoft.com/"

# 必要认证 Cookie 白名单，过滤掉超大且无用的 SSO 重定向临时 Cookie (如 OParams, ESTSWCTXFLOWTOKEN)
ALLOWED_COOKIE_KEYS = {
    "__Host-MSAAUTHP",
    "KievRPSSecAuth",
    "MUID",
    "MUIDB",
    "_C_Auth",
    "hasBeenMsalAuthenticated",
    "ANON",
    "NAP",
    "MSPPre",
    "MSPCID",
    "WLSSC",
    "brcap",
    "fpc",
    "PPLState",
    "cf_clearance",
    "__cf_bm",
    "__Host-copilot-anon"
}

# --- in-page JavaScript -----------------------------------------------------

# Create a conversation. Runs in the page so cookies/Cloudflare apply.
_CREATE_CONVERSATION_JS = """
async () => {
  const res = await fetch('/c/api/conversations', {
    method: 'POST',
    credentials: 'include',
    headers: {'content-type': 'application/json'},
  });
  const text = await res.text();
  if (!res.ok) return {ok: false, status: res.status, text: text};
  let data = {};
  try { data = JSON.parse(text); } catch (e) {}
  return {ok: true, id: data.id || data.conversationId || null, raw: text};
}
"""

# Discover the Copilot chat MSAL access token from localStorage. The cache holds
# several tokens for different scopes; the chat WebSocket only accepts the one
# scoped 'ChatAI.ReadWrite' — a wrong-audience token (e.g. the Graph
# User.Read/Files.Read token) makes the WS upgrade 401. We therefore PREFER the
# ChatAI token and only fall back to the first token found if none matches.
# Returns null for anonymous sessions (anonymous chat may still work via cookies).
_FIND_TOKEN_JS = """
() => {
  try {
    for (let i = 0; i < localStorage.length; i++) {
      const k = localStorage.key(i);
      const v = localStorage.getItem(k);
      if (v && v.indexOf('"credentialType":"AccessToken"') !== -1) {
        try {
          const o = JSON.parse(v);
          if (o && o.secret) {
            // Match the chat scope (e.g. '<resource>/chatai.readwrite', case-insensitive)
            if (o.target && o.target.toLowerCase().indexOf('chatai') !== -1) return o.secret;
          }
        } catch (e) {}
      }
    }
  } catch (e) {}
  return null;
}
"""

# Strict token finder: only returns a token if its MSAL scope target matches ChatAI.
_FIND_STRICT_CHAT_TOKEN_JS = """
() => {
  try {
    for (let i = 0; i < localStorage.length; i++) {
      const k = localStorage.key(i);
      const v = localStorage.getItem(k);
      if (v && v.indexOf('"credentialType":"AccessToken"') !== -1) {
        try {
          const o = JSON.parse(v);
          if (o && o.secret && o.target && o.target.toLowerCase().indexOf('chatai') !== -1) {
            return o.secret;
          }
        } catch (e) {}
      }
    }
  } catch (e) {}
  return null;
}
"""

# Open the chat WebSocket and wire handlers that push into a window-scoped
# buffer. Returns immediately; messages accumulate while Python polls.
_START_STREAM_JS = """
([conversationId, accessToken, prompt]) => {
  const state = {queue: [], done: false, error: null, started: false};
  window.__copilot = state;
  let url = 'wss://copilot.microsoft.com/c/api/chat?api-version=2';
  if (accessToken) url += '&accessToken=' + encodeURIComponent(accessToken);
  let ws;
  try { ws = new WebSocket(url); } catch (e) { state.error = 'ws-init: ' + e; state.done = true; return false; }
  window.__copilotWs = ws;
  ws.onopen = () => {
    ws.send(JSON.stringify({
      event: 'send',
      conversationId: conversationId,
      content: [{type: 'text', text: prompt}],
      mode: 'chat'
    }));
  };
  ws.onmessage = (ev) => {
    let msg;
    try { msg = JSON.parse(ev.data); } catch (e) { return; }
    const e = msg.event;
    if (e === 'appendText') { state.started = true; if (msg.text) state.queue.push(msg.text); }
    else if (e === 'done') { state.done = true; try { ws.close(); } catch (x) {} }
    else if (e === 'error') { state.error = JSON.stringify(msg); state.done = true; try { ws.close(); } catch (x) {} }
  };
  ws.onerror = () => { state.error = state.error || 'websocket error'; state.done = true; };
  ws.onclose = () => { state.done = true; };
  return true;
}
"""

# Drain the buffer and report status in one round-trip.
_POLL_JS = """
() => {
  const s = window.__copilot || {queue: [], done: true, error: 'not started', started: false};
  const q = s.queue;
  s.queue = [];
  return {q: q, done: s.done, error: s.error, started: s.started};
}
"""


class BrowserCopilot:
    """Drives Microsoft Copilot through a real Playwright browser.

    Parameters
    ----------
    profile_dir:
        Directory for the persistent Chromium profile (cookies, Cloudflare
        clearance, sign-in). Reused across runs.
    headless:
        Run without a visible window. Use ``False`` (or :meth:`login`) for the
        first interactive sign-in, then ``True`` afterwards.
    """

    label = "Microsoft Copilot (browser)"
    default_model = "Copilot"

    def __init__(
        self,
        profile_dir: str = DEFAULT_PROFILE_DIR,
        headless: bool = True,
        nav_timeout: int = 60,
        proxy: Optional[str] = None,
    ):
        self.profile_dir = str(Path(profile_dir).resolve())
        self.headless = headless
        self.nav_timeout = nav_timeout
        # Copilot consumer chat is geo-restricted. If you are outside a supported
        # region, route the browser through a proxy/VPN in a supported region,
        # e.g. proxy="http://user:pass@host:port" or "socks5://host:port".
        if proxy is None:
            import os
            proxy = (
                os.environ.get("HTTPS_PROXY")
                or os.environ.get("HTTP_PROXY")
                or os.environ.get("ALL_PROXY")
                or os.environ.get("https_proxy")
                or os.environ.get("http_proxy")
                or os.environ.get("all_proxy")
            )
        if proxy and "socks5h://" in proxy:
            proxy = proxy.replace("socks5h://", "socks5://")
        self.proxy = proxy

        self._pw = None
        self._context = None
        self._page = None

    # -- lifecycle ----------------------------------------------------------

    def start(self, headless: Optional[bool] = None) -> "BrowserCopilot":
        """Launch the persistent browser context and open Copilot."""
        if self._context is not None:
            return self
        if headless is not None:
            self.headless = headless
        try:
            self._pw = sync_playwright().start()
            launch_kwargs = dict(
                headless=self.headless,
                args=[
                    "--disable-blink-features=AutomationControlled",
                    "--excludeSwitches=enable-automation",
                ],
                user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
            )
            if self.proxy:
                launch_kwargs["proxy"] = self._parse_proxy(self.proxy)
            self._context = self._pw.chromium.launch_persistent_context(
                self.profile_dir,
                **launch_kwargs,
            )
            self._page = self._context.pages[0] if self._context.pages else self._context.new_page()
            self._page.set_default_timeout(self.nav_timeout * 1000)
            self._page.goto(COPILOT_URL, wait_until="domcontentloaded")
            # Give Cloudflare a moment to clear on first paint.
            try:
                self._page.wait_for_load_state("networkidle", timeout=5000)
            except PlaywrightError:
                pass
        except PlaywrightError as exc:
            self.close()
            raise ConnectionError(f"Failed to start browser: {exc}") from exc
        return self

    @staticmethod
    def _parse_proxy(proxy: str) -> dict:
        """Turn a ``scheme://user:pass@host:port`` string into Playwright form."""
        from urllib.parse import urlparse

        u = urlparse(proxy)
        server = f"{u.scheme}://{u.hostname}:{u.port}" if u.port else f"{u.scheme}://{u.hostname}"
        cfg = {"server": server}
        if u.username:
            cfg["username"] = u.username
        if u.password:
            cfg["password"] = u.password
        return cfg

    @staticmethod
    def _release_port(port: int) -> None:
        """Find and terminate any process listening on the target CDP port."""
        import subprocess
        try:
            output = subprocess.check_output(
                ["netstat", "-ano"],
                stderr=subprocess.DEVNULL,
                text=True
            )
            pids = set()
            for line in output.splitlines():
                if f":{port}" in line and "LISTENING" in line:
                    parts = line.strip().split()
                    if len(parts) >= 5:
                        pid = parts[-1]
                        if pid.isdigit() and int(pid) > 0:
                            pids.add(int(pid))
            
            for pid in pids:
                print(f"[Warning] 检测到调试端口 {port} 被 PID {pid} 占用，正在强制结束进程树以释放端口...")
                subprocess.run(
                    ["taskkill", "/F", "/T", "/PID", str(pid)],
                    stdout=subprocess.DEVNULL,
                    stderr=subprocess.DEVNULL
                )
        except Exception as e:
            print(f"[Warning] 释放调试端口 {port} 失败: {e}")

    def region_blocked(self) -> bool:
        """True if Copilot is showing the 'Not available in your region' notice."""
        if self._page is None:
            return False
        try:
            text = self._page.evaluate("() => document.body ? document.body.innerText : ''")
        except PlaywrightError:
            return False
        return "available in your region" in (text or "").lower()

    def close(self) -> None:
        for attr, closer in (("_context", lambda c: c.close()), ("_pw", lambda p: p.stop())):
            obj = getattr(self, attr, None)
            if obj is not None:
                try:
                    closer(obj)
                except Exception:
                    pass
                setattr(self, attr, None)
        self._page = None

    def __enter__(self) -> "BrowserCopilot":
        return self.start()

    def __exit__(self, *exc) -> None:
        self.close()

    # -- auth ---------------------------------------------------------------

    def _find_chrome_binary(self) -> Optional[str]:
        import os
        possible_paths = [
            r"C:\Program Files\Google\Chrome\Application\chrome.exe",
            r"C:\Program Files (x86)\Google\Chrome\Application\chrome.exe",
            r"C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe",
            r"C:\Program Files\Microsoft\Edge\Application\msedge.exe",
            os.path.expandvars(r"%LOCALAPPDATA%\Google\Chrome\Application\chrome.exe"),
            os.path.expandvars(r"%LOCALAPPDATA%\Microsoft\Edge\Application\msedge.exe"),
        ]
        for path in possible_paths:
            if os.path.exists(path):
                return path
        return None

    def login(self, path: str = DEFAULT_AUTH_FILE) -> dict:
        """Open a native browser window for interactive Microsoft sign-in.

        Launches the system Google Chrome / Edge with CDP (Chrome DevTools Protocol)
        debugging enabled. Captures the signed-in session dynamically without
        blocking the console on user input.
        """
        import subprocess
        import socket
        import os
        
        # Delete old token file to prevent stale cache reuse
        if os.path.exists(path):
            try:
                os.remove(path)
                print(f"已清理旧的本地凭证缓存文件: {path}")
            except Exception as e:
                print(f"清理旧凭证文件失败: {e}")

        self.close()
        binary = self._find_chrome_binary()
        if not binary:
            raise RuntimeError("Google Chrome or Microsoft Edge was not found on your system.")
            
        port = 9527
        self._release_port(port)
        user_data_dir = self.profile_dir
        
        args = [
            binary,
            f"--remote-debugging-port={port}",
            f"--user-data-dir={user_data_dir}",
            "--remote-allow-origins=*",
            "--no-first-run",
            "--no-default-browser-check",
        ]
        if self.proxy:
            args.append(f"--proxy-server={self.proxy}")
            
        print(f"\nLaunching native browser: {binary} with CDP port {port}...")
        chrome_process = subprocess.Popen(args, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        
        # Wait for CDP port to open
        for _ in range(30):
            try:
                with socket.create_connection(("127.0.0.1", port), timeout=0.5):
                    break
            except OSError:
                time.sleep(0.2)
        else:
            try:
                chrome_process.terminate()
            except Exception:
                pass
            raise RuntimeError("Failed to connect to browser CDP port.")
            
        # Connect over CDP
        # Temporarily clear environment proxies to prevent Playwright driver from redirecting localhost CDP connection
        proxies_cache = {}
        for key in ["HTTP_PROXY", "HTTPS_PROXY", "http_proxy", "https_proxy"]:
            if key in os.environ:
                proxies_cache[key] = os.environ[key]
                del os.environ[key]
        os.environ["NO_PROXY"] = "127.0.0.1,localhost"

        from playwright.sync_api import sync_playwright
        self._pw = sync_playwright().start()
        auth = {}
        try:
            print("Connecting to native browser context via CDP...")
            cdp_browser = self._pw.chromium.connect_over_cdp(f"http://127.0.0.1:{port}")
            self._context = cdp_browser.contexts[0] if cdp_browser.contexts else cdp_browser.new_context()
            self._page = self._context.pages[0] if self._context.pages else self._context.new_page()
            
            # Navigate to Copilot if not already there
            if "copilot.microsoft.com" not in self._page.url:
                print(f"正在导航至 {COPILOT_URL}...")
                try:
                    self._page.goto(COPILOT_URL, wait_until="domcontentloaded", timeout=60000)
                except Exception as e:
                    print(f"[Warning] 导航到 {COPILOT_URL} 发生超时或失败: {e}，将尝试直接进入捕获流程...")
            
            # Clear old localStorage tokens to force MSAL to request fresh ones
            try:
                self._page.evaluate("localStorage.clear()")
                self._page.reload(wait_until="domcontentloaded", timeout=60000)
                print("已清除浏览器 LocalStorage 中的历史 Token 缓存并刷新页面。")
            except Exception as e:
                print(f"[Warning] 清空浏览器 LocalStorage 或刷新页面超时/失败: {e}，将尝试直接进入捕获流程...")
            
            print("\n======================================================================")
            print("  [操作提示] 请在弹出的真实浏览器窗口中进行以下操作:")
            print("  1. 登录您的微软账户 (如果尚未登录)。")
            print("  2. 在聊天框中随意发送一条对话消息 (例如：'你好')，并等待回答完毕。")
            print("  3. 确认网页对话正常后，回到此控制台，按 [Enter] 回车键保存凭证并退出。")
            print("======================================================================\n")
            
            input("请在完成上述操作后，按 [Enter] 回车键以捕获并保存凭证...")
            
            print("正在检索并同步有效的会话凭证...")
            token = None
            for attempt in range(5):
                try:
                    token = self._page.evaluate(_FIND_STRICT_CHAT_TOKEN_JS)
                    if not token:
                        token = self._page.evaluate(_FIND_TOKEN_JS)
                    break
                except Exception as e:
                    if "destroyed" in str(e) or "navigation" in str(e):
                        print(f"[Warning] 页面正处于重定向或导航状态，等待 1 秒后进行第 {attempt+1}/5 次重试...")
                        time.sleep(1)
                        continue
                    raise
                
            cookies_list = self._context.cookies()
            cookies = {c["name"]: c["value"] for c in cookies_list if c["name"] in ALLOWED_COOKIE_KEYS and any(dom in c.get("domain", "") for dom in ["microsoft.com", "live.com", "bing.com", "microsoftonline.com"])}
            
            auth = {}
            if token and cookies:
                auth = {
                    "cookies": cookies,
                    "access_token": token,
                    "saved_at": time.time(),
                }
                dest = Path(path)
                dest.parent.mkdir(parents=True, exist_ok=True)
                dest.write_text(json.dumps(auth, indent=2), encoding="utf-8")
                print("[Success] 成功自动捕获并保存了有效的 Copilot 会话 Token 凭证!")
                print(f"Auth snapshot saved to {path}")
            else:
                if not token:
                    print("[Warning] 未能在浏览器中提取到有效的 ChatAI 聊天 Token。")
                if not cookies:
                    print("[Warning] 未能在浏览器中提取到任何核心的认证 Cookie。")
                
        finally:
            # Restore environment proxies
            for key, val in proxies_cache.items():
                os.environ[key] = val
            os.environ.pop("NO_PROXY", None)
            
            self.close()
            try:
                chrome_process.terminate()
                chrome_process.wait(timeout=2)
            except Exception:
                pass
                
        print(f"Session saved to {self.profile_dir}")
        return auth

    def access_token(self) -> Optional[str]:
        """Return the page's MSAL access token, or ``None`` if anonymous."""
        self._ensure_started()
        try:
            return self._page.evaluate(_FIND_TOKEN_JS)
        except PlaywrightError:
            return None

    def cookies(self) -> Dict[str, str]:
        """Return the signed-in Microsoft cookies as a name->value dict."""
        self._ensure_started()
        try:
            raw = self._context.cookies()
        except PlaywrightError:
            return {}
        return {c["name"]: c["value"] for c in raw if c["name"] in ALLOWED_COOKIE_KEYS and any(dom in c.get("domain", "") for dom in ["microsoft.com", "live.com", "bing.com", "microsoftonline.com"])}

    def export_auth(self, path: str = DEFAULT_AUTH_FILE, stamp: Optional[float] = None) -> dict:
        """Snapshot the signed-in cookies + access token to ``path`` as JSON.

        ``stamp`` is the epoch seconds to record as ``saved_at`` (pass
        ``time.time()`` from the caller). Returns the auth dict.
        """
        auth = {
            "cookies": self.cookies(),
            "access_token": self.access_token(),
            "saved_at": stamp if stamp is not None else 0,
        }
        dest = Path(path)
        dest.parent.mkdir(parents=True, exist_ok=True)
        dest.write_text(json.dumps(auth, indent=2), encoding="utf-8")
        return auth

    # -- chat ---------------------------------------------------------------

    def create_completion(
        self,
        prompt: str,
        stream: bool = False,
        timeout: int = 900,
        **kwargs,
    ) -> Generator[str, None, None]:
        """Stream a Copilot reply to ``prompt``. Mirrors ``Copilot.create_completion``.

        Yields text chunks as they arrive. ``stream`` is accepted for API
        compatibility; chunks are always produced incrementally.
        """
        self._ensure_started()

        if self.region_blocked():
            raise RuntimeError(
                "Microsoft Copilot is not available in your region. "
                "Route the browser through a proxy/VPN in a supported region, e.g.:\n"
                "    BrowserCopilot(proxy='http://user:pass@host:port')\n"
                "or 'socks5://host:port'. See README for details."
            )

        conv = self._page.evaluate(_CREATE_CONVERSATION_JS)
        if not conv.get("ok"):
            status = conv.get("status")
            body = (conv.get("text") or "")[:500]
            if status in (401, 403):
                raise RuntimeError(
                    f"Conversation create returned HTTP {status}. "
                    f"Run login() / `python -m copilot login` to sign in. Body: {body}"
                )
            raise RuntimeError(f"Conversation create failed (HTTP {status}): {body}")

        conversation_id = conv.get("id")
        if not conversation_id:
            raise RuntimeError(f"No conversation id in response: {conv.get('raw')!r}")

        token = self._page.evaluate(_FIND_TOKEN_JS)

        started_ok = self._page.evaluate(_START_STREAM_JS, [conversation_id, token, prompt])
        if started_ok is False:
            state = self._page.evaluate(_POLL_JS)
            raise ConnectionError(f"WebSocket failed to start: {state.get('error')}")

        yield from self._pump(timeout)

    # -- internals ----------------------------------------------------------

    def _pump(self, timeout: int) -> Generator[str, None, None]:
        deadline = time.time() + timeout
        any_text = False
        while True:
            state = self._page.evaluate(_POLL_JS)
            for chunk in state.get("q") or []:
                if chunk:
                    any_text = True
                    yield chunk
            if state.get("error"):
                raise RuntimeError(f"Copilot error: {state['error']}")
            if state.get("done") and not state.get("q"):
                break
            if time.time() > deadline:
                raise TimeoutError(f"No 'done' within {timeout}s")
            time.sleep(0.08)

        if not any_text and not state.get("started"):
            raise RuntimeError("Invalid response: stream produced no text")

    def _ensure_started(self) -> None:
        if self._context is None or self._page is None:
            self.start()