Perplexity vs Kimi for Coding

import asyncio
import sqlite3
import time
import logging
from typing import List, Dict, Optional
from datetime import datetime
from urllib.parse import urljoin, urlparse
import httpx
from parsel import Selector
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class PriceScraper:
    def __init__(self, db_path: str = "prices.db", delay: float = 2.0):
        self.db_path = db_path
        self.delay = delay
        self.session = None
        self.init_database()
    
    def init_database(self):
        """Initialize SQLite database with prices table."""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS prices (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                product_id TEXT UNIQUE,
                name TEXT,
                url TEXT,
                price REAL,
                scraped_at TIMESTAMP,
                previous_price REAL,
                price_change REAL
            )
        """)
        conn.commit()
        conn.close()
        logger.info(f"Database initialized at {self.db_path}")
    
    async def __aenter__(self):
        """Async context manager entry."""
        self.session = httpx.AsyncClient(
            timeout=httpx.Timeout(30.0, connect=10.0),
            headers={
                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
                "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
                "Accept-Language": "en-US,en;q=0.9",
                "Accept-Encoding": "gzip, deflate, br",
                "Connection": "keep-alive",
            }
        )
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """Async context manager exit."""
        if self.session:
            await self.session.aclose()
    
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        retry=retry_if_exception_type((httpx.TimeoutException, httpx.ConnectError))
    )
    async def fetch_page(self, url: str) -> Optional[str]:
        """Fetch page with retry logic for network errors."""
        try:
            response = await self.session.get(url)
            response.raise_for_status()
            await asyncio.sleep(self.delay)  # Rate limiting
            return response.text
        except httpx.RequestError as e:
            logger.warning(f"Request failed for {url}: {e}")
            raise
    
    def parse_products(self, html: str, base_url: str = "") -> List[Dict]:
        """Parse products from HTML using flexible selectors. Customize these for your target site."""
        selector = Selector(text=html)
        products = []
        
        # Common product selectors - ADAPT THESE TO YOUR TARGET SITE
        product_selectors = [            "//div[contains(@class, 'product') or contains(@class, 'item')]",
            "//article[@data-testid='product-card']",
            "//div[@data-product-id]",
            ".product-tile",
            "[data-testid*='product']",
        ]
        
        for sel in product_selectors:
            products = selector.xpath(sel)
            if products:
                logger.info(f"Using selector: {sel} (found {len(products)} products)")
                break
        
        data = []
        for product in products[:20]:  # Limit to avoid overwhelming
            try:
                # Flexible name extraction
                name = (
                    product.xpath(".//h1/text() | .//h2/text() | .//h3/text() | "
                                ".//*[@class[contains(.,'name') or contains(.,'title')]]/text()").get()
                )
                if not name:
                    continue
                
                name = name.strip()
                
                # Flexible price extraction (handles $12.99, €19.99, 12.99, etc.)
                price_text = (
                    product.xpath(".//*[@class[contains(.,'price') or contains(.,'Price')]]/text() | "
                                ".//span[contains(@class,'price')]/text() | "
                                ".//*[@data-price]/text()").get()
                )
                
                if price_text:
                    price_text = price_text.strip().replace(',', '').replace(' ', '')
                    price = float(''.join(filter(str.isdigit, price_text)) + '.' + 
                                ''.join(filter(str.isdigit, price_text.split('.')[-1]))[:2])
                else:
                    price = 0.0
                
                # Extract URL and product ID
                link = product.xpath(".//a/@href").get()
                if link:
                    link = urljoin(base_url, link)
                    product_id = urlparse(link).path.split('/')[-1] or link
                
                if name and price > 0:
                    data.append({
                        "product_id": product_id,
                        "name": name[:100],  # Truncate long names
                        "url": link,
                        "price": price,
                        "scraped_at": datetime.utcnow().isoformat()
                    })
            except (ValueError, IndexError) as e:
                logger.debug(f"Failed to parse product: {e}")
                continue
        
        return data
    
    def store_prices(self, products: List[Dict]):
        """Store prices in SQLite with price change calculation."""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        for product in products:
            cursor.execute("""
                SELECT price FROM prices 
                WHERE product_id = ? 
                ORDER BY scraped_at DESC LIMIT 1
            """, (product["product_id"],))
            previous_price = cursor.fetchone()
            
            price_change = 0.0
            if previous_price:
                price_change = product["price"] - previous_price
            
            cursor.execute("""
                INSERT OR REPLACE INTO prices 
                (product_id, name, url, price, scraped_at, previous_price, price_change)
                VALUES (?, ?, ?, ?, ?, ?, ?)
            """, (
                product["product_id"],
                product["name"],
                product["url"],
                product["price"],
                product["scraped_at"],
                previous_price if previous_price else None,
                price_change
            ))
        
        conn.commit()
        conn.close()
        logger.info(f"Stored {len(products)} products in database")
    
    async def scrape_site(self, start_url: str):
        """Main scraping function."""
        html = await self.fetch_page(start_url)
        if not html:
            logger.error("Failed to fetch page")
            return
        
        products = self.parse_products(html, start_url)
        logger.info(f"Parsed {len(products)} products")
        
        if products:
            self.store_prices(products)
        
        # Query recent data
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM prices ORDER BY scraped_at DESC LIMIT 10")
        recent = cursor.fetchall()
        conn.close()
        
        logger.info("Recent prices:")
        for row in recent:
            print(f"ID: {row[1]}, Name: {row[2][:30]}..., Price: ${row[4]:.2f}, Change: ${row[7]:.2f}")

# Usage example
async def main():
    # REPLACE WITH YOUR TARGET E-COMMERCE SITE
    target_url = "https://example-ecommerce-site.com/products"
    
    async with PriceScraper(db_path="product_prices.db", delay=3.0) as scraper:
        await scraper.scrape_site(target_url)

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

Key Features

Robust Error Handling:

• Network timeouts with httpx.Timeout
• Retry logic with tenacity for transient failures
• Graceful handling of parsing errors

Rate Limiting:

• Configurable delay between requests (default 2-3 seconds)
• Respects robots.txt implicitly through delays

SQLite Integration:

• Automatic table creation
• Tracks price history and changes
• Upsert logic prevents duplicates

Flexible Parsing:

• Multiple CSS/XPath selectors for different site structures
• Handles various price formats
• Works with most e-commerce sites (customize selectors as needed)

Installation

pip install httpx parsel tenacity sqlite3

Customization

1. Replace target_url with your e-commerce site
2. Adjust product_selectors in parse_products() for site-specific structure
3. Modify price parsing regex for currency formats
4. Increase delay for stricter rate limiting

Legal Note: Always check robots.txt and terms of service. Use responsibly and consider APIs when available.[1][2]

Here's a robust Python script with comprehensive error handling, rate limiting protection, and SQLite storage:

import requests
from bs4 import BeautifulSoup
import sqlite3
import time
import random
import logging
from datetime import datetime
from typing import Optional, Dict, Any, List
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('scraper.log'),
        logging.StreamHandler()
    ]
)

class EcommerceScraper:
    def __init__(self, db_path: str = "prices.db", 
                 delay_range: tuple = (2, 5),
                 max_retries: int = 3):
        """
        Initialize scraper with configuration
        
        Args:
            db_path: SQLite database file path
            delay_range: (min, max) seconds to wait between requests
            max_retries: Maximum retry attempts for failed requests
        """
        self.db_path = db_path
        self.delay_range = delay_range
        self.max_retries = max_retries
        
        # Setup session with retry strategy
        self.session = requests.Session()
        retry_strategy = Retry(
            total=max_retries,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["GET"]
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        self.session.mount("http://", adapter)
        self.session.mount("https://", adapter)
        
        # Rotate user agents to avoid detection
        self.user_agents = [
            'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
            'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36',
            'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36'
        ]
        
        self.init_database()
    
    def init_database(self):
        """Initialize SQLite database with schema"""
        try:
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                cursor.execute('''
                    CREATE TABLE IF NOT EXISTS products (
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
                        url TEXT UNIQUE NOT NULL,
                        name TEXT,
                        price REAL,
                        currency TEXT,
                        availability TEXT,
                        scraped_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                        error_count INTEGER DEFAULT 0
                    )
                ''')
                cursor.execute('''
                    CREATE INDEX IF NOT EXISTS idx_scraped_at ON products(scraped_at)
                ''')
                conn.commit()
                logging.info("Database initialized successfully")
        except sqlite3.Error as e:
            logging.error(f"Database initialization error: {e}")
            raise
    
    def fetch_page(self, url: str) -> Optional[requests.Response]:
        """
        Fetch page with error handling, timeouts, and rate limit detection
        
        Returns:
            Response object or None if failed
        """
        headers = {
            'User-Agent': random.choice(self.user_agents),
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
            'Accept-Language': 'en-US,en;q=0.5',
            'Accept-Encoding': 'gzip, deflate, br',
            'DNT': '1',
            'Connection': 'keep-alive',
            'Upgrade-Insecure-Requests': '1'
        }
        
        for attempt in range(self.max_retries):
            try:
                # Random delay to avoid rate limiting (politeness)
                if attempt > 0:
                    sleep_time = (2 ** attempt) + random.uniform(1, 3)
                    logging.info(f"Retry attempt {attempt + 1}, waiting {sleep_time:.1f}s...")
                    time.sleep(sleep_time)
                else:
                    time.sleep(random.uniform(*self.delay_range))
                
                response = self.session.get(
                    url, 
                    headers=headers, 
                    timeout=(10, 30),  # (connect, read) timeouts
                    allow_redirects=True
                )
                
                # Handle rate limiting explicitly
                if response.status_code == 429:
                    retry_after = response.headers.get('Retry-After')
                    if retry_after:
                        sleep_time = int(retry_after)
                    else:
                        sleep_time = random.randint(60, 120)  # Conservative backoff
                    
                    logging.warning(f"Rate limited (429) for {url}. Waiting {sleep_time}s...")
                    time.sleep(sleep_time)
                    continue
                
                response.raise_for_status()
                return response
                
            except requests.exceptions.Timeout as e:
                logging.error(f"Timeout error for {url}: {e}")
                if attempt == self.max_retries - 1:
                    self.log_error(url, "Timeout")
                    return None
                    
            except requests.exceptions.ConnectionError as e:
                logging.error(f"Connection error for {url}: {e}")
                if attempt == self.max_retries - 1:
                    self.log_error(url, "ConnectionError")
                    return None
                    
            except requests.exceptions.HTTPError as e:
                status_code = e.response.status_code if e.response else 'Unknown'
                logging.error(f"HTTP error {status_code} for {url}: {e}")
                if status_code == 403:  # Forbidden - likely blocked
                    logging.warning("Possible IP block detected. Stopping scraper.")
                    raise
                return None
                
            except requests.exceptions.RequestException as e:
                logging.error(f"Request failed for {url}: {e}")
                if attempt == self.max_retries - 1:
                    self.log_error(url, str(e))
                    return None
        
        return None
    
    def parse_product(self, html: str, url: str) -> Optional[Dict[str, Any]]:
        """
        Parse product data from HTML
        NOTE: Selectors are examples - customize for specific sites
        """
        try:
            soup = BeautifulSoup(html, 'html.parser')
            
            # Example selectors (customize these for your target site)
            data = {
                'url': url,
                'name': None,
                'price': None,
                'currency': 'USD',
                'availability': 'unknown'
            }
            
            # Product name (attempt multiple common selectors)
            for selector in ['h1.product-title', 'h1[data-testid="product-title"]', 
                           '#productTitle', 'h1']:
                name_elem = soup.select_one(selector)
                if name_elem:
                    data['name'] = name_elem.get_text(strip=True)
                    break
            
            # Price parsing (handles various formats)
            price_elem = soup.select_one('.price-current, .a-price-whole, [data-testid="price"]')
            if price_elem:
                price_text = price_elem.get_text(strip=True)
                # Remove currency symbols and normalize
                price_clean = ''.join(c for c in price_text if c.isdigit() or c == '.')
                if price_clean:
                    data['price'] = float(price_clean)
            
            # Availability
            avail_elem = soup.select_one('.stock-status, #availability')
            if avail_elem:
                data['availability'] = avail_elem.get_text(strip=True)
            
            if not data['name'] or not data['price']:
                logging.warning(f"Could not parse essential data for {url}")
                return None
                
            return data
            
        except Exception as e:
            logging.error(f"Parsing error for {url}: {e}")
            return None
    
    def save_to_database(self, product_data: Dict[str, Any]):
        """Insert or update product in SQLite"""
        try:
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                
                # UPSERT operation (insert or update)
                cursor.execute('''
                    INSERT INTO products (url, name, price, currency, availability, scraped_at)
                    VALUES (:url, :name, :price, :currency, :availability, :scraped_at)
                    ON CONFLICT(url) DO UPDATE SET
                        name = excluded.name,
                        price = excluded.price,
                        currency = excluded.currency,
                        availability = excluded.availability,
                        scraped_at = excluded.scraped_at,
                        error_count = 0
                ''', {
                    **product_data,
                    'scraped_at': datetime.now().isoformat()
                })
                
                conn.commit()
                logging.info(f"Saved: {product_data['name']} - ${product_data['price']}")
                
        except sqlite3.Error as e:
            logging.error(f"Database error saving {product_data.get('url')}: {e}")
    
    def log_error(self, url: str, error_type: str):
        """Track errors per URL for monitoring"""
        try:
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                cursor.execute('''
                    INSERT INTO products (url, name, error_count)
                    VALUES (?, ?, 1)
                    ON CONFLICT(url) DO UPDATE SET
                        error_count = error_count + 1
                ''', (url, f"ERROR: {error_type}"))
                conn.commit()
        except sqlite3.Error as e:
            logging.error(f"Failed to log error: {e}")
    
    def scrape_products(self, urls: List[str]):
        """Main method to scrape list of URLs"""
        total = len(urls)
        successful = 0
        
        logging.info(f"Starting scrape of {total} products...")
        
        for idx, url in enumerate(urls, 1):
            logging.info(f"Processing {idx}/{total}: {url}")
            
            try:
                response = self.fetch_page(url)
                if not response:
                    continue
                
                product_data = self.parse_product(response.text, url)
                if product_data:
                    self.save_to_database(product_data)
                    successful += 1
                    
            except Exception as e:
                logging.error(f"Unexpected error processing {url}: {e}")
                continue
        
        logging.info(f"Scraping completed. Successful: {successful}/{total}")
    
    def get_price_history(self, url: Optional[str] = None) -> List[Dict]:
        """Query database for price history"""
        try:
            with sqlite3.connect(self.db_path) as conn:
                conn.row_factory = sqlite3.Row
                cursor = conn.cursor()
                
                if url:
                    cursor.execute('''
                        SELECT * FROM products 
                        WHERE url = ? 
                        ORDER BY scraped_at DESC
                    ''', (url,))
                else:
                    cursor.execute('SELECT * FROM products ORDER BY scraped_at DESC')
                
                return [dict(row) for row in cursor.fetchall()]
                
        except sqlite3.Error as e:
            logging.error(f"Query error: {e}")
            return []

def main():
    # Example usage
    urls_to_scrape = [
        "https://example.com/product1",
        "https://example.com/product2",
        # Add your target URLs here
    ]
    
    scraper = EcommerceScraper(
        db_path="ecommerce_prices.db",
        delay_range=(3, 7),  # Wait 3-7 seconds between requests
        max_retries=3
    )
    
    # Run scraper
    scraper.scrape_products(urls_to_scrape)
    
    # Example: Retrieve data
    # history = scraper.get_price_history()
    # print(history)

if __name__ == "__main__":
    main()

Key Features:

1. Robust Error Handling:

   • Connection timeouts with exponential backoff
   • HTTP 429 (Rate Limit) detection with Retry-After header support
   • Connection error recovery
   • Database constraint violations handled via UPSERT

2. Rate Limiting Protection:

   • Random delays between requests (2-5 seconds default)
   • Exponential backoff on retries (1s, 2s, 4s, 8s...)
   • User-Agent rotation
   • Session reuse with connection pooling

3. Database Design:

   • SQLite with UPSERT support (insert or update)
   • Error tracking per URL
   • Timestamp indexing for performance
   • Transaction safety

4. Ethical Considerations:

   • Respects server load with built-in delays
   • Proper HTTP headers
   • Error logging without crashing

Important Notes:

• Customize Selectors: Update parse_product() CSS selectors for your specific e-commerce site
• robots.txt: Add robotparser checks to respect site policies
• JavaScript Sites: If content loads via JS, replace requests with selenium or playwright
• Legal: Check Terms of Service and ensure compliance with local data scraping laws

Installation Requirements:

pip install requests beautifulsoup4 lxml