Scraping Amazon Product Data: The Complete Guide

import requests
from bs4 import BeautifulSoup

headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
    'Accept-Language': 'en-US,en;q=0.9',
    'Accept-Encoding': 'gzip, deflate, br',
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
    'Connection': 'keep-alive',
}

def scrape_amazon_product(asin):
    url = f'https://www.amazon.com/dp/{asin}'
    response = requests.get(url, headers=headers)

    if response.status_code != 200:
        return None

    soup = BeautifulSoup(response.text, 'html.parser')

    return {
        'title': soup.select_one('#productTitle').text.strip() if soup.select_one('#productTitle') else None,
        'price': soup.select_one('.a-price .a-offscreen').text if soup.select_one('.a-price .a-offscreen') else None,
        'rating': soup.select_one('.a-icon-star-small .a-icon-alt').text if soup.select_one('.a-icon-star-small .a-icon-alt') else None,
    }

from playwright.sync_api import sync_playwright
import random
import time

class AmazonScraper:
    def __init__(self):
        self.playwright = sync_playwright().start()
        self.browser = self.playwright.chromium.launch(
            headless=True,
            args=['--disable-blink-features=AutomationControlled']
        )
        self.context = self.browser.new_context(
            viewport={'width': 1920, 'height': 1080},
            user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        )

    def scrape_product(self, asin):
        page = self.context.new_page()

        try:
            url = f'https://www.amazon.com/dp/{asin}'
            page.goto(url, wait_until='domcontentloaded')

            # Random delay to appear human
            time.sleep(random.uniform(2, 4))

            # Wait for key elements
            page.wait_for_selector('#productTitle', timeout=10000)

            data = {
                'asin': asin,
                'title': page.locator('#productTitle').text_content().strip(),
                'price': self._extract_price(page),
                'rating': self._extract_rating(page),
                'review_count': self._extract_review_count(page),
                'bsr': self._extract_bsr(page),
                'availability': self._extract_availability(page),
            }

            return data

        except Exception as e:
            print(f"Error scraping {asin}: {e}")
            return None
        finally:
            page.close()

    def _extract_price(self, page):
        selectors = [
            '.a-price .a-offscreen',
            '#priceblock_ourprice',
            '#priceblock_dealprice',
            '.a-price-whole'
        ]
        for selector in selectors:
            element = page.locator(selector).first
            if element.count() > 0:
                return element.text_content().strip()
        return None

    def _extract_rating(self, page):
        element = page.locator('[data-hook="rating-out-of-text"]').first
        if element.count() > 0:
            text = element.text_content()
            return float(text.split()[0])
        return None

    def _extract_review_count(self, page):
        element = page.locator('#acrCustomerReviewText').first
        if element.count() > 0:
            text = element.text_content()
            return int(text.replace(',', '').split()[0])
        return None

    def _extract_bsr(self, page):
        element = page.locator('#productDetails_detailBullets_sections1 tr:has-text("Best Sellers Rank")').first
        if element.count() > 0:
            return element.text_content().strip()
        return None

    def _extract_availability(self, page):
        element = page.locator('#availability').first
        if element.count() > 0:
            return element.text_content().strip()
        return None

    def close(self):
        self.browser.close()
        self.playwright.stop()

import random

class ProxyRotator:
    def __init__(self, proxies):
        self.proxies = proxies
        self.current_index = 0

    def get_next(self):
        proxy = self.proxies[self.current_index]
        self.current_index = (self.current_index + 1) % len(self.proxies)
        return proxy

    def get_random(self):
        return random.choice(self.proxies)

# Use residential proxies for Amazon
proxies = [
    'http://user:pass@residential1.proxy.com:8080',
    'http://user:pass@residential2.proxy.com:8080',
    # Add more proxies
]

rotator = ProxyRotator(proxies)

# In Playwright
context = browser.new_context(
    proxy={'server': rotator.get_next()}
)

import random
import time

def human_delay():
    """Random delay that mimics human browsing patterns."""
    base_delay = random.uniform(3, 7)
    # Occasionally take longer breaks
    if random.random() < 0.1:
        base_delay += random.uniform(10, 30)
    time.sleep(base_delay)

def scrape_with_delays(asins):
    results = []
    for asin in asins:
        result = scrape_product(asin)
        results.append(result)
        human_delay()
    return results

import random

USER_AGENTS = [
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0',
]

VIEWPORTS = [
    {'width': 1920, 'height': 1080},
    {'width': 1366, 'height': 768},
    {'width': 1536, 'height': 864},
    {'width': 1440, 'height': 900},
]

def create_stealth_context(browser):
    return browser.new_context(
        user_agent=random.choice(USER_AGENTS),
        viewport=random.choice(VIEWPORTS),
        locale='en-US',
        timezone_id='America/New_York',
        permissions=['geolocation'],
        geolocation={'latitude': 40.7128, 'longitude': -74.0060},
    )

import json
import csv
from datetime import datetime
from dataclasses import dataclass, asdict
from typing import Optional, List

@dataclass
class AmazonProduct:
    asin: str
    title: Optional[str]
    price: Optional[float]
    currency: str = 'USD'
    rating: Optional[float]
    review_count: Optional[int]
    bsr_rank: Optional[int]
    bsr_category: Optional[str]
    availability: Optional[str]
    seller: Optional[str]
    is_prime: bool = False
    scraped_at: str = None

    def __post_init__(self):
        if not self.scraped_at:
            self.scraped_at = datetime.utcnow().isoformat()

class AmazonDataPipeline:
    def __init__(self, output_format='json'):
        self.output_format = output_format
        self.products: List[AmazonProduct] = []

    def add(self, product: AmazonProduct):
        self.products.append(product)

    def save(self, filename):
        if self.output_format == 'json':
            self._save_json(filename)
        elif self.output_format == 'csv':
            self._save_csv(filename)

    def _save_json(self, filename):
        with open(filename, 'w') as f:
            json.dump([asdict(p) for p in self.products], f, indent=2)

    def _save_csv(self, filename):
        if not self.products:
            return
        with open(filename, 'w', newline='') as f:
            writer = csv.DictWriter(f, fieldnames=asdict(self.products[0]).keys())
            writer.writeheader()
            for product in self.products:
                writer.writerow(asdict(product))

# Usage
scraper = AmazonScraper()
pipeline = AmazonDataPipeline(output_format='json')

asins = ['B09V3KXJPB', 'B0BDJFV5V5', 'B09JQMJHXY']

for asin in asins:
    data = scraper.scrape_product(asin)
    if data:
        product = AmazonProduct(**data)
        pipeline.add(product)

pipeline.save('amazon_products.json')
scraper.close()