""" Unified HTTP client using Playwright for all requests Provides session sharing across all tools """ import asyncio import json import time from typing import Any, Dict, List from mcp.types import TextContent from ..config import ensure_browser_session, logger class UnifiedHTTPClient: """Unified HTTP client using Playwright browser session.""" def __init__(self): self.page = None self.stealth_enabled = False self.in_use = False # Для параллельного использования self.is_parallel = False # Маркер параллельного клиента async def ensure_session(self, stealth: bool = False): """Ensure browser session is available and synchronized with authenticated_session.""" from ..config import authenticated_session, logger # Ensure page exists if not self.page or (stealth and not self.stealth_enabled): self.page = await ensure_browser_session(stealth) self.stealth_enabled = stealth # CRITICAL FIX: Synchronize cookies from authenticated_session await self._sync_cookies_from_session() return self.page async def _sync_cookies_from_session(self): """Synchronize cookies from authenticated_session to browser context.""" from ..config import authenticated_session, logger if not self.page: return # Check if authenticated session is active and has cookies if authenticated_session.get("active") and authenticated_session.get("browser_cookies"): browser_cookies = authenticated_session.get("browser_cookies", []) if browser_cookies: try: # Get current context context = self.page.context # Clear existing cookies to avoid conflicts await context.clear_cookies() # Add cookies from authenticated session playwright_cookies = [] 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"] # Skip empty cookies if cookie["name"] and cookie["value"]: playwright_cookies.append(cookie) except Exception as cookie_error: logger.debug( f"Failed to prepare cookie {cookie_data.get('name', 'unknown')}: {cookie_error}" ) continue # Add all cookies to context if playwright_cookies: await context.add_cookies(playwright_cookies) logger.debug( f"🔄 Synchronized {len(playwright_cookies)} cookies from authenticated_session to browser context" ) except Exception as e: logger.error(f"Failed to sync cookies from authenticated_session: {e}") else: logger.debug("No cookies in authenticated_session to sync") else: logger.debug("Authenticated session not active or no cookies available") async def get_with_session( self, url: str, timeout: int = 10, headers: dict[str, Any] = None, debug: bool = False, ) -> dict[str, Any]: """GET request using Playwright browser session for cookie extraction, then requests for efficiency.""" from ..config import logger # Ensure session and sync cookies BEFORE making request page = await self.ensure_session() try: # Log session synchronization logger.debug(f"🔄 Making authenticated request to {url}") if debug else None response = await page.goto(url, wait_until="domcontentloaded", timeout=timeout * 1000) # Get cookies from browser session cookies = await page.context.cookies() cookies_dict = {c["name"]: c["value"] for c in cookies} # Get final URL and page data final_url = page.url html = await page.content() title = await page.title() # Create result with session info result = { "method": "browser_session_api", "url": url, "final_url": final_url, "status_code": response.status, "content_type": response.headers.get("content-type", ""), "html": html, "size": len(html), "cookies_count": len(cookies), "authenticated": len(cookies) > 0, "page_title": title, "cookies": cookies_dict, # For API requests "response_headers": dict(response.headers) if debug else {}, } if debug: result["debug_info"] = { "browser_cookies": len(cookies), "page_title": title, "url_changed": final_url != url, } return result except Exception as e: logger.error(f"Browser session error for {url}: {e}") raise async def get_json(self, url: str, timeout: int = 10) -> dict[str, Any]: """GET request expecting JSON response.""" result = await self.get(url, timeout) if result["content_type"].startswith("application/json"): try: result["json_data"] = json.loads(result["html"]) result["html"] = "" # Clear HTML since we have JSON except: pass return result async def post( self, url: str, data: dict[str, Any] = None, headers: dict[str, Any] = None, timeout: int = 10, stealth: bool = False, ) -> dict[str, Any]: """POST request using unified browser session.""" page = await self.ensure_session(stealth) # Set headers if provided if headers: for key, value in headers.items(): await page.set_extra_http_headers({key: value}) try: # Prepare form data form_data = {} if data: form_data = data # Navigate with POST data response = await page.request.post(url, data=form_data, timeout=timeout * 1000) # Get response data html = await response.text() status = response.status # Get cookies info cookies = await page.context.cookies() cookies_count = len(cookies) return { "method": "playwright", "url": url, "status_code": status, "content_type": response.headers.get("content-type", ""), "html": html, "size": len(html), "cookies_count": cookies_count, "authenticated": cookies_count > 0, "response_headers": dict(response.headers) if timeout > 10 else {}, } except Exception as e: logger.error(f"HTTP POST error for {url}: {e}") raise async def get_cookies_info(self) -> dict[str, Any]: """Get current session cookies information.""" try: page = await self.ensure_session() cookies = await page.context.cookies() return { "cookies_count": len(cookies), "cookies": [ { "name": c["name"], "domain": c.get("domain", ""), "path": c.get("path", "/"), "secure": c.get("secure", False), "httpOnly": c.get("httpOnly", False), } for c in cookies ], } except Exception as e: logger.error(f"Error getting cookies: {e}") return {"cookies_count": 0, "cookies": []} async def clear_cookies(self): """Clear all cookies from browser context.""" try: page = await self.ensure_session() await page.context.clear_cookies() logger.info("Browser cookies cleared") except Exception as e: logger.error(f"Error clearing cookies: {e}") class ParallelBrowserManager: """Manager for multiple browser instances for parallel operations.""" def __init__(self): self.browser_pool = [] self.max_browsers = 5 # Максимум 5 браузеров одновременно self.playwright_instance = None async def initialize(self): """Initialize Playwright instance.""" if not self.playwright_instance: from playwright.async_api import async_playwright self.playwright_instance = await async_playwright().start() async def get_browser_client(self, stealth: bool = False) -> UnifiedHTTPClient: """Get or create browser client from pool.""" await self.initialize() # Найти свободный браузер for client in self.browser_pool: if not client.in_use: client.in_use = True return client # Создать новый браузер если пул не полон if len(self.browser_pool) < self.max_browsers: client = UnifiedHTTPClient() client.in_use = True client.is_parallel = True # Маркер для параллельного использования self.browser_pool.append(client) # Создать отдельный браузер для параллельного использования if not self.playwright_instance: from playwright.async_api import async_playwright self.playwright_instance = await async_playwright().start() browser = await self.playwright_instance.chromium.launch( headless=True, # Для параллельных задач используем headless args=["--no-sandbox", "--disable-dev-shm-usage"], ) context = await browser.new_context( viewport={"width": 1920, "height": 1080}, user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36", ) if stealth: await context.add_init_script( """ Object.defineProperty(navigator, 'webdriver', {get: () => undefined}); """ ) page = await context.new_page() client.page = page client.browser = browser client.context = context return client # Если все браузеры заняты, подождать с таймаутом max_wait_time = 30 # максимум 30 секунд ожидания wait_start = asyncio.get_event_loop().time() while True: await asyncio.sleep(0.1) # Проверить таймаут if asyncio.get_event_loop().time() - wait_start > max_wait_time: raise TimeoutError(f"Failed to get browser client after {max_wait_time} seconds") for client in self.browser_pool: if not client.in_use: client.in_use = True return client async def release_browser_client(self, client: UnifiedHTTPClient): """Release browser client back to pool.""" client.in_use = False async def cleanup(self): """Clean up all browser instances.""" for client in self.browser_pool: try: if hasattr(client, "browser") and client.browser: await client.browser.close() except: pass if self.playwright_instance: try: await self.playwright_instance.stop() except: pass self.browser_pool.clear() # Global instances http_client = UnifiedHTTPClient() parallel_manager = ParallelBrowserManager() async def get_http_client() -> UnifiedHTTPClient: """Get global HTTP client instance.""" return http_client async def get_parallel_browser_client(stealth: bool = False) -> UnifiedHTTPClient: """Get browser client for parallel operations.""" return await parallel_manager.get_browser_client(stealth) async def release_parallel_browser_client(client: UnifiedHTTPClient): """Release browser client after parallel operation.""" await parallel_manager.release_browser_client(client)