pryscraper/tls_fingerprint.py

"""Pry — TLS Fingerprint Impersonation using curl_cffi.
This is the SINGLE BIGGEST BOTTLE-NECK in modern scraping. Sites like Cloudflare,
Akamai, DataDome, and PerimeterX use TLS fingerprinting (JA3/JA4) to detect bots.
curl_cffi can impersonate real browser TLS handshakes, bypassing these checks."""

import asyncio
import logging
import time
from typing import Any

logger = logging.getLogger(__name__)

# Check if curl_cffi is available
try:
    from curl_cffi.requests import AsyncSession
    _has_curl_cffi = True
except ImportError:
    _has_curl_cffi = False

# Browser fingerprints that curl_cffi can impersonate
BROWSER_FINGERPRINTS = [
    "chrome", "chrome99", "chrome100", "chrome101", "chrome104", "chrome107", "chrome110",
    "chrome116", "chrome119", "chrome120", "chrome123", "chrome124", "chrome131", "chrome133",
    "chrome99_android", "chrome131_android",
    "edge", "edge99", "edge101", "edge122", "edge127",
    "firefox", "firefox109", "firefox117", "firefox120", "firefox123", "firefox124", "firefox132",
    "safari", "safari15_3", "safari15_5", "safari16_5", "safari17_0", "safari17_2_ios",
    "tor",
]


class TLSScraper:
    """Scraper that uses curl_cffi to impersonate real browser TLS fingerprints.
    Bypasses JA3/JA4 fingerprinting that blocks 80%+ of bot traffic."""

    def __init__(self, default_impersonate: str = "chrome131"):
        self.default_impersonate = (
            default_impersonate if default_impersonate in BROWSER_FINGERPRINTS else "chrome"
        )
        if not _has_curl_cffi:
            logger.warning(
                "curl_cffi_not_installed",
                extra={"hint": "pip install curl_cffi"},
            )

    async def fetch(
        self,
        url: str,
        impersonate: str = "",
        headers: dict[str, str] | None = None,
        proxy: str = "",
        timeout: int = 30,
        cookies: dict[str, str] | None = None,
        follow_redirects: bool = True,
    ) -> dict[str, Any]:
        """Fetch a URL with TLS fingerprint impersonation.

        Args:
            url: The URL to fetch
            impersonate: Browser fingerprint to impersonate (e.g., "chrome131", "firefox132", "safari17_0")
            headers: Additional HTTP headers
            proxy: Proxy URL (e.g., "http://user:pass@proxy.example.com:8080")
            timeout: Request timeout in seconds
            cookies: Cookies to send with the request
            follow_redirects: Whether to follow HTTP redirects

        Returns: {status_code, text, headers, cookies, url, elapsed}
        """
        if not _has_curl_cffi:
            return {
                "success": False,
                "error": "curl_cffi not installed. Run: pip install curl_cffi",
            }

        impersonate = impersonate or self.default_impersonate
        if impersonate not in BROWSER_FINGERPRINTS:
            impersonate = "chrome"

        default_headers = {
            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
            "Accept-Language": "en-US,en;q=0.9",
            "Accept-Encoding": "gzip, deflate, br",
            "Connection": "keep-alive",
            "Upgrade-Insecure-Requests": "1",
        }
        if headers:
            default_headers.update(headers)

        try:
            start = time.time()
            async with AsyncSession(
                impersonate=impersonate,
                headers=default_headers,
                cookies=cookies or {},
                proxy=proxy or None,
                timeout=timeout,
                follow_redirects=follow_redirects,
            ) as session:
                resp = await session.get(url)
                elapsed = time.time() - start
                return {
                    "success": True,
                    "status_code": resp.status_code,
                    "text": resp.text,
                    "content": resp.content,
                    "headers": dict(resp.headers),
                    "cookies": dict(resp.cookies),
                    "url": str(resp.url),
                    "elapsed": round(elapsed, 2),
                    "impersonate": impersonate,
                }
        except Exception as e:
            return {
                "success": False,
                "error": str(e)[:300],
                "impersonate": impersonate,
            }

    async def fetch_with_rotation(
        self,
        url: str,
        fingerprints: list[str] | None = None,
        headers: dict[str, str] | None = None,
        proxy: str = "",
        timeout: int = 30,
    ) -> dict[str, Any]:
        """Try multiple browser fingerprints until one succeeds (anti-fingerprint rotation)."""
        fingerprints = fingerprints or BROWSER_FINGERPRINTS[:10]
        for fp in fingerprints:
            result = await self.fetch(
                url, impersonate=fp, headers=headers, proxy=proxy, timeout=timeout
            )
            if result.get("success") and result.get("status_code", 0) < 400:
                result["matched_fingerprint"] = fp
                return result
        return {
            "success": False,
            "error": "All fingerprints failed",
            "attempts": len(fingerprints),
        }

    def is_available(self) -> bool:
        return _has_curl_cffi


async def _test_fetch() -> None:
    """Smoke test: verifies the scraper can be instantiated and is_available works."""
    s = TLSScraper()
    logger.info("tls_scraper_available", extra={"available": s.is_available()})


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(_test_fetch())