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)

            # 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"):
                screenshot_bytes = await page.screenshot(
                    full_page=True,
                    type="jpeg",
                    quality=config.SCREENSHOT_QUALITY,
                )
                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}