amazon_keyboard_scraper.py

"""
Amazon PC Component Scraper Script (Async)

This script extracts product data from Amazon's search results for PC components
using async/parallel execution. It scrapes 10 pages for each of 8 component categories
and stores the data in Elasticsearch.

Features:
- Scrapes 8 PC component categories (CPU, GPU, RAM, Motherboard, SSD, PSU, Case, Cooler)
- 10 pages per component (80 total pages)
- Async parallel execution for fast scraping (pages within component run in parallel)
- Extracts: name, price, review stars, number of reviews, Prime availability
- Stores data in Elasticsearch with component category tags
- Performs analytical queries on the dataset
- Robust error handling for scraping and storage failures

Usage:
    python amazon_keyboard_scraper.py

Requirements:
    - Elasticsearch running on localhost:9200
    - ScrapeGraphAI API key (set as SGAI_API_KEY environment variable)
    - scrapegraph-py SDK with AsyncClient support
"""

import sys
import os
import time
import re
import traceback
import asyncio
from typing import List, Dict, Any, Optional
from datetime import datetime
from pydantic import BaseModel

# Add parent directory to path
sys.path.insert(0, os.path.abspath(os.path.dirname(__file__)))

from src.scrapegraph_demo import Config, ElasticsearchClient
from src.scrapegraph_demo.models import Product


# Pydantic models for ScrapeGraph API output schema
class ProductItem(BaseModel):
    """Single product item from Amazon search results"""
    name: str
    price: float
    rating: Optional[float] = None
    review_count: Optional[int] = None
    prime: bool = False
    product_url: str = ""
    asin: str = ""


class ProductList(BaseModel):
    """List of products from search results"""
    products: List[ProductItem]


class AmazonPCComponentScraper:
    """Async scraper for Amazon PC components with Elasticsearch integration"""

    # PC component categories to scrape
    PC_COMPONENTS = [
        "cpu",
        "gpu",
        "motherboard",
        "ram",
        "ssd",
        "power supply",
        "pc case",
        "cpu cooler"
    ]

    # Default API key (can be overridden with SGAI_API_KEY environment variable)
    DEFAULT_API_KEY = "sgai-763dcc80-3a64-417f-b9bf-b98c8f50cc4b"
    MARKETPLACE = "Amazon IT"
    PAGES_PER_COMPONENT = 10
    
    def __init__(self):
        """Initialize the scraper configuration"""
        # Get API key from environment or use default
        # Users can set SGAI_API_KEY environment variable to override the default
        self.api_key = os.environ.get('SGAI_API_KEY', self.DEFAULT_API_KEY)
        os.environ['SGAI_API_KEY'] = self.api_key

        # Load configuration
        self.config = Config.from_env()

        # Initialize Elasticsearch client
        self.es_client = ElasticsearchClient(self.config)

        # Statistics (per component)
        self.component_stats = {}
        self.total_scraped = 0
        self.total_stored = 0
        self.failed_pages = []
        self.failed_products = []
    
    def _generate_search_url(self, component: str, page_num: int) -> str:
        """
        Generate Amazon search URL for a component and page

        Args:
            component: PC component category (e.g., "cpu", "gpu")
            page_num: Page number (1-10)

        Returns:
            Amazon search URL
        """
        return f"https://www.amazon.it/s?k={component.replace(' ', '+')}&page={page_num}"

    async def scrape_page(self, client, component: str, page_num: int) -> List[Product]:
        """
        Scrape a single page of Amazon search results (async)

        Args:
            client: AsyncClient instance
            component: PC component category
            page_num: Page number to scrape (1-10)

        Returns:
            List of Product objects (empty list if scraping fails)
        """
        page_url = self._generate_search_url(component, page_num)
        print(f"  📄 [{component}] Page {page_num}/{self.PAGES_PER_COMPONENT}: {page_url}")

        products = []

        try:
            products = await self._scrape_with_api(client, page_url, component, page_num)
            print(f"  ✓ [{component}] Page {page_num}: Found {len(products)} products")
            self.total_scraped += len(products)
        except Exception as e:
            print(f"  ✗ [{component}] Page {page_num} error: {str(e)}")
            self.failed_pages.append({
                "component": component,
                "page": page_num,
                "error": str(e)
            })

        return products
    
    async def _scrape_with_api(self, client, page_url: str, component: str, page_num: int) -> List[Product]:
        """Scrape using ScrapeGraph API (async)"""

        # Define the prompt for extracting product information
        prompt = f"""
        Extract all {component} products from this Amazon search results page.
        For each product, extract:
        - name: Product name/title
        - price: Price in EUR (numeric value only, without currency symbol)
        - rating: Star rating out of 5 (e.g., 4.5)
        - review_count: Number of customer reviews/ratings
        - prime: Whether the product has Prime delivery (true/false)
        - product_url: Relative or full product URL
        - asin: Amazon ASIN (product ID) if visible

        Return a list of products with these fields.
        """

        # Call ScrapeGraph API with Pydantic model as output schema (async)
        response = await client.smartscraper(
            website_url=page_url,
            user_prompt=prompt,
            output_schema=ProductList
        )
        
        # Parse response
        products = []
        if response and isinstance(response, dict) and 'result' in response:
            result_data = response.get('result', {})
            if isinstance(result_data, dict) and 'products' in result_data:
                products_list = result_data.get('products', [])
                
                for idx, product_data in enumerate(products_list):
                    if not isinstance(product_data, dict):
                        continue
                    
                    try:
                        # Extract product URL
                        product_url = product_data.get('product_url', '')
                        if product_url and not product_url.startswith('http'):
                            product_url = f"https://www.amazon.it{product_url}"
                        
                        # Extract ASIN or generate product ID
                        asin = product_data.get('asin', '')
                        if not asin and product_url:
                            # Try to extract ASIN from URL
                            match = re.search(r'/dp/([A-Z0-9]{10})', product_url)
                            if match:
                                asin = match.group(1)
                            else:
                                asin = f"AMZIT-P{page_num}-{idx}"
                        
                        # Create Product object
                        product = Product(
                            product_id=asin,
                            name=product_data.get('name', f'Unknown {component.title()}'),
                            price=float(product_data.get('price', 0.0)),
                            currency="EUR",
                            url=product_url or page_url,
                            marketplace=self.MARKETPLACE,
                            rating=product_data.get('rating'),
                            review_count=product_data.get('review_count'),
                            availability="Prime" if product_data.get('prime', False) else "Standard",
                            specifications={
                                "prime_eligible": product_data.get('prime', False),
                                "page_number": page_num,
                                "component_type": component
                            },
                            category=component.title(),
                            scraped_at=datetime.utcnow()
                        )
                        
                        products.append(product)
                        
                    except Exception as e:
                        print(f"  ⚠ Warning: Error parsing product {idx} on page {page_num}: {e}")
                        self.failed_products.append({
                            "page": page_num,
                            "index": idx,
                            "error": str(e)
                        })
        
        return products

    def store_products(self, products: List[Product]) -> int:
        """
        Store products in Elasticsearch
        
        Args:
            products: List of Product objects to store
            
        Returns:
            Number of successfully stored products
        """
        if not products:
            return 0
        
        try:
            success, failed = self.es_client.index_products(products)
            self.total_stored += success
            
            if failed:
                print(f"  ⚠ Warning: Failed to store {len(failed)} products")
                self.failed_products.extend(failed)
            
            return success
            
        except Exception as e:
            print(f"  ✗ Error storing products in Elasticsearch: {e}")
            return 0
    
    async def scrape_component(self, client, component: str) -> List[Product]:
        """
        Scrape all pages for a single component (async with parallel execution)

        Args:
            client: AsyncClient instance
            component: PC component category

        Returns:
            List of all products for this component
        """
        print(f"\n{'='*70}")
        print(f"  Scraping Component: {component.upper()}")
        print(f"{'='*70}")

        # Create tasks for all pages (parallel execution)
        tasks = [
            self.scrape_page(client, component, page_num)
            for page_num in range(1, self.PAGES_PER_COMPONENT + 1)
        ]

        # Execute all page scrapes in parallel
        component_start = time.time()
        results = await asyncio.gather(*tasks, return_exceptions=True)
        component_time = time.time() - component_start

        # Collect all products from successful scrapes
        all_products = []
        successful_pages = 0
        failed_pages = 0

        for i, result in enumerate(results, 1):
            if isinstance(result, Exception):
                print(f"  ✗ [{component}] Page {i} failed with exception: {result}")
                failed_pages += 1
            elif isinstance(result, list):
                all_products.extend(result)
                if result:
                    successful_pages += 1

        # Store component statistics
        self.component_stats[component] = {
            "products_found": len(all_products),
            "successful_pages": successful_pages,
            "failed_pages": failed_pages,
            "time_seconds": component_time
        }

        print(f"\n  [{component.upper()}] Summary:")
        print(f"    Products found: {len(all_products)}")
        print(f"    Successful pages: {successful_pages}/{self.PAGES_PER_COMPONENT}")
        print(f"    Time: {component_time:.2f} seconds")

        return all_products

    async def scrape_all_components(self):
        """Scrape all PC components using async parallel execution"""
        print("\n" + "="*70)
        print("  Amazon PC Component Scraper - Starting Extraction")
        print("="*70)
        print(f"Components: {', '.join(self.PC_COMPONENTS)}")
        print(f"Pages per component: {self.PAGES_PER_COMPONENT}")
        print(f"Total pages to scrape: {len(self.PC_COMPONENTS) * self.PAGES_PER_COMPONENT}")
        print(f"Marketplace: {self.MARKETPLACE}")
        print("="*70)

        start_time = time.time()

        # Import AsyncClient
        try:
            from scrapegraph_py import AsyncClient
            print("✓ ScrapeGraph AsyncClient imported")
        except Exception as e:
            raise RuntimeError(
                f"Failed to import AsyncClient: {e}\n"
                f"Please ensure scrapegraph-py is installed and up to date."
            )

        # Use AsyncClient context manager
        async with AsyncClient(api_key=self.api_key) as client:
            print("✓ AsyncClient initialized\n")

            # Process each component sequentially (pages within component are parallel)
            for component in self.PC_COMPONENTS:
                products = await self.scrape_component(client, component)

                # Store products in Elasticsearch
                if products:
                    stored = self.store_products(products)
                    print(f"  ✓ [{component.upper()}] Stored {stored}/{len(products)} products in Elasticsearch\n")

        elapsed_time = time.time() - start_time

        # Print final summary
        self._print_final_summary(elapsed_time)

    def _print_final_summary(self, elapsed_time: float):
        """Print final scraping summary"""
        print("\n" + "="*70)
        print("  FINAL SCRAPING SUMMARY")
        print("="*70)

        # Component-wise breakdown
        print("\n  Component Breakdown:")
        print(f"  {'Component':<20} {'Products':<12} {'Success':<10} {'Failed':<10} {'Time':<10}")
        print("  " + "-" * 68)

        for component in self.PC_COMPONENTS:
            if component in self.component_stats:
                stats = self.component_stats[component]
                print(f"  {component.title():<20} {stats['products_found']:<12} "
                      f"{stats['successful_pages']:<10} {stats['failed_pages']:<10} "
                      f"{stats['time_seconds']:.1f}s")

        # Overall statistics
        print("\n  Overall Statistics:")
        print(f"    Total products scraped: {self.total_scraped}")
        print(f"    Total products stored: {self.total_stored}")
        print(f"    Total failed pages: {len(self.failed_pages)}")
        print(f"    Total time: {elapsed_time:.2f} seconds")
        print(f"    Average time per component: {elapsed_time/len(self.PC_COMPONENTS):.2f} seconds")
        print("="*70)

        if self.failed_pages:
            print("\n⚠ Failed pages details:")
            for failed in self.failed_pages[:10]:  # Show first 10
                print(f"  - [{failed['component']}] Page {failed['page']}: {failed['error']}")
            if len(self.failed_pages) > 10:
                print(f"  ... and {len(self.failed_pages) - 10} more")
    
    def run_queries(self):
        """Run various queries on the scraped dataset"""
        print("\n" + "="*70)
        print("  Query Analysis on PC Component Dataset")
        print("="*70)
        
        # Query 1: Top-rated products
        print("\n📊 Query 1: Top 10 Highest-Rated Products")
        print("-" * 70)
        top_rated = self._query_top_rated(10)
        self._display_products(top_rated, show_rating=True)

        # Query 2: Most-reviewed products
        print("\n📊 Query 2: Top 10 Most-Reviewed Products")
        print("-" * 70)
        most_reviewed = self._query_most_reviewed(10)
        self._display_products(most_reviewed, show_reviews=True)
        
        # Query 3: Price distribution
        print("\n📊 Query 3: Price Distribution Statistics")
        print("-" * 70)
        price_stats = self._query_price_distribution()
        self._display_price_stats(price_stats)
        
        # Query 4: Prime vs Non-Prime comparison
        print("\n📊 Query 4: Prime vs Non-Prime Products")
        print("-" * 70)
        prime_stats = self._query_prime_comparison()
        self._display_prime_stats(prime_stats)
        
        # Query 5: Products by price range
        print("\n📊 Query 5: Products by Price Range")
        print("-" * 70)
        price_ranges = self._query_price_ranges()
        self._display_price_ranges(price_ranges)
        
        # Query 6: Top brands
        print("\n📊 Query 6: Top 10 Brands by Product Count")
        print("-" * 70)
        top_brands = self._query_top_brands(10)
        self._display_top_brands(top_brands)
        
        # Query 7: Best value products (high rating, low price)
        print("\n📊 Query 7: Best Value Products (Rating >= 4.5, Price < 100 EUR)")
        print("-" * 70)
        best_value = self._query_best_value()
        self._display_products(best_value, show_rating=True, show_price=True)
        
        print("\n" + "="*70)
        print("  Query Analysis Complete")
        print("="*70)
    
    def _query_top_rated(self, limit: int) -> List[Product]:
        """Query top-rated products"""
        search_body = {
            "query": {
                "bool": {
                    "must": [{"exists": {"field": "rating"}}],
                    "filter": [{"range": {"rating": {"gte": 4.0}}}]
                }
            },
            "sort": [
                {"rating": {"order": "desc"}},
                {"review_count": {"order": "desc"}}
            ],
            "size": limit
        }
        
        response = self.es_client.client.search(
            index=self.es_client.INDEX_NAME,
            body=search_body
        )
        
        products = []
        for hit in response["hits"]["hits"]:
            products.append(Product(**hit["_source"]))
        
        return products
    
    def _query_most_reviewed(self, limit: int) -> List[Product]:
        """Query most-reviewed products"""
        search_body = {
            "query": {
                "bool": {
                    "must": [{"exists": {"field": "review_count"}}]
                }
            },
            "sort": [{"review_count": {"order": "desc"}}],
            "size": limit
        }
        
        response = self.es_client.client.search(
            index=self.es_client.INDEX_NAME,
            body=search_body
        )
        
        products = []
        for hit in response["hits"]["hits"]:
            products.append(Product(**hit["_source"]))
        
        return products
    
    def _query_price_distribution(self) -> Dict[str, Any]:
        """Query price statistics"""
        search_body = {
            "size": 0,
            "aggs": {
                "price_stats": {
                    "stats": {"field": "price"}
                },
                "price_histogram": {
                    "histogram": {
                        "field": "price",
                        "interval": 25
                    }
                }
            }
        }
        
        response = self.es_client.client.search(
            index=self.es_client.INDEX_NAME,
            body=search_body
        )
        
        return {
            "stats": response["aggregations"]["price_stats"],
            "histogram": response["aggregations"]["price_histogram"]["buckets"]
        }
    
    def _query_prime_comparison(self) -> Dict[str, Any]:
        """Compare Prime vs Non-Prime products"""
        # Prime products
        prime_body = {
            "query": {
                "term": {"availability": "Prime"}
            },
            "size": 0,
            "aggs": {
                "avg_price": {"avg": {"field": "price"}},
                "avg_rating": {"avg": {"field": "rating"}},
                "count": {"value_count": {"field": "product_id"}}
            }
        }
        
        prime_response = self.es_client.client.search(
            index=self.es_client.INDEX_NAME,
            body=prime_body
        )
        
        # Non-Prime products
        non_prime_body = {
            "query": {
                "term": {"availability": "Standard"}
            },
            "size": 0,
            "aggs": {
                "avg_price": {"avg": {"field": "price"}},
                "avg_rating": {"avg": {"field": "rating"}},
                "count": {"value_count": {"field": "product_id"}}
            }
        }
        
        non_prime_response = self.es_client.client.search(
            index=self.es_client.INDEX_NAME,
            body=non_prime_body
        )
        
        return {
            "prime": {
                "count": prime_response["hits"]["total"]["value"],
                "avg_price": prime_response["aggregations"]["avg_price"]["value"],
                "avg_rating": prime_response["aggregations"]["avg_rating"]["value"]
            },
            "non_prime": {
                "count": non_prime_response["hits"]["total"]["value"],
                "avg_price": non_prime_response["aggregations"]["avg_price"]["value"],
                "avg_rating": non_prime_response["aggregations"]["avg_rating"]["value"]
            }
        }
    
    def _query_price_ranges(self) -> List[Dict[str, Any]]:
        """Query products by price range"""
        ranges = [
            {"label": "Budget (<30 EUR)", "min": 0, "max": 30},
            {"label": "Mid-Range (30-60 EUR)", "min": 30, "max": 60},
            {"label": "Premium (60-100 EUR)", "min": 60, "max": 100},
            {"label": "High-End (>100 EUR)", "min": 100, "max": 1000}
        ]
        
        results = []
        for range_config in ranges:
            search_body = {
                "query": {
                    "range": {
                        "price": {
                            "gte": range_config["min"],
                            "lt": range_config["max"]
                        }
                    }
                },
                "size": 0
            }
            
            response = self.es_client.client.search(
                index=self.es_client.INDEX_NAME,
                body=search_body
            )
            
            results.append({
                "label": range_config["label"],
                "count": response["hits"]["total"]["value"]
            })
        
        return results
    
    def _query_top_brands(self, limit: int) -> List[Dict[str, Any]]:
        """Query top brands by product count"""
        search_body = {
            "size": 0,
            "aggs": {
                "brands": {
                    "terms": {
                        "field": "brand.keyword",
                        "size": limit
                    },
                    "aggs": {
                        "avg_rating": {"avg": {"field": "rating"}},
                        "avg_price": {"avg": {"field": "price"}}
                    }
                }
            }
        }
        
        response = self.es_client.client.search(
            index=self.es_client.INDEX_NAME,
            body=search_body
        )
        
        brands = []
        for bucket in response["aggregations"]["brands"]["buckets"]:
            brands.append({
                "brand": bucket["key"],
                "count": bucket["doc_count"],
                "avg_rating": bucket["avg_rating"]["value"],
                "avg_price": bucket["avg_price"]["value"]
            })
        
        return brands
    
    def _query_best_value(self) -> List[Product]:
        """Query best value products (high rating, low price)"""
        search_body = {
            "query": {
                "bool": {
                    "must": [
                        {"range": {"rating": {"gte": 4.5}}},
                        {"range": {"price": {"lt": 100}}}
                    ]
                }
            },
            "sort": [
                {"rating": {"order": "desc"}},
                {"price": {"order": "asc"}}
            ],
            "size": 10
        }
        
        response = self.es_client.client.search(
            index=self.es_client.INDEX_NAME,
            body=search_body
        )
        
        products = []
        for hit in response["hits"]["hits"]:
            products.append(Product(**hit["_source"]))
        
        return products
    
    def _display_products(self, products: List[Product], show_rating: bool = False,
                         show_reviews: bool = False, show_price: bool = False):
        """Display product list"""
        if not products:
            print("  No products found.")
            return
        
        for i, product in enumerate(products, 1):
            parts = [f"  {i}. {product.name[:60]}"]
            
            if show_price:
                parts.append(f"€{product.price:.2f}")
            
            if show_rating and product.rating:
                parts.append(f"★ {product.rating:.1f}")
            
            if show_reviews and product.review_count:
                parts.append(f"({product.review_count:,} reviews)")
            
            print(" - ".join(parts))
    
    def _display_price_stats(self, price_data: Dict[str, Any]):
        """Display price statistics"""
        stats = price_data["stats"]
        print(f"  Average Price: €{stats['avg']:.2f}")
        print(f"  Min Price: €{stats['min']:.2f}")
        print(f"  Max Price: €{stats['max']:.2f}")
        print(f"  Total Products: {stats['count']}")
        
        print("\n  Price Distribution:")
        for bucket in price_data["histogram"]:
            if bucket["doc_count"] > 0:
                price_range = f"€{bucket['key']:.0f}-€{bucket['key']+25:.0f}"
                bar = "█" * (bucket["doc_count"] // 5)
                print(f"  {price_range:20} | {bar} ({bucket['doc_count']})")
    
    def _display_prime_stats(self, prime_data: Dict[str, Any]):
        """Display Prime vs Non-Prime comparison"""
        prime = prime_data["prime"]
        non_prime = prime_data["non_prime"]
        
        print("  Prime Products:")
        print(f"    Count: {prime['count']}")
        print(f"    Avg Price: €{prime['avg_price']:.2f}")
        print(f"    Avg Rating: {prime['avg_rating']:.2f} ★")
        
        print("\n  Non-Prime Products:")
        print(f"    Count: {non_prime['count']}")
        print(f"    Avg Price: €{non_prime['avg_price']:.2f}")
        print(f"    Avg Rating: {non_prime['avg_rating']:.2f} ★")
        
        total = prime['count'] + non_prime['count']
        if total > 0:
            prime_pct = (prime['count'] / total) * 100
            print(f"\n  Prime Availability: {prime_pct:.1f}% of products")
    
    def _display_price_ranges(self, ranges: List[Dict[str, Any]]):
        """Display price range distribution"""
        total = sum(r["count"] for r in ranges)
        
        for range_data in ranges:
            count = range_data["count"]
            pct = (count / total * 100) if total > 0 else 0
            bar = "█" * int(pct / 2)
            print(f"  {range_data['label']:30} | {bar} {count:4} ({pct:.1f}%)")
    
    def _display_top_brands(self, brands: List[Dict[str, Any]]):
        """Display top brands"""
        for i, brand_data in enumerate(brands, 1):
            print(f"  {i}. {brand_data['brand']:20} "
                  f"- {brand_data['count']:3} products "
                  f"- Avg: €{brand_data['avg_price']:.2f} "
                  f"- Rating: {brand_data['avg_rating']:.2f} ★")
    
    def close(self):
        """Close Elasticsearch connection"""
        self.es_client.close()


async def main():
    """Main async function to run the PC component scraper"""
    scraper = None

    try:
        # Initialize scraper
        scraper = AmazonPCComponentScraper()
        print("✓ Scraper initialized\n")

        # Scrape all PC components (async with parallel execution)
        await scraper.scrape_all_components()

        # Run queries if we have data
        if scraper.total_stored > 0:
            scraper.run_queries()
        else:
            print("\n⚠ No products were stored. Skipping query analysis.")

    except KeyboardInterrupt:
        print("\n\n⚠ Scraping interrupted by user.")

    except Exception as e:
        print(f"\n✗ Fatal error: {e}")
        traceback.print_exc()

    finally:
        # Clean up
        if scraper:
            scraper.close()
            print("\n✓ Connections closed.")


if __name__ == "__main__":
    asyncio.run(main())