daniel156161
076914e5b7
- Split client, native, remote, serve, markdown, and SDK internals into focused packages with direct imports. - Move local and remote transport framing/protocol helpers behind clearer module boundaries. - Break up the extension injected DOM logic into a separate content dispatch bundle and dedicated content modules. - Add explicit client handling for passive remote discovery without noisy PQ warnings. - Keep behavior covered with updated unit, integration, and extension tests.
96 lines
2.9 KiB
Python
96 lines
2.9 KiB
Python
"""Browser target discovery and local socket resolution."""
|
|
import os
|
|
import socket
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from browser_cli.endpoints import display_browser_name
|
|
from browser_cli.errors import BrowserNotConnected
|
|
from browser_cli.platform import endpoint_for_alias, is_windows, registry_path
|
|
from browser_cli.registry import load_registry
|
|
|
|
REGISTRY_PATH = registry_path()
|
|
|
|
@dataclass(frozen=True)
|
|
class BrowserTarget:
|
|
profile: str
|
|
display_name: str
|
|
socket_path: str
|
|
remote: str | None = None
|
|
|
|
def is_reachable_unix_endpoint(endpoint: str) -> bool:
|
|
"""Return True when a Unix socket path exists and accepts connections."""
|
|
path = Path(endpoint)
|
|
if not path.exists():
|
|
return False
|
|
try:
|
|
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
|
|
sock.settimeout(0.2)
|
|
sock.connect(endpoint)
|
|
return True
|
|
except OSError:
|
|
return False
|
|
|
|
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 is_reachable_unix_endpoint(v)}
|
|
|
|
def active_local_browser_targets() -> list[BrowserTarget]:
|
|
if not REGISTRY_PATH.exists():
|
|
return []
|
|
reg = load_registry(REGISTRY_PATH)
|
|
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 is_active_local_profile(profile: str | None) -> bool:
|
|
"""Return True when profile names a reachable local browser endpoint."""
|
|
if not profile:
|
|
return False
|
|
if REGISTRY_PATH.exists():
|
|
reg = load_registry(REGISTRY_PATH)
|
|
if profile in active_endpoints(reg):
|
|
return True
|
|
if not is_windows():
|
|
try:
|
|
return Path(endpoint_for_alias(profile)).exists()
|
|
except Exception:
|
|
return False
|
|
return False
|
|
|
|
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():
|
|
reg = load_registry(REGISTRY_PATH)
|
|
if target in reg:
|
|
return reg[target]
|
|
return endpoint_for_alias(target)
|
|
|
|
try:
|
|
active = active_local_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."
|
|
)
|