diff --git a/brow/src/brow/cli.py b/brow/src/brow/cli.py index e7749ea..dd8377a 100644 --- a/brow/src/brow/cli.py +++ b/brow/src/brow/cli.py @@ -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() @@ -159,7 +172,12 @@ 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 = {} @@ -167,6 +185,8 @@ def snapshot_cmd(search: Optional[str] = None, locator: Optional[str] = None, s: 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"]) @@ -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") @@ -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__": diff --git a/brow/src/brow/routes/browser.py b/brow/src/brow/routes/browser.py index 8d929a9..7199592 100644 --- a/brow/src/brow/routes/browser.py +++ b/brow/src/brow/routes/browser.py @@ -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 @@ -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 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 @@ -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)