""" Navigation tools: pagination, parallel fetching, infinite scroll """ import asyncio import json from typing import Any, Dict, List from bs4 import BeautifulSoup from mcp.types import TextContent from ..config import ( PLAYWRIGHT_AVAILABLE, authenticated_session, browser_instance, logger, playwright_instance, ) from ..utils.detection import detect_js_site from .fetch import fetch_html_tool async def _setup_context_with_session(context): """Setup Playwright context with cookies from authenticated session.""" if authenticated_session.get("active"): browser_cookies = authenticated_session.get("browser_cookies", []) if browser_cookies: logger.debug( f"Setting up Playwright context with {len(browser_cookies)} cookies from browser_login" ) # Convert cookies to Playwright format and add to context for cookie_data in browser_cookies: try: cookie = { "name": cookie_data.get("name", ""), "value": cookie_data.get("value", ""), "domain": cookie_data.get("domain", ""), "path": cookie_data.get("path", "/"), "httpOnly": cookie_data.get("httpOnly", False), "secure": cookie_data.get("secure", False), } # Add expires if present if cookie_data.get("expires"): cookie["expires"] = cookie_data["expires"] await context.add_cookies([cookie]) except Exception as cookie_error: logger.debug(f"Failed to add cookie to Playwright context: {cookie_error}") continue logger.debug(f"Successfully added {len(browser_cookies)} cookies to Playwright context") async def parallel_fetch_tool(args: dict[str, Any]) -> list[TextContent]: """Fetch multiple URLs in parallel for improved performance.""" urls = args["urls"] max_workers = args.get("max_workers", 5) timeout = args.get("timeout", 10) use_browser = args.get("use_browser", False) extract_data = args.get("extract_data", {}) if not isinstance(urls, list) or len(urls) == 0: return [TextContent(type="text", text="URLs must be a non-empty list")] # Limit URLs to prevent abuse if len(urls) > 50: urls = urls[:50] logger.warning("Limited to first 50 URLs for performance") try: # Semaphore to limit concurrent requests semaphore = asyncio.Semaphore(max_workers) async def fetch_single_url(url: str) -> dict[str, Any]: """Fetch a single URL with dedicated browser instance for true parallelism.""" try: # Get dedicated browser client for this URL from ..utils.http_client import ( get_parallel_browser_client, release_parallel_browser_client, ) browser_client = await get_parallel_browser_client(stealth=False) try: # Use human-like navigation for each URL result_data = await browser_client.navigate_like_human( url, timeout, None, False, False ) # Add extraction data if specified if extract_data and "html" in result_data: html = result_data["html"] soup = BeautifulSoup(html, "lxml") extracted = {} for key, selector in extract_data.items(): try: elements = soup.select(selector) extracted[key] = [ elem.get_text(strip=True) for elem in elements[:10] ] except: extracted[key] = [] result_data["extracted_data"] = extracted # Add parallel execution info result_data["parallel_execution"] = { "dedicated_browser": True, "browser_instance": browser_client.is_parallel, "human_simulation": True, } return {"url": url, "success": True, "data": result_data} finally: # Always release browser client back to pool await release_parallel_browser_client(browser_client) except Exception as e: return {"url": url, "success": False, "error": str(e)} # Execute all fetches in parallel logger.info(f"Starting parallel fetch of {len(urls)} URLs with {max_workers} workers") start_time = asyncio.get_event_loop().time() tasks = [fetch_single_url(url) for url in urls] results = await asyncio.gather(*tasks, return_exceptions=True) end_time = asyncio.get_event_loop().time() # Process results successful_fetches = [] failed_fetches = [] for result in results: if isinstance(result, Exception): failed_fetches.append({"url": "unknown", "success": False, "error": str(result)}) elif result.get("success"): successful_fetches.append(result) else: failed_fetches.append(result) # Compile final result final_result = { "total_urls": len(urls), "successful": len(successful_fetches), "failed": len(failed_fetches), "max_workers": max_workers, "total_time": round(end_time - start_time, 2), "average_time_per_url": round((end_time - start_time) / len(urls), 2), "extract_data_selectors": extract_data, "successful_fetches": successful_fetches, "failed_fetches": failed_fetches, "status": f"Processed {len(urls)} URLs: {len(successful_fetches)} successful, {len(failed_fetches)} failed", } return [ TextContent(type="text", text=json.dumps(final_result, indent=2, ensure_ascii=False)) ] except Exception as e: return [TextContent(type="text", text=f"Parallel fetch failed: {str(e)}")] async def auto_paginate_tool(args: dict[str, Any]) -> list[TextContent]: """Automatically navigate through pagination.""" url = args["url"] max_pages = args.get("max_pages", 10) wait_between = args.get("wait_between", 2) next_selector = args.get("next_selector") extract_data = args.get("extract_data", {}) if not PLAYWRIGHT_AVAILABLE: return [ TextContent( type="text", text="Playwright not available. Install with: pip install playwright", ) ] try: # 🔧 ARCHITECTURAL FIX: Use unified browser session instead of manual creation from ..utils.http_client import get_http_client http_client = await get_http_client() page = await http_client.ensure_session() # This already includes session sync all_data = [] current_page = 1 # Navigate to first page await page.goto(url) while current_page <= max_pages: logger.info(f"Processing page {current_page}") # Extract data from current page page_data = { "page_number": current_page, "url": page.url, "extracted_data": {}, } # Extract specified data for key, selector in extract_data.items(): try: elements = await page.locator(selector).all() page_data["extracted_data"][key] = [ await elem.inner_text() for elem in elements[:50] # Limit to 50 items ] except: page_data["extracted_data"][key] = [] # Get page title and basic info page_data["title"] = await page.title() page_data["html_size"] = len(await page.content()) all_data.append(page_data) # Try to find next page button next_button = None if next_selector: next_button = page.locator(next_selector).first else: # Auto-detect next button next_selectors = [ 'a[rel="next"]', 'a:has-text("Next")', 'a:has-text("→")', 'a:has-text(">")', 'button:has-text("Next")', ".next a", ".pagination a:last-child", ] for selector in next_selectors: try: next_button = page.locator(selector).first if await next_button.count() > 0: break except: continue # Check if next button exists and is clickable if not next_button or await next_button.count() == 0: logger.info("No more pages found") break try: # Check if button is disabled is_disabled = await next_button.get_attribute("disabled") if is_disabled: break # Click next button await next_button.click() await page.wait_for_load_state("domcontentloaded") await asyncio.sleep(wait_between) current_page += 1 except Exception as e: logger.warning(f"Failed to navigate to next page: {e}") break result = { "starting_url": url, "pages_processed": len(all_data), "max_pages": max_pages, "extraction_selectors": extract_data, "pages_data": all_data, "status": f"Successfully processed {len(all_data)} pages", } await context.close() return [TextContent(type="text", text=json.dumps(result, indent=2, ensure_ascii=False))] except Exception as e: return [TextContent(type="text", text=f"Auto pagination failed: {str(e)}")] async def infinite_scroll_tool(args: dict[str, Any]) -> list[TextContent]: """Handle infinite scroll pages with automatic loading.""" url = args["url"] trigger_selector = args.get("trigger_selector") scroll_pause_time = args.get("scroll_pause_time", 2) max_scrolls = args.get("max_scrolls", 10) extract_data = args.get("extract_data", {}) stop_condition = args.get("stop_condition") if not PLAYWRIGHT_AVAILABLE: return [ TextContent( type="text", text="Playwright not available. Install with: pip install playwright", ) ] try: # 🔧 ARCHITECTURAL FIX: Use unified browser session instead of manual creation from ..utils.http_client import get_http_client http_client = await get_http_client() page = await http_client.ensure_session() # This already includes session sync # Navigate to page await page.goto(url) await page.wait_for_load_state("domcontentloaded") all_data = [] scroll_count = 0 previous_content_count = 0 while scroll_count < max_scrolls: logger.info(f"Infinite scroll iteration {scroll_count + 1}") # Extract data from current state current_data = { "scroll_iteration": scroll_count + 1, "url": page.url, "extracted_data": {}, } # Extract specified data for key, selector in extract_data.items(): try: elements = await page.locator(selector).all() current_data["extracted_data"][key] = [ await elem.inner_text() for elem in elements ] except: current_data["extracted_data"][key] = [] all_data.append(current_data) # Check stop condition if stop_condition: try: stop_elements = await page.locator(stop_condition).count() if stop_elements > 0: logger.info("Stop condition met") break except: pass # Check if content is still loading current_content_count = sum( len(data) for data in current_data["extracted_data"].values() ) if current_content_count == previous_content_count and scroll_count > 0: logger.info("No new content loaded, stopping") break previous_content_count = current_content_count # Try to trigger more content loading if trigger_selector: try: # Click load more button if exists trigger = page.locator(trigger_selector).first if await trigger.count() > 0: await trigger.click() await page.wait_for_load_state("domcontentloaded") else: # No more trigger button, stop break except: # Trigger failed, try scrolling await page.evaluate("window.scrollTo(0, document.body.scrollHeight)") else: # Auto-scroll to bottom await page.evaluate("window.scrollTo(0, document.body.scrollHeight)") # Wait for new content to load await asyncio.sleep(scroll_pause_time) scroll_count += 1 result = { "url": url, "scroll_iterations": len(all_data), "max_scrolls": max_scrolls, "trigger_selector": trigger_selector, "stop_condition": stop_condition, "total_items_extracted": sum( len(data) for iteration in all_data for data in iteration["extracted_data"].values() ), "scroll_data": all_data, "status": f"Completed {len(all_data)} scroll iterations", } await context.close() return [TextContent(type="text", text=json.dumps(result, indent=2, ensure_ascii=False))] except Exception as e: return [TextContent(type="text", text=f"Infinite scroll failed: {str(e)}")]