# curl_cffi ships with pre-compiled libcurl-impersonate binaries — no build dependencies needed
pip install curl-cffi beautifulsoup4 lxml tenacity

"""
Amazon Product Data Scraper (curl_cffi edition)
Tested 2026 — bypasses TLS/JA4 fingerprint detection
Requirements: pip install curl-cffi beautifulsoup4 lxml tenacity
"""

import json
import random
import time
import logging
from dataclasses import dataclass, field
from typing import Optional

from curl_cffi import requests as cffi_requests
from bs4 import BeautifulSoup
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

# ── Logging ───────────────────────────────────────────────────────────────────
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger(__name__)


# ── Data Model ────────────────────────────────────────────────────────────────
@dataclass
class AmazonProduct:
    asin: str
    title: Optional[str] = None
    brand: Optional[str] = None
    price: Optional[str] = None
    price_whole: Optional[str] = None
    list_price: Optional[str] = None
    rating: Optional[str] = None
    review_count: Optional[str] = None
    bullet_points: list = field(default_factory=list)
    bsr: list = field(default_factory=list)
    main_image: Optional[str] = None
    marketplace: str = "www.amazon.com"
    url: Optional[str] = None
    is_captcha: bool = False
    scraped_at: Optional[str] = None


# ── Request Header Templates ──────────────────────────────────────────────────
# curl_cffi's impersonate parameter handles TLS automatically.
# We still control HTTP-layer headers to match real browser behavior.
HEADERS_TEMPLATES = [
    {
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
        "Accept-Language": "en-US,en;q=0.9",
        "Accept-Encoding": "gzip, deflate, br",
        "Cache-Control": "max-age=0",
        "Sec-Fetch-Dest": "document",
        "Sec-Fetch-Mode": "navigate",
        "Sec-Fetch-Site": "none",
        "Sec-Fetch-User": "?1",
        "Upgrade-Insecure-Requests": "1",
    },
    {
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
        "Accept-Language": "en-GB,en;q=0.5",
        "Accept-Encoding": "gzip, deflate, br",
        "DNT": "1",
        "Connection": "keep-alive",
        "Upgrade-Insecure-Requests": "1",
        "Sec-Fetch-Dest": "document",
        "Sec-Fetch-Mode": "navigate",
        "Sec-Fetch-Site": "cross-site",
    },
]

# curl_cffi supported browser impersonation targets
IMPERSONATE_TARGETS = [
    "chrome120",
    "chrome124",
    "chrome131",
]


def _get_random_headers() -> dict:
    return random.choice(HEADERS_TEMPLATES)


def _get_random_impersonate() -> str:
    return random.choice(IMPERSONATE_TARGETS)


def _is_captcha_page(soup: BeautifulSoup) -> bool:
    """
    Detect whether Amazon returned a CAPTCHA / robot-check page.
    Amazon's silent CAPTCHA returns HTTP 200 but the body is a
    verification form, not product data. Always check before parsing.
    """
    captcha_indicators = [
        soup.select_one("form[action='/errors/validateCaptcha']"),
        soup.select_one("#captchacharacters"),
        soup.find("input", {"id": "captchacharacters"}),
    ]
    if any(captcha_indicators):
        return True

    # Check page title for bot-check keywords
    title_tag = soup.find("title")
    if title_tag and title_tag.string:
        title_text = title_tag.string.lower()
        if any(kw in title_text for kw in ["robot check", "sorry!", "automated access", "api gateway"]):
            return True

    # If the page is unusually short and has no #productTitle, assume interception
    page_text = soup.get_text(strip=True)
    if len(page_text) < 2000 and not soup.select_one("#productTitle"):
        return True

    return False


def _extract_price(soup: BeautifulSoup) -> tuple[Optional[str], Optional[str]]:
    """
    Extract current price and list/strike-through price.
    Amazon distributes price data across multiple DOM nodes —
    a multi-level fallback chain is required for reliable extraction.

    Returns: (current price string, list price string)
    """
    price = None
    list_price = None

    # Method 1: New price container (dominant since 2023)
    price_block = soup.select_one(".a-price[data-a-color='price'] .a-offscreen")
    if price_block:
        price = price_block.get_text(strip=True)

    # Method 2: Legacy whole + fraction split format
    if not price:
        whole = soup.select_one("span.a-price-whole")
        fraction = soup.select_one("span.a-price-fraction")
        if whole:
            price = f"${whole.get_text(strip=True)}"
            if fraction:
                price = f"${whole.get_text(strip=True)}{fraction.get_text(strip=True)}"

    # Method 3: Deal / sale price block
    if not price:
        deal_tag = soup.select_one("#priceblock_dealprice, #priceblock_saleprice")
        if deal_tag:
            price = deal_tag.get_text(strip=True)

    # Method 4: Core price display module
    if not price:
        core_tag = soup.select_one("#corePriceDisplay_desktop_feature_div .a-offscreen")
        if core_tag:
            price = core_tag.get_text(strip=True)

    # Extract list/strike-through price
    list_price_tag = soup.select_one(".a-price[data-a-color='secondary'] .a-offscreen")
    if not list_price_tag:
        list_price_tag = soup.select_one("#listPrice, .a-text-price .a-offscreen")
    if list_price_tag:
        list_price = list_price_tag.get_text(strip=True)

    return price, list_price


def _extract_bsr(soup: BeautifulSoup) -> list:
    """
    Extract Best Sellers Rank entries.
    A product may rank simultaneously in a parent category and 1–2 sub-categories.
    """
    bsr_list = []

    bsr_section = soup.select_one("#detailBulletsWrapper_feature_div, #productDetails_feature_div")
    if bsr_section:
        for li in bsr_section.select("li, tr"):
            text = li.get_text(" ", strip=True)
            if "Best Sellers Rank" in text or "Amazon Best Sellers Rank" in text:
                bsr_list.append(text)
                break

    if not bsr_list:
        sales_rank = soup.select_one("#SalesRank")
        if sales_rank:
            bsr_list.append(sales_rank.get_text(" ", strip=True))

    return bsr_list


def _extract_images(soup: BeautifulSoup) -> Optional[str]:
    """
    Amazon's main image URLs are embedded in a JSON object inside a
    <script> tag (the colorImages variable) — far more reliable than
    parsing <img> src attributes, which are often low-resolution thumbnails.
    """
    import re
    scripts = soup.find_all("script", type="text/javascript")
    for script in scripts:
        if script.string and "'colorImages'" in script.string:
            match = re.search(r'"hiRes":"(https://[^"]+)"', script.string)
            if match:
                return match.group(1)
            match = re.search(r'"large":"(https://[^"]+)"', script.string)
            if match:
                return match.group(1)
    # Fallback: grab data-old-hires from #landingImage
    img_tag = soup.select_one("#landingImage[data-old-hires]")
    if img_tag:
        return img_tag.get("data-old-hires")
    return None


def _parse_product_page(html: str, asin: str, marketplace: str, url: str) -> AmazonProduct:
    """
    Parse product HTML into an AmazonProduct data object.
    """
    from datetime import datetime, timezone

    soup = BeautifulSoup(html, "lxml")
    product = AmazonProduct(asin=asin, marketplace=marketplace, url=url)
    product.scraped_at = datetime.now(timezone.utc).isoformat()

    # CAPTCHA check — always run first
    if _is_captcha_page(soup):
        product.is_captcha = True
        logger.warning(f"CAPTCHA detected for ASIN {asin}")
        return product

    # Title
    title_tag = soup.select_one("#productTitle")
    product.title = title_tag.get_text(strip=True) if title_tag else None

    # Brand
    brand_tag = soup.select_one("#bylineInfo, #brand")
    if brand_tag:
        brand_text = brand_tag.get_text(strip=True)
        product.brand = (
            brand_text
            .replace("Brand: ", "")
            .replace("Visit the ", "")
            .replace(" Store", "")
            .strip()
        )

    # Price
    product.price, product.list_price = _extract_price(soup)

    # Rating
    rating_tag = soup.select_one("span[data-hook='rating-out-of-text'], #acrPopover")
    if rating_tag:
        product.rating = (rating_tag.get("title") or rating_tag.get_text()).strip()

    # Review count
    review_tag = soup.select_one("#acrCustomerReviewText")
    product.review_count = review_tag.get_text(strip=True) if review_tag else None

    # Bullet points
    bullets = []
    for li in soup.select("#feature-bullets ul li:not(.aok-hidden) span.a-list-item"):
        text = li.get_text(strip=True)
        if text and len(text) > 5:
            bullets.append(text)
    product.bullet_points = bullets

    # BSR
    product.bsr = _extract_bsr(soup)

    # Main image
    product.main_image = _extract_images(soup)

    return product


# ── Core Scrape Function (with retry) ─────────────────────────────────────────
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=2, min=4, max=30),
    retry=retry_if_exception_type((Exception,)),
    reraise=True,
)
def scrape_amazon_product(
    asin: str,
    marketplace: str = "www.amazon.com",
    proxy: Optional[str] = None,
) -> AmazonProduct:
    """
    Scrape a single Amazon product page using curl_cffi.

    Args:
        asin: Amazon product identifier (e.g. 'B0CXXX')
        marketplace: Amazon domain (default US storefront)
        proxy: Proxy URL in format 'http://user:pass@host:port' (optional)

    Returns:
        AmazonProduct data object

    Raises:
        tenacity.RetryError — after 3 failed attempts
    """
    url = f"https://{marketplace}/dp/{asin}"
    headers = _get_random_headers()
    impersonate = _get_random_impersonate()

    # Polite delay: 2.5–6 second random interval, simulating real reading pace
    delay = random.uniform(2.5, 6.0)
    logger.info(f"Fetching ASIN {asin} | impersonate={impersonate} | delay={delay:.1f}s")
    time.sleep(delay)

    proxies = {"https": proxy, "http": proxy} if proxy else None

    response = cffi_requests.get(
        url,
        headers=headers,
        impersonate=impersonate,
        proxies=proxies,
        timeout=20,
        allow_redirects=True,
    )

    if response.status_code not in (200, 301, 302):
        raise ValueError(f"Unexpected status {response.status_code} for ASIN {asin}")

    return _parse_product_page(response.text, asin, marketplace, url)


# ── Entry Point ───────────────────────────────────────────────────────────────
if __name__ == "__main__":
    TEST_ASIN = "B0CXZXZXZX"  # Replace with a real ASIN
    PROXY = None  # e.g. "http://user:[email protected]:8000"

    product = scrape_amazon_product(TEST_ASIN, proxy=PROXY)

    if product.is_captcha:
        print("⚠️  CAPTCHA hit — rotate proxy or increase request delay")
    else:
        print(json.dumps(product.__dict__, indent=2, ensure_ascii=False))

{
  "asin": "B0CXZXZXZX",
  "title": "Example Product — Premium Edition, 500ml, 6-Pack (SGS Certified)",
  "brand": "ExampleBrand",
  "price": "$24.99",
  "list_price": "$29.99",
  "rating": "4.3 out of 5 stars",
  "review_count": "2,841 ratings",
  "bullet_points": [
    "SGS-certified food-grade material, meets US FDA standards",
    "Dishwasher safe, BPA-free, lead-free, cadmium-free",
    "Fits standard car cup holders (3.5-inch diameter), keeps drinks hot/cold 12 hours",
    "18-month manufacturer warranty, no-questions-asked replacement",
    "Ships from US warehouse, Prime next-day delivery"
  ],
  "bsr": ["#1,243 in Kitchen & Dining (#18 in Water Bottles)"],
  "main_image": "https://m.media-amazon.com/images/I/61xxxx.jpg",
  "marketplace": "www.amazon.com",
  "url": "https://www.amazon.com/dp/B0CXZXZXZX",
  "is_captcha": false,
  "scraped_at": "2026-06-10T01:00:00+00:00"
}

"""
Amazon Batch Async Scraper (Production Grade)
curl_cffi AsyncSession + asyncio Semaphore rate limiting + proxy rotation + CSV output
Requirements: pip install curl-cffi beautifulsoup4 lxml tenacity
"""

import asyncio
import csv
import json
import random
import logging
from dataclasses import asdict
from typing import Optional

from curl_cffi.requests import AsyncSession

# Reuse AmazonProduct, _parse_product_page, HEADERS_TEMPLATES from above

logger = logging.getLogger(__name__)


class ProxyRotator:
    """
    Simple proxy rotator with random selection and failure marking.
    In production, prefer a rotating endpoint from Bright Data or Smartproxy
    over maintaining a static IP list manually.
    """

    def __init__(self, proxy_list: list[str]):
        self._proxies = proxy_list
        self._failed: set = set()

    def get(self) -> Optional[str]:
        available = [p for p in self._proxies if p not in self._failed]
        if not available:
            logger.warning("All proxies exhausted, falling back to direct connection (high risk)")
            return None
        return random.choice(available)

    def mark_failed(self, proxy: str):
        self._failed.add(proxy)
        logger.warning(f"Proxy marked failed: {proxy[:30]}...")


async def fetch_one(
    session: AsyncSession,
    asin: str,
    proxy_rotator: ProxyRotator,
    semaphore: asyncio.Semaphore,
    marketplace: str = "www.amazon.com",
) -> dict:
    """
    Fetch a single ASIN under Semaphore concurrency control.
    """
    url = f"https://{marketplace}/dp/{asin}"
    proxy = proxy_rotator.get()
    impersonate = random.choice(["chrome120", "chrome124", "chrome131"])
    headers = random.choice(HEADERS_TEMPLATES)

    async with semaphore:
        # Human-like delay: 3–8 seconds between requests
        await asyncio.sleep(random.uniform(3.0, 8.0))

        try:
            proxies = {"https": proxy, "http": proxy} if proxy else None
            response = await session.get(
                url,
                headers=headers,
                impersonate=impersonate,
                proxies=proxies,
                timeout=25,
            )

            if response.status_code == 503:
                if proxy:
                    proxy_rotator.mark_failed(proxy)
                raise ValueError(f"503 for {asin} — proxy rotated")

            product = _parse_product_page(response.text, asin, marketplace, url)
            status = "captcha" if product.is_captcha else "success"
            logger.info(f"[{status.upper()}] ASIN {asin}")
            return asdict(product)

        except Exception as e:
            logger.error(f"[FAILED] ASIN {asin}: {e}")
            return {"asin": asin, "error": str(e)}


async def scrape_batch(
    asins: list[str],
    proxy_list: list[str],
    output_file: str = "amazon_products.json",
    max_concurrency: int = 3,
    marketplace: str = "www.amazon.com",
):
    """
    Batch async scrape entry point.

    max_concurrency=3 is a conservative setting for residential proxies.
    With a high-quality rotating proxy service, you can push to 5–8,
    but exceeding that tends to increase CAPTCHA rates significantly.
    """
    proxy_rotator = ProxyRotator(proxy_list)
    semaphore = asyncio.Semaphore(max_concurrency)

    async with AsyncSession() as session:
        tasks = [
            fetch_one(session, asin, proxy_rotator, semaphore, marketplace)
            for asin in asins
        ]
        results = await asyncio.gather(*tasks, return_exceptions=False)

    # Write JSON output
    with open(output_file, "w", encoding="utf-8") as f:
        json.dump(results, f, indent=2, ensure_ascii=False)

    # Write CSV output (for Excel / database import)
    csv_file = output_file.replace(".json", ".csv")
    if results:
        fieldnames = [k for k in results[0].keys() if k != "bullet_points"]
        with open(csv_file, "w", newline="", encoding="utf-8") as f:
            writer = csv.DictWriter(f, fieldnames=fieldnames + ["bullet_points_str"])
            writer.writeheader()
            for row in results:
                row_copy = {k: v for k, v in row.items() if k != "bullet_points"}
                row_copy["bullet_points_str"] = " | ".join(row.get("bullet_points", []))
                writer.writerow(row_copy)

    success = sum(1 for r in results if not r.get("error") and not r.get("is_captcha"))
    captcha = sum(1 for r in results if r.get("is_captcha"))
    failed = sum(1 for r in results if r.get("error"))
    logger.info(f"Done: {success} success, {captcha} CAPTCHA, {failed} failed | Output: {output_file}")
    return results


# ── Usage Example ─────────────────────────────────────────────────────────────
if __name__ == "__main__":
    ASINS_TO_SCRAPE = [
        "B0CXZ1", "B0CXZ2", "B0CXZ3", "B0CXZ4", "B0CXZ5",
    ]

    PROXY_LIST = [
        "http://user1:[email protected]:8001",
        "http://user2:[email protected]:8002",
        "http://user3:[email protected]:8003",
    ]

    asyncio.run(
        scrape_batch(
            asins=ASINS_TO_SCRAPE,
            proxy_list=PROXY_LIST,
            output_file="output/amazon_products.json",
            max_concurrency=3,
        )
    )

"""
Pangolinfo Amazon Scraper API — Python Example
No proxy management. No CAPTCHA handling. No selector maintenance.
"""

import requests
import json
from typing import Optional


API_KEY = "your_pangolinfo_api_key"
BASE_URL = "https://api.pangolinfo.com/v1"


def get_amazon_product(
    asin: str,
    marketplace: str = "US",
    include_fields: Optional[str] = None,
) -> dict:
    """
    Retrieve Amazon product data via Pangolinfo API.

    Args:
        asin: Amazon product identifier
        marketplace: Storefront code (US, UK, DE, JP, CA, FR, IT, ES, etc.)
        include_fields: Comma-separated field names to return (omit for all fields)

    Returns:
        Fully structured product data dict
    """
    params = {
        "asin": asin,
        "marketplace": marketplace,
    }
    if include_fields:
        params["include_fields"] = include_fields

    response = requests.get(
        f"{BASE_URL}/amazon/product",
        params=params,
        headers={"Authorization": f"Bearer {API_KEY}"},
        timeout=30,
    )
    response.raise_for_status()
    return response.json()


# ── Async Batch Version ───────────────────────────────────────────────────────
import asyncio
import aiohttp


async def get_amazon_product_async(
    session: aiohttp.ClientSession,
    asin: str,
    marketplace: str = "US",
) -> dict:
    url = f"{BASE_URL}/amazon/product"
    params = {"asin": asin, "marketplace": marketplace}
    headers = {"Authorization": f"Bearer {API_KEY}"}

    async with session.get(url, params=params, headers=headers) as resp:
        return await resp.json()


async def batch_fetch(asins: list[str], marketplace: str = "US") -> list[dict]:
    async with aiohttp.ClientSession() as session:
        tasks = [get_amazon_product_async(session, asin, marketplace) for asin in asins]
        return await asyncio.gather(*tasks)


# ── Usage ─────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
    # Single product
    result = get_amazon_product("B0CXZXZXZX", marketplace="US")
    print(json.dumps(result, indent=2, ensure_ascii=False))

    # Batch (5 ASINs, concurrent)
    asins = ["B0001", "B0002", "B0003", "B0004", "B0005"]
    results = asyncio.run(batch_fetch(asins))
    print(f"Fetched {len(results)} products")

{
  "success": true,
  "data": {
    "asin": "B0CXZXZXZX",
    "title": "Example Product — Premium Edition, 500ml, 6-Pack",
    "brand": "ExampleBrand",
    "price": {
      "current": 24.99,
      "currency": "USD",
      "list_price": 29.99,
      "deal_price": null,
      "coupon": "Extra 5% off with coupon"
    },
    "rating": 4.3,
    "review_count": 2841,
    "bullet_points": [
      "SGS-certified food-grade material",
      "Dishwasher safe, BPA-free"
    ],
    "bsr": [
      {"rank": 1243, "category": "Kitchen & Dining"},
      {"rank": 18, "category": "Water Bottles"}
    ],
    "images": {
      "main": "https://m.media-amazon.com/images/I/xxx.jpg",
      "gallery": [
        "https://m.media-amazon.com/images/I/yyy.jpg",
        "https://m.media-amazon.com/images/I/zzz.jpg"
      ]
    },
    "dimensions": {
      "length_inches": 9.8,
      "width_inches": 3.5,
      "height_inches": 3.5,
      "weight_pounds": 0.88
    },
    "availability": "In Stock",
    "is_prime": true,
    "marketplace": "US",
    "scraped_at": "2026-06-10T01:00:00Z"
  },
  "credits_used": 1,
  "credits_remaining": 9999
}