# curl_cffi 包含预编译的 libcurl-impersonate 二进制，无需额外编译依赖
pip install curl-cffi beautifulsoup4 lxml tenacity

"""
亚马逊商品数据采集器（curl_cffi 版本）
2026 年可用，绕过 TLS/JA4 指纹检测
依赖: 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.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger(__name__)


# ── 数据结构 ──────────────────────────────────────────────────────────────────
@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


# ── 请求头模板 ─────────────────────────────────────────────────────────────────
# 注意：curl_cffi 的 impersonate 会自动设置与浏览器一致的 TLS 握手，
# 但 HTTP 层的请求头仍需我们手动控制以匹配浏览器行为
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 支持的浏览器指纹选项（选 Chrome 最新稳定版系列）
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:
    """
    检测是否命中亚马逊的 CAPTCHA / 机器人验证页面。
    亚马逊静默 CAPTCHA 返回 HTTP 200，但页面是验证表单而非商品数据。
    """
    captcha_indicators = [
        soup.select_one("form[action='/errors/validateCaptcha']"),
        soup.select_one("#captchacharacters"),
        soup.find("input", {"id": "captchacharacters"}),
    ]
    if any(captcha_indicators):
        return True

    # 检查页面标题是否是 "Robot Check" 或 "Sorry"
    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

    # 如果页面极短且没有 #productTitle，很可能是拦截页
    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]]:
    """
    提取当前价格和划线价（list price）。
    亚马逊价格字段分散在多个 DOM 节点，需要多级回退链。
    
    返回: (当前价格字符串, 划线价字符串)
    """
    price = None
    list_price = None

    # 方法 1：新版价格容器（2023 年后主流）
    price_block = soup.select_one(".a-price[data-a-color='price'] .a-offscreen")
    if price_block:
        price = price_block.get_text(strip=True)

    # 方法 2：旧版 whole + fraction 分体格式
    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)}"

    # 方法 3：Deal 价格
    if not price:
        deal_tag = soup.select_one("#priceblock_dealprice, #priceblock_saleprice")
        if deal_tag:
            price = deal_tag.get_text(strip=True)

    # 方法 4：核心价格 Box
    if not price:
        core_tag = soup.select_one("#corePriceDisplay_desktop_feature_div .a-offscreen")
        if core_tag:
            price = core_tag.get_text(strip=True)

    # 提取划线价（原价）
    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:
    """
    提取 Best Sellers Rank（BSR）排名列表。
    一个商品可能同时在父类目和 1-2 个子类目有排名。
    """
    bsr_list = []

    # 方法 1：Product details 表格中的 Best Sellers Rank 行
    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

    # 方法 2：独立 #SalesRank 容器（部分类目）
    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]:
    """
    亚马逊主图藏在页面内嵌的 JSON 数据中（colorImages 变量），
    直接解析 script 标签比抓 img src 更可靠。
    """
    import re
    scripts = soup.find_all("script", type="text/javascript")
    for script in scripts:
        if script.string and "'colorImages'" in script.string:
            # 提取第一张主图 URL（hiRes 或 large）
            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)
    # 回退：抓 #landingImage 的 data-old-hires
    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:
    """
    解析商品 HTML，返回 AmazonProduct 数据对象。
    """
    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
    if _is_captcha_page(soup):
        product.is_captcha = True
        logger.warning(f"CAPTCHA detected for ASIN {asin}")
        return product

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

    # 品牌
    brand_tag = soup.select_one("#bylineInfo, #brand")
    if brand_tag:
        brand_text = brand_tag.get_text(strip=True)
        # 去掉 "Brand: " / "Visit the XXX Store" 前缀
        product.brand = brand_text.replace("Brand: ", "").replace("Visit the ", "").replace(" Store", "").strip()

    # 价格
    product.price, product.list_price = _extract_price(soup)

    # 评分
    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_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)

    # 主图
    product.main_image = _extract_images(soup)

    return product


# ── 核心采集函数（带重试） ──────────────────────────────────────────────────────
@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:
    """
    用 curl_cffi 采集单个亚马逊商品页面。
    
    参数:
        asin: 亚马逊商品标识符（如 'B0CXXX'）
        marketplace: 亚马逊域名（默认美国站）
        proxy: 代理 URL，格式 'http://user:pass@host:port'（可选）
    
    返回:
        AmazonProduct 数据对象
    
    异常:
        tenacity.RetryError - 重试 3 次后仍失败
    """
    url = f"https://{marketplace}/dp/{asin}"
    headers = _get_random_headers()
    impersonate = _get_random_impersonate()

    # 礼貌性延迟：2.5–6 秒随机间隔，模拟真实阅读节奏
    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)


# ── 入口示例 ──────────────────────────────────────────────────────────────────
if __name__ == "__main__":
    TEST_ASIN = "B0CXZXZXZX"  # 替换为真实 ASIN
    PROXY = None  # 可填入住宅代理，如 "http://user:[email protected]:8000"

    product = scrape_amazon_product(TEST_ASIN, proxy=PROXY)

    if product.is_captcha:
        print("⚠️  命中 CAPTCHA，需要更换代理或增加延迟")
    else:
        print(json.dumps(product.__dict__, indent=2, ensure_ascii=False))

{
  "asin": "B0CXZXZXZX",
  "title": "示例商品——高级版，500ml，6件装（SGS 认证）",
  "brand": "ExampleBrand",
  "price": "$24.99",
  "list_price": "$29.99",
  "rating": "4.3 out of 5 stars",
  "review_count": "2,841 ratings",
  "bullet_points": [
    "SGS 认证高品质食品级材料，通过美国 FDA 标准",
    "可洗碗机清洗，不含 BPA、铅、镉",
    "适配标准汽车杯架（3.5 英寸直径），保温 12 小时",
    "18 个月制造商保修，无条件退换",
    "美国仓库发货，Prime 次日达"
  ],
  "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"
}

"""
亚马逊批量异步采集器（生产级）
curl_cffi AsyncSession + asyncio Semaphore 限速 + 代理轮换 + CSV 输出
依赖: pip install curl-cffi beautifulsoup4 lxml tenacity
"""

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

from curl_cffi.requests import AsyncSession
from tenacity import retry, stop_after_attempt, wait_exponential

# 复用上面定义的 AmazonProduct、_parse_product_page、HEADERS_TEMPLATES 等

logger = logging.getLogger(__name__)


class ProxyRotator:
    """
    简单的代理轮换器，支持随机选取和失败标记。
    生产环境建议使用 Bright Data、Smartproxy 等服务的 rotating endpoint，
    而非手动维护 IP 列表。
    """

    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("所有代理均已失败，使用直连（高风险）")
            return None
        return random.choice(available)

    def mark_failed(self, proxy: str):
        self._failed.add(proxy)
        logger.warning(f"代理 {proxy[:30]}... 标记为失败")


async def fetch_one(
    session: AsyncSession,
    asin: str,
    proxy_rotator: ProxyRotator,
    semaphore: asyncio.Semaphore,
    marketplace: str = "www.amazon.com",
) -> dict:
    """
    采集单个 ASIN，在 Semaphore 控制下运行（限制并发数量）。
    """
    url = f"https://{marketplace}/dp/{asin}"
    proxy = proxy_rotator.get()
    impersonate = random.choice(["chrome120", "chrome124", "chrome131"])
    headers = random.choice(HEADERS_TEMPLATES)

    async with semaphore:
        # 人性化延迟：每个请求前等 3-8 秒
        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",
):
    """
    批量异步采集入口。
    
    max_concurrency=3 是针对住宅代理的保守设置。
    如果使用高质量轮换代理服务，可以适当提升到 5-8，但不建议更高。
    """
    proxy_rotator = ProxyRotator(proxy_list)
    semaphore = asyncio.Semaphore(max_concurrency)
    results = []

    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)

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

    # 同时写出 CSV（方便导入 Excel / 数据库）
    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"完成: {success} 成功, {captcha} CAPTCHA, {failed} 失败 | 输出: {output_file}")
    return results


# ── 使用示例 ──────────────────────────────────────────────────────────────────
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 亚马逊数据采集 API 示例
无需管理代理、无需处理 CAPTCHA、无需维护 Selector
"""

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:
    """
    通过 Pangolinfo API 获取亚马逊商品数据。
    
    参数:
        asin: 亚马逊商品标识符
        marketplace: 站点代码（US, UK, DE, JP, CA, FR, IT, ES 等）
        include_fields: 需要的字段，逗号分隔（不传则返回全部）
    
    返回:
        完全结构化的商品数据 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()


# ── 批量采集（异步版）────────────────────────────────────────────────────────
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)


# ── 使用示例 ──────────────────────────────────────────────────────────────────
if __name__ == "__main__":
    # 单个商品
    result = get_amazon_product("B0CXZXZXZX", marketplace="US")
    print(json.dumps(result, indent=2, ensure_ascii=False))

    # 批量（5 个 ASIN 并发）
    asins = ["B0001", "B0002", "B0003", "B0004", "B0005"]
    results = asyncio.run(batch_fetch(asins))
    print(f"获取到 {len(results)} 条商品数据")

{
  "success": true,
  "data": {
    "asin": "B0CXZXZXZX",
    "title": "示例商品——高级版，500ml，6件装",
    "brand": "ExampleBrand",
    "price": {
      "current": 24.99,
      "currency": "USD",
      "list_price": 29.99,
      "deal_price": null,
      "coupon": "额外 5% 优惠券"
    },
    "rating": 4.3,
    "review_count": 2841,
    "bullet_points": [
      "SGS 认证高品质食品级材料",
      "可洗碗机清洗，不含 BPA"
    ],
    "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
}