Automatisation/browser-cli
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases 5 Wiki Activity

add moveing of tabs and groups, multi browser support, auto complite into terminal, extract html and adding testing

Browse Source
This commit is contained in:
Daniel Dolezal
2026-04-09 01:41:01 +02:00
parent 0cb2f1cb3f
commit ab4ba97886
19 changed files with 1069 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
browser_cli/cli.py
+100 -14
View File
@@ -59,22 +59,58 @@ main.add_command(session_group)
@main.command("clients")
def cmd_clients():
"""Show connected browser clients."""
try:
clients = send_command("clients.list")
except BrowserNotConnected as e:
console.print(f"[red]Error:[/red] {e}")
import json as _json
from browser_cli.client import REGISTRY_PATH, DEFAULT_SOCKET
# Build a map of profile → socket path from the registry
profiles: dict[str, str] = {}
if REGISTRY_PATH.exists():
try:
profiles = _json.loads(REGISTRY_PATH.read_text())
except Exception:
pass
if not profiles:
profiles = {"default": DEFAULT_SOCKET}
all_clients = []
for profile_name, sock_path in profiles.items():
try:
result = send_command("clients.list", profile=profile_name)
for c in (result or []):
c.setdefault("profile", profile_name)
all_clients.append(c)
except (BrowserNotConnected, RuntimeError):
# Socket registered but browser no longer connected
all_clients.append({"profile": profile_name, "name": "—", "version": "—", "platform": "disconnected"})
if not all_clients:
console.print("[yellow]No browser clients found[/yellow]")
sys.exit(1)
from rich.table import Table
table = Table(show_header=True, header_style="bold cyan")
table.add_column("Profile")
table.add_column("Browser")
table.add_column("Version")
table.add_column("Platform")
for c in (clients or []):
table.add_row(c.get("name", ""), c.get("version", ""), c.get("platform", ""))
for c in all_clients:
table.add_row(c.get("profile", "default"), c.get("name", ""), c.get("version", ""), c.get("platform", ""))
console.print(table)
@main.command("rename-profile")
@click.argument("alias")
def cmd_rename_profile(alias):
"""Set the profile alias used to identify this browser instance."""
try:
send_command("clients.rename_profile", {"alias": alias})
except BrowserNotConnected as e:
console.print(f"[red]Error:[/red] {e}")
sys.exit(1)
console.print(f"[green]Profile renamed to '{alias}'[/green]")
console.print(" Restart the browser for the change to take effect.")
# ── install ────────────────────────────────────────────────────────────────────
@main.command("install")
@@ -82,17 +118,19 @@ def cmd_clients():
def cmd_install(browser):
"""Register the native messaging host and print extension load instructions."""
# Find the native_host.py path
native_host_script = Path(__file__).parent / "native_host.py"
if not native_host_script.exists():
console.print(f"[red]Cannot find native_host.py at {native_host_script}[/red]")
# Find the venv entry point for the native host (stable regardless of project location)
venv_script = Path(sys.executable).parent / "browser-cli-native-host"
if not venv_script.exists():
console.print(f"[red]Cannot find browser-cli-native-host in venv ({venv_script})[/red]")
console.print(" Run [cyan]uv sync[/cyan] first to install entry points.")
sys.exit(1)
# Build a wrapper shell script so it's executable by Chrome
wrapper_path = Path(__file__).parent.parent / "browser-cli-native-host"
python_exe = sys.executable
# Install wrapper to ~/.local/bin so the manifest path never changes
local_bin = Path.home() / ".local" / "bin"
local_bin.mkdir(parents=True, exist_ok=True)
wrapper_path = local_bin / "browser-cli-native-host"
wrapper_content = f"""#!/bin/sh
exec "{python_exe}" "{native_host_script}" "$@"
exec "{venv_script}" "$@"
"""
wrapper_path.write_text(wrapper_content)
wrapper_path.chmod(wrapper_path.stat().st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH)
@@ -143,5 +181,53 @@ exec "{python_exe}" "{native_host_script}" "$@"
console.print(" After restarting Chrome, try: [cyan]browser-cli tabs list[/cyan]")
# ── completion ─────────────────────────────────────────────────────────────────
@main.command("completion")
@click.argument("shell", type=click.Choice(["zsh", "bash", "fish"]))
@click.option("--script", is_flag=True, help="Output the raw completion script instead of instructions")
def cmd_completion(shell, script):
"""Print shell completion setup instructions (or output the script with --script)."""
if script:
from click.shell_completion import BashComplete, ZshComplete, FishComplete
cls = {"zsh": ZshComplete, "bash": BashComplete, "fish": FishComplete}[shell]
comp = cls(main, {}, "browser-cli", "_BROWSER_CLI_COMPLETE")
click.echo(comp.source())
return
exe = sys.executable.replace("/python", "/browser-cli").replace("/python3", "/browser-cli")
if not Path(exe).exists():
exe = "browser-cli"
env_var = "_BROWSER_CLI_COMPLETE"
if shell == "zsh":
console.print("[bold]Quickest setup — generate the file once:[/bold]")
console.print()
console.print(f" [cyan]uv run browser-cli completion zsh --script > ~/.zfunc/_browser-cli[/cyan]")
console.print()
console.print(" Then add these lines to [bold]~/.zshrc[/bold] (before any compinit call):")
console.print(" [cyan]fpath=(~/.zfunc $fpath)[/cyan]")
console.print(" [cyan]autoload -Uz compinit && compinit[/cyan]")
console.print()
console.print(" Reload: [cyan]exec zsh[/cyan]")
console.print()
console.print("[bold]Alternative — eval on every shell start (simpler but slower):[/bold]")
console.print(f' [cyan]eval "$({env_var}=zsh_source {exe})"[/cyan]')
elif shell == "bash":
console.print("[bold]Quickest setup — generate the file once:[/bold]")
console.print()
console.print(f" [cyan]uv run browser-cli completion bash --script > ~/.bash_completion.d/browser-cli[/cyan]")
console.print()
console.print(" Reload: [cyan]source ~/.bashrc[/cyan]")
console.print()
console.print("[bold]Alternative — eval on every shell start:[/bold]")
console.print(f' [cyan]eval "$({env_var}=bash_source {exe})"[/cyan]')
elif shell == "fish":
console.print("[bold]Setup:[/bold]")
console.print()
console.print(f" [cyan]uv run browser-cli completion fish --script > ~/.config/fish/completions/browser-cli.fish[/cyan]")
if __name__ == "__main__":
main()
browser_cli/client.py
+44 -5
View File
@@ -1,22 +1,59 @@
"""
Unix socket client — sends commands to the native host relay socket.
Used by both the CLI and the public Python API.
Profile selection order:
1. Explicit `profile` argument to send_command()
2. BROWSER_CLI_PROFILE environment variable
3. First entry in /tmp/browser-cli-registry.json
4. Fallback: /tmp/browser-cli-default.sock
"""
import json
import os
import socket
import struct
import uuid
from pathlib import Path
from typing import Any
SOCKET_PATH = "/tmp/browser-cli.sock"
REGISTRY_PATH = Path("/tmp/browser-cli-registry.json")
DEFAULT_SOCKET = "/tmp/browser-cli-default.sock"
class BrowserNotConnected(Exception):
"""Raised when the native host socket is not available."""
def send_command(command: str, args: dict | None = None) -> Any:
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
safe = target.replace(" ", "_").replace("/", "_")
return f"/tmp/browser-cli-{safe}.sock"
# Auto-detect: use first registered entry
if REGISTRY_PATH.exists():
try:
reg = json.loads(REGISTRY_PATH.read_text())
if reg:
return next(iter(reg.values()))
except Exception:
pass
return DEFAULT_SOCKET
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,
@@ -27,16 +64,18 @@ def send_command(command: str, args: dict | None = None) -> Any:
try:
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
sock.connect(SOCKET_PATH)
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(
"Cannot connect to browser.\n"
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 chrome\n"
" 3. Your browser is running"
" 3. Your browser is running\n"
" Tip: use BROWSER_CLI_PROFILE=<name> to select a specific profile"
)
result = json.loads(response)
browser_cli/commands/extract.py
+7
View File
@@ -66,3 +66,10 @@ def extract_json(selector):
"""Parse and pretty-print JSON content inside SELECTOR."""
data = _handle("extract.json", {"selector": selector})
console.print_json(json.dumps(data))
@extract_group.command("html")
def extract_html():
"""Print the full HTML of the active tab to stdout."""
html = _handle("extract.html")
click.echo(html or "")
browser_cli/commands/groups.py
+17
View File
@@ -100,3 +100,20 @@ def group_add_tab(group, url):
tab_id = result.get("tabId") if isinstance(result, dict) else result
label = url or "new tab"
console.print(f"[green]Opened {label}[/green] in group '{group}' (tab id: {tab_id})")
@group_group.command("move")
@click.argument("group")
@click.option("--forward", is_flag=True, help="Move group one position to the right")
@click.option("--backward", is_flag=True, help="Move group one position to the left")
def group_move(group, forward, backward):
"""Move a tab group forward or backward (name or ID)."""
if not forward and not backward:
console.print("[red]Specify --forward or --backward[/red]")
raise SystemExit(1)
result = _handle("group.move", {"group": group, "forward": forward, "backward": backward})
if isinstance(result, dict) and not result.get("moved"):
console.print(f"[yellow]Group '{group}' is already at the {'end' if forward else 'start'}[/yellow]")
else:
direction = "forward" if forward else "backward"
console.print(f"[green]Group '{group}' moved {direction}[/green]")
browser_cli/commands/navigate.py
+1 -1
View File
@@ -72,7 +72,7 @@ def cmd_forward(tab_id):
@nav_group.command("focus")
@click.argument("pattern")
def cmd_focus(pattern):
"""Jump to the first tab whose URL matches PATTERN."""
"""Jump to a tab by URL pattern or tab ID."""
result = _handle("navigate.focus", {"pattern": pattern})
if result:
console.print(f"[green]Focused:[/green] {result.get('url', result)}")
browser_cli/commands/tabs.py
+9 -4
View File
@@ -64,12 +64,17 @@ def tabs_close(tab_id, inactive, duplicates):
@tabs_group.command("move")
@click.argument("tab_id", type=int)
@click.option("--forward", is_flag=True, help="Move one position to the right")
@click.option("--backward", is_flag=True, help="Move one position to the left")
@click.option("--group", "group_id", type=int, default=None, help="Move to tab group ID")
@click.option("--window", "window_id", type=int, default=None, help="Move to window ID")
@click.option("--index", type=int, default=None, help="Position index in target")
def tabs_move(tab_id, group_id, window_id, index):
"""Move a tab to a different window or group."""
_handle("tabs.move", {"tabId": tab_id, "groupId": group_id, "windowId": window_id, "index": index})
@click.option("--index", type=int, default=None, help="Absolute position index in target")
def tabs_move(tab_id, forward, backward, group_id, window_id, index):
"""Move a tab. Use --forward/--backward for relative movement."""
_handle("tabs.move", {
"tabId": tab_id, "forward": forward, "backward": backward,
"groupId": group_id, "windowId": window_id, "index": index,
})
console.print("[green]Tab moved[/green]")
browser_cli/native_host.py
+71 -15
View File
@@ -5,6 +5,10 @@ Native Messaging Host for browser-cli.
Chrome launches this process when the extension calls connectNative().
It relays messages between the Chrome extension (via stdin/stdout using the
Native Messaging protocol) and the CLI (via a Unix domain socket).
Multi-browser support: the extension sends a "hello" message on startup
with a profile alias. The host uses that alias to create a unique socket
path and registers it in a shared registry file.
"""
import json
import os
@@ -16,7 +20,10 @@ import threading
import uuid
from pathlib import Path
SOCKET_PATH = "/tmp/browser-cli.sock"
REGISTRY_PATH = Path("/tmp/browser-cli-registry.json")
DEFAULT_ALIAS = "default"
SOCKET_PATH: str = "" # set after hello handshake
PENDING: dict[str, queue.Queue] = {}
PENDING_LOCK = threading.Lock()
@@ -40,16 +47,48 @@ def write_native_message(stream, msg: dict) -> None:
stream.flush()
# --- Registry helpers ---
def _registry_add(alias: str, sock_path: str) -> None:
try:
reg = json.loads(REGISTRY_PATH.read_text()) if REGISTRY_PATH.exists() else {}
reg[alias] = sock_path
REGISTRY_PATH.write_text(json.dumps(reg))
except Exception:
pass
def _registry_remove(alias: str) -> None:
try:
if not REGISTRY_PATH.exists():
return
reg = json.loads(REGISTRY_PATH.read_text())
reg.pop(alias, None)
REGISTRY_PATH.write_text(json.dumps(reg))
except Exception:
pass
def _socket_path_for(alias: str) -> str:
safe = alias.replace(" ", "_").replace("/", "_")
return f"/tmp/browser-cli-{safe}.sock"
# --- Thread A: read messages from extension (stdin) ---
def stdin_reader():
def stdin_reader(alias: str):
stdin = sys.stdin.buffer
while True:
msg = read_native_message(stdin)
if msg is None:
# Extension disconnected — clean up socket and exit
_cleanup()
# Extension disconnected — clean up and exit
_cleanup(alias)
os._exit(0)
# Profile alias handshake
if msg.get("type") == "hello":
continue # already handled during startup
msg_id = msg.get("id")
if msg_id:
with PENDING_LOCK:
@@ -60,13 +99,13 @@ def stdin_reader():
# --- Thread B: accept CLI socket connections ---
def socket_server():
path = Path(SOCKET_PATH)
def socket_server(sock_path: str):
path = Path(sock_path)
if path.exists():
path.unlink()
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.bind(SOCKET_PATH)
sock.bind(sock_path)
sock.listen(16)
while True:
@@ -92,10 +131,8 @@ def handle_cli_connection(conn: socket.socket) -> None:
with PENDING_LOCK:
PENDING[msg_id] = response_queue
# Forward command to extension via stdout
write_native_message(sys.stdout.buffer, cmd)
# Wait for extension's response (30 s timeout)
try:
result = response_queue.get(timeout=30)
except queue.Empty:
@@ -139,20 +176,39 @@ def _recv_exact(conn: socket.socket, n: int) -> bytes | None:
return buf
def _cleanup():
def _cleanup(alias: str):
try:
Path(SOCKET_PATH).unlink(missing_ok=True)
Path(_socket_path_for(alias)).unlink(missing_ok=True)
except Exception:
pass
_registry_remove(alias)
def main():
# Start socket server thread
t = threading.Thread(target=socket_server, daemon=True)
stdin = sys.stdin.buffer
# Wait for the hello handshake to learn the profile alias
first_msg = read_native_message(stdin)
if first_msg and first_msg.get("type") == "hello":
alias = first_msg.get("alias") or DEFAULT_ALIAS
else:
# No hello — fall back to default, re-queue message if it was a command
alias = DEFAULT_ALIAS
if first_msg:
msg_id = first_msg.get("id")
if msg_id:
q: queue.Queue = queue.Queue()
with PENDING_LOCK:
PENDING[msg_id] = q
write_native_message(sys.stdout.buffer, first_msg)
sock_path = _socket_path_for(alias)
_registry_add(alias, sock_path)
t = threading.Thread(target=socket_server, args=(sock_path,), daemon=True)
t.start()
# Read extension messages on main thread (blocks until extension disconnects)
stdin_reader()
stdin_reader(alias)
if __name__ == "__main__":