#pip install browser-use langchain-openai langchain-ollama playwright #playwright install ## comandi da CLI """ browser-use open https://example.com # Navigate to URL browser-use state # See clickable elements browser-use click 5 # Click element by index browser-use type "Hello" # Type text browser-use screenshot page.png # Take screenshot browser-use close # Close browser """ import asyncio import aiofiles import os import logging import sys import re from datetime import datetime from browser_use import Agent, ChatAzureOpenAI, Browser import base64, os, json from browser_use.browser import BrowserProfile, BrowserSession from browser_use.browser.profile import ViewportSize # highlight_elements goes on BrowserProfile, not Browser browser_profile = BrowserProfile( headless=False, #Visible window opens on screen highlight_elements=True, # draws colored bounding boxes on interactive elements # dom_highlight_elements=True, # alternative: DOM-based highlighting (takes priority if both set) window_size=ViewportSize(width=900, height=1100), # bigger window to compensate for OS chrome viewport=ViewportSize(width=768, height=1024), # ✅ forces exact page content size ) browser_session = BrowserSession(browser_profile=browser_profile) # ─── LOGGING SETUP ──────────────────────────────────────────────────────────── os.environ["BROWSER_USE_LOGGING_LEVEL"] = "debug" # ── BBox tracker — captures the count from the log message ────────────────── class BBoxTracker: """Intercepts the BBox filtering log line and stores the latest count.""" last_excluded: int = 0 history: list[dict] = [] def record(self, step: int, count: int): self.last_excluded = count self.history.append({"step": step, "excluded_nodes": count}) bbox_tracker = BBoxTracker() _BBOX_RE = re.compile(r"BBox filtering excluded (\d+) nodes", re.IGNORECASE) class BBoxInterceptHandler(logging.Handler): """Silently watches every log record for the BBox line.""" def emit(self, record: logging.LogRecord): msg = record.getMessage() m = _BBOX_RE.search(msg) if m: bbox_tracker.last_excluded = int(m.group(1)) # Colored formatter class ColoredFormatter(logging.Formatter): COLORS = { logging.DEBUG: "\033[36m", logging.INFO: "\033[32m", logging.WARNING: "\033[33m", logging.ERROR: "\033[31m", logging.CRITICAL: "\033[35m", } RESET = "\033[0m" BOLD = "\033[1m" def format(self, record): color = self.COLORS.get(record.levelno, self.RESET) record.levelname = f"{color}{self.BOLD}{record.levelname:<8}{self.RESET}" record.name = f"\033[90m{record.name}{self.RESET}" return super().format(record) console_handler = logging.StreamHandler(sys.stdout) console_handler.setLevel(logging.DEBUG) console_handler.setFormatter(ColoredFormatter( fmt="%(asctime)s │ %(levelname)s │ %(name)s\n └─ %(message)s\n", datefmt="%H:%M:%S", )) root_logger = logging.getLogger() root_logger.setLevel(logging.DEBUG) root_logger.handlers.clear() root_logger.addHandler(console_handler) root_logger.addHandler(BBoxInterceptHandler()) # ← silent tracker for noisy in ("httpx", "httpcore", "playwright", "urllib3", "asyncio"): logging.getLogger(noisy).setLevel(logging.WARNING) logger = logging.getLogger("agent_runner") # ─── STEP CALLBACKS ─────────────────────────────────────────────────────────── step_count = 0 step_start_time: datetime | None = None os.makedirs("./screenshots_on_step", exist_ok=True) import asyncio import aiofiles import base64 import io import os from PIL import Image, ImageDraw, ImageFont async def on_step(state, model_output, step_num): if not state.screenshot: return img = Image.open(io.BytesIO(base64.b64decode(state.screenshot))) draw = ImageDraw.Draw(img) selector_map = state.dom_state.selector_map colors = ["#FF0000", "#FF6600", "#0066FF", "#009900", "#9900CC"] for idx, node in selector_map.items(): # ✅ node IS the EnhancedDOMTreeNode directly — no .original_node rect = node.absolute_position if rect is None: continue x, y, w, h = rect.x, rect.y, rect.width, rect.height if w <= 0 or h <= 0: continue color = colors[idx % len(colors)] draw.rectangle([x, y, x + w, y + h], outline=color, width=2) label = str(idx) draw.rectangle([x, y - 14, x + len(label) * 8 + 4, y], fill=color) draw.text((x + 2, y - 13), label, fill="white") path = f"./screenshots_on_step/step_{step_num:03d}.png" buf = io.BytesIO() img.save(buf, format="PNG") async with aiofiles.open(path, "wb") as f: await f.write(buf.getvalue()) print(f"[step {step_num}] annotated: {path} ({len(selector_map)} elements) | {state.url}") async def on_step_start(agent): global step_count, step_start_time step_count += 1 step_start_time = datetime.now() bbox_tracker.last_excluded = 0 # reset for this step print(f"\n\033[1;34m{'━'*60}\033[0m") print(f"\033[1;34m 🚀 STEP {step_count} — {step_start_time.strftime('%H:%M:%S')}\033[0m") print(f"\033[1;34m{'━'*60}\033[0m") async def on_step_end(agent): global step_start_time elapsed = (datetime.now() - step_start_time).total_seconds() if step_start_time else 0 # record bbox stats for this step bbox_tracker.record(step_count, bbox_tracker.last_excluded) history = agent.history.history last = history[-1] if history else None url = "N/A" if last and last.state and hasattr(last.state, "url"): url = last.state.url actions = "N/A" if last and last.model_output and hasattr(last.model_output, "action"): acts = last.model_output.action if isinstance(acts, list): actions = ", ".join( type(a).__name__ if not isinstance(a, dict) else list(a.keys())[0] for a in acts ) else: actions = str(acts) result_text = "N/A" if last and last.result: results = last.result if isinstance(last.result, list) else [last.result] parts = [] for r in results: if hasattr(r, "extracted_content") and r.extracted_content: parts.append(f"📄 {str(r.extracted_content)[:120]}") elif hasattr(r, "error") and r.error: parts.append(f"❌ {str(r.error)[:120]}") result_text = " | ".join(parts) if parts else "✔ done" # BBox colour: green if low, yellow if medium, red if high bbox_n = bbox_tracker.last_excluded if bbox_n < 20: bbox_color = "\033[32m" elif bbox_n < 60: bbox_color = "\033[33m" else: bbox_color = "\033[31m" print(f"\033[32m ✅ STEP {step_count} COMPLETE ({elapsed:.2f}s)\033[0m") print(f"\033[90m URL :\033[0m {url}") print(f"\033[90m Actions:\033[0m {actions}") print(f"\033[90m Result :\033[0m {result_text}") print(f"\033[90m BBox :\033[0m {bbox_color}{bbox_n} nodes pruned (redundant children of links/buttons)\033[0m") print(f"\033[1;34m{'━'*60}\033[0m\n") # ✅ highlight_elements draws bounding boxes from DOM on the live browser #browser = Browser( # headless=False, # highlight_elements=True, #) # ─── MAIN ───────────────────────────────────────────────────────────────────── async def run_azure_agent(): logger.info("Initializing Azure OpenAI LLM...") llm = ChatAzureOpenAI( model="gpt-4o", azure_endpoint="https://hiis-accessibility-fonderia.cognitiveservices.azure.com/", azure_deployment="gpt-4o", api_version="2025-01-01-preview", api_key="" ) agent = Agent( #task=( # "Open the URL https://github.com/rasbt and check if the page offers " # "an alternative function to the interactive map of content in the year " # "(contribution heatmap data visualization). 'Equivalent' is to be interpreted " # "according to the WCAG Success Criterion 2.5.8 Target Size (Minimum): " # "'The function can be achieved through a different control on the same page " # "that meets this criterion' - Answer YES or NO and justify your answer." #) task=( "Open the URL https://github.com/rasbt and check if the page offers " "an 'Equivalent' function to the ”Contribution graph cells”: The dense grid of small squares in " "the '1,535 contributions in the last year' heatmap calendar." "'Equivalent' is to be interpreted according to the WCAG Success Criterion 2.5.8 Target Size (Minimum): " "'The function can be achieved through a different control on the same page " "that meets this criterion'. - Answer YES or NO and justify your answer." ), llm=llm, #on_step_start=on_step_start, #on_step_end=on_step_end, save_conversation_path="./logs/session.json",# all screenshots embedded here save_conversation_path_encoding="utf-8", # default # Optional: keep full history (don't prune steps) max_history_items=None, # None = keep ALL steps in memory use_vision=True, browser_session=browser_session, #browser # pass session, not Browser() vision_detail_level="high", # 'low', 'high', or 'auto' generate_gif="./logs/session.gif", register_new_step_callback=on_step, # real-time callback ) print(f"\n\033[1;35m{'═'*60}\033[0m") print(f"\033[1;35m 🤖 BROWSER-USE AGENT STARTING\033[0m") print(f"\033[1;35m{'═'*60}\033[0m\n") result = await agent.run(max_steps=20) #await browser.close() """ #### non va ### import os import base64 import json os.makedirs("screenshots_from_results", exist_ok=True) for i, shot in enumerate(result.screenshots()): # screenshots() returns base64 strings img_bytes = base64.b64decode(shot) path = f"screenshots_from_results/step_{i+1:02d}.png" with open(path, "wb") as f: f.write(img_bytes) print(f"Saved {path}") """ # ── BBox summary across all steps ────────────────────────────────────── print(f"\n\033[1;33m{'═'*60}\033[0m") print(f"\033[1;33m 📦 BBOX FILTERING SUMMARY\033[0m") print(f"\033[1;33m{'═'*60}\033[0m") total = sum(s["excluded_nodes"] for s in bbox_tracker.history) for s in bbox_tracker.history: bar = "█" * min(s["excluded_nodes"] // 5, 30) print(f" Step {s['step']:>2} │ {s['excluded_nodes']:>4} nodes {bar}") print(f" {'─'*40}") print(f" Total pruned across all steps: {total} nodes") print(f"\033[1;33m{'═'*60}\033[0m\n") print(f"\n\033[1;35m{'═'*60}\033[0m") print(f"\033[1;35m 🏁 FINAL RESULT\033[0m") print(f"\033[1;35m{'═'*60}\033[0m") print(result) if __name__ == "__main__": asyncio.run(run_azure_agent())