browser-cli/browser_cli/sdk/tabs.py

"""Tabs namespace: ``b.tabs.*``."""
from __future__ import annotations

from collections.abc import Callable, Iterable

from browser_cli.models import BrowserCounts, Tab
from browser_cli.sdk.base import Namespace

# Keep SDK-driven bulk closes comfortably below the native-host response
# timeout. The extension can close larger batches, but real browsers may take
# much longer when hundreds of visible tabs are involved.
BULK_CLOSE_CHUNK_SIZE = 50

class TabsNS(Namespace):
  """List, open, close, move, and inspect browser tabs."""

  def 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.
    """
    return self.multi_list("tabs.list", {}, self.tab_from_target)

  def open(
    self,
    url: str,
    *,
    wait: bool = False,
    timeout: float = 30.0,
    background: bool = False,
    focus: bool = False,
    window: str | None = None,
    group: str | None = None,
  ) -> Tab:
    """Open *url* in a new tab and return a bound :class:`Tab`.

    Set ``wait=True`` to block until the page reaches ``readyState=complete``.
    Pass ``focus=True`` to explicitly bring the created tab/window forward.
    """
    if wait:
      return self._c.nav.open_wait(url, timeout=timeout, background=background, focus=focus, window=window, group=group)
    return self.require_tab(
      self.command("navigate.open", {"url": url, "background": background or not focus, "focus": focus, "window": window, "group": group}),
      "navigate.open returned unexpected data",
    )

  def get(self, tab_id: int) -> Tab:
    """Return a specific tab by ID."""
    return self.status(tab_id)

  def active(self) -> Tab:
    """Return the active tab."""
    return self.status()

  def query(self, search: str) -> list[Tab]:
    """Search tabs by URL or title."""
    return [self.tab_from(t) for t in (self.command("tabs.query", {"search": search}) or [])]

  def first(self, search: str) -> Tab | None:
    """Return the first tab matching *search*, or ``None``."""
    matches = self.query(search)
    return matches[0] if matches else None

  def close(
    self,
    tab_id: int | None = None,
    *,
    tab_ids: Iterable[int | Tab] | None = None,
    inactive: bool = False,
    duplicates: bool = False,
    gentle_mode: str = "auto",
  ) -> int:
    """Close tab(s). Returns the number of tabs closed.

    Pass ``tab_ids`` to close many tabs in a single round-trip. Accepts tab
    IDs or :class:`Tab` objects. ``gentle_mode`` (auto/normal/gentle/ultra)
    controls throttling of large close operations.
    """
    ids = None
    if tab_ids is not None:
      ids = [t.id if isinstance(t, Tab) else t for t in tab_ids]
    if ids is not None and len(ids) > BULK_CLOSE_CHUNK_SIZE and not inactive and not duplicates and tab_id is None:
      closed = 0
      for start in range(0, len(ids), BULK_CLOSE_CHUNK_SIZE):
        chunk = ids[start:start + BULK_CLOSE_CHUNK_SIZE]
        result = self.command("tabs.close", {
          "tabId": None,
          "tabIds": chunk,
          "inactive": False,
          "duplicates": False,
          "gentleMode": gentle_mode,
        })
        closed += self.field(result, "closed", len(chunk))
      return closed

    result = self.command("tabs.close", {
      "tabId": tab_id,
      "tabIds": ids,
      "inactive": inactive,
      "duplicates": duplicates,
      "gentleMode": gentle_mode,
    })
    return self.field(result, "closed", 1)

  def close_inactive(self) -> int:
    """Close all inactive tabs. Returns count closed."""
    return self.field(self.command("tabs.close", {"inactive": True}), "closed", 0)

  def close_duplicates(self) -> int:
    """Close duplicate tabs. Returns count closed."""
    return self.field(self.command("tabs.close", {"duplicates": True}), "closed", 0)

  def 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.command("tabs.move", {
      "tabId": tab_id, "forward": forward, "backward": backward,
      "groupId": group_id, "windowId": window_id, "index": index,
    })

  def activate(self, tab_id: int) -> None:
    """Switch browser focus to a tab by ID."""
    self.command("tabs.active", {"tabId": tab_id})

  def status(self, tab_id: int | None = None) -> Tab:
    """Return status for the active tab or a specific tab."""
    return self.require_tab(self.command("tabs.status", {"tabId": tab_id}), "No tab status returned")

  def mute(self, tab_id: int | None = None) -> int:
    """Mute the active tab or a specific tab. Returns the target tab ID."""
    return self.toggle_tab("tabs.mute", tab_id)

  def unmute(self, tab_id: int | None = None) -> int:
    """Unmute the active tab or a specific tab. Returns the target tab ID."""
    return self.toggle_tab("tabs.unmute", tab_id)

  def pin(self, tab_id: int | None = None) -> int:
    """Pin the active tab or a specific tab. Returns the target tab ID."""
    return self.toggle_tab("tabs.pin", tab_id)

  def unpin(self, tab_id: int | None = None) -> int:
    """Unpin the active tab or a specific tab. Returns the target tab ID."""
    return self.toggle_tab("tabs.unpin", tab_id)

  def 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."""
    return self.require_tab(
      self.command("tabs.watch_url", {"pattern": pattern, "tabId": tab_id, "timeout": int(timeout * 1000)}),
      "tabs.watch_url returned unexpected 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"``.
    """
    return self.require_tab(
      self.command("navigate.wait", {
        "tabId": tab_id,
        "timeout": int(timeout * 1000),
        "readyState": ready_state,
      }),
      "navigate.wait returned unexpected data",
    )

  def 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.command("tabs.screenshot", {"tabId": tab_id, "format": format, "quality": quality})
    return self.field(result, "dataUrl", "", fallback=str(result))

  def active_in_window(self, window_id: int) -> Tab:
    """Return active tab for a specific browser window."""
    return self.require_tab(
      self.command("tabs.active_in_window", {"windowId": window_id}),
      f"No active tab found for window {window_id}",
    )

  def filter(
    self,
    pattern_or_filter: str | Callable[[Tab], bool] | Callable[[list[Tab]], Iterable[Tab]],
  ) -> list[Tab]:
    """Return tabs filtered by URL pattern or a Python callable."""
    if isinstance(pattern_or_filter, str):
      return [self.tab_from(t) for t in (self.command("tabs.filter", {"pattern": pattern_or_filter}) or [])]
    return self.apply_tab_filter(pattern_or_filter)

  def count(self, pattern: str | None = None) -> "int | BrowserCounts":
    """Count open tabs, optionally filtered by URL pattern.

    Returns ``BrowserCounts`` in implicit multi-browser mode.
    """
    return self.multi_count("tabs.count", {"pattern": pattern})

  def html(self, tab_id: int | None = None) -> str:
    """Return the full HTML source of the active (or specified) tab."""
    return self.command("tabs.html", {"tabId": tab_id}) or ""

  def dedupe(self, *, gentle_mode: str = "auto") -> int:
    """Close duplicate tabs (keep the first occurrence of each URL). Returns count closed."""
    return self.field(self.command("tabs.dedupe", {"gentleMode": gentle_mode}), "closed", 0)

  def sort(self, by: str = "domain", *, gentle_mode: str = "auto") -> None:
    """Sort tabs within each window. *by* is one of 'domain', 'title', 'time'."""
    self.command("tabs.sort", {"by": by, "gentleMode": gentle_mode})

  def merge_windows(self, *, gentle_mode: str = "auto") -> int:
    """Move all tabs into the focused window. Returns count moved."""
    return self.field(self.command("tabs.merge_windows", {"gentleMode": gentle_mode}), "moved", 0)