4b2abbbfc5
Security: - serve.py: server now sends nonce challenge before accepting any command; clients sign nonce + SHA256(canonical_payload) with Ed25519 key - New --authorized-keys FILE option for serve; token auth still works as fallback - Connection limit: BoundedSemaphore(64) in serve.py - Secure file creation with os.open(..., 0o600) for token/key files - New auth.py module: keygen, file key load/save, SSH agent protocol (pure Python), sign/verify helpers compatible with both file keys and agent-held keys (YubiKey, TPM, gpg-agent) Features: - YubiKey support via SSH agent protocol — no new runtime deps, just $SSH_AUTH_SOCK - New `browser-cli auth` command group: keygen, trust, show, keys - Global --key PATH flag (or BROWSER_CLI_KEY env) selects signing key; pass "agent" or "agent:<selector>" to use SSH agent key - BrowserCLI Python API gains key= parameter Bug fixes (11 issues across two review passes): - client.py: check response is not None before json.loads - native_host.py: _read_exact_stream loop handles EINTR short reads; fix Windows Listener leak on accept error - __init__.py: open_wait / tabs_watch_url raise RuntimeError instead of silent None - extension/tabs.ts: dedupe skips tabs without URL; tabsSort uses pendingUrl fallback - extension/session.ts: removeListener before addListener prevents duplicate handlers Breaking: TCP serve protocol now sends a challenge frame first (v0.9.0) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
716 lines
30 KiB
Python
716 lines
30 KiB
Python
"""
|
|
browser_cli — Python API for controlling your running browser.
|
|
|
|
Usage:
|
|
from browser_cli import BrowserCLI
|
|
b = BrowserCLI()
|
|
|
|
tabs = b.tabs_list() # list[Tab]
|
|
tabs[0].close()
|
|
tabs[0].move(forward=True)
|
|
|
|
groups = b.group_list() # list[Group]
|
|
groups[0].tabs()
|
|
groups[0].add_tab("https://example.com")
|
|
|
|
# When multiple browser instances are active, pass the alias:
|
|
b = BrowserCLI(browser="brave")
|
|
"""
|
|
from collections.abc import Callable, Iterable
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from browser_cli.client import BrowserNotConnected, active_browser_targets, remote_browser_targets, send_command
|
|
from browser_cli.models import Group, Tab
|
|
|
|
__all__ = ["BrowserCLI", "BrowserCounts", "BrowserNotConnected", "Tab", "Group"]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BrowserCounts:
|
|
"""Aggregated per-browser counts returned in implicit multi-browser mode."""
|
|
total: int
|
|
by_browser: dict[str, int]
|
|
|
|
|
|
class BrowserCLI:
|
|
def __init__(self, browser: str | None = None, remote: str | None = None, token: str | None = None, key: str | None = None):
|
|
"""
|
|
Args:
|
|
browser: Profile alias to target. Required when multiple browser
|
|
instances are active. Equivalent to ``--browser`` on the CLI.
|
|
remote: Connect to a remote browser exposed via ``browser-cli serve``.
|
|
Format: ``"host:port"`` (e.g. ``"192.168.1.10:8765"``).
|
|
Can be combined with ``browser`` to route to a specific
|
|
remote profile.
|
|
token: Auth token for the remote serve instance (legacy token auth).
|
|
key: Path to Ed25519 private key PEM for pubkey auth. When set,
|
|
overrides token auth. Defaults to ``~/.config/browser-cli/client.key.pem``
|
|
if that file exists.
|
|
"""
|
|
self._browser = browser
|
|
self._remote = remote
|
|
self._token = token
|
|
self._key = Path(key) if key else None
|
|
|
|
def _cmd(self, command: str, args: dict | None = None):
|
|
return send_command(command, args, profile=self._browser, remote=self._remote, token=self._token, key=self._key)
|
|
|
|
def _multi_browser_targets(self):
|
|
if self._browser is not None:
|
|
return []
|
|
if self._remote:
|
|
targets = remote_browser_targets(self._remote, self._token)
|
|
else:
|
|
targets = active_browser_targets()
|
|
if len(targets) <= 1 and not any(target.remote for target in targets):
|
|
return []
|
|
return targets
|
|
|
|
def _collect_multi_browser(self, command: str, args: dict | None = None):
|
|
results = []
|
|
targets = self._multi_browser_targets()
|
|
for target in targets:
|
|
try:
|
|
if target.remote:
|
|
data = send_command(command, args, profile=target.profile, remote=target.remote, token=target.token, key=self._key)
|
|
else:
|
|
data = send_command(command, args, profile=target.profile)
|
|
except (BrowserNotConnected, RuntimeError):
|
|
continue
|
|
results.append((target, data))
|
|
if results:
|
|
return results
|
|
if targets:
|
|
raise BrowserNotConnected(
|
|
"Cannot resolve a browser socket automatically.\n"
|
|
"Make sure the browser is running with the browser-cli extension enabled,\n"
|
|
"or pass --browser <alias> / set BROWSER_CLI_PROFILE to a known alias."
|
|
)
|
|
return []
|
|
|
|
# ── Internal factories ────────────────────────────────────────────────
|
|
|
|
def _make_tab(
|
|
self,
|
|
data: dict,
|
|
*,
|
|
browser_profile: str | None = None,
|
|
browser_name: str | None = None,
|
|
browser_remote: str | None = None,
|
|
browser_token: str | None = None,
|
|
) -> Tab:
|
|
tab = Tab(
|
|
id=data["id"],
|
|
window_id=data.get("windowId", 0),
|
|
active=data.get("active", False),
|
|
muted=data.get("muted", False),
|
|
title=data.get("title") or "",
|
|
url=data.get("url") or "",
|
|
group_id=data.get("groupId") or None,
|
|
browser=browser_name,
|
|
)
|
|
tab._browser = self if browser_profile is None else BrowserCLI(
|
|
browser=browser_profile,
|
|
remote=browser_remote,
|
|
token=browser_token,
|
|
)
|
|
return tab
|
|
|
|
def _make_group(
|
|
self,
|
|
data: dict,
|
|
*,
|
|
browser_profile: str | None = None,
|
|
browser_name: str | None = None,
|
|
browser_remote: str | None = None,
|
|
browser_token: str | None = None,
|
|
) -> Group:
|
|
group = Group(
|
|
id=data["id"],
|
|
title=data.get("title") or "",
|
|
color=data.get("color") or "",
|
|
collapsed=data.get("collapsed", False),
|
|
tab_count=data.get("tabCount", 0),
|
|
browser=browser_name,
|
|
)
|
|
group._browser = self if browser_profile is None else BrowserCLI(
|
|
browser=browser_profile,
|
|
remote=browser_remote,
|
|
token=browser_token,
|
|
)
|
|
return group
|
|
|
|
# ── Navigation ────────────────────────────────────────────────────────
|
|
|
|
def open(self, url: str, *, background: bool = False, window: str | None = None, group: str | None = None) -> None:
|
|
self._cmd("navigate.open", {"url": url, "background": background, "window": window, "group": group})
|
|
|
|
def reload(self, tab_id: int | None = None) -> None:
|
|
self._cmd("navigate.reload", {"tabId": tab_id})
|
|
|
|
def hard_reload(self, tab_id: int | None = None) -> None:
|
|
self._cmd("navigate.hard_reload", {"tabId": tab_id})
|
|
|
|
def back(self, tab_id: int | None = None) -> None:
|
|
self._cmd("navigate.back", {"tabId": tab_id})
|
|
|
|
def forward(self, tab_id: int | None = None) -> None:
|
|
self._cmd("navigate.forward", {"tabId": tab_id})
|
|
|
|
def focus_url(self, pattern: str) -> None:
|
|
self._cmd("navigate.focus", {"pattern": pattern})
|
|
|
|
def navigate_tab(self, tab_id: int, url: str) -> None:
|
|
"""Navigate a specific tab to *url*."""
|
|
self._cmd("navigate.to", {"tabId": tab_id, "url": url})
|
|
|
|
def open_wait(
|
|
self,
|
|
url: str,
|
|
*,
|
|
timeout: float = 30.0,
|
|
background: bool = False,
|
|
window: str | None = None,
|
|
group: str | None = None,
|
|
) -> "Tab":
|
|
"""Open URL in a new tab and block until fully loaded. Returns the Tab."""
|
|
data = self._cmd("navigate.open_wait", {
|
|
"url": url, "timeout": int(timeout * 1000),
|
|
"background": background, "window": window, "group": group,
|
|
})
|
|
if not isinstance(data, dict) or "id" not in data:
|
|
raise RuntimeError("navigate.open_wait returned unexpected data")
|
|
return self._make_tab(data)
|
|
|
|
def wait_for_load(
|
|
self,
|
|
tab_id: int | None = None,
|
|
*,
|
|
timeout: float = 30.0,
|
|
ready_state: str = "complete",
|
|
) -> Tab:
|
|
"""Block until the tab finishes loading. Returns the Tab when ready.
|
|
|
|
Args:
|
|
tab_id: Tab to watch. Defaults to the active tab.
|
|
timeout: Max seconds to wait before raising ``RuntimeError``.
|
|
ready_state: ``"complete"`` (default) or ``"interactive"``.
|
|
"""
|
|
data = self._cmd("navigate.wait", {
|
|
"tabId": tab_id,
|
|
"timeout": int(timeout * 1000),
|
|
"readyState": ready_state,
|
|
})
|
|
if not isinstance(data, dict) or "id" not in data:
|
|
raise RuntimeError("navigate.wait returned unexpected data")
|
|
return self._make_tab(data)
|
|
|
|
# ── Search ────────────────────────────────────────────────────────────
|
|
|
|
def search(
|
|
self, engine: str, query: str, *,
|
|
background: bool = False, window: str | None = None, group: str | None = None,
|
|
) -> None:
|
|
"""Open a search query in the given engine (e.g. 'google', 'youtube', 'ddg')."""
|
|
from urllib.parse import quote_plus
|
|
from browser_cli.commands.search import ENGINES
|
|
template = ENGINES.get(engine)
|
|
if template is None:
|
|
raise ValueError(f"Unknown search engine '{engine}'. Available: {', '.join(ENGINES)}")
|
|
url = template.format(query=quote_plus(query))
|
|
self._cmd("navigate.open", {"url": url, "background": background, "window": window, "group": group})
|
|
|
|
# ── Tabs ──────────────────────────────────────────────────────────────
|
|
|
|
def tabs_list(self) -> list[Tab]:
|
|
"""Return all open tabs across all windows.
|
|
|
|
When multiple browsers are active and no browser was specified, each Tab
|
|
includes ``tab.browser`` naming its source browser.
|
|
"""
|
|
multi_results = self._collect_multi_browser("tabs.list", {})
|
|
if multi_results:
|
|
return [
|
|
self._make_tab(
|
|
tab,
|
|
browser_profile=target.profile,
|
|
browser_name=target.display_name,
|
|
browser_remote=target.remote,
|
|
browser_token=target.token,
|
|
)
|
|
for target, tabs in multi_results
|
|
for tab in (tabs or [])
|
|
]
|
|
return [self._make_tab(t) for t in (self._cmd("tabs.list", {}) or [])]
|
|
|
|
def tabs_close(self, tab_id: int | None = None, *, inactive: bool = False, duplicates: bool = False) -> int:
|
|
"""Close tab(s). Returns the number of tabs closed."""
|
|
result = self._cmd("tabs.close", {"tabId": tab_id, "inactive": inactive, "duplicates": duplicates})
|
|
return result.get("closed", 1) if isinstance(result, dict) else 1
|
|
|
|
def tabs_move(
|
|
self, tab_id: int, *,
|
|
forward: bool = False, backward: bool = False,
|
|
group_id: int | None = None, window_id: int | None = None, index: int | None = None,
|
|
) -> None:
|
|
self._cmd("tabs.move", {
|
|
"tabId": tab_id, "forward": forward, "backward": backward,
|
|
"groupId": group_id, "windowId": window_id, "index": index,
|
|
})
|
|
|
|
def tabs_active(self, tab_id: int) -> None:
|
|
"""Switch browser focus to a tab by ID."""
|
|
self._cmd("tabs.active", {"tabId": tab_id})
|
|
|
|
def tabs_status(self, tab_id: int | None = None) -> Tab:
|
|
"""Return status for the active tab or a specific tab."""
|
|
data = self._cmd("tabs.status", {"tabId": tab_id})
|
|
if not isinstance(data, dict) or "id" not in data:
|
|
raise RuntimeError("No tab status returned")
|
|
return self._make_tab(data)
|
|
|
|
def tabs_mute(self, tab_id: int | None = None) -> int:
|
|
"""Mute the active tab or a specific tab. Returns the target tab ID."""
|
|
result = self._cmd("tabs.mute", {"tabId": tab_id})
|
|
return result.get("tabId", tab_id) if isinstance(result, dict) else int(tab_id or 0)
|
|
|
|
def tabs_unmute(self, tab_id: int | None = None) -> int:
|
|
"""Unmute the active tab or a specific tab. Returns the target tab ID."""
|
|
result = self._cmd("tabs.unmute", {"tabId": tab_id})
|
|
return result.get("tabId", tab_id) if isinstance(result, dict) else int(tab_id or 0)
|
|
|
|
def tabs_pin(self, tab_id: int | None = None) -> int:
|
|
"""Pin the active tab or a specific tab. Returns the target tab ID."""
|
|
result = self._cmd("tabs.pin", {"tabId": tab_id})
|
|
return result.get("tabId", tab_id) if isinstance(result, dict) else int(tab_id or 0)
|
|
|
|
def tabs_unpin(self, tab_id: int | None = None) -> int:
|
|
"""Unpin the active tab or a specific tab. Returns the target tab ID."""
|
|
result = self._cmd("tabs.unpin", {"tabId": tab_id})
|
|
return result.get("tabId", tab_id) if isinstance(result, dict) else int(tab_id or 0)
|
|
|
|
def tabs_watch_url(
|
|
self,
|
|
pattern: str,
|
|
*,
|
|
tab_id: int | None = None,
|
|
timeout: float = 30.0,
|
|
) -> "Tab":
|
|
"""Block until the tab URL matches regex pattern. Returns the Tab."""
|
|
data = self._cmd("tabs.watch_url", {"pattern": pattern, "tabId": tab_id, "timeout": int(timeout * 1000)})
|
|
if not isinstance(data, dict) or "id" not in data:
|
|
raise RuntimeError("tabs.watch_url returned unexpected data")
|
|
return self._make_tab(data)
|
|
|
|
def tabs_screenshot(
|
|
self,
|
|
tab_id: int | None = None,
|
|
*,
|
|
format: str = "png",
|
|
quality: int | None = None,
|
|
) -> str:
|
|
"""Capture the visible area of a tab. Returns a base64 data URL.
|
|
|
|
Args:
|
|
tab_id: Tab to capture. Defaults to the active tab.
|
|
format: ``"png"`` (default) or ``"jpeg"``.
|
|
quality: JPEG quality 0-100 (ignored for PNG).
|
|
"""
|
|
result = self._cmd("tabs.screenshot", {"tabId": tab_id, "format": format, "quality": quality})
|
|
return result.get("dataUrl", "") if isinstance(result, dict) else str(result)
|
|
|
|
def window_active_tab(self, window_id: int) -> Tab:
|
|
"""Return active tab for a specific browser window."""
|
|
data = self._cmd("tabs.active_in_window", {"windowId": window_id})
|
|
if not isinstance(data, dict) or "id" not in data:
|
|
raise RuntimeError(f"No active tab found for window {window_id}")
|
|
return self._make_tab(data)
|
|
|
|
def tabs_filter(self, pattern_or_filter: str | Callable[[Tab], bool] | Callable[[list[Tab]], Iterable[Tab]]) -> list[Tab]:
|
|
"""Return tabs filtered by pattern or a Python callable."""
|
|
if isinstance(pattern_or_filter, str):
|
|
return [self._make_tab(t) for t in (self._cmd("tabs.filter", {"pattern": pattern_or_filter}) or [])]
|
|
return self._apply_tab_filter(pattern_or_filter)
|
|
|
|
def tabs_count(self, pattern: str | None = None) -> int | BrowserCounts:
|
|
"""Count open tabs, optionally filtered by URL pattern.
|
|
|
|
Returns ``BrowserCounts`` in implicit multi-browser mode.
|
|
"""
|
|
multi_results = self._collect_multi_browser("tabs.count", {"pattern": pattern})
|
|
if multi_results:
|
|
by_browser = {target.display_name: int(count or 0) for target, count in multi_results}
|
|
return BrowserCounts(total=sum(by_browser.values()), by_browser=by_browser)
|
|
return self._cmd("tabs.count", {"pattern": pattern})
|
|
|
|
def tabs_query(self, search: str) -> list[Tab]:
|
|
"""Search tabs by URL or title."""
|
|
return [self._make_tab(t) for t in (self._cmd("tabs.query", {"search": search}) or [])]
|
|
|
|
def tabs_html(self, tab_id: int | None = None) -> str:
|
|
"""Return the full HTML source of the active (or specified) tab."""
|
|
return self._cmd("tabs.html", {"tabId": tab_id}) or ""
|
|
|
|
def tabs_dedupe(self) -> int:
|
|
"""Close duplicate tabs (keep the first occurrence of each URL). Returns count closed."""
|
|
result = self._cmd("tabs.dedupe", {})
|
|
return result.get("closed", 0) if isinstance(result, dict) else 0
|
|
|
|
def tabs_sort(self, by: str = "domain") -> None:
|
|
"""Sort tabs within each window. *by* is one of 'domain', 'title', 'time'."""
|
|
self._cmd("tabs.sort", {"by": by})
|
|
|
|
def tabs_merge_windows(self) -> int:
|
|
"""Move all tabs into the focused window. Returns count moved."""
|
|
result = self._cmd("tabs.merge_windows", {})
|
|
return result.get("moved", 0) if isinstance(result, dict) else 0
|
|
|
|
def tabs_close_inactive(self) -> int:
|
|
"""Close all inactive tabs. Returns count closed."""
|
|
result = self._cmd("tabs.close", {"inactive": True})
|
|
return result.get("closed", 0) if isinstance(result, dict) else 0
|
|
|
|
def tabs_close_duplicates(self) -> int:
|
|
"""Close duplicate tabs. Returns count closed."""
|
|
result = self._cmd("tabs.close", {"duplicates": True})
|
|
return result.get("closed", 0) if isinstance(result, dict) else 0
|
|
|
|
# ── Tab Groups ────────────────────────────────────────────────────────
|
|
|
|
def group_list(self) -> list[Group]:
|
|
"""Return all tab groups.
|
|
|
|
When multiple browsers are active and no browser was specified, each Group
|
|
includes ``group.browser`` naming its source browser.
|
|
"""
|
|
multi_results = self._collect_multi_browser("group.list", {})
|
|
if multi_results:
|
|
return [
|
|
self._make_group(
|
|
group,
|
|
browser_profile=target.profile,
|
|
browser_name=target.display_name,
|
|
browser_remote=target.remote,
|
|
browser_token=target.token,
|
|
)
|
|
for target, groups in multi_results
|
|
for group in (groups or [])
|
|
]
|
|
return [self._make_group(g) for g in (self._cmd("group.list", {}) or [])]
|
|
|
|
def group_tabs(self, group_id: int) -> list[Tab]:
|
|
"""Return all tabs inside a group."""
|
|
return [self._make_tab(t) for t in (self._cmd("group.tabs", {"groupId": group_id}) or [])]
|
|
|
|
def group_count(self) -> int | BrowserCounts:
|
|
"""Return the number of tab groups.
|
|
|
|
Returns ``BrowserCounts`` in implicit multi-browser mode.
|
|
"""
|
|
multi_results = self._collect_multi_browser("group.count", {})
|
|
if multi_results:
|
|
by_browser = {target.display_name: int(count or 0) for target, count in multi_results}
|
|
return BrowserCounts(total=sum(by_browser.values()), by_browser=by_browser)
|
|
return self._cmd("group.count", {})
|
|
|
|
def group_query(self, search: str) -> list[Group]:
|
|
"""Search groups by name."""
|
|
return [self._make_group(g) for g in (self._cmd("group.query", {"search": search}) or [])]
|
|
|
|
def group_close(self, group_id: int) -> None:
|
|
"""Ungroup (and close) a tab group by ID."""
|
|
self._cmd("group.close", {"groupId": group_id})
|
|
|
|
def group_create(self, name: str) -> Group:
|
|
"""Create a new tab group with *name*. Returns the created Group."""
|
|
data = self._cmd("group.open", {"name": name})
|
|
return self._make_group(data) if isinstance(data, dict) else Group(id=data, title=name, color="", collapsed=False, tab_count=0)
|
|
|
|
def group_add_tab(self, group: str | int, url: str | None = None) -> int | None:
|
|
"""Open a new tab (optionally at URL) inside a group. Returns the new tab ID."""
|
|
result = self._cmd("group.add_tab", {"group": str(group), "url": url})
|
|
return result.get("tabId") if isinstance(result, dict) else result
|
|
|
|
def group_move(self, group: str | int, *, forward: bool = False, backward: bool = False) -> None:
|
|
"""Move a tab group forward or backward."""
|
|
self._cmd("group.move", {"group": str(group), "forward": forward, "backward": backward})
|
|
|
|
# ── Windows ───────────────────────────────────────────────────────────
|
|
|
|
def windows_list(self) -> list[dict]:
|
|
"""Return browser windows.
|
|
|
|
In implicit multi-browser mode each window dict includes a ``browser`` key.
|
|
"""
|
|
multi_results = self._collect_multi_browser("windows.list", {})
|
|
if multi_results:
|
|
return [
|
|
{**window, "browser": target.display_name}
|
|
for target, windows in multi_results
|
|
for window in (windows or [])
|
|
]
|
|
return self._cmd("windows.list", {}) or []
|
|
|
|
def windows_rename(self, window_id: int, name: str) -> None:
|
|
self._cmd("windows.rename", {"windowId": window_id, "name": name})
|
|
|
|
def windows_close(self, window_id: int) -> None:
|
|
self._cmd("windows.close", {"windowId": window_id})
|
|
|
|
def windows_open(self, url: str | None = None) -> dict:
|
|
"""Open a new browser window, optionally on a URL."""
|
|
return self._cmd("windows.open", {"url": url}) or {}
|
|
|
|
# ── DOM ───────────────────────────────────────────────────────────────
|
|
|
|
def dom_query(self, selector: str) -> list[dict]:
|
|
return self._cmd("dom.query", {"selector": selector}) or []
|
|
|
|
def dom_click(self, selector: str) -> None:
|
|
self._cmd("dom.click", {"selector": selector})
|
|
|
|
def dom_type(self, selector: str, text: str) -> None:
|
|
self._cmd("dom.type", {"selector": selector, "text": text})
|
|
|
|
def dom_attr(self, selector: str, attr: str) -> list[str]:
|
|
return self._cmd("dom.attr", {"selector": selector, "attr": attr}) or []
|
|
|
|
def dom_text(self, selector: str) -> list[str]:
|
|
return self._cmd("dom.text", {"selector": selector}) or []
|
|
|
|
def dom_exists(self, selector: str) -> bool:
|
|
return self._cmd("dom.exists", {"selector": selector}) or False
|
|
|
|
def dom_scroll(self, selector: str | None = None, *, x: int | None = None, y: int | None = None) -> None:
|
|
"""Scroll to a CSS selector or to pixel coordinates."""
|
|
self._cmd("dom.scroll", {"selector": selector, "x": x, "y": y})
|
|
|
|
def dom_select(self, selector: str, value: str) -> None:
|
|
"""Set the value of a <select> element."""
|
|
self._cmd("dom.select", {"selector": selector, "value": value})
|
|
|
|
def dom_eval(self, code: str, tab_id: int | None = None):
|
|
"""Evaluate JavaScript in the page's main world and return the result."""
|
|
return self._cmd("dom.eval", {"code": code, "tabId": tab_id})
|
|
|
|
def dom_key(self, key: str, selector: str | None = None) -> None:
|
|
"""Dispatch a keyboard event. key examples: 'Enter', 'Tab', 'Escape', 'ArrowDown'."""
|
|
self._cmd("dom.key", {"key": key, "selector": selector})
|
|
|
|
def dom_hover(self, selector: str) -> None:
|
|
"""Dispatch mouseover/mouseenter on an element."""
|
|
self._cmd("dom.hover", {"selector": selector})
|
|
|
|
def dom_check(self, selector: str) -> None:
|
|
"""Check a checkbox."""
|
|
self._cmd("dom.check", {"selector": selector})
|
|
|
|
def dom_uncheck(self, selector: str) -> None:
|
|
"""Uncheck a checkbox."""
|
|
self._cmd("dom.uncheck", {"selector": selector})
|
|
|
|
def dom_clear(self, selector: str) -> None:
|
|
"""Clear the value of an input element."""
|
|
self._cmd("dom.clear", {"selector": selector})
|
|
|
|
def dom_focus(self, selector: str) -> None:
|
|
"""Focus an element."""
|
|
self._cmd("dom.focus", {"selector": selector})
|
|
|
|
def dom_submit(self, selector: str) -> None:
|
|
"""Submit the form containing the matched element."""
|
|
self._cmd("dom.submit", {"selector": selector})
|
|
|
|
def dom_poll(
|
|
self,
|
|
selector: str,
|
|
pattern: str,
|
|
*,
|
|
attr: str | None = None,
|
|
timeout: float = 30.0,
|
|
interval: float = 0.5,
|
|
tab_id: int | None = None,
|
|
) -> dict:
|
|
"""Poll selector's text/value until it matches regex pattern.
|
|
|
|
Returns ``{"selector": ..., "value": ..., "pattern": ...}`` when matched.
|
|
"""
|
|
return self._cmd("dom.poll", {
|
|
"selector": selector,
|
|
"pattern": pattern,
|
|
"attr": attr,
|
|
"timeout": int(timeout * 1000),
|
|
"interval": int(interval * 1000),
|
|
"tabId": tab_id,
|
|
})
|
|
|
|
def dom_wait_for(
|
|
self,
|
|
selector: str,
|
|
*,
|
|
timeout: float = 10.0,
|
|
visible: bool = False,
|
|
hidden: bool = False,
|
|
tab_id: int | None = None,
|
|
) -> dict:
|
|
"""Wait until a CSS selector appears (or disappears) in the DOM.
|
|
|
|
Args:
|
|
selector: CSS selector to watch.
|
|
timeout: Max seconds to wait before raising ``RuntimeError``.
|
|
visible: Wait until the element has non-zero dimensions.
|
|
hidden: Wait until the element is absent or has ``offsetParent == null``.
|
|
tab_id: Tab to watch. Defaults to the active tab.
|
|
"""
|
|
return self._cmd("dom.wait_for", {
|
|
"selector": selector,
|
|
"timeout": int(timeout * 1000),
|
|
"visible": visible,
|
|
"hidden": hidden,
|
|
"tabId": tab_id,
|
|
})
|
|
|
|
# ── Page ─────────────────────────────────────────────────────────────
|
|
|
|
def page_info(self) -> dict:
|
|
"""Return title, URL, readyState, lang, and meta tags of the active tab."""
|
|
return self._cmd("page.info", {}) or {}
|
|
|
|
# ── Storage ───────────────────────────────────────────────────────────
|
|
|
|
def storage_get(
|
|
self,
|
|
key: str | None = None,
|
|
*,
|
|
type: str = "local",
|
|
tab_id: int | None = None,
|
|
) -> str | dict | None:
|
|
"""Get a localStorage/sessionStorage entry (or all entries if key omitted)."""
|
|
return self._cmd("storage.get", {"key": key, "type": type, "tabId": tab_id})
|
|
|
|
def storage_set(
|
|
self,
|
|
key: str,
|
|
value: str,
|
|
*,
|
|
type: str = "local",
|
|
tab_id: int | None = None,
|
|
) -> None:
|
|
"""Set a localStorage/sessionStorage entry."""
|
|
self._cmd("storage.set", {"key": key, "value": value, "type": type, "tabId": tab_id})
|
|
|
|
# ── Cookies ───────────────────────────────────────────────────────────
|
|
|
|
def cookies_list(
|
|
self,
|
|
*,
|
|
url: str | None = None,
|
|
domain: str | None = None,
|
|
name: str | None = None,
|
|
) -> list[dict]:
|
|
"""List cookies, optionally filtered by url, domain, or name."""
|
|
return self._cmd("cookies.list", {"url": url, "domain": domain, "name": name}) or []
|
|
|
|
def cookies_get(self, url: str, name: str) -> dict | None:
|
|
"""Get a single cookie by url and name."""
|
|
return self._cmd("cookies.get", {"url": url, "name": name})
|
|
|
|
def cookies_set(
|
|
self,
|
|
url: str,
|
|
name: str,
|
|
value: str,
|
|
*,
|
|
domain: str | None = None,
|
|
path: str | None = None,
|
|
secure: bool | None = None,
|
|
http_only: bool | None = None,
|
|
expiration_date: float | None = None,
|
|
same_site: str | None = None,
|
|
) -> dict:
|
|
"""Set a cookie. Returns the created cookie dict."""
|
|
return self._cmd("cookies.set", {
|
|
"url": url, "name": name, "value": value,
|
|
"domain": domain, "path": path,
|
|
"secure": secure, "httpOnly": http_only,
|
|
"expirationDate": expiration_date, "sameSite": same_site,
|
|
})
|
|
|
|
# ── Extract ───────────────────────────────────────────────────────────
|
|
|
|
def extract_links(self) -> list[dict]:
|
|
return self._cmd("extract.links", {}) or []
|
|
|
|
def extract_images(self) -> list[dict]:
|
|
return self._cmd("extract.images", {}) or []
|
|
|
|
def extract_text(self) -> str:
|
|
return self._cmd("extract.text", {}) or ""
|
|
|
|
def extract_json(self, selector: str):
|
|
return self._cmd("extract.json", {"selector": selector})
|
|
|
|
def extract_markdown(self, selector: str | None = None) -> str:
|
|
return self._cmd("extract.markdown", {"selector": selector}) or ""
|
|
|
|
# ── Session ───────────────────────────────────────────────────────────
|
|
|
|
def session_save(self, name: str) -> None:
|
|
self._cmd("session.save", {"name": name})
|
|
|
|
def session_load(self, name: str) -> None:
|
|
self._cmd("session.load", {"name": name})
|
|
|
|
def session_diff(self, name_a: str, name_b: str) -> dict:
|
|
return self._cmd("session.diff", {"nameA": name_a, "nameB": name_b}) or {}
|
|
|
|
def session_list(self) -> list[dict]:
|
|
"""Return saved sessions.
|
|
|
|
In implicit multi-browser mode each session dict includes a ``browser`` key.
|
|
"""
|
|
multi_results = self._collect_multi_browser("session.list", {})
|
|
if multi_results:
|
|
return [
|
|
{**session, "browser": target.display_name}
|
|
for target, sessions in multi_results
|
|
for session in (sessions or [])
|
|
]
|
|
return self._cmd("session.list", {}) or []
|
|
|
|
def session_remove(self, name: str) -> None:
|
|
self._cmd("session.remove", {"name": name})
|
|
|
|
def session_auto_save(self, enabled: bool) -> None:
|
|
self._cmd("session.auto_save", {"enabled": enabled})
|
|
|
|
# ── Misc ──────────────────────────────────────────────────────────────
|
|
|
|
def clients(self) -> list[dict]:
|
|
return self._cmd("clients.list", {})
|
|
|
|
def _apply_tab_filter(self, filter_fn: Callable[[Tab], bool] | Callable[[list[Tab]], Iterable[Tab]]) -> list[Tab]:
|
|
tabs = self.tabs_list()
|
|
|
|
try:
|
|
transformed = filter_fn(tabs)
|
|
except (AttributeError, TypeError):
|
|
return [tab for tab in tabs if filter_fn(tab)]
|
|
|
|
if isinstance(transformed, list):
|
|
return transformed
|
|
if isinstance(transformed, tuple):
|
|
return list(transformed)
|
|
if isinstance(transformed, set):
|
|
return list(transformed)
|
|
if transformed is tabs:
|
|
return tabs
|
|
if isinstance(transformed, bool):
|
|
return [tab for tab in tabs if filter_fn(tab)]
|
|
|
|
try:
|
|
return list(transformed)
|
|
except TypeError:
|
|
return [tab for tab in tabs if filter_fn(tab)]
|