Amazon Price Scraping: Validate Prices, Diagnose Blocks, and Export to 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 normalized_host(url: str) -> str:
    host = (urlparse(url).hostname or "").lower()
    return host.removeprefix("www.")


def normalized_text(value: str) -> str:
    return " ".join(value.split())


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


def wait_for_page_outcome(page):
    page.wait_for_function(
        """() => Boolean(
            document.querySelector('#productTitle') ||
            document.querySelector('input#captchacharacters') ||
            document.querySelector("form[action*='validateCaptcha']") ||
            document.querySelector("form[name='signIn']") ||
            document.querySelector('input#ap_email')
        )""",
        timeout=12_000,
    )


def detect_page_type(page, final_url: str) -> str:
    lower_url = final_url.lower()
    captcha = page.locator(
        "input#captchacharacters, form[action*='validateCaptcha'], img[src*='captcha']"
    )
    login = page.locator("form[name='signIn'], input#ap_email")

    if "validatecaptcha" in lower_url or captcha.count():
        return "captcha"
    if "/ap/signin" in lower_url or login.count():
        return "login_wall"
    if page.locator("#productTitle").count():
        return "product"
    return "unknown"


def extract_asin(page, final_url: str):
    asin_input = page.locator("input#ASIN").first
    if asin_input.count():
        value = asin_input.get_attribute("value")
        if value:
            return value.upper()

    match = re.search(r"/(?:dp|gp/product)/([A-Z0-9]{10})", final_url, re.I)
    return match.group(1).upper() if match else None


def parse_strict_usd_display(raw_price: str) -> str:
    value = normalized_text(raw_price)
    match = STRICT_USD_DISPLAY.fullmatch(value)
    if not match:
        raise CollectionError(
            "price_format",
            "price text was not one unambiguous US-style dollar amount",
        )
    amount = match.group("whole").replace(",", "")
    return str(Decimal(f"{amount}.{match.group('fraction')}"))


def extract_currency_code(offer_container):
    marker = offer_container.locator("[itemprop='priceCurrency']").first
    if not marker.count():
        return None
    value = marker.get_attribute("content") or marker.text_content() or ""
    value = normalized_text(value).upper()
    return value if re.fullmatch(r"[A-Z]{3}", value) else None


def extract_offer_candidate(page):
    container_selectors = ("#apex_desktop", "#buybox", "#rightCol")

    page.wait_for_function(
        """() => Boolean(
            document.querySelector('#apex_desktop .a-price') ||
            document.querySelector('#buybox .a-price') ||
            document.querySelector('#rightCol .a-price') ||
            document.querySelector('input#captchacharacters') ||
            document.querySelector("form[name='signIn']")
        )""",
        timeout=10_000,
    )

    page_type = detect_page_type(page, page.url)
    if page_type != "product":
        raise CollectionError(
            "offer_wait",
            f"page changed to {page_type} while waiting for the offer",
        )

    for container_selector in container_selectors:
        container = page.locator(container_selector).first
        if not container.count() or not container.is_visible():
            continue

        nodes = container.locator(
            ".a-price:not(.a-text-price) .a-offscreen"
        )
        observed = []
        for index in range(nodes.count()):
            node = nodes.nth(index)
            price_box = node.locator("xpath=..")
            if price_box.count() and price_box.is_visible():
                text = normalized_text(node.text_content() or "")
                if text and text not in observed:
                    observed.append(text)

        if len(observed) > 1:
            raise CollectionError(
                "offer_ambiguity",
                f"multiple active price candidates appeared in {container_selector}",
            )
        if len(observed) == 1:
            return container, observed[0], container_selector

    raise CollectionError(
        "offer_validation",
        "no single active price candidate was found in the offer containers",
    )


def extract_variation_state(page):
    controls = page.locator("[id^='variation_']")
    selections = page.locator("[id^='variation_'] .selection")

    if controls.count() == 0:
        return None, "not_applicable"

    values = []
    for index in range(selections.count()):
        text = normalized_text(selections.nth(index).inner_text())
        if text and text not in values:
            values.append(text)

    if not values:
        return None, "unresolved"
    return " | ".join(values), "observed"


def classify_browser_error(exc, stage):
    message = str(exc).lower()
    if "407" in message or "proxy authentication" in message:
        return "proxy_authentication"
    if "name_not_resolved" in message or "dns" in message:
        return "dns_resolution"
    if "certificate" in message or "ssl" in message or "tls" in message:
        return "tls_handshake"
    if "connection refused" in message or "connection reset" in message:
        return "connection"
    if isinstance(exc, PlaywrightTimeoutError):
        return f"{stage}_timeout"
    return stage


def collect_price(url: str, expected_asin: str, expected_marketplace: str) -> dict:
    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 = normalized_text(
        os.environ.get("EXPECTED_DELIVERY_LABEL", "")
    )
    storage_state = os.environ.get("PLAYWRIGHT_STORAGE_STATE")
    state_loaded = False

    context_options = {"locale": "en-US"}
    if storage_state:
        storage_path = Path(storage_state)
        if not storage_path.is_file():
            raise CollectionError(
                "context_setup",
                "PLAYWRIGHT_STORAGE_STATE does not point to a file",
            )
        context_options["storage_state"] = str(storage_path)
        state_loaded = True

    with sync_playwright() as playwright:
        browser = None
        context = None
        stage = "browser_launch"
        try:
            browser = playwright.chromium.launch(headless=True, proxy=proxy)
            stage = "context_setup"
            context = browser.new_context(**context_options)
            page = context.new_page()
            network = {"requests": 0}
            page.on(
                "request",
                lambda _request: network.__setitem__(
                    "requests", network["requests"] + 1
                ),
            )

            stage = "navigation"
            response = page.goto(
                url,
                wait_until="domcontentloaded",
                timeout=30_000,
            )
            if response is None:
                raise CollectionError("navigation", "no HTTP response was returned")

            status = response.status
            retry_after = response.headers.get("retry-after")
            final_url = page.url

            if status < 200 or status >= 300:
                raise CollectionError(
                    "http_response",
                    f"target returned HTTP {status}",
                    status,
                    retry_after,
                )

            stage = "page_identity"
            wait_for_page_outcome(page)
            page_type = detect_page_type(page, page.url)
            if page_type != "product":
                raise CollectionError(
                    "page_identity",
                    f"expected product page, received {page_type}",
                    status,
                    retry_after,
                )

            marketplace = normalized_host(final_url)
            if marketplace != expected_marketplace:
                raise CollectionError(
                    "marketplace",
                    f"expected {expected_marketplace}, received {marketplace}",
                    status,
                )

            observed_asin = extract_asin(page, final_url)
            if observed_asin != expected_asin.upper():
                raise CollectionError(
                    "product_identity",
                    f"expected ASIN {expected_asin}, received {observed_asin}",
                    status,
                )

            title = normalized_text(
                page.locator("#productTitle").inner_text(timeout=10_000)
            )

            stage = "offer_wait"
            offer, raw_price, offer_selector = extract_offer_candidate(page)
            numeric_price = parse_strict_usd_display(raw_price)
            currency_code = extract_currency_code(offer)
            seller, _ = first_visible_text(
                offer, ["#sellerProfileTriggerId", "#merchant-info"]
            )
            delivery, _ = first_visible_text(page, ["#glow-ingress-line2"])
            variation, variation_status = extract_variation_state(page)

            reasons = []
            if not expected_currency:
                reasons.append("expected currency was not declared")
            elif currency_code != expected_currency:
                reasons.append(
                    "ISO currency code was not independently observed or did not match"
                )
            if not seller:
                reasons.append("seller was not found inside the selected offer container")
            if variation_status == "unresolved":
                reasons.append("variation controls existed but the active selection was unresolved")
            if not state_loaded:
                reasons.append("no prepared browser storage state was loaded")
            if not expected_delivery:
                reasons.append("expected delivery label was not declared")
                delivery_status = "undeclared"
            elif not delivery:
                reasons.append("delivery label was not visible")
                delivery_status = "missing"
            elif expected_delivery.casefold() not in delivery.casefold():
                reasons.append("displayed delivery label did not match the expected context")
                delivery_status = "mismatch"
            else:
                delivery_status = "observed_match"

            validation_status = (
                "candidate_context_complete"
                if not reasons
                else "candidate_manual_review"
            )

            return {
                "requested_url": url,
                "final_url": final_url,
                "asin": observed_asin,
                "title": title,
                "marketplace": marketplace,
                "variation": variation,
                "variation_status": variation_status,
                "seller": seller,
                "raw_price": raw_price,
                "numeric_price": numeric_price,
                "currency_symbol": "$",
                "currency_code": currency_code,
                "expected_currency": expected_currency or None,
                "delivery_location": delivery,
                "delivery_status": delivery_status,
                "offer_container": offer_selector,
                "captured_at_utc": datetime.now(timezone.utc).isoformat(),
                "http_status": status,
                "top_level_navigations": 1,
                "browser_request_count": network["requests"],
                "validation_status": validation_status,
                "missing_reason": "; ".join(reasons) or None,
            }
        except CollectionError:
            raise
        except (PlaywrightTimeoutError, PlaywrightError) as exc:
            category = classify_browser_error(exc, stage)
            raise CollectionError(category, "browser operation failed") from exc
        finally:
            if context is not None:
                context.close()
            if browser is not None:
                browser.close()


def main():
    captured = collect_price(
        os.environ["AMAZON_PRODUCT_URL"],
        os.environ["EXPECTED_ASIN"],
        os.environ.get("EXPECTED_MARKETPLACE", "amazon.com"),
    )

    expected_keys = set(EXPORT_COLUMNS)
    actual_keys = set(captured)
    if actual_keys != expected_keys:
        missing = sorted(expected_keys - actual_keys)
        extra = sorted(actual_keys - expected_keys)
        raise RuntimeError(f"record schema mismatch: missing={missing}, extra={extra}")

    frame = pd.DataFrame([captured], columns=EXPORT_COLUMNS)

    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    state = captured["validation_status"].replace("_", "-")
    base = Path(f"amazon-price-snapshot-{state}-{stamp}")
    frame.to_csv(base.with_suffix(".csv"), index=False)
    frame.to_excel(base.with_suffix(".xlsx"), index=False)
    print(f"Saved {state} observation to {base}.csv and {base}.xlsx")


if __name__ == "__main__":
    try:
        main()
    except CollectionError as exc:
        details = [f"stage={exc.stage}"]
        if exc.status is not None:
            details.append(f"status={exc.status}")
        if exc.retry_after:
            details.append(f"retry_after={exc.retry_after}")
        print(f"Collection stopped: {exc}. " + ", ".join(details))
        raise SystemExit(2)

python amazon_price_snapshot.py

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