How to Scrape Amazon Reviews From Public Product Pages

python -m venv .venv
.\.venv\Scripts\Activate.ps1
python -m pip install beautifulsoup4==4.15.0 lxml==6.1.3 playwright==1.63.0
python -m playwright install chromium

python3 -m venv .venv
source .venv/bin/activate
python -m pip install beautifulsoup4==4.15.0 lxml==6.1.3 playwright==1.63.0
python -m playwright install chromium

from __future__ import annotations

import argparse
import csv
import hashlib
import json
import os
import platform
import re
import sys
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from importlib.metadata import version
from pathlib import Path
from typing import Any
from urllib.parse import urlparse

from bs4 import BeautifulSoup
from playwright.sync_api import (
    Browser,
    Error as PlaywrightError,
    Page,
    TimeoutError as PlaywrightTimeoutError,
    sync_playwright,
)


SCRIPT_VERSION = "1.1.0"
ASIN_RE = re.compile(
    r"/(?:dp|gp/product|product-reviews)/([A-Z0-9]{10})(?:[/?]|$)", re.I
)
ALLOWED_HOSTS = {"amazon.com", "www.amazon.com"}
REVIEW_SELECTORS = {
    "container": ['div[data-hook="review"]', 'div[id^="customer_review-"]'],
    "rating": [
        '[data-hook="review-star-rating"]',
        '[data-hook="review-star-rating-view-point"]',
    ],
    "title": ['[data-hook="reviewTitle"]', '[data-hook="review-title"]'],
    "body": [
        '[data-hook="reviewText"]',
        '[data-hook="reviewRichContentContainer"]',
        '[data-hook="review-body"]',
    ],
    "date": ['[data-hook="review-date"]'],
    "verified": ['[data-hook="avp-badge"]'],
    "helpful": ['[data-hook="helpful-vote-statement"]'],
}


@dataclass
class Review:
    asin: str
    review_id: str | None
    rating: float | None
    title: str | None
    body: str
    date_text: str | None
    review_date: str | None
    verified_purchase: bool | None
    helpful_votes: int | None
    source_url: str


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Export reviews visible on a public Amazon product page."
    )
    parser.add_argument(
        "url",
        nargs="?",
        default="https://www.amazon.com/dp/B07FZ8S74R",
        help="Public Amazon product URL containing a 10-character ASIN.",
    )
    parser.add_argument(
        "--output-dir",
        type=Path,
        default=Path(__file__).resolve().parent / "output",
    )
    parser.add_argument(
        "--check-pagination",
        action="store_true",
        help="Check one signed-out page-2 URL and report the observed access state.",
    )
    parser.add_argument(
        "--capture-screenshots",
        action="store_true",
        help="Save unedited browser screenshots for verification.",
    )
    parser.add_argument("--headed", action="store_true")
    return parser.parse_args()


def validate_destination(url: str) -> None:
    parsed = urlparse(url)
    if (parsed.scheme != "https" or parsed.hostname not in ALLOWED_HOSTS
            or parsed.username is not None or parsed.password is not None
            or parsed.port not in {None, 443}):
        raise ValueError("Use an HTTPS amazon.com or www.amazon.com URL without credentials or a custom port.")


def validate_target(url: str) -> tuple[str, str]:
    validate_destination(url)
    parsed = urlparse(url)
    match = ASIN_RE.search(parsed.path)
    if not match:
        raise ValueError("No 10-character ASIN was found in the URL path.")
    asin = match.group(1).upper()
    return asin, f"https://{parsed.hostname}/dp/{asin}"


def guard_navigation(route: Any) -> None:
    """Allow approved documents only; stop HTTP redirects; allow CDN resources."""
    if route.request.is_navigation_request():
        try:
            validate_destination(route.request.url)
        except ValueError:
            route.abort("blockedbyclient")
            return
    # Never follow a document redirect. Playwright routing must not be assumed
    # to intercept every hop in a redirect chain.
    if route.request.is_navigation_request():
        response = route.fetch(max_redirects=0, timeout=30_000)
        if 300 <= response.status < 400 and response.headers.get("location"):
            route.fulfill(status=403, content_type="text/html", body="<title>Redirect stopped</title><p>This tutorial does not follow document redirects.</p>")
        else:
            route.fulfill(response=response)
    else:
        route.continue_()


def clean_text(value: str | None) -> str | None:
    if value is None:
        return None
    cleaned = " ".join(value.split())
    return cleaned or None


def first_text(node: Any, selectors: list[str]) -> str | None:
    for selector in selectors:
        match = node.select_one(selector)
        if match:
            value = clean_text(match.get_text(" ", strip=True))
            if value:
                return value
    return None


def parse_rating(text: str | None) -> float | None:
    match = re.search(r"([0-5](?:[.,]\d)?)", text or "")
    return float(match.group(1).replace(",", ".")) if match else None


def parse_helpful_votes(text: str | None) -> int | None:
    if not text:
        return None
    if re.search(r"\bone\s+person\b", text, re.I):
        return 1
    match = re.search(r"([\d,]+)", text)
    return int(match.group(1).replace(",", "")) if match else None


def parse_review_date(text: str | None) -> str | None:
    """Normalize the en-US date while preserving Amazon's display text."""
    match = re.search(r"\bon\s+([A-Za-z]+\s+\d{1,2},\s+\d{4})$", text or "")
    if not match:
        return None
    try:
        return datetime.strptime(match.group(1), "%B %d, %Y").date().isoformat()
    except ValueError:
        return None


def review_id_from(container: Any) -> str | None:
    html_id = clean_text(container.get("id"))
    if html_id and html_id.startswith("customer_review-"):
        return html_id.removeprefix("customer_review-")
    review_link = container.select_one('a[href*="customer-reviews/srp/-/"]')
    if review_link:
        match = re.search(
            r"/customer-reviews/srp/-/([A-Z0-9]+)", review_link.get("href", "")
        )
        if match:
            return match.group(1)
    return None


def parse_reviews(html: str, asin: str, source_url: str) -> tuple[list[Review], int]:
    soup = BeautifulSoup(html, "lxml")
    containers = []
    for selector in REVIEW_SELECTORS["container"]:
        containers = soup.select(selector)
        if containers:
            break

    reviews: list[Review] = []
    seen: set[str] = set()
    for container in containers:
        body = first_text(container, REVIEW_SELECTORS["body"])
        if not body:
            continue

        review_id = review_id_from(container)
        title = first_text(container, REVIEW_SELECTORS["title"])
        date_text = first_text(container, REVIEW_SELECTORS["date"])
        dedupe_key = review_id or hashlib.sha256(
            f"{title}|{body}|{date_text}".encode("utf-8")
        ).hexdigest()
        if dedupe_key in seen:
            continue
        seen.add(dedupe_key)

        reviews.append(
            Review(
                asin=asin,
                review_id=review_id,
                rating=parse_rating(first_text(container, REVIEW_SELECTORS["rating"])),
                title=title,
                body=body,
                date_text=date_text,
                review_date=parse_review_date(date_text),
                verified_purchase=(
                    True
                    if any(container.select_one(s) for s in REVIEW_SELECTORS["verified"])
                    else None
                ),
                helpful_votes=parse_helpful_votes(
                    first_text(container, REVIEW_SELECTORS["helpful"])
                ),
                source_url=source_url,
            )
        )
    return reviews, len(containers)


def classify_page(page: Page, html: str) -> str:
    validate_destination(page.url)
    soup = BeautifulSoup(html, "lxml")
    for node in soup.select("script, style, template, noscript"):
        node.decompose()
    # Inspect page text and challenge controls, not arbitrary script keywords.
    sample = f"{page.title()}\n{soup.get_text(' ', strip=True)}".lower()
    if re.search(
        r"click the button below.{0,200}continue shopping", sample, re.DOTALL
    ):
        return "continue_shopping_challenge"
    if "robot check" in sample or "enter the characters you see below" in sample:
        return "robot_check"
    if soup.select_one('form[action*="validateCaptcha"], form[action*="validatecaptcha"], input#captchacharacters'):
        return "captcha"
    if "/ap/signin" in page.url or (
        "sign in or create account" in sample and soup.select_one('input[name="email"]')
    ):
        return "sign_in"
    if "page not found" in sample or "dogs of amazon" in sample:
        return "not_found"
    if re.search(r"sign in.{0,200}to see customer reviews", sample, re.DOTALL):
        return "review_sign_in_required"
    if re.search(r"/dp/[A-Z0-9]{10}", page.url, re.I):
        return "product_page"
    if re.search(r"/product-reviews/[A-Z0-9]{10}", page.url, re.I):
        return "review_page"
    return "unexpected_page"


def check_http_status(response: Any, report: dict[str, Any]) -> None:
    status = response.status if response else None
    report["http_status"] = status
    if status == 429:
        report["page_type"] = "rate_limited"
        report["retry_after"] = response.headers.get("retry-after")
        raise RuntimeError("HTTP 429: stopped without retrying. Respect Retry-After before any later authorized run.")
    if status is None or status >= 400:
        report["page_type"] = {403: "access_denied", 407: "proxy_authentication_required", 404: "not_found"}.get(status, "http_error")
        raise RuntimeError(f"HTTP {status}: stopped before review extraction.")


def launch_browser(playwright: Any, headed: bool) -> Browser:
    options: dict[str, Any] = {"headless": not headed}
    chrome = Path(r"C:\Program Files\Google\Chrome\Application\chrome.exe")
    if chrome.exists():
        options["executable_path"] = str(chrome)

    proxy_server = os.getenv("ROLA_PROXY_SERVER")
    if proxy_server:
        options["proxy"] = {
            "server": proxy_server,
            "username": os.getenv("ROLA_PROXY_USERNAME", ""),
            "password": os.getenv("ROLA_PROXY_PASSWORD", ""),
        }
    return playwright.chromium.launch(**options)


def scroll_until_reviews(page: Page, max_steps: int = 60) -> bool:
    review_locator = page.locator(', '.join(REVIEW_SELECTORS["container"]))
    for _ in range(max_steps):
        if review_locator.count() > 0:
            return True
        state = page.evaluate(
            """
            () => {
              const step = Math.max(600, Math.floor(window.innerHeight * 0.75));
              window.scrollBy(0, step);
              return {
                y: window.scrollY,
                max: Math.max(0, document.documentElement.scrollHeight - window.innerHeight)
              };
            }
            """
        )
        page.wait_for_timeout(250)
        if state["y"] >= state["max"]:
            break

    try:
        review_locator.first.wait_for(state="attached", timeout=15_000)
        return True
    except PlaywrightTimeoutError:
        return False


def save_browser_evidence(page: Page, output_dir: Path, name: str) -> None:
    evidence_dir = output_dir / "browser_evidence"
    evidence_dir.mkdir(parents=True, exist_ok=True)
    (evidence_dir / f"{name}.html").write_text(page.content(), encoding="utf-8")
    page.screenshot(path=str(evidence_dir / f"{name}.png"), full_page=False)


def record_key(row: dict[str, Any]) -> str:
    return row["review_id"] or hashlib.sha256(
        f"{row['title']}|{row['body']}|{row['date_text']}".encode("utf-8")
    ).hexdigest()


def csv_record(row: dict[str, Any]) -> dict[str, str]:
    return {key: "" if value is None else str(value) for key, value in row.items()}


def verify_exports(rows: list[dict[str, Any]], output_dir: Path) -> dict[str, Any]:
    with (output_dir / "amazon_reviews.csv").open("r", newline="", encoding="utf-8-sig") as handle:
        reader = csv.DictReader(handle)
        if reader.fieldnames != list(Review.__dataclass_fields__):
            raise RuntimeError("Round-trip validation failed: CSV schema differs.")
        csv_rows = list(reader)
    json_rows = json.loads((output_dir / "amazon_reviews.json").read_text(encoding="utf-8"))
    if not rows or len(csv_rows) != len(rows) or len(json_rows) != len(rows):
        raise RuntimeError("Round-trip validation failed: empty input or row counts differ.")
    expected_csv = [csv_record(row) for row in rows]
    # JSON comparison preserves types (True and 1 must not compare as equivalent).
    if json.dumps(json_rows, sort_keys=True) != json.dumps(rows, sort_keys=True) or csv_rows != expected_csv:
        raise RuntimeError("Round-trip validation failed: exported fields differ.")
    for label, loaded in (("JSON", json_rows), ("CSV", csv_rows)):
        keys = [record_key(row) for row in loaded]
        if len(keys) != len(set(keys)) or any(not row["body"].strip() for row in loaded):
            raise RuntimeError(f"Round-trip validation failed: {label} duplicate keys or empty body.")
    return {
        "csv_rows_reopened": len(csv_rows),
        "json_rows_reopened": len(json_rows),
        "unique_record_keys": len({record_key(row) for row in json_rows}),
        "all_fields_compared": True,
        "round_trip_validated": True,
    }


def export_and_verify(reviews: list[Review], output_dir: Path) -> dict[str, Any]:
    rows = [asdict(review) for review in reviews]
    json_path = output_dir / "amazon_reviews.json"
    csv_path = output_dir / "amazon_reviews.csv"
    json_path.write_text(json.dumps(rows, indent=2, ensure_ascii=False), encoding="utf-8")

    fieldnames = list(Review.__dataclass_fields__)
    with csv_path.open("w", newline="", encoding="utf-8-sig") as handle:
        writer = csv.DictWriter(handle, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(rows)

    return verify_exports(rows, output_dir)


def check_public_page_two(
    page: Page,
    asin: str,
    origin: str,
    report: dict[str, Any],
    output_dir: Path,
) -> None:
    url = f"{origin}/product-reviews/{asin}/?pageNumber=2"
    response = None
    state_report: dict[str, Any] = {}
    try:
        response = page.goto(url, wait_until="commit", timeout=30_000)
        validate_destination(page.url)
        check_http_status(response, state_report)
        page.locator("body").wait_for(state="attached", timeout=15_000)
        page.wait_for_timeout(3_000)
        html = page.content()
        page_type = classify_page(page, html)
        reviews, containers = (parse_reviews(html, asin, page.url)
                               if page_type in {"product_page", "review_page"} else ([], 0))
        report.update(
            {
                "public_pagination_tested": True,
                "pagination_http_status": response.status if response else None,
                "pagination_final_url": page.url,
                "pagination_page_type": page_type,
                "pagination_review_containers": containers,
                "pagination_reviews_parsed": len(reviews),
                "public_pagination_available": bool(reviews),
            }
        )
        save_browser_evidence(page, output_dir, "pagination_page_2")
    except (PlaywrightError, RuntimeError, ValueError) as exc:
        report.update(
            {
                "public_pagination_tested": True,
                "public_pagination_available": False,
                "pagination_error": type(exc).__name__,
                "pagination_page_type": state_report.get("page_type", "navigation_error"),
                "pagination_http_status": state_report.get("http_status"),
                "pagination_retry_after": state_report.get("retry_after"),
            }
        )
        try:
            save_browser_evidence(page, output_dir, "pagination_timeout")
        except PlaywrightError:
            pass


def run() -> int:
    args = parse_args()
    asin, normalized_url = validate_target(args.url)
    # A separate run directory prevents old successful exports being mistaken
    # for the output of a later failed run.
    args.output_dir = args.output_dir / datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
    args.output_dir.mkdir(parents=True, exist_ok=False)
    parsed_target = urlparse(normalized_url)
    origin = f"{parsed_target.scheme}://{parsed_target.hostname}"
    report: dict[str, Any] = {
        "script_version": SCRIPT_VERSION,
        "tested_at_utc": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        "python_version": sys.version.split()[0],
        "os": platform.platform(),
        "dependencies": {name: version(name) for name in ("playwright", "beautifulsoup4", "lxml")},
        "script_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
        "automatic_retries": 0,
        "success": False,
        "target_url": normalized_url,
        "asin": asin,
        "access_state": "signed_out_fresh_context",
        "proxy_configured": bool(os.getenv("ROLA_PROXY_SERVER")),
    }

    print(f"Script version: {SCRIPT_VERSION}")
    print(f"Target ASIN: {asin}")
    print(f"Opening: {normalized_url}")

    with sync_playwright() as playwright:
        browser = None
        page = None
        response = None
        try:
            browser = launch_browser(playwright, args.headed)
            report["browser_version"] = browser.version
            context = browser.new_context(
                locale="en-US", timezone_id="America/New_York",
                viewport={"width": 1440, "height": 900}, service_workers="block",
            )
            context.route("**/*", guard_navigation)
            page = context.new_page()
            response = page.goto(
                normalized_url, wait_until="domcontentloaded", timeout=90_000
            )
            validate_destination(page.url)
            check_http_status(response, report)
            page.locator("body").wait_for(state="attached", timeout=15_000)
            page_type = classify_page(page, page.content())
            report["page_type"] = page_type
            if page_type != "product_page":
                raise RuntimeError(f"Stopped before scrolling: page_type={page_type}.")
            try:
                page.locator("#productTitle:visible").first.wait_for(
                    state="visible", timeout=15_000
                )
            except PlaywrightTimeoutError:
                pass

            if args.capture_screenshots:
                page.screenshot(
                    path=str(args.output_dir / "product_page_viewport.png"),
                    full_page=False,
                )

            review_loaded = scroll_until_reviews(page)
            if review_loaded:
                first_review = page.locator(', '.join(REVIEW_SELECTORS["container"])).first
                first_review.scroll_into_view_if_needed()
                page.wait_for_timeout(1_000)
                if args.capture_screenshots:
                    page.screenshot(
                        path=str(args.output_dir / "review_page_viewport.png"),
                        full_page=False,
                    )
                    review_area = page.locator("#customer-reviews_feature_div")
                    if review_area.count() and review_area.first.is_visible():
                        review_area.first.screenshot(
                            path=str(args.output_dir / "review_section.png")
                        )

            html = page.content()
            page_type = classify_page(page, html)
            reviews, container_count = (parse_reviews(html, asin, page.url)
                                        if page_type == "product_page" else ([], 0))
            report.update(
                {
                    "http_status": response.status if response else None,
                    "page_title": page.title(),
                    "final_url": page.url,
                    "page_type": page_type,
                    "review_locator_loaded": review_loaded,
                    "review_containers_found": container_count,
                    "reviews_parsed": len(reviews),
                    "containers_not_exported": container_count - len(reviews),
                    "public_pagination_tested": False,
                    "public_pagination_available": None,
                }
            )

            if page_type != "product_page" or not reviews:
                save_browser_evidence(page, args.output_dir, "failed_product_page")
                raise RuntimeError(
                    f"No usable public reviews were found; page_type={page_type}. "
                    "Inspect browser_evidence before retrying."
                )

            report["export_validation"] = export_and_verify(reviews, args.output_dir)
            snapshot = args.output_dir / "source.html"
            snapshot.write_text(html, encoding="utf-8")
            report["source_html_sha256"] = hashlib.sha256(snapshot.read_bytes()).hexdigest()
            report["export_sha256"] = {name: hashlib.sha256((args.output_dir / name).read_bytes()).hexdigest()
                                       for name in ("amazon_reviews.csv", "amazon_reviews.json")}
            report["success"] = True
            report["field_quality"] = {
                "missing_review_id": sum(r.review_id is None for r in reviews),
                "missing_rating": sum(r.rating is None for r in reviews),
                "missing_title": sum(r.title is None for r in reviews),
                "missing_body": sum(not r.body for r in reviews),
                "missing_date_text": sum(r.date_text is None for r in reviews),
                "missing_review_date": sum(r.review_date is None for r in reviews),
                "missing_verified_purchase": sum(
                    r.verified_purchase is None for r in reviews
                ),
                "missing_helpful_votes": sum(r.helpful_votes is None for r in reviews),
            }

            if args.check_pagination:
                pagination_page = context.new_page()
                try:
                    check_public_page_two(
                        pagination_page, asin, origin, report, args.output_dir
                    )
                finally:
                    pagination_page.close()
        except (PlaywrightError, RuntimeError, ValueError, OSError) as exc:
            report["success"] = False
            report["error_type"] = type(exc).__name__
            # Do not persist raw exception strings, which can include proxy details.
            report.setdefault("page_type", "navigation_or_browser_error")
            if page is not None:
                try:
                    save_browser_evidence(page, args.output_dir, "failed_run")
                except (PlaywrightError, OSError):
                    report["evidence_capture_failed"] = True
            raise
        finally:
            (args.output_dir / "verification_report.json").write_text(
                json.dumps(report, indent=2, ensure_ascii=False), encoding="utf-8"
            )
            if browser is not None:
                browser.close()
            print(f"Run record: {args.output_dir.resolve()}")

    print(f"HTTP status: {report.get('http_status')}")
    print(f"Page type: {report.get('page_type')}")
    print(f"Review containers found: {report.get('review_containers_found')}")
    print(f"Reviews parsed: {report.get('reviews_parsed')}")
    print(
        "CSV round-trip validated: "
        f"{report.get('export_validation', {}).get('round_trip_validated')}"
    )
    if args.check_pagination:
        print(f"Pagination page type: {report.get('pagination_page_type')}")
        print(
            "Public pagination available: "
            f"{report.get('public_pagination_available')}"
        )
    print(f"Saved: {args.output_dir.resolve()}")
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(run())
    except (ValueError, RuntimeError, PlaywrightError, OSError) as exc:
        message = str(exc) if isinstance(exc, (ValueError, RuntimeError)) else "Browser, network, or file operation failed. See the run report; check proxy authentication, DNS, connection, and timeout settings."
        print(f"ERROR: {message}", file=sys.stderr)
        raise SystemExit(2)

python amazon_review_scraper.py `
  "https://www.amazon.com/dp/B07FZ8S74R" `
  --capture-screenshots `
  --output-dir output

python amazon_review_scraper.py "https://www.amazon.com/dp/B07FZ8S74R" --capture-screenshots --output-dir output

$Env:ROLA_PROXY_SERVER="http://HOST:PORT"
$Env:ROLA_PROXY_USERNAME="YOUR_USERNAME"
$Env:ROLA_PROXY_PASSWORD="YOUR_PASSWORD"
python amazon_review_scraper.py "https://www.amazon.com/dp/B07FZ8S74R"