import asyncio import base64 import shutil import sys import time from typing import Any, Literal, Optional, Union from urllib.parse import urlparse from camoufox.async_api import AsyncCamoufox from app.config import config from app.libs.log import log def _detect_headless_mode() -> Union[bool, Literal["virtual"]]: if sys.platform.startswith("linux") and shutil.which("Xvfb"): return "virtual" return True def _resolve_headless_mode() -> Union[bool, Literal["virtual"]]: env_mode = config.HEADLESS_MODE if env_mode == "virtual": return "virtual" if env_mode in ("true", "True", "1"): return True if env_mode in ("false", "False", "0"): return False return _detect_headless_mode() class CamoufoxWrapper: _instance: Optional["CamoufoxWrapper"] = None def __new__(cls) -> "CamoufoxWrapper": if cls._instance is None: cls._instance = super().__new__(cls) return cls._instance def _build_proxy_config( self, proxy: str, proxy_auth: Optional[dict[str, Any]] = None ) -> dict[str, Any]: proxy_url = urlparse(proxy if proxy.startswith("http") else f"http://{proxy}") config_proxy: dict[str, Any] = { "server": f"{proxy_url.scheme}://{proxy_url.hostname}:{proxy_url.port}", } if proxy_auth: if proxy_auth.get("username"): config_proxy["username"] = proxy_auth["username"] if proxy_auth.get("password"): config_proxy["password"] = proxy_auth["password"] return config_proxy async def _handle_mercadolibre(self, page, url: str, browser) -> str: """Hacks específicos para el portal de Mercado Libre.""" html = await page.content() # 1. Detectar captcha "Para continuar" if "Para continuar" in html: log.logger.debug("MeLi captcha detected. Reloading in new page...") page = await browser.new_page() page.set_default_navigation_timeout(60000) await page.goto(url, timeout=60000) html = await page.content() # 2. Espera adicional de network idle await page.wait_for_load_state("networkidle", timeout=2000) # 3. Rate limit detection (429) html = await page.content() rate_limit_pattern = '{"message":"local_rate_limited","status":429}' if rate_limit_pattern in html: log.logger.debug("Rate limit detected for MeLi. Reloading page...") await page.reload() html = await page.content() # 4. Remover banner de cookies try: await page.wait_for_selector("div.cookie-consent-banner-opt-out", timeout=1000) await page.evaluate( "() => { const banner = document.querySelector('div.cookie-consent-banner-opt-out'); if (banner) banner.remove(); }" ) except Exception: pass # 5. Click "todas las características" try: await page.wait_for_selector( "button.ui-pdp-collapsable__action.ui-vpp-highlighted-specs__striped-collapsed__action[data-testid='action-collapsable-target']", timeout=1000, ) await page.click( "button.ui-pdp-collapsable__action.ui-vpp-highlighted-specs__striped-collapsed__action[data-testid='action-collapsable-target']" ) except Exception: pass # 6. Click "toda la descripción" try: await page.wait_for_selector( "button.ui-pdp-collapsable__action.ui-pdp-description-collapse[data-testid='action-collapsable-target']", timeout=1000, ) await page.click( "button.ui-pdp-collapsable__action.ui-pdp-description-collapse[data-testid='action-collapsable-target']" ) except Exception: pass # HTML final actualizado html = await page.content() return html async def visit(self, url: str, options: dict[str, Any]) -> dict[str, Any]: start_time = time.time() headless = _resolve_headless_mode() log.logger.info(f"Using headless mode: {headless}") camoufox_options: dict[str, Any] = { "headless": headless, } proxy = options.get("proxy") if proxy: log.logger.debug(f"Using proxy: {proxy}") camoufox_options["proxy"] = self._build_proxy_config( proxy, options.get("proxyAuth") ) camoufox_options["geoip"] = True headers = options.get("headers") if headers: if "host" in headers: del headers["host"] if "content-length" in headers: del headers["content-length"] log.logger.debug(f"Using headers: {headers}") # Intentar inicializar AsyncCamoufox (con fallback si es virtual) try: browser_cls = AsyncCamoufox(**camoufox_options) except Exception as launch_exc: if headless == "virtual": log.logger.warning( f"Failed to launch with virtual display: {launch_exc}. Falling back to headless=True" ) camoufox_options["headless"] = True browser_cls = AsyncCamoufox(**camoufox_options) else: raise # Usar context manager para toda la lógica del browser async with browser_cls as browser: page = await browser.new_page() if headers: await page.set_extra_http_headers(headers) await page.goto(url, timeout=120000, wait_until=options.get("waitUntil", "load")) await page.wait_for_load_state("networkidle", timeout=120000) # Extra wait after networkidle (for SPAs that need hydration time) extra_wait = options.get("waitTime", 0) or 0 if extra_wait > 0: log.logger.info(f"Extra wait: {extra_wait}s after networkidle") await asyncio.sleep(extra_wait) # Hacks específicos para Mercado Libre is_meli = "mercadolibre.cl" in url if is_meli: log.logger.debug("Applying MeLi-specific hacks") html = await self._handle_mercadolibre(page, url, browser) else: html = await page.content() screenshot: Optional[str] = None if ( options.get("screenshot") or options.get("screenshotSelector") or options.get("clickSelector") ): await self._click_element(page, options) screenshot_bytes = await self._take_screenshot(page, options) if screenshot_bytes: screenshot = base64.b64encode(screenshot_bytes).decode("utf-8") duration = time.time() - start_time log.logger.info(f"HTML Size: {len(html) / 1000:.2f} KB from '{url}'") log.logger.info(f"Execution time: {duration:.2f}s") return {"html": html, "screenshot": screenshot} def _get_screenshot_quality(self) -> int: return config.SCREENSHOT_QUALITY async def _take_screenshot( self, page, options: dict[str, Any] ) -> Optional[bytes]: screenshot_options = { "type": "jpeg", "quality": self._get_screenshot_quality(), } selector = options.get("screenshotSelector") selector_timeout = options.get("selectorTimeout", 1000) if selector: log.logger.info(f"Taking screenshot of selector: {selector}") try: await page.wait_for_selector(selector, timeout=selector_timeout) element = await page.query_selector(selector) if element: await self._apply_screenshot_delay(options) return await element.screenshot(**screenshot_options) log.logger.warning( "Selector not found after waitForSelector, falling back to full page screenshot" ) except Exception as err: log.logger.warning( f'Selector "{selector}" not found within timeout, falling back to full page screenshot: {err}' ) await self._apply_screenshot_delay(options) return await page.screenshot(full_page=True, **screenshot_options) async def _apply_screenshot_delay(self, options: dict[str, Any]) -> None: delay = options.get("screenshotDelay", 0) or 0 if delay > 0: log.logger.info(f"Waiting {delay} ms before taking screenshot") await asyncio.sleep(delay / 1000) async def _click_element(self, page, options: dict[str, Any]) -> None: selector = options.get("clickSelector") if not selector: return log.logger.info(f"Clicking element: {selector}") try: selector_timeout = options.get("selectorTimeout", 1000) await page.wait_for_selector(selector, timeout=selector_timeout) await page.click(selector) except Exception as err: log.logger.warning( f'Click selector "{selector}" not found or not clickable: {err}' )