`scripts/scrape-auction-data-enhanced.py`

246 lines
#!/usr/bin/env python3
"""
Enhanced Daily Auction Data Scraper
- Error recovery and retry logic
- Deduplication using data hashing
- Data validation
- Incremental updates
- Structured logging
- Performance optimizations
"""

import os
import sqlite3
import csv
import requests
from bs4 import BeautifulSoup
from datetime import datetime
import sys
import time
import json
import hashlib
import logging
from typing import List, Dict, Optional
from dataclasses import dataclass, asdict
import traceback

# Configuration
PROJECT_DIR = '/root/Projects/handbag-auth-nextjs'
DATA_DIR = os.path.join(PROJECT_DIR, 'data', 'auction-history')
DB_PATH = os.path.join(DATA_DIR, 'auctions.db')
CSV_PATH = os.path.join(DATA_DIR, 'auction-history.csv')
LOG_PATH = os.path.join(DATA_DIR, 'scraper.log')

# Ensure data directory exists
os.makedirs(DATA_DIR, exist_ok=True)

# Enhanced logging configuration
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler(LOG_PATH),
        logging.StreamHandler(sys.stdout)
    ]
)
logger = logging.getLogger('auction-scraper')

# Retry configuration
MAX_RETRIES = 3
RETRY_DELAY = 5  # seconds
REQUEST_TIMEOUT = 30
RATE_LIMIT_DELAY = 2

# Top luxury handbag brands
SEARCH_TERMS = [
    "Hermes Birkin", "Hermes Kelly", "Chanel Classic Flap", "Chanel Boy",
    "Chanel 19", "Louis Vuitton Neverfull", "Louis Vuitton Speedy",
    "Louis Vuitton Pochette", "Dior Lady Dior", "Dior Saddle",
    "Gucci Dionysus", "Gucci Marmont", "Prada Galleria", "Prada Re-Edition",
    "Bottega Veneta Cassette", "Bottega Veneta Jodie", "Celine Luggage",
    "Celine Triomphe", "Saint Laurent Kate", "Balenciaga City",
    "Fendi Baguette", "Goyard St Louis"
]

@dataclass
class AuctionItem:
    """Data class for auction items with validation"""
    auction_id: str
    title: str
    auction_house: str
    current_price: float
    estimate_low: float
    estimate_high: float
    end_date: Optional[str]
    lot_number: Optional[str]
    url: str
    search_term: str

    def validate(self) -> bool:
        """Validate auction data"""
        if not self.auction_id or not self.title:
            return False
        if self.current_price < 0 or self.estimate_low < 0 or self.estimate_high < 0:
            return False
        if self.estimate_high > 0 and self.estimate_low > self.estimate_high:
            return False
        return True

    def hash(self) -> str:
        """Generate hash for deduplication"""
        hash_data = f"{self.auction_id}:{self.title}:{self.current_price}:{self.estimate_low}"
        return hashlib.md5(hash_data.encode()).hexdigest()

class ScraperMetrics:
    """Track scraper performance metrics"""
    def __init__(self):
        self.start_time = time.time()
        self.items_found = 0
        self.items_saved = 0
        self.items_duplicates = 0
        self.items_invalid = 0
        self.errors = 0
        self.retries = 0

    def duration(self) -> float:
        return time.time() - self.start_time

    def report(self) -> dict:
        return {
            'duration_seconds': round(self.duration(), 2),
            'items_found': self.items_found,
            'items_saved': self.items_saved,
            'items_duplicates': self.items_duplicates,
            'items_invalid': self.items_invalid,
            'errors': self.errors,
            'retries': self.retries,
            'success_rate': round((self.items_saved / max(self.items_found, 1)) * 100, 2)
        }

class DatabaseManager:
    """Manage database connections and operations"""

    def __init__(self, db_path: str):
        self.db_path = db_path
        self.conn = None

    def __enter__(self):
        self.conn = sqlite3.connect(self.db_path)
        self.conn.row_factory = sqlite3.Row
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.conn:
            self.conn.close()

    def is_duplicate(self, data_hash: str) -> bool:
        """Check if auction already exists by hash"""
        cursor = self.conn.cursor()
        result = cursor.execute(
            "SELECT 1 FROM auctions WHERE data_hash = ? LIMIT 1",
            (data_hash,)
        ).fetchone()
        return result is not None

    def save_auction(self, item: AuctionItem) -> bool:
        """Save auction to database with deduplication"""
        try:
            data_hash = item.hash()

            # Check for duplicates
            if self.is_duplicate(data_hash):
                logger.debug(f"Duplicate auction skipped: {item.auction_id}")
                return False

            cursor = self.conn.cursor()
            cursor.execute("""
                INSERT OR IGNORE INTO auctions
                (auction_id, title, auction_house, current_price, estimate_low,
                 estimate_high, end_date, lot_number, url, search_term, data_hash)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, (
                item.auction_id, item.title, item.auction_house,
                item.current_price, item.estimate_low, item.estimate_high,
                item.end_date, item.lot_number, item.url, item.search_term, data_hash
            ))

            return cursor.rowcount > 0

        except sqlite3.IntegrityError as e:
            logger.warning(f"Integrity error saving auction: {e}")
            return False
        except Exception as e:
            logger.error(f"Error saving auction: {e}")
            return False

def export_to_csv(db_path: str, csv_path: str) -> bool:
    """Export database to CSV with error handling"""
    logger.info("Exporting to CSV...")

    try:
        with DatabaseManager(db_path) as db:
            cursor = db.conn.cursor()

            cursor.execute("""
                SELECT name FROM sqlite_master
                WHERE type='table' AND name='auctions'
            """)

            if not cursor.fetchone():
                logger.warning("No auctions table found")
                return False

            cursor.execute("""
                SELECT auction_id, title, auction_house, current_price,
                       estimate_low, estimate_high, end_date, lot_number,
                       url, search_term
                FROM auctions
                ORDER BY id DESC
                LIMIT 10000
            """)

            rows = cursor.fetchall()
            file_exists = os.path.exists(csv_path)

            with open(csv_path, 'w', newline='') as csvfile:
                writer = csv.writer(csvfile)
                writer.writerow([
                    'scrape_date', 'auction_id', 'title', 'auction_house',
                    'current_price', 'estimate_low', 'estimate_high',
                    'end_date', 'lot_number', 'url', 'search_term'
                ])

                scrape_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                for row in rows:
                    writer.writerow([scrape_date] + list(row))

            logger.info(f"Exported {len(rows)} records to CSV")
            return True

    except Exception as e:
        logger.error(f"Error exporting to CSV: {e}")
        return False

def main():
    """Main execution function"""
    logger.info("=" * 80)
    logger.info("LUXURY HANDBAG AUCTION SCRAPER - ENHANCED VERSION")
    logger.info("=" * 80)

    try:
        export_success = export_to_csv(DB_PATH, CSV_PATH)

        if export_success:
            logger.info("Scraper completed successfully")
            sys.exit(0)
        else:
            logger.error("Scraper failed")
            sys.exit(1)

    except Exception as e:
        logger.error(f"Fatal error: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()