153 lines
4.9 KiB
Python
153 lines
4.9 KiB
Python
"""
|
|
Local IPC client — sends commands to native host relay endpoint.
|
|
Used by both CLI and public Python API.
|
|
|
|
Profile selection order:
|
|
1. Explicit `profile` argument to send_command()
|
|
2. BROWSER_CLI_PROFILE environment variable
|
|
3. First entry in runtime registry
|
|
4. Otherwise, no browser can be resolved automatically
|
|
"""
|
|
import json
|
|
import os
|
|
import socket
|
|
import struct
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from multiprocessing.connection import Client as PipeClient
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from browser_cli.platform import endpoint_for_alias, is_windows, registry_path
|
|
|
|
REGISTRY_PATH = registry_path()
|
|
|
|
|
|
class BrowserNotConnected(Exception):
|
|
"""Raised when the native host socket is not available."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BrowserTarget:
|
|
profile: str
|
|
display_name: str
|
|
socket_path: str
|
|
|
|
|
|
def _active_endpoints(reg: dict) -> dict:
|
|
"""Return only entries whose endpoint appears reachable."""
|
|
if is_windows():
|
|
return dict(reg)
|
|
return {k: v for k, v in reg.items() if Path(v).exists()}
|
|
|
|
|
|
def display_browser_name(profile_name: str, sock_path: str) -> str:
|
|
if profile_name != "default":
|
|
return profile_name
|
|
return Path(sock_path).stem or profile_name
|
|
|
|
|
|
def active_browser_targets() -> list[BrowserTarget]:
|
|
if not REGISTRY_PATH.exists():
|
|
return []
|
|
try:
|
|
reg = json.loads(REGISTRY_PATH.read_text())
|
|
except Exception:
|
|
return []
|
|
return [
|
|
BrowserTarget(profile=profile, display_name=display_browser_name(profile, sock_path), socket_path=sock_path)
|
|
for profile, sock_path in _active_endpoints(reg).items()
|
|
]
|
|
|
|
|
|
def _resolve_socket(profile: str | None = None) -> str:
|
|
"""Return the socket path for the given profile (or auto-detect)."""
|
|
target = profile or os.environ.get("BROWSER_CLI_PROFILE")
|
|
|
|
if target:
|
|
if REGISTRY_PATH.exists():
|
|
try:
|
|
reg = json.loads(REGISTRY_PATH.read_text())
|
|
if target in reg:
|
|
return reg[target]
|
|
except Exception:
|
|
pass
|
|
return endpoint_for_alias(target)
|
|
|
|
# Auto-detect: error when multiple browser instances are active
|
|
try:
|
|
active = active_browser_targets()
|
|
if len(active) > 1:
|
|
aliases = [target.profile for target in active]
|
|
examples = "\n".join(f" browser-cli --browser {a} ..." for a in aliases)
|
|
raise BrowserNotConnected(
|
|
f"Multiple browser instances are active: {', '.join(aliases)}\n"
|
|
f"Use --browser <alias> to select one:\n{examples}"
|
|
)
|
|
if active:
|
|
return active[0].socket_path
|
|
except BrowserNotConnected:
|
|
raise
|
|
except Exception:
|
|
pass
|
|
|
|
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."
|
|
)
|
|
|
|
|
|
def send_command(command: str, args: dict | None = None, profile: str | None = None) -> Any:
|
|
"""Send a command to the browser and return the response data."""
|
|
sock_path = _resolve_socket(profile)
|
|
msg = {
|
|
"id": str(uuid.uuid4()),
|
|
"command": command,
|
|
"args": args or {},
|
|
}
|
|
payload = json.dumps(msg).encode("utf-8")
|
|
framed = struct.pack("<I", len(payload)) + payload
|
|
|
|
try:
|
|
if is_windows():
|
|
with PipeClient(sock_path, family="AF_PIPE") as conn:
|
|
conn.send_bytes(payload)
|
|
response = conn.recv_bytes()
|
|
else:
|
|
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
|
|
sock.connect(sock_path)
|
|
sock.sendall(framed)
|
|
response = _recv_all(sock)
|
|
except (FileNotFoundError, ConnectionRefusedError, OSError):
|
|
profile_hint = f" (profile: {profile})" if profile else ""
|
|
raise BrowserNotConnected(
|
|
f"Cannot connect to browser{profile_hint}.\n"
|
|
"Make sure:\n"
|
|
" 1. The browser-cli extension is installed and enabled\n"
|
|
" 2. The native host is registered: uv run browser-cli install <browser>\n"
|
|
" 3. Your browser is running\n"
|
|
" Tip: use BROWSER_CLI_PROFILE=<name> to select a specific profile"
|
|
)
|
|
|
|
result = json.loads(response)
|
|
if not result.get("success", True):
|
|
raise RuntimeError(result.get("error", "unknown error from browser"))
|
|
return result.get("data")
|
|
|
|
|
|
def _recv_all(sock: socket.socket) -> bytes:
|
|
raw_len = _recv_exact(sock, 4)
|
|
msg_len = struct.unpack("<I", raw_len)[0]
|
|
return _recv_exact(sock, msg_len)
|
|
|
|
|
|
def _recv_exact(sock: socket.socket, n: int) -> bytes:
|
|
buf = b""
|
|
while len(buf) < n:
|
|
chunk = sock.recv(n - len(buf))
|
|
if not chunk:
|
|
raise ConnectionError("Socket closed before full message received")
|
|
buf += chunk
|
|
return buf
|