亚马逊价格抓取：Python、代理与Excel导出

python -m pip install playwright pandas openpyxl
python -m playwright install chromium
python --version
python -m playwright --version
python -m pip show pandas openpyxl

export ROLA_HTTP_PROXY_SERVER="http://CURRENT_ROLA_HOST:PORT"
export ROLA_PROXY_USERNAME="replace-with-account-username"
export ROLA_PROXY_PASSWORD="replace-with-account-password"
export AMAZON_PRODUCT_URL="https://www.amazon.com/dp/B0XXXXXXXXX"
export EXPECTED_ASIN="B0XXXXXXXXX"
export EXPECTED_MARKETPLACE="amazon.com"
export EXPECTED_CURRENCY="USD"
export EXPECTED_DELIVERY_LABEL="10001"
export PLAYWRIGHT_STORAGE_STATE="amazon-context.json"

import os
import re
from datetime import datetime, timezone
from decimal import Decimal
from pathlib import Path
from urllib.parse import urlparse
import pandas as pd
from playwright.sync_api import Error as PlaywrightError, TimeoutError as PlaywrightTimeoutError, sync_playwright

EXPORT_COLUMNS = [
    "requested_url", "final_url", "asin", "title", "marketplace",
    "variation", "variation_status", "seller", "raw_price", "numeric_price",
    "currency_symbol", "currency_code", "expected_currency", "delivery_location",
    "delivery_status", "offer_container", "captured_at_utc", "http_status",
    "top_level_navigations", "browser_request_count", "validation_status",
    "missing_reason",
]
STRICT_USD_DISPLAY = re.compile(r"^\$(?P<whole>(?:0|[1-9]\d{0,2}(?:,\d{3})*))\.(?P<fraction>\d{2})$")

class CollectionError(RuntimeError):
    def __init__(self, stage, message, status=None, retry_after=None):
        super().__init__(message); self.stage = stage
        self.status = status; self.retry_after = retry_after

def text(value):
    return " ".join((value or "").split())

def host(url):
    return (urlparse(url).hostname or "").lower().removeprefix("www.")

def page_type(page):
    if page.locator("input#captchacharacters, form[action*='validateCaptcha'], img[src*='captcha']").count():
        return "captcha"
    if "/ap/signin" in page.url.lower() or page.locator("form[name='signIn'], input#ap_email").count():
        return "login_wall"
    if page.locator("#productTitle").count():
        return "product"
    return "unknown"

def parse_usd(raw):
    value = text(raw); match = STRICT_USD_DISPLAY.fullmatch(value)
    if not match:
        raise CollectionError("price_format", "价格不是明确的美国美元格式")
    amount = match.group("whole").replace(",", "")
    return str(Decimal(f"{amount}.{match.group('fraction')}"))

def visible(scope, selectors):
    for selector in selectors:
        loc = scope.locator(selector).first
        if loc.count() and loc.is_visible():
            value = text(loc.inner_text())
            if value: return value
    return None

def collect_price(url, expected_asin, expected_marketplace):
    proxy = {"server": os.environ["ROLA_HTTP_PROXY_SERVER"],
             "username": os.environ["ROLA_PROXY_USERNAME"],
             "password": os.environ["ROLA_PROXY_PASSWORD"]}
    expected_currency = os.environ.get("EXPECTED_CURRENCY", "").upper()
    expected_delivery = text(os.environ.get("EXPECTED_DELIVERY_LABEL", ""))
    state = os.environ.get("PLAYWRIGHT_STORAGE_STATE")
    options = {"locale": "en-US"}
    if state:
        if not Path(state).is_file(): raise CollectionError("context_setup", "找不到storage state文件")
        options["storage_state"] = state
    stage = "browser_launch"
    with sync_playwright() as pw:
        browser = pw.chromium.launch(headless=True, proxy=proxy)
        context = browser.new_context(**options); page = context.new_page()
        requests = {"count": 0}; page.on("request", lambda _: requests.__setitem__("count", requests["count"] + 1))
        try:
            stage = "navigation"
            response = page.goto(url, wait_until="domcontentloaded", timeout=30000)
            if response is None: raise CollectionError("navigation", "没有返回HTTP响应")
            status = response.status; final_url = page.url
            if not 200 <= status < 300: raise CollectionError("http_response", f"目标返回HTTP {status}", status, response.headers.get("retry-after"))
            stage = "page_identity"
            page.wait_for_timeout(1500)
            if page_type(page) != "product": raise CollectionError("page_identity", f"收到{page_type(page)}而不是商品页", status)
            if host(final_url) != expected_marketplace: raise CollectionError("marketplace", "最终站点不匹配", status)
            asin = (page.locator("input#ASIN").get_attribute("value") or "").upper()
            if asin != expected_asin.upper(): raise CollectionError("product_identity", "ASIN不匹配", status)
            stage = "offer_validation"
            offer = page.locator("#apex_desktop, #buybox, #rightCol").filter(has=page.locator(".a-price")).first
            raw = visible(offer, [".a-price:not(.a-text-price) .a-offscreen"])
            if not raw: raise CollectionError("offer_validation", "没有找到唯一活动报价")
            currency = visible(offer, ["[itemprop='priceCurrency']"])
            seller = visible(offer, ["#sellerProfileTriggerId", "#merchant-info"])
            delivery = visible(page, ["#glow-ingress-line2"])
            reasons = []
            if currency != expected_currency: reasons.append("币种未独立确认或不匹配")
            if not seller: reasons.append("报价容器内没有卖家")
            if not state: reasons.append("未加载准备好的浏览器状态")
            if not expected_delivery: delivery_status = "undeclared"; reasons.append("未声明配送标签")
            elif not delivery: delivery_status = "missing"; reasons.append("未显示配送标签")
            elif expected_delivery.casefold() not in delivery.casefold(): delivery_status = "mismatch"; reasons.append("配送标签不匹配")
            else: delivery_status = "observed_match"
            record = {"requested_url": url, "final_url": final_url, "asin": asin,
                "title": text(page.locator("#productTitle").inner_text()), "marketplace": host(final_url),
                "variation": None, "variation_status": "manual_review", "seller": seller,
                "raw_price": raw, "numeric_price": parse_usd(raw), "currency_symbol": "$",
                "currency_code": currency, "expected_currency": expected_currency or None,
                "delivery_location": delivery, "delivery_status": delivery_status,
                "offer_container": "#apex_desktop/#buybox/#rightCol", "captured_at_utc": datetime.now(timezone.utc).isoformat(),
                "http_status": status, "top_level_navigations": 1, "browser_request_count": requests["count"],
                "validation_status": "candidate_context_complete" if not reasons else "candidate_manual_review",
                "missing_reason": "; ".join(reasons) or None}
            return record
        except (PlaywrightTimeoutError, PlaywrightError) as exc:
            message = str(exc).lower()
            if "407" in message or "proxy authentication" in message:
                stage = "proxy_authentication"
            elif "dns" in message or "name_not_resolved" in message:
                stage = "dns_resolution"
            elif "tls" in message or "ssl" in message or "certificate" in message:
                stage = "tls_handshake"
            elif "connection refused" in message or "connection reset" in message:
                stage = "connection"
            elif isinstance(exc, PlaywrightTimeoutError):
                stage = f"{stage}_timeout"
            raise CollectionError(stage, "browser operation failed") from exc
        finally:
            context.close(); browser.close()

if __name__ == "__main__":
    try:
        record = collect_price(os.environ["AMAZON_PRODUCT_URL"], os.environ["EXPECTED_ASIN"], os.environ.get("EXPECTED_MARKETPLACE", "amazon.com"))
        if set(record) != set(EXPORT_COLUMNS): raise RuntimeError("record schema mismatch")
        frame = pd.DataFrame([record], columns=EXPORT_COLUMNS)
        stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
        base = Path(f"amazon-price-snapshot-{record['validation_status']}-{stamp}")
        frame.to_csv(base.with_suffix(".csv"), index=False); frame.to_excel(base.with_suffix(".xlsx"), index=False)
        print(f"已保存{base}.csv和{base}.xlsx")
    except CollectionError as exc:
        print(f"采集停止：{exc}；stage={exc.stage} status={exc.status} retry_after={exc.retry_after}")
        raise SystemExit(2)

python amazon_price_snapshot.py

python -m playwright codegen --save-storage=amazon-context.json https://www.amazon.com/