How to Scrape Amazon Reviews With Python (2026)

https://www.amazon.com/dp/B07FZ8S74R

python -m venv .venv
.\.venv\Scripts\Activate.ps1
python -m pip install -r requirements.txt
python -m playwright install chromium

python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt
python -m playwright install chromium

ASIN_RE = re.compile(
    r"/(?:dp|gp/product|product-reviews)/([A-Z0-9]{10})(?:[/?]|$)", re.I
)

ALLOWED_AMAZON_MARKETPLACES = {
    "amazon.com",
    "amazon.ca",
    "amazon.co.uk",
    "amazon.de",
    "amazon.fr",
    "amazon.co.jp",
}

def scroll_until_reviews(page, max_steps=60):
    reviews = page.locator('div[data-hook="review"], div[id^="customer_review-"]')
    for _ in range(max_steps):
        if reviews.count() > 0:
            return True
        page.evaluate(
            """
            () => window.scrollBy(
              0, Math.max(600, Math.floor(window.innerHeight * 0.75))
            )
            """
        )
        page.wait_for_timeout(250)
    reviews.first.wait_for(state="attached", timeout=15_000)
    return True

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)

from __future__ import annotations

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

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


SCRIPT_VERSION = "1.0.2"
ASIN_RE = re.compile(
    r"/(?:dp|gp/product|product-reviews)/([A-Z0-9]{10})(?:[/?]|$)", re.I
)
ALLOWED_AMAZON_MARKETPLACES = {
    "amazon.ae",
    "amazon.ca",
    "amazon.cn",
    "amazon.co.jp",
    "amazon.co.uk",
    "amazon.com",
    "amazon.com.au",
    "amazon.com.be",
    "amazon.com.br",
    "amazon.com.mx",
    "amazon.com.tr",
    "amazon.de",
    "amazon.eg",
    "amazon.es",
    "amazon.fr",
    "amazon.ie",
    "amazon.in",
    "amazon.it",
    "amazon.nl",
    "amazon.pl",
    "amazon.sa",
    "amazon.se",
    "amazon.sg",
}
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_target(url: str) -> tuple[str, str]:
    parsed = urlparse(url)
    if parsed.scheme not in {"http", "https"} or not parsed.hostname:
        raise ValueError("Enter a complete HTTP or HTTPS URL.")
    hostname = parsed.hostname.lower().rstrip(".")
    marketplace = hostname.removeprefix("www.")
    if marketplace not in ALLOWED_AMAZON_MARKETPLACES:
        raise ValueError("The hostname must be an explicitly supported Amazon marketplace.")
    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"{parsed.scheme}://{hostname}/dp/{asin}"


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 spreadsheet_safe(value: Any) -> Any:
    """Prevent spreadsheet software from evaluating untrusted text as a formula."""
    if isinstance(value, str) and value.startswith(("=", "+", "-", "@")):
        return "'" + value
    return value


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:
    sample = f"{page.title()}\n{html}".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 "validatecaptcha" in sample or ("captcha" in sample and "amazon" in sample):
        return "captcha"
    if "/ap/signin" in page.url or (
        "sign in or create account" in sample and 'name="email"' in html.lower()
    ):
        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"
    return "unexpected_page"


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 review_containers_locator(page: Page):
    return page.locator('div[data-hook="review"], div[id^="customer_review-"]')

def scroll_until_reviews(page: Page, max_steps: int = 60) -> bool:
    review_locator = review_containers_locator(page)
    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 export_and_verify(reviews: list[Review], output_dir: Path) -> dict[str, Any]:
    rows = [asdict(review) for review in reviews]
    spreadsheet_rows = [
        {key: spreadsheet_safe(value) for key, value in row.items()}
        for row in rows
    ]
    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(spreadsheet_rows)

    with csv_path.open("r", newline="", encoding="utf-8-sig") as handle:
        csv_rows = list(csv.DictReader(handle))
    json_rows = json.loads(json_path.read_text(encoding="utf-8"))
    if len(csv_rows) != len(reviews) or len(json_rows) != len(reviews):
        raise RuntimeError("Round-trip validation failed: exported row counts differ.")

    source_keys = [
        row["review_id"]
        or hashlib.sha256(
            f"{row['title']}|{row['body']}|{row['date_text']}".encode("utf-8")
        ).hexdigest()
        for row in rows
    ]
    if len(source_keys) != len(set(source_keys)):
        raise RuntimeError("Round-trip validation failed: duplicate review keys remain.")
    if any(not row["body"] for row in csv_rows):
        raise RuntimeError("Round-trip validation failed: an exported body is empty.")

    return {
        "csv_rows_reopened": len(csv_rows),
        "json_rows_reopened": len(json_rows),
        "unique_record_keys": len(set(source_keys)),
        "round_trip_validated": True,
    }


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
    try:
        response = page.goto(url, wait_until="commit", timeout=30_000)
        page.locator("body").wait_for(state="attached", timeout=15_000)
        page.wait_for_timeout(3_000)
        html = page.content()
        reviews, containers = parse_reviews(html, asin, page.url)
        report.update(
            {
                "public_pagination_tested": True,
                "pagination_http_status": response.status if response else None,
                "pagination_final_url": page.url,
                "pagination_page_type": classify_page(page, html),
                "pagination_review_containers": containers,
                "pagination_reviews_parsed": len(reviews),
                "public_pagination_available": bool(reviews),
            }
        )
        save_browser_evidence(page, output_dir, "pagination_page_2")
    except PlaywrightTimeoutError:
        report.update(
            {
                "public_pagination_tested": True,
                "public_pagination_available": False,
                "pagination_error": "navigation_timeout",
            }
        )
        try:
            save_browser_evidence(page, output_dir, "pagination_timeout")
        except PlaywrightTimeoutError:
            pass


def run() -> int:
    args = parse_args()
    args.output_dir.mkdir(parents=True, exist_ok=True)
    asin, normalized_url = validate_target(args.url)
    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],
        "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 = launch_browser(playwright, args.headed)
        context = browser.new_context(
            locale="en-US",
            timezone_id="America/New_York",
            viewport={"width": 1440, "height": 900},
        )
        page = context.new_page()
        response = None
        try:
            response = page.goto(
                normalized_url, wait_until="domcontentloaded", timeout=90_000
            )
            page.locator("body").wait_for(state="attached", timeout=15_000)
            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 = review_containers_locator(page).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)
            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)
            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()
        finally:
            (args.output_dir / "verification_report.json").write_text(
                json.dumps(report, indent=2, ensure_ascii=False), encoding="utf-8"
            )
            browser.close()

    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, PlaywrightTimeoutError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        raise SystemExit(2)

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

HTTP status: 200
Page type: product_page
Review containers found: 8
Reviews parsed: 8
CSV rows reopened: 8
Unique record keys: 8
Round-trip validated: True

$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"