How to Scrape Amazon Without Getting Blocked (2026 Guide)

python3 -m venv amazon-scraper
source amazon-scraper/bin/activate
pip install curl_cffi selectolax fake-useragent

from curl_cffi import requests
from fake_useragent import UserAgent
import random
import time

ua = UserAgent(browsers=['Chrome'], os=['Windows', 'macOS'])

def get_headers():
    user_agent = ua.random
    return {
        'User-Agent': user_agent,
        'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8',
        'Accept-Language': 'en-US,en;q=0.9',
        'Accept-Encoding': 'gzip, deflate, br',
        'DNT': '1',
        'Connection': 'keep-alive',
        'Upgrade-Insecure-Requests': '1',
        'Sec-Fetch-Dest': 'document',
        'Sec-Fetch-Mode': 'navigate',
        'Sec-Fetch-Site': 'none',
        'Sec-Fetch-User': '?1',
        'Cache-Control': 'max-age=0',
    }

def scrape_amazon(url, proxy=None):
    session = requests.Session(impersonate='chrome')
    proxies = {'https': proxy} if proxy else None
    
    response = session.get(
        url,
        headers=get_headers(),
        proxies=proxies,
        timeout=30
    )
    return response

import random
from collections import defaultdict
from datetime import datetime, timedelta

class ProxyManager:
    def __init__(self, proxies: list[str]):
        self.proxies = proxies
        self.cooldowns: dict[str, datetime] = {}
        self.failures: dict[str, int] = defaultdict(int)
    
    def get_proxy(self) -> str | None:
        available = [
            p for p in self.proxies
            if p not in self.cooldowns 
            or datetime.now() > self.cooldowns[p]
        ]
        if not available:
            return None
        # Weight toward proxies with fewer failures
        weights = [1 / (self.failures[p] + 1) for p in available]
        return random.choices(available, weights=weights, k=1)[0]
    
    def report_failure(self, proxy: str, is_captcha: bool = False):
        self.failures[proxy] += 1
        cooldown = timedelta(minutes=30) if is_captcha else timedelta(minutes=5)
        self.cooldowns[proxy] = datetime.now() + cooldown
    
    def report_success(self, proxy: str):
        self.failures[proxy] = max(0, self.failures[proxy] - 1)

import random
import time

def human_delay(min_seconds=1.5, max_seconds=5.0):
    """Generate a human-like delay using a log-normal distribution."""
    # Log-normal produces occasional longer pauses, mimicking
    # a real user who sometimes reads the page longer
    delay = random.lognormvariate(0.5, 0.5)
    delay = max(min_seconds, min(delay, max_seconds))
    time.sleep(delay)

def scrape_product_list(urls: list[str], proxy_manager: ProxyManager):
    results = []
    for url in urls:
        proxy = proxy_manager.get_proxy()
        if not proxy:
            print('All proxies on cooldown. Waiting...')
            time.sleep(60)
            proxy = proxy_manager.get_proxy()
        
        try:
            response = scrape_amazon(url, proxy=proxy)
            if 'captcha' in response.text.lower():
                proxy_manager.report_failure(proxy, is_captcha=True)
                continue
            proxy_manager.report_success(proxy)
            results.append(response.text)
        except Exception as e:
            proxy_manager.report_failure(proxy)
            print(f'Error: {e}')
        
        human_delay()
    return results

from selectolax.parser import HTMLParser

def parse_product_page(html: str) -> dict:
    tree = HTMLParser(html)
    
    def text(selector: str) -> str:
        node = tree.css_first(selector)
        return node.text(strip=True) if node else ''
    
    def attr(selector: str, attribute: str) -> str:
        node = tree.css_first(selector)
        return node.attributes.get(attribute, '') if node else ''
    
    return {
        'title': text('#productTitle'),
        'price': text('.a-price .a-offscreen'),
        'rating': text('#acrPopover .a-icon-alt'),
        'review_count': text('#acrCustomerReviewText'),
        'availability': text('#availability span'),
        'brand': text('#bylineInfo'),
        'asin': attr('[data-asin]', 'data-asin'),
        'image_url': attr('#landingImage', 'src'),
        'features': [
            li.text(strip=True) 
            for li in tree.css('#feature-bullets li span.a-list-item')
        ],
    }

def detect_block(response) -> str | None:
    """Returns block type or None if response is clean."""
    if response.status_code == 503:
        return 'rate_limited'
    
    body = response.text.lower()
    if 'captcha' in body or 'validatecaptcha' in body:
        return 'captcha'
    if 'sorry, we just need to make sure' in body:
        return 'robot_check'
    if '[email protected]' in body:
        return 'hard_block'
    if response.url and 'robot-check' in str(response.url):
        return 'robot_check'
    
    return None