Scraping Amazon Product Data: Methods, Tools, and Best Practices

1import boto3
2import json
3
4class AmazonProductAPI:
5    def __init__(self, access_key, secret_key, partner_tag):
6        self.client = boto3.client(
7            'paapi5',
8            access_key=access_key,
9            secret_key=secret_key,
10            region_name='us-east-1'
11        )
12        self.partner_tag = partner_tag
13    
14    def search_items(self, keywords):
15        try:
16            response = self.client.search_items(
17                PartnerTag=self.partner_tag,
18                PartnerType='Associates',
19                Keywords=keywords,
20                Resources=[
21                    'Images.Primary.Medium',
22                    'ItemInfo.Title',
23                    'Offers.Listings.Price',
24                    'CustomerReviews.Count',
25                    'CustomerReviews.StarRating'
26                ]
27            )
28            return self._parse_response(response)
29        except Exception as e:
30            print(f"API Error: {e}")
31            return None
32    
33    def _parse_response(self, response):
34        products = []
35        for item in response.get('SearchResult', {}).get('Items', []):
36            product = {
37                'asin': item.get('ASIN'),
38                'title': item.get('ItemInfo', {}).get('Title', {}).get('DisplayValue'),
39                'price': item.get('Offers', {}).get('Listings', [{}])[0].get('Price', {}).get('DisplayAmount'),
40                'rating': item.get('CustomerReviews', {}).get('StarRating'),
41                'review_count': item.get('CustomerReviews', {}).get('Count'),
42                'url': item.get('DetailPageURL')
43            }
44            products.append(product)
45        return products
46
47# Usage
48api = AmazonProductAPI(
49    access_key='YOUR_ACCESS_KEY',
50    secret_key='YOUR_SECRET_KEY',
51    partner_tag='YOUR_PARTNER_TAG'
52)
53products = api.search_items('wireless headphones')
54
55

1import requests
2from bs4 import BeautifulSoup
3import time
4import random
5import re
6from urllib.parse import urljoin, urlparse
7import json
8
9class AmazonScraper:
10    def __init__(self):
11        self.session = requests.Session()
12        self.set_headers()
13    
14    def set_headers(self):
15        self.headers = {
16            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
17            'Accept-Language': 'en-US,en;q=0.9',
18            'Accept-Encoding': 'gzip, deflate, br',
19            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
20            'Connection': 'keep-alive',
21            'Upgrade-Insecure-Requests': '1',
22        }
23        self.session.headers.update(self.headers)
24    
25    def random_delay(self):
26        time.sleep(random.uniform(2, 4))
27    
28    def scrape_product_page(self, url):
29        """Scrape individual product page"""
30        try:
31            self.random_delay()
32            response = self.session.get(url, timeout=10)
33            response.raise_for_status()
34            
35            soup = BeautifulSoup(response.content, 'html.parser')
36            
37            product_data = {
38                'url': url,
39                'timestamp': time.strftime('%Y-%m-%d %H:%M:%S'),
40                'title': self.extract_title(soup),
41                'price': self.extract_price(soup),
42                'availability': self.extract_availability(soup),
43                'rating': self.extract_rating(soup),
44                'review_count': self.extract_review_count(soup),
45                'description': self.extract_description(soup),
46                'images': self.extract_images(soup),
47                'specifications': self.extract_specifications(soup)
48            }
49            
50            return product_data
51            
52        except requests.RequestException as e:
53            print(f"Error fetching {url}: {e}")
54            return None
55    
56    def extract_title(self, soup):
57        # Multiple selectors for title
58        selectors = [
59            '#productTitle',
60            'h1.a-size-large',
61            '.a-size-medium.a-spacing-none'
62        ]
63        for selector in selectors:
64            element = soup.select_one(selector)
65            if element:
66                return element.get_text().strip()
67        return None
68    
69    def extract_price(self, soup):
70        # Multiple price selectors
71        price_selectors = [
72            '.a-price-whole',
73            '.a-price .a-offscreen',
74            '#priceblock_dealprice',
75            '#priceblock_ourprice'
76        ]
77        
78        for selector in price_selectors:
79            element = soup.select_one(selector)
80            if element:
81                price_text = element.get_text().strip()
82                # Clean price text
83                price = re.search(r'[\d,]+\.?\d*', price_text)
84                if price:
85                    return price.group()
86        return None
87    
88    def extract_rating(self, soup):
89        rating_element = soup.select_one('.a-icon-alt')
90        if rating_element:
91            rating_text = rating_element.get_text()
92            match = re.search(r'(\d+\.\d+)', rating_text)
93            if match:
94                return match.group(1)
95        return None
96    
97    def extract_review_count(self, soup):
98        review_element = soup.select_one('#acrCustomerReviewText')
99        if review_element:
100            count_text = review_element.get_text()
101            numbers = re.findall(r'\d+', count_text.replace(',', ''))
102            if numbers:
103                return int(numbers[0])
104        return None
105    
106    def extract_availability(self, soup):
107        availability_selectors = [
108            '#availability .a-color-success',
109            '#availability .a-color-price',
110            '#outOfStock'
111        ]
112        for selector in availability_selectors:
113            element = soup.select_one(selector)
114            if element:
115                return element.get_text().strip()
116        return "Available"
117    
118    def extract_description(self, soup):
119        # Try multiple description locations
120        description_selectors = [
121            '#productDescription',
122            '#feature-bullets',
123            '.a-plus-content'
124        ]
125        for selector in description_selectors:
126            element = soup.select_one(selector)
127            if element:
128                return element.get_text().strip()[:1000]  # Limit length
129        return None
130    
131    def extract_images(self, soup):
132        images = []
133        image_elements = soup.select('#landingImage, .a-dynamic-image')
134        for img in image_elements:
135            src = img.get('src') or img.get('data-src')
136            if src and 'http' in src:
137                images.append(src)
138        return images
139    
140    def extract_specifications(self, soup):
141        specs = {}
142        # Technical specifications table
143        table = soup.select_one('#productDetails_techSpec_section_1')
144        if table:
145            rows = table.select('tr')
146            for row in rows:
147                th = row.select_one('th')
148                td = row.select_one('td')
149                if th and td:
150                    key = th.get_text().strip()
151                    value = td.get_text().strip()
152                    specs[key] = value
153        return specs
154
155# Usage example
156scraper = AmazonScraper()
157product_data = scraper.scrape_product_page('https://www.amazon.com/dp/B08N5WRWNW')
158print(json.dumps(product_data, indent=2))
159
160

1
2import scrapy
3from scrapy.crawler import CrawlerProcess
4from scrapy.utils.project import get_project_settings
5import json
6import time
7
8class AmazonSearchSpider(scrapy.Spider):
9    name = 'amazon_search'
10    
11    custom_settings = {
12        'DOWNLOAD_DELAY': 3,
13        'CONCURRENT_REQUESTS_PER_DOMAIN': 1,
14        'ROBOTSTXT_OBEY': False,  # Note: This should typically be True
15        'USER_AGENT': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
16        'FEEDS': {
17            'products.json': {
18                'format': 'json',
19                'encoding': 'utf8',
20                'store_empty': False,
21                'fields': None,
22                'indent': 4,
23            }
24        }
25    }
26    
27    def __init__(self, keywords=None, *args, **kwargs):
28        super(AmazonSearchSpider, self).__init__(*args, **kwargs)
29        self.keywords = keywords or ['laptop']
30    
31    def start_requests(self):
32        base_url = "https://www.amazon.com/s"
33        
34        for keyword in self.keywords:
35            url = f"{base_url}?k={keyword.replace(' ', '+')}"
36            yield scrapy.Request(
37                url=url,
38                callback=self.parse_search_results,
39                meta={'keyword': keyword}
40            )
41    
42    def parse_search_results(self, response):
43        products = response.css('div[data-component-type="s-search-result"]')
44        
45        for product in products:
46            product_url = product.css('h2 a::attr(href)').get()
47            if product_url:
48                full_url = response.urljoin(product_url)
49                
50                yield scrapy.Request(
51                    url=full_url,
52                    callback=self.parse_product_page,
53                    meta={
54                        'keyword': response.meta['keyword'],
55                        'search_position': products.index(product) + 1
56                    }
57                )
58        
59        # Pagination
60        next_page = response.css('a.s-pagination-next::attr(href)').get()
61        if next_page:
62            yield response.follow(
63                next_page,
64                callback=self.parse_search_results,
65                meta={'keyword': response.meta['keyword']}
66            )
67    
68    def parse_product_page(self, response):
69        def extract_with_css(selector):
70            return response.css(selector).get(default='').strip()
71        
72        def extract_rating(text):
73            import re
74            match = re.search(r'(\d+\.\d+)', text)
75            return match.group(1) if match else None
76        
77        yield {
78            'keyword': response.meta['keyword'],
79            'search_position': response.meta['search_position'],
80            'url': response.url,
81            'title': extract_with_css('#productTitle::text'),
82            'price': extract_with_css('.a-price-whole::text'),
83            'rating': extract_rating(extract_with_css('.a-icon-alt::text')),
84            'review_count': extract_with_css('#acrCustomerReviewText::text'),
85            'timestamp': time.strftime('%Y-%m-%d %H:%M:%S'),
86        }
87
88# Run the spider
89if __name__ == "__main__":
90    process = CrawlerProcess(get_project_settings())
91    process.crawl(AmazonSearchSpider, keywords=['wireless headphones', 'smartwatch'])
92    process.start()
93
94

1
2from selenium import webdriver
3from selenium.webdriver.common.by import By
4from selenium.webdriver.support.ui import WebDriverWait
5from selenium.webdriver.support import expected_conditions as EC
6from selenium.webdriver.chrome.options import Options
7import time
8import json
9
10class SeleniumAmazonScraper:
11    def __init__(self, headless=True):
12        self.setup_driver(headless)
13    
14    def setup_driver(self, headless):
15        chrome_options = Options()
16        if headless:
17            chrome_options.add_argument('--headless')
18        chrome_options.add_argument('--no-sandbox')
19        chrome_options.add_argument('--disable-dev-shm-usage')
20        chrome_options.add_argument('--disable-blink-features=AutomationControlled')
21        chrome_options.add_experimental_option("excludeSwitches", ["enable-automation"])
22        chrome_options.add_experimental_option('useAutomationExtension', False)
23        chrome_options.add_argument('--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36')
24        
25        self.driver = webdriver.Chrome(options=chrome_options)
26        self.driver.execute_script("Object.defineProperty(navigator, 'webdriver', {get: () => undefined})")
27        self.wait = WebDriverWait(self.driver, 10)
28    
29    def scrape_product(self, url):
30        try:
31            self.driver.get(url)
32            
33            # Wait for critical elements
34            self.wait.until(
35                EC.presence_of_element_located((By.ID, "productTitle"))
36            )
37            
38            # Scroll to load dynamic content
39            self.driver.execute_script("window.scrollTo(0, document.body.scrollHeight/2);")
40            time.sleep(2)
41            
42            product_data = self.extract_product_data()
43            return product_data
44            
45        except Exception as e:
46            print(f"Error scraping {url}: {e}")
47            return None
48    
49    def extract_product_data(self):
50        return {
51            'title': self.get_element_text(By.ID, 'productTitle'),
52            'price': self.get_element_text(By.CSS_SELECTOR, '.a-price-whole'),
53            'rating': self.extract_rating(),
54            'review_count': self.get_element_text(By.ID, 'acrCustomerReviewText'),
55            'availability': self.get_availability(),
56            'description': self.get_description(),
57        }
58    
59    def get_element_text(self, by, selector):
60        try:
61            element = self.driver.find_element(by, selector)
62            return element.text.strip()
63        except:
64            return None
65    
66    def extract_rating(self):
67        try:
68            rating_element = self.driver.find_element(By.CSS_SELECTOR, '.a-icon-alt')
69            import re
70            match = re.search(r'(\d+\.\d+)', rating_element.text)
71            return match.group(1) if match else None
72        except:
73            return None
74    
75    def get_availability(self):
76        try:
77            availability = self.driver.find_element(By.ID, 'availability')
78            return availability.text.strip()
79        except:
80            return "Available"
81    
82    def get_description(self):
83        try:
84            # Try different description locations
85            selectors = [
86                '#productDescription',
87                '#feature-bullets',
88                '.a-plus-content'
89            ]
90            for selector in selectors:
91                try:
92                    element = self.driver.find_element(By.CSS_SELECTOR, selector)
93                    return element.text.strip()[:1000]
94                except:
95                    continue
96            return None
97        except:
98            return None
99    
100    def close(self):
101        self.driver.quit()
102
103# Usage
104scraper = SeleniumAmazonScraper()
105product_data = scraper.scrape_product('https://www.amazon.com/dp/B08N5WRWNW')
106print(json.dumps(product_data, indent=2))
107scraper.close()
108

1import random
2from fp.fp import FreeProxy
3
4class AntiDetectScraper:
5    def __init__(self):
6        self.session = requests.Session()
7        self.setup_stealth()
8    
9    def setup_stealth(self):
10        # Rotate user agents
11        user_agents = [
12            'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
13            'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
14            'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
15        ]
16        
17        self.session.headers.update({
18            'User-Agent': random.choice(user_agents),
19            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
20            'Accept-Language': 'en-US,en;q=0.5',
21            'Accept-Encoding': 'gzip, deflate, br',
22            'DNT': '1',
23            'Connection': 'keep-alive',
24            'Upgrade-Insecure-Requests': '1',
25            'Sec-Fetch-Dest': 'document',
26            'Sec-Fetch-Mode': 'navigate',
27            'Sec-Fetch-Site': 'none',
28            'Cache-Control': 'max-age=0',
29        })
30    
31    def get_proxy(self):
32        try:
33            proxy = FreeProxy(rand=True).get()
34            return {'http': proxy, 'https': proxy}
35        except:
36            return None
37    
38    def human_like_delay(self):
39        time.sleep(random.uniform(3, 7))
40    
41    def scrape_with_retry(self, url, max_retries=3):
42        for attempt in range(max_retries):
43            try:
44                self.human_like_delay()
45                
46                # Rotate proxy every few requests
47                if attempt % 2 == 0:
48                    proxy = self.get_proxy()
49                else:
50                    proxy = None
51                
52                response = self.session.get(
53                    url, 
54                    timeout=10,
55                    proxies=proxy
56                )
57                
58                if response.status_code == 200:
59                    if "[email protected]" in response.text:
60                        print("Bot detection triggered")
61                        continue
62                    return response
63                
64            except Exception as e:
65                print(f"Attempt {attempt + 1} failed: {e}")
66        
67        return None
68

1class CaptchaSolver:
2    def __init__(self, api_key):
3        self.api_key = api_key
4    
5    def check_captcha(self, response_text):
6        captcha_indicators = [
7            "Type the characters you see in this image",
8            "Enter the characters you see below",
9            "captcha",
10            "CAPTCHA"
11        ]
12        return any(indicator in response_text for indicator in captcha_indicators)
13    
14    def solve_captcha(self, image_url):
15        # Integrate with captcha solving service
16        # This is a placeholder for services like 2captcha, anti-captcha
17        pass
18

1import re
2from datetime import datetime
3
4class DataParser:
5    @staticmethod
6    def clean_price(price_text):
7        if not price_text:
8            return None
9        # Remove currency symbols and commas
10        cleaned = re.sub(r'[^\d.]', '', price_text)
11        try:
12            return float(cleaned)
13        except ValueError:
14            return None
15    
16    @staticmethod
17    def extract_rating(rating_text):
18        if not rating_text:
19            return None
20        match = re.search(r'(\d+\.\d+)', rating_text)
21        return float(match.group(1)) if match else None
22    
23    @staticmethod
24    def extract_review_count(count_text):
25        if not count_text:
26            return None
27        numbers = re.findall(r'\d+', count_text.replace(',', ''))
28        return int(numbers[0]) if numbers else None
29    
30    @staticmethod
31    def parse_date(date_text):
32        # Parse various date formats
33        formats = [
34            '%B %d, %Y',
35            '%b %d, %Y',
36            '%Y-%m-%d'
37        ]
38        for fmt in formats:
39            try:
40                return datetime.strptime(date_text, fmt)
41            except ValueError:
42                continue
43        return None
44

1class DataValidator:
2    @staticmethod
3    def validate_product_data(product_data):
4        required_fields = ['title', 'price', 'url']
5        missing_fields = [field for field in required_fields if not product_data.get(field)]
6        
7        if missing_fields:
8            return False, f"Missing fields: {missing_fields}"
9        
10        # Validate price format
11        if product_data.get('price'):
12            try:
13                float(product_data['price'])
14            except ValueError:
15                return False, "Invalid price format"
16        
17        return True, "Valid"
18

1import time
2from collections import deque
3from threading import Lock
4
5class RateLimiter:
6    def __init__(self, max_requests, time_window):
7        self.max_requests = max_requests
8        self.time_window = time_window
9        self.requests = deque()
10        self.lock = Lock()
11    
12    def wait_if_needed(self):
13        with self.lock:
14            now = time.time()
15            
16            # Remove old requests
17            while self.requests and self.requests[0] < now - self.time_window:
18                self.requests.popleft()
19            
20            # Check if we've exceeded the limit
21            if len(self.requests) >= self.max_requests:
22                sleep_time = self.requests[0] + self.time_window - now
23                if sleep_time > 0:
24                    time.sleep(sleep_time)
25            
26            # Add current request
27            self.requests.append(now)
28
29# Usage
30limiter = RateLimiter(max_requests=5, time_window=60)  # 5 requests per minute
31
32def respectful_request(url):
33    limiter.wait_if_needed()
34    # Make request here
35

1import logging
2from tenacity import retry, stop_after_attempt, wait_exponential
3
4logging.basicConfig(level=logging.INFO)
5logger = logging.getLogger(__name__)
6
7class RobustScraper:
8    @retry(
9        stop=stop_after_attempt(3),
10        wait=wait_exponential(multiplier=1, min=4, max=10)
11    )
12    def scrape_with_retry(self, url):
13        try:
14            response = self.session.get(url, timeout=10)
15            response.raise_for_status()
16            
17            if self.is_blocked(response):
18                logger.warning("Request blocked, retrying...")
19                raise Exception("Blocked by Amazon")
20            
21            return response
22            
23        except requests.RequestException as e:
24            logger.error(f"Request failed: {e}")
25            raise
26    
27    def is_blocked(self, response):
28        block_indicators = [
29            "To discuss automated access to Amazon data",
30            "api-services-support@amazon.com",
31            "Sorry, we just need to make sure you're not a robot"
32        ]
33        return any(indicator in response.text for indicator in block_indicators)
34

1import sqlite3
2import csv
3import json
4from contextlib import contextmanager
5
6class DataStorage:
7    def __init__(self, db_path='amazon_data.db'):
8        self.db_path = db_path
9        self.init_database()
10    
11    def init_database(self):
12        with self.get_connection() as conn:
13            conn.execute('''
14                CREATE TABLE IF NOT EXISTS products (
15                    id INTEGER PRIMARY KEY AUTOINCREMENT,
16                    asin TEXT UNIQUE,
17                    title TEXT,
18                    price REAL,
19                    rating REAL,
20                    review_count INTEGER,
21                    url TEXT,
22                    timestamp DATETIME,
23                    created_at DATETIME DEFAULT CURRENT_TIMESTAMP
24                )
25            ''')
26    
27    @contextmanager
28    def get_connection(self):
29        conn = sqlite3.connect(self.db_path)
30        try:
31            yield conn
32            conn.commit()
33        finally:
34            conn.close()
35    
36    def save_product(self, product_data):
37        with self.get_connection() as conn:
38            conn.execute('''
39                INSERT OR REPLACE INTO products 
40                (asin, title, price, rating, review_count, url, timestamp)
41                VALUES (?, ?, ?, ?, ?, ?, ?)
42            ''', (
43                product_data.get('asin'),
44                product_data.get('title'),
45                product_data.get('price'),
46                product_data.get('rating'),
47                product_data.get('review_count'),
48                product_data.get('url'),
49                product_data.get('timestamp')
50            ))
51    
52    def export_to_csv(self, filename):
53        with self.get_connection() as conn:
54            cursor = conn.execute('SELECT * FROM products')
55            with open(filename, 'w', newline='', encoding='utf-8') as f:
56                writer = csv.writer(f)
57                writer.writerow([description[0] for description in cursor.description])
58                writer.writerows(cursor)
59

1import requests
2
3def scrape_with_scraperapi(url, api_key):
4    payload = {
5        'api_key': api_key,
6        'url': url,
7        'country_code': 'us',
8        'render': 'true'
9    }
10    
11    response = requests.get('http://api.scraperapi.com', params=payload)
12    return response.content
13

1import keepa
2
3def use_keepa_api(access_key, asins):
4    api = keepa.Keepa(access_key)
5    products = api.query(asins, offers=20)
6    return products
7

1# AWS Lambda example for serverless scraping
2import boto3
3import json
4
5def lambda_handler(event, context):
6    # Scraping logic here
7    product_data = scrape_product(event['url'])
8    
9    # Store in S3
10    s3 = boto3.client('s3')
11    s3.put_object(
12        Bucket='amazon-data-bucket',
13        Key=f"products/{event['asin']}.json",
14        Body=json.dumps(product_data)
15    )
16    
17    return {
18        'statusCode': 200,
19        'body': json.dumps({'message': 'Success'})
20    }
21