browser-cli/browser_cli/commands/clients.py

"""Click commands for inspecting connected browser clients."""
from __future__ import annotations

import os
import sys

import click
from rich.console import Console

from browser_cli.client import (
  BrowserNotConnected,
  REGISTRY_PATH,
  active_browser_targets,
  display_browser_name,
  remote_browser_targets,
  remote_target_for_alias,
  send_command,
)
from browser_cli.registry import load_registry

console = Console()

def _rename_target_profile(target_browser: str | None) -> str | None:
  if target_browser:
    return target_browser

  active = active_browser_targets()
  if len(active) == 1:
    return active[0].profile
  return None

def _ensure_unique_browser_alias(alias: str, target_browser: str | None) -> None:
  target_profile = _rename_target_profile(target_browser)
  profiles: dict[str, str] = load_registry(REGISTRY_PATH)

  if alias in profiles and alias != target_profile:
    raise click.ClickException(f"Browser alias '{alias}' already exists")

def _append_clients(into, label, *, profile=None, remote=None, key=None, quiet_remote_warning=False):
  """Query clients.list for one target and append each, tagged with *label*."""
  if quiet_remote_warning:
    result = send_command(
      "clients.list",
      profile=profile,
      remote=remote,
      key=key,
      suppress_pq_warning=True,
    )
  else:
    result = send_command("clients.list", profile=profile, remote=remote, key=key)
  for c in (result or []):
    c["profile"] = label
    into.append(c)

@click.group("clients", invoke_without_command=True)
@click.pass_context
def clients_group(ctx):
  """Inspect and manage connected browser clients."""
  if ctx.invoked_subcommand is not None:
    return

  all_clients = []

  browser_alias = (ctx.obj or {}).get("browser")
  remote = (ctx.obj or {}).get("remote") or os.environ.get("BROWSER_CLI_REMOTE")
  key = (ctx.obj or {}).get("key")

  if not remote and browser_alias:
    _collect_remote_alias_clients(all_clients, browser_alias, key)
  elif remote:
    _collect_explicit_remote_clients(all_clients, browser_alias, remote, key)
  else:
    _collect_local_and_saved_remote_clients(all_clients)

  if not all_clients:
    console.print("[yellow]No browser clients found. Start a browser with the extension enabled first.[/yellow]")
    sys.exit(1)

  _print_clients(all_clients)

def _collect_remote_alias_clients(all_clients: list, browser_alias: str, key) -> None:
  resolved = remote_target_for_alias(browser_alias)
  if not resolved:
    return
  try:
    targets = remote_browser_targets(resolved.remote)
  except (BrowserNotConnected, RuntimeError) as e:
    console.print(f"[red]Error:[/red] {e}")
    sys.exit(1)
  for target in targets:
    try:
      _append_clients(all_clients, target.display_name, profile=target.profile, remote=resolved.remote, key=key)
    except (BrowserNotConnected, RuntimeError):
      continue

def _collect_explicit_remote_clients(all_clients: list, browser_alias: str | None, remote: str, key) -> None:
  try:
    result = send_command("clients.list", profile=browser_alias, remote=remote, key=key)
    for c in (result or []):
      c["profile"] = c.get("profile") or browser_alias or "remote"
      all_clients.append(c)
  except (BrowserNotConnected, RuntimeError) as e:
    console.print(f"[red]Error:[/red] {e}")
    sys.exit(1)

def _collect_local_and_saved_remote_clients(all_clients: list) -> None:
  profiles: dict[str, str] = load_registry(REGISTRY_PATH) if REGISTRY_PATH.exists() else {}

  for profile_name, sock_path in profiles.items():
    display_profile = display_browser_name(profile_name, sock_path)
    try:
      _append_clients(all_clients, display_profile, profile=profile_name)
    except (BrowserNotConnected, RuntimeError):
      all_clients.append({
        "profile": display_profile,
        "name": "—",
        "version": "—",
        "extensionVersion": "disconnected",
      })

  targets = active_browser_targets(suppress_pq_warning=True)

  for target in targets:
    if target.remote is None:
      continue
    try:
      _append_clients(
        all_clients,
        target.display_name,
        profile=target.profile,
        remote=target.remote,
        quiet_remote_warning=True,
      )
    except (BrowserNotConnected, RuntimeError):
      continue

def _print_clients(all_clients: list) -> None:
  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("Extension Version")
  for c in all_clients:
    table.add_row(
      c.get("profile", ""),
      c.get("name", ""),
      c.get("version", ""),
      c.get("extensionVersion", ""),
    )
  console.print(table)

@clients_group.command("rename")
@click.option(
  "--browser",
  "target_browser",
  default=None,
  metavar="ALIAS",
  help="Browser profile alias to rename. Overrides the global --browser option for this command.",
)
@click.argument("alias")
def cmd_clients_rename(target_browser, alias):
  """Set the profile alias used to identify this browser instance."""
  try:
    _ensure_unique_browser_alias(alias, target_browser)
    send_command("clients.rename_profile", {"alias": alias}, profile=target_browser)
  except BrowserNotConnected as e:
    console.print(f"[red]Error:[/red] {e}")
    sys.exit(1)
  console.print(f"[green]Profile renamed to '{alias}'[/green]")