How to Scrape Amazon Without Getting Blocked

import requests
import time
import random

PROXY_URL = "http://YOUR_USERNAME:[email protected]:8080"

HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36",
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
    "Accept-Language": "en-US,en;q=0.9",
    "Accept-Encoding": "gzip, deflate, br",
    "Connection": "keep-alive",
    "Upgrade-Insecure-Requests": "1",
    "Sec-Fetch-Dest": "document",
    "Sec-Fetch-Mode": "navigate",
    "Sec-Fetch-Site": "none",
    "Sec-Fetch-User": "?1",
}

def fetch_amazon(url: str, retries: int = 3) -> str | None:
    for attempt in range(retries):
        try:
            response = requests.get(
                url,
                headers=HEADERS,
                proxies={"http": PROXY_URL, "https": PROXY_URL},
                timeout=30,
            )
            if response.status_code == 200 and "captcha" not in response.url:
                return response.text
            if response.status_code == 503:
                time.sleep(2 ** attempt + random.uniform(0, 1))
                continue
        except requests.RequestException:
            time.sleep(1)
    return None

from bs4 import BeautifulSoup
import re

def parse_product(html: str) -> dict:
    soup = BeautifulSoup(html, "lxml")

    def get_text(selector: str) -> str:
        el = soup.select_one(selector)
        return el.get_text(strip=True) if el else ""

    # Title
    title = get_text("#productTitle")

    # Price — Amazon uses multiple possible selectors depending on variant
    price_raw = (
        get_text(".a-price .a-offscreen") or
        get_text("#priceblock_ourprice") or
        get_text("#priceblock_dealprice") or
        get_text(".apexPriceToPay .a-offscreen")
    )
    price = re.sub(r"[^\d.]", "", price_raw)

    # Rating
    rating_raw = get_text(".a-icon-star .a-icon-alt")
    rating_match = re.search(r"([\d.]+) out of", rating_raw)
    rating = rating_match.group(1) if rating_match else ""

    # Review count
    reviews_raw = get_text("#acrCustomerReviewText")
    review_count = re.sub(r"[^\d]", "", reviews_raw)

    # ASIN
    asin_match = re.search(r'/dp/([A-Z0-9]{10})', soup.find("link", {"rel": "canonical"})["href"]) if soup.find("link", {"rel": "canonical"}) else None
    asin = asin_match.group(1) if asin_match else ""

    return {
        "title": title,
        "price": price,
        "rating": rating,
        "review_count": review_count,
        "asin": asin,
    }


# Usage
html = fetch_amazon("https://www.amazon.com/dp/B0CHWRXH8B")
if html:
    product = parse_product(html)
    print(product)

import asyncio
import httpx
import random
import time
from typing import Optional

PROXY_URL = "http://USERNAME:[email protected]:8080"
BASE_URL = "https://www.amazon.com/dp/{asin}"

async def fetch_asin(client: httpx.AsyncClient, asin: str, semaphore: asyncio.Semaphore) -> dict:
    async with semaphore:
        url = BASE_URL.format(asin=asin)
        try:
            response = await client.get(url, timeout=30)
            if response.status_code == 200 and "captcha" not in str(response.url):
                data = parse_product(response.text)
                data["asin"] = asin
                data["status"] = "ok"
            else:
                data = {"asin": asin, "status": f"blocked_{response.status_code}"}
        except Exception as e:
            data = {"asin": asin, "status": f"error_{type(e).__name__}"}

        # Randomized delay between requests
        await asyncio.sleep(random.uniform(1.5, 4.0))
        return data

async def scrape_asins(asin_list: list[str], concurrency: int = 5) -> list[dict]:
    semaphore = asyncio.Semaphore(concurrency)
    async with httpx.AsyncClient(
        proxy=PROXY_URL,
        headers=HEADERS,
        follow_redirects=True,
    ) as client:
        tasks = [fetch_asin(client, asin, semaphore) for asin in asin_list]
        results = await asyncio.gather(*tasks)
    return list(results)

# Example: scrape 20 ASINs
asins = ["B0CHWRXH8B", "B09B8V1LZ3", "B08J5F3G18"]  # replace with your list
results = asyncio.run(scrape_asins(asins, concurrency=5))
print(f"Scraped {len([r for r in results if r.get('status') == 'ok'])} / {len(asins)} products")

def is_captcha(response: requests.Response) -> bool:
    return (
        "captcha" in response.url or
        "api-services-support" in response.url or
        "Type the characters you see in this image" in response.text or
        "Enter the characters you see below" in response.text
    )

def is_blocked(response: requests.Response) -> bool:
    return response.status_code in (403, 503) or is_captcha(response)

# In your scraping loop:
response = requests.get(url, ...)
if is_blocked(response):
    # The IP got flagged — the rotating gateway will assign a new one next request
    time.sleep(random.uniform(5, 15))
    continue