Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 47 additions & 4 deletions brow/src/brow/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -140,6 +140,19 @@ def session_delete(sid: str):
typer.echo(f"Deleted session {sid}")


@session_app.command("cleanup")
def session_cleanup():
ensure_daemon()
c = _client()
sessions = run_async(c.get("/sessions"))
if not sessions:
typer.echo("No sessions to clean up")
return
for s in sessions:
run_async(c.delete(f"/sessions/{s['id']}"))
typer.echo(f"Cleaned up {len(sessions)} session(s)")


@app.command("navigate")
def navigate(url: str, s: Optional[str] = session_opt, timeout: int = 30000):
ensure_daemon()
Expand All @@ -159,14 +172,21 @@ def wait_cmd(selector: Optional[str] = None, load: bool = False, s: Optional[str


@app.command("snapshot")
def snapshot_cmd(search: Optional[str] = None, locator: Optional[str] = None, s: Optional[str] = session_opt):
def snapshot_cmd(
search: Optional[str] = None,
locator: Optional[str] = None,
compact: bool = typer.Option(False, "--compact", help="Show only interactive elements"),
s: Optional[str] = session_opt,
):
ensure_daemon()
c = _client()
params = {}
if search:
params["search"] = search
if locator:
params["locator"] = locator
if compact:
params["compact"] = "true"
result = run_async(c.get(f"/browser/{s}/snapshot", params=params))
typer.echo(result["tree"])

Expand Down Expand Up @@ -444,10 +464,27 @@ def hover_cmd(selector: str, s: Optional[str] = session_opt, timeout: int = 3000


@app.command("scroll")
def scroll_cmd(pixels: int = 0, s: Optional[str] = session_opt):
def scroll_cmd(
pixels: int = 0,
until: Optional[str] = typer.Option(None, "--until", help="Scroll until selector is visible"),
max_attempts: int = typer.Option(10, "--max-attempts"),
s: Optional[str] = session_opt,
):
ensure_daemon()
c = _client()
run_async(c.post(f"/browser/{s}/scroll", json={"pixels": pixels}))
if until:
result = run_async(
c.post(
f"/browser/{s}/scroll-until",
json={"until": until, "pixels": pixels or 800, "max_attempts": max_attempts},
)
)
if result.get("found"):
typer.echo(f"Found after {result['attempts']} scroll(s)")
else:
typer.echo(f"Not found after {result['attempts']} scroll(s)", err=True)
else:
run_async(c.post(f"/browser/{s}/scroll", json={"pixels": pixels}))


@app.command("scroll-to")
Expand Down Expand Up @@ -547,13 +584,19 @@ def state_list():

@app.command("eval")
def eval_cmd(code: str, s: Optional[str] = session_opt, timeout: int = 30000):
import json

ensure_daemon()
c = _client()
result = run_async(c.post(f"/eval/{s}", json={"code": code, "timeout": timeout}))
if result.get("stdout"):
typer.echo(result["stdout"], nl=False)
if result.get("result") is not None:
typer.echo(result["result"])
val = result["result"]
if isinstance(val, (dict, list)):
typer.echo(json.dumps(val, indent=2, ensure_ascii=False))
else:
typer.echo(str(val))


if __name__ == "__main__":
Expand Down
39 changes: 37 additions & 2 deletions brow/src/brow/routes/browser.py
Original file line number Diff line number Diff line change
Expand Up @@ -297,6 +297,13 @@ class ScrollReq(BaseModel):
selector: Optional[str] = None


class ScrollUntilReq(BaseModel):
until: str
pixels: int = 800
max_attempts: int = 10
timeout: int = DEFAULT_TIMEOUT


class DragReq(BaseModel):
source: str
target: str
Expand Down Expand Up @@ -362,14 +369,25 @@ async def get_url(req: Request, sid: str):


@router.get("/snapshot")
async def snapshot(req: Request, sid: str, search: Optional[str] = None, locator: Optional[str] = None):
async def snapshot(
req: Request, sid: str, search: Optional[str] = None, locator: Optional[str] = None, compact: bool = False
):
session = _get_session(req, sid)
page = _get_page(session)
formatted, truncated, node_count = await _take_snapshot(page, search=search)
lines = formatted.split("\n")
large = len(lines) > 500
if (compact or large) and not search:
interactive_lines = [ln for ln in lines if "[" in ln and "]" in ln]
context_lines = [ln for ln in lines if any(k in ln for k in ("heading", "navigation", "main", "form"))]
kept = list(dict.fromkeys(interactive_lines + context_lines))[:300]
header = f"[Showing {len(kept)} of {len(lines)} lines — use --search <regex> to filter]\n"
formatted = header + "\n".join(kept)
truncated = True
resp = {"tree": formatted}
if truncated:
resp["truncated"] = True
resp["hint"] = f"Page has {node_count}+ nodes. Use search param to filter, e.g. search='Item 100'"
resp["hint"] = f"Page has {node_count}+ nodes ({len(lines)} lines). Use --search to filter."
return resp


Expand Down Expand Up @@ -642,6 +660,23 @@ async def scroll(req: Request, sid: str, body: ScrollReq):
return {"ok": True}


@router.post("/scroll-until")
async def scroll_until(req: Request, sid: str, body: ScrollUntilReq):
session = _get_session(req, sid)
page = _get_page(session)
for attempt in range(body.max_attempts):
try:
loc = page.locator(body.until)
if await loc.count() > 0 and await loc.first.is_visible():
await loc.first.scroll_into_view_if_needed()
return {"ok": True, "found": True, "attempts": attempt + 1}
except Exception:
pass
await page.evaluate(f"window.scrollBy(0, {body.pixels})")
await page.wait_for_timeout(500)
return {"ok": True, "found": False, "attempts": body.max_attempts}


@router.post("/drag")
async def drag(req: Request, sid: str, body: DragReq):
session = _get_session(req, sid)
Expand Down