This repository has no description
3.5 kB
102 lines
1"""Browser CLI client - Unix socket client for browser automation."""
2
3import asyncio
4import json
5import os
6import tempfile
7from pathlib import Path
8from typing import Any
9
10from browser_cli.errors import BrowserConnectionError, CommandError
11
12
13class BrowserClient:
14 """Unix socket client for browser automation."""
15
16 def __init__(self, socket_path: str | None = None) -> None:
17 """Initialize the browser CLI client."""
18 if socket_path:
19 self.socket_path = socket_path
20 else:
21 runtime_dir = os.environ.get("XDG_RUNTIME_DIR")
22 if not runtime_dir:
23 runtime_dir = tempfile.gettempdir()
24 self.socket_path = str(Path(runtime_dir) / "browser-cli.sock")
25
26 self.message_counter = 0
27
28 async def send_command(
29 self,
30 command: str,
31 params: dict[str, Any] | None = None,
32 tab_id: str | None = None,
33 ) -> dict[str, Any]:
34 """Send command to browser and wait for response."""
35 self.message_counter += 1
36 message_id = str(self.message_counter)
37
38 message: dict[str, Any] = {
39 "command": command,
40 "params": params or {},
41 "id": message_id,
42 }
43
44 if tab_id:
45 message["tabId"] = tab_id
46
47 try:
48 reader, writer = await asyncio.open_unix_connection(
49 self.socket_path,
50 limit=10 * 1024 * 1024, # 10MB for screenshots
51 )
52
53 writer.write((json.dumps(message) + "\n").encode("utf-8"))
54 await writer.drain()
55
56 response_line = await reader.readline()
57 if not response_line:
58 msg = "No response from server"
59 raise CommandError(msg)
60
61 response = json.loads(response_line.decode("utf-8").strip())
62
63 writer.close()
64 await writer.wait_closed()
65
66 if response.get("id") == message_id:
67 if response.get("success"):
68 result: dict[str, Any] = response.get("result", {})
69 return result
70 error_msg = response.get("error", "Unknown error")
71 raise CommandError(error_msg) from None
72
73 msg = f"Invalid response ID. Expected: {message_id}, Got: {response.get('id')}"
74 raise CommandError(msg)
75
76 except (ConnectionRefusedError, FileNotFoundError) as e:
77 msg = (
78 "Cannot connect to browser extension. Make sure:\n"
79 "1. Firefox is running\n"
80 "2. The Browser CLI extension is installed\n"
81 "3. The extension is enabled on the current tab\n"
82 f"\nSocket path: {self.socket_path}"
83 )
84 raise BrowserConnectionError(msg) from e
85
86 async def exec_js(
87 self,
88 code: str,
89 tab_id: str | None = None,
90 ) -> dict[str, object] | list[object] | str | int | float | bool | None:
91 """Execute JavaScript code in the browser and return result."""
92 result = await self.send_command("exec", {"code": code}, tab_id)
93 js_result: dict[str, object] | list[object] | str | int | float | bool | None = result.get(
94 "result",
95 )
96 return js_result
97
98 async def list_tabs(self) -> list[dict[str, Any]]:
99 """List all managed tabs."""
100 result = await self.send_command("list-tabs")
101 tabs: list[dict[str, Any]] = result.get("tabs", [])
102 return tabs