api-camoufox/app/use_cases/camoufox/wrapper.py

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}'
            )