Function handle_single_command

backend/extensions/browser_server.py:225–438

Handle a single command from a WebSocket connection

(websocket)

    }

async def handle_single_command(websocket):
    """Handle a single command from a WebSocket connection"""
    global global_history, global_current_url, global_forms
    
    async with browser_lock:
        await initialize_browser()
        
        try:
            # Wait for the command
            command = await asyncio.wait_for(websocket.recv(), timeout=30.0)
            command = command.strip()
            response = {}
            
            logger.info(f"Processing command: {command}")
            
            # Navigate to URL
            if command.startswith(("url:", "go:", "open:")):
                url = command.split(":", 1)[1].strip()
                if not url.startswith(('http://', 'https://')):
                    url = "https://" + url
                
                try:
                    await global_page.goto(url, wait_until='domcontentloaded', timeout=15000)
                    content = await get_structured_data(global_page, global_page.url)
                    global_current_url = global_page.url
                    global_forms = content["forms"]
                    global_history.append(global_current_url)
                    response = content
                except Exception as e:
                    response = {"error": f"Failed to load {url}: {str(e)}"}
            
            # Follow link by index
            elif command.startswith("link:"):
                try:
                    idx = int(command.split(":", 1)[1].strip())
                    current_content = await get_structured_data(global_page, global_page.url)
                    if 0 <= idx < len(current_content["links"]):
                        target_url = current_content["links"][idx]["url"]
                        await global_page.goto(target_url, wait_until='domcontentloaded', timeout=15000)
                        content = await get_structured_data(global_page, global_page.url)
                        global_current_url = global_page.url
                        global_forms = content["forms"]
                        global_history.append(global_current_url)
                        response = content
                    else:
                        response = {"error": f"Link index {idx} out of range"}
                except (ValueError, IndexError) as e:
                    response = {"error": f"Invalid link command: {str(e)}"}
            
            # Search on page
            elif command.startswith("search:"):
                search_term = command.split(":", 1)[1].strip()
                html = await global_page.content()
                soup = BeautifulSoup(html, "html.parser")
                text = soup.get_text().lower()
                
                if search_term.lower() in text:
                    # Find context around the search term