USCardForum MCP Server

"""Cloudflare bypass utilities.

Provides functions and constants for bypassing Cloudflare protection
using multiple strategies in order:
1. Playwright with stealth (real browser - most effective)
2. curl_cffi (real browser TLS fingerprints)
3. cloudscraper (multiple browser profiles)
"""
from __future__ import annotations

import logging
import subprocess
import time
from typing import Any

import cloudscraper
import requests

logger = logging.getLogger(__name__)

# Track if we've already tried to install Playwright browsers
_playwright_browsers_installed = False

# Cloudflare-related status codes that might be worth retrying
CLOUDFLARE_RETRY_CODES = {403, 429, 503, 520, 521, 522, 523, 524}

# Browser profiles for cloudscraper
BROWSER_PROFILES = [
    {"browser": "chrome", "platform": "linux", "desktop": True},
    {"browser": "chrome", "platform": "windows", "desktop": True},
    {"browser": "chrome", "platform": "darwin", "desktop": True},
    {"browser": "firefox", "platform": "linux", "desktop": True},
    {"browser": "firefox", "platform": "windows", "desktop": True},
]

# curl_cffi browser impersonation profiles (TLS fingerprints)
CURL_CFFI_IMPERSONATES = [
    "chrome120",
    "chrome119",
    "chrome110",
    "chrome107",
    "chrome104",
    "edge101",
    "safari15_5",
]

# Common headers that make requests look more like a real browser
BROWSER_HEADERS = {
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8",
    "Accept-Language": "en-US,en;q=0.9",
    "Accept-Encoding": "gzip, deflate, br",
    "Connection": "keep-alive",
    "Upgrade-Insecure-Requests": "1",
    "Sec-Fetch-Dest": "document",
    "Sec-Fetch-Mode": "navigate",
    "Sec-Fetch-Site": "none",
    "Sec-Fetch-User": "?1",
    "Cache-Control": "max-age=0",
}


def _ensure_playwright_browsers() -> bool:
    """Ensure Playwright browsers are installed.
    
    Attempts to install Chromium if not already installed.
    Only tries once per process to avoid repeated installation attempts.
    
    Returns:
        True if browsers are available, False otherwise
    """
    global _playwright_browsers_installed
    
    if _playwright_browsers_installed:
        return True
    
    try:
        from playwright.sync_api import sync_playwright
    except ImportError:
        logger.warning("Playwright package not installed")
        return False
    
    # Try to launch browser to check if it's installed
    try:
        with sync_playwright() as p:
            browser = p.chromium.launch(headless=True)
            browser.close()
            _playwright_browsers_installed = True
            logger.info("Playwright browsers already installed")
            return True
    except Exception as e:
        logger.info(f"Playwright browsers not installed, attempting installation: {e}")
    
    # Try to install browsers
    try:
        logger.info("Installing Playwright Chromium browser...")
        result = subprocess.run(
            ["playwright", "install", "chromium"],
            capture_output=True,
            text=True,
            timeout=300,  # 5 minute timeout
        )
        if result.returncode == 0:
            logger.info("Playwright Chromium installed successfully")
            _playwright_browsers_installed = True
            return True
        else:
            logger.warning(f"Playwright install failed: {result.stderr}")
            return False
    except subprocess.TimeoutExpired:
        logger.warning("Playwright installation timed out")
        return False
    except FileNotFoundError:
        logger.warning("playwright command not found")
        return False
    except Exception as e:
        logger.warning(f"Failed to install Playwright browsers: {e}")
        return False


def create_cloudflare_session(
    delay: int = 3,
    browser: str = "chrome",
    platform: str = "linux",
) -> requests.Session:
    """Create a cloudscraper session configured for Cloudflare bypass.

    Args:
        delay: Delay in seconds for challenge solving (default: 3)
        browser: Browser to emulate (chrome, firefox)
        platform: Platform to emulate (linux, windows, darwin)

    Returns:
        A cloudscraper session ready for Cloudflare-protected sites
    """
    session = cloudscraper.create_scraper(
        browser={"browser": browser, "platform": platform, "desktop": True},
        delay=delay,
    )
    session.headers.update(BROWSER_HEADERS)
    return session


class CurlCffiSessionWrapper:
    """Wrapper to make curl_cffi.Session behave like requests.Session.
    
    This allows curl_cffi to be used as a drop-in replacement.
    """
    
    def __init__(self, session: Any = None, impersonate: str = "chrome120"):
        """Initialize wrapper with existing session or create new one.
        
        Args:
            session: Existing curl_cffi Session to wrap (reuses cookies)
            impersonate: Browser to impersonate if creating new session
        """
        if session is not None:
            self._session = session
        else:
            from curl_cffi.requests import Session
            self._session = Session(impersonate=impersonate)
        self._impersonate = impersonate
        self.headers = dict(BROWSER_HEADERS)
        self.cookies = self._session.cookies
    
    def get(self, url: str, **kwargs) -> Any:
        """Make GET request."""
        kwargs.setdefault("timeout", 15)
        # Merge headers
        headers = dict(self.headers)
        if "headers" in kwargs:
            headers.update(kwargs["headers"])
        kwargs["headers"] = headers
        return self._session.get(url, **kwargs)
    
    def post(self, url: str, **kwargs) -> Any:
        """Make POST request."""
        kwargs.setdefault("timeout", 15)
        headers = dict(self.headers)
        if "headers" in kwargs:
            headers.update(kwargs["headers"])
        kwargs["headers"] = headers
        return self._session.post(url, **kwargs)
    
    def request(self, method: str, url: str, **kwargs) -> Any:
        """Make any HTTP request."""
        kwargs.setdefault("timeout", 15)
        headers = dict(self.headers)
        if "headers" in kwargs:
            headers.update(kwargs["headers"])
        kwargs["headers"] = headers
        return self._session.request(method, url, **kwargs)


def _create_session_with_curl_cffi(
    base_url: str,
    timeout_seconds: float = 15.0,
) -> Any | None:
    """Create a session using curl_cffi with browser TLS fingerprints.
    
    curl_cffi impersonates real browser TLS fingerprints, which is very
    effective against Cloudflare's TLS fingerprinting.
    
    Args:
        base_url: The base URL to test against
        timeout_seconds: Timeout for test requests
        
    Returns:
        A curl_cffi session wrapper (reusing the session that passed), or None if failed
    """
    try:
        from curl_cffi.requests import Session
    except ImportError:
        logger.warning("curl_cffi not available, skipping TLS fingerprint bypass")
        return None
    
    base_url = base_url.rstrip("/")
    
    for impersonate in CURL_CFFI_IMPERSONATES:
        try:
            session = Session(impersonate=impersonate)
            
            # Test if this impersonation works
            test_resp = session.get(
                f"{base_url}/",
                timeout=timeout_seconds,
                allow_redirects=True,
            )
            
            if test_resp.status_code == 200:
                logger.info(f"Cloudflare bypass successful with curl_cffi impersonate={impersonate}")
                # IMPORTANT: Return wrapper with the SAME session that passed (keeps cookies!)
                wrapper = CurlCffiSessionWrapper(session=session, impersonate=impersonate)
                return wrapper
            elif test_resp.status_code == 403:
                logger.warning(f"curl_cffi {impersonate} got 403, trying next...")
                continue
        except Exception as e:
            logger.warning(f"curl_cffi {impersonate} failed: {e}, trying next...")
            continue
    
    logger.warning("All curl_cffi impersonations failed")
    return None


def _create_session_with_playwright(
    base_url: str,
    timeout_seconds: float = 30.0,
    headless: bool = True,
) -> requests.Session | None:
    """Create a requests session using cookies obtained from Playwright with stealth.

    Uses a real browser to solve Cloudflare challenges and transfers cookies
    to a requests session. Will attempt to install browsers if not present.

    Args:
        base_url: The base URL to navigate to
        timeout_seconds: Timeout for page load
        headless: Run browser in headless mode (default: True)

    Returns:
        A requests.Session with Cloudflare cookies, or None if failed
    """
    # Ensure browsers are installed (this also verifies imports work)
    if not _ensure_playwright_browsers():
        logger.warning("Playwright browsers not available")
        return None

    base_url = base_url.rstrip("/")
    session = requests.Session()
    session.headers.update(BROWSER_HEADERS)

    try:
        # Import here after ensuring browsers are installed
        from playwright.sync_api import sync_playwright
        from playwright_stealth import Stealth
        
        stealth = Stealth()
        
        with sync_playwright() as p:
            # Launch browser with stealth settings
            browser = p.chromium.launch(
                headless=headless,
                args=[
                    "--disable-blink-features=AutomationControlled",
                    "--disable-dev-shm-usage",
                    "--no-sandbox",
                    "--disable-setuid-sandbox",
                    "--disable-infobars",
                    "--window-size=1920,1080",
                    "--start-maximized",
                ],
            )

            # Create context with realistic viewport and user agent
            context = 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/120.0.0.0 Safari/537.36",
                locale="en-US",
                timezone_id="America/New_York",
            )

            page = context.new_page()

            # Apply stealth to avoid detection
            stealth.apply_stealth_sync(page)

            # Navigate to the site and wait for Cloudflare challenge to complete
            logger.info(f"Playwright: Navigating to {base_url}")
            page.goto(base_url, wait_until="networkidle", timeout=timeout_seconds * 1000)

            # Wait additional time for any JS challenges
            page.wait_for_timeout(3000)

            # Check if we got past Cloudflare
            content = page.content().lower()
            if "cloudflare" in content and ("challenge" in content or "checking" in content):
                # Wait longer for challenge to complete
                logger.info("Playwright: Waiting for Cloudflare challenge...")
                page.wait_for_timeout(5000)

            # Get cookies from browser
            cookies = context.cookies()
            logger.info(f"Playwright: Got {len(cookies)} cookies")

            # Transfer cookies to requests session
            for cookie in cookies:
                session.cookies.set(
                    cookie["name"],
                    cookie["value"],
                    domain=cookie.get("domain", ""),
                    path=cookie.get("path", "/"),
                )

            # Get the user agent used
            user_agent = page.evaluate("navigator.userAgent")
            session.headers["User-Agent"] = user_agent

            browser.close()

            # Test if session works
            test_resp = session.get(
                f"{base_url}/",
                timeout=timeout_seconds,
                allow_redirects=True,
            )

            if test_resp.status_code == 200:
                logger.info("Playwright: Session created successfully")
                return session
            else:
                logger.warning(f"Playwright: Test request got status {test_resp.status_code}")
                return None

    except ImportError as e:
        logger.warning(f"Playwright import failed: {e}")
        return None
    except Exception as e:
        logger.error(f"Playwright fallback failed: {e}")
        return None


def create_cloudflare_session_with_fallback(
    base_url: str,
    timeout_seconds: float = 15.0,
    use_playwright: bool = True,
    use_curl_cffi: bool = True,
    use_cloudscraper: bool = True,
) -> Any:
    """Create a session with Cloudflare bypass, trying multiple strategies.

    Order of attempts (Playwright first as most effective):
    1. Playwright with stealth (real browser - best success rate)
    2. curl_cffi with real browser TLS fingerprints
    3. cloudscraper with different browser profiles

    Args:
        base_url: The base URL to test against
        timeout_seconds: Timeout for test requests
        use_playwright: Try Playwright first (default: True)
        use_curl_cffi: Try curl_cffi (default: True)
        use_cloudscraper: Try cloudscraper (default: True)

    Returns:
        A session configured to bypass Cloudflare
    """
    base_url = base_url.rstrip("/")

    # Strategy 1: Try Playwright with stealth (most effective)
    if use_playwright:
        logger.info("Trying Playwright with stealth (most effective)...")
        playwright_session = _create_session_with_playwright(
            base_url,
            timeout_seconds=30.0,
            headless=True,
        )
        if playwright_session is not None:
            return playwright_session
        logger.warning("Playwright failed, trying next strategy...")

    # Strategy 2: Try curl_cffi with TLS fingerprinting
    if use_curl_cffi:
        logger.info("Trying curl_cffi with TLS fingerprints...")
        curl_session = _create_session_with_curl_cffi(base_url, timeout_seconds)
        if curl_session is not None:
            return curl_session
        logger.warning("curl_cffi failed, trying next strategy...")

    # Strategy 3: Try cloudscraper with different profiles
    if use_cloudscraper:
        logger.info("Trying cloudscraper profiles...")
        for profile in BROWSER_PROFILES:
            try:
                session = cloudscraper.create_scraper(
                    browser=profile,
                    delay=3,
                )
                session.headers.update(BROWSER_HEADERS)

                test_resp = session.get(
                    f"{base_url}/",
                    timeout=timeout_seconds,
                    allow_redirects=True,
                )
                if test_resp.status_code == 200:
                    logger.info(f"Cloudflare bypass successful with cloudscraper profile: {profile}")
                    return session
                elif test_resp.status_code == 403:
                    logger.warning(f"Cloudscraper profile {profile} got 403, trying next...")
                    continue
            except Exception as e:
                logger.warning(f"Cloudscraper profile {profile} failed: {e}, trying next...")
                continue

    # Ultimate fallback: return cloudscraper session even if it didn't work
    logger.warning("All bypass methods failed, using basic cloudscraper as fallback")
    session = cloudscraper.create_scraper(
        browser={"browser": "chrome", "platform": "linux", "desktop": True},
        delay=5,
    )
    session.headers.update(BROWSER_HEADERS)
    return session


def warm_up_session(
    session: Any,
    base_url: str,
    timeout_seconds: float = 15.0,
    with_delay: bool = True,
) -> bool:
    """Warm up a session to obtain Cloudflare cookies.

    Args:
        session: The session to warm up
        base_url: Base URL of the site
        timeout_seconds: Request timeout
        with_delay: Add delays between requests

    Returns:
        True if at least one warm-up request succeeded
    """
    base_url = base_url.rstrip("/")
    warmup_urls = [
        f"{base_url}/",
        f"{base_url}/about",
    ]

    success = False
    for i, url in enumerate(warmup_urls):
        try:
            resp = session.get(
                url,
                timeout=timeout_seconds,
                allow_redirects=True,
            )
            if resp.status_code == 200:
                success = True
                logger.debug(f"Warm-up successful: {url}")
            else:
                logger.warning(f"Warm-up got status {resp.status_code}: {url}")

            # Add delay between requests
            if with_delay and i < len(warmup_urls) - 1:
                time.sleep(0.5)
        except Exception as e:
            logger.warning(f"Warm-up failed for {url}: {e}")

    return success


def extended_warm_up(
    session: Any,
    base_url: str,
    timeout_seconds: float = 15.0,
) -> None:
    """Extended warm-up with delays and multiple page visits.

    More aggressive than basic warm-up to help with Cloudflare challenges.

    Args:
        session: The session to warm up
        base_url: Base URL of the site
        timeout_seconds: Request timeout
    """
    base_url = base_url.rstrip("/")
    warmup_urls = [
        f"{base_url}/",
        f"{base_url}/about",
        f"{base_url}/categories",
        f"{base_url}/top",
    ]

    for i, url in enumerate(warmup_urls):
        try:
            resp = session.get(
                url,
                timeout=timeout_seconds,
                allow_redirects=True,
            )
            if resp.status_code == 200:
                logger.debug(f"Warm-up successful for {url}")
            else:
                logger.warning(f"Warm-up got status {resp.status_code} for {url}")

            # Add delay between requests to avoid rate limiting
            if i < len(warmup_urls) - 1:
                time.sleep(0.5)
        except Exception as e:
            logger.warning(f"Warm-up failed for {url}: {e}")


def is_cloudflare_challenge(response: Any) -> bool:
    """Check if a response is a Cloudflare challenge page.

    Args:
        response: The response to check

    Returns:
        True if this looks like a Cloudflare challenge
    """
    if response.status_code != 200:
        return False

    content_type = response.headers.get("Content-Type", "")
    if "text/html" not in content_type:
        return False

    content_lower = response.text[:1000].lower()
    return "cloudflare" in content_lower or "challenge" in content_lower


def is_cloudflare_error(status_code: int) -> bool:
    """Check if a status code is a Cloudflare-related error.

    Args:
        status_code: HTTP status code

    Returns:
        True if this is a Cloudflare-related error code
    """
    return status_code in CLOUDFLARE_RETRY_CODES