← back to Handbag Auth Nextjs
scripts/scrape-auction-data.py
337 lines
#!/usr/bin/env python3
"""
Daily Auction Data Scraper
Scrapes LiveAuctioneers for luxury handbag auction results
Saves to CSV for historical tracking
Runs daily at 6am via cron
"""
import os
import sqlite3
import csv
import requests
from bs4 import BeautifulSoup
from datetime import datetime
import sys
import time
import json
# 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)
# Top 20 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"
]
def log(message):
"""Log message with timestamp"""
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
log_message = f"[{timestamp}] {message}\n"
print(log_message.strip())
with open(LOG_PATH, 'a') as f:
f.write(log_message)
def init_database():
"""Initialize SQLite database with auctions table"""
log("Initializing database...")
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS auctions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
auction_id TEXT UNIQUE,
title TEXT,
auction_house TEXT,
current_price REAL,
estimate_low REAL,
estimate_high REAL,
end_date TEXT,
lot_number TEXT,
url TEXT,
search_term TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.commit()
conn.close()
log("✓ Database initialized")
def scrape_liveauctioneers_search(search_term):
"""Scrape LiveAuctioneers search results for a given term"""
log(f"Searching LiveAuctioneers for: {search_term}")
try:
# LiveAuctioneers search URL
base_url = "https://www.liveauctioneers.com/search/"
params = {
'keyword': search_term,
'sort': '-relevance',
'status': 'archive' # Get sold items
}
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
}
response = requests.get(base_url, params=params, headers=headers, timeout=30)
response.raise_for_status()
soup = BeautifulSoup(response.content, 'html.parser')
# This is a simplified scraper - LiveAuctioneers uses dynamic loading
# For production, you'd need Playwright/Selenium or API access
# For now, we'll create sample data structure
results = []
# Try to find auction items (this is a placeholder - real scraping would need JS rendering)
items = soup.find_all('div', class_='item') or soup.find_all('a', class_='lot-tile')
log(f"Found {len(items)} items for {search_term}")
# Since LiveAuctioneers requires JS rendering, we'll use a hybrid approach
# For demo purposes, we'll log the attempt
return results
except requests.RequestException as e:
log(f"✗ Error fetching {search_term}: {str(e)}")
return []
except Exception as e:
log(f"✗ Unexpected error for {search_term}: {str(e)}")
return []
def scrape_all_sources():
"""Scrape all auction sources"""
log("Starting auction scraping...")
all_results = []
for term in SEARCH_TERMS:
results = scrape_liveauctioneers_search(term)
all_results.extend(results)
# Rate limiting - be respectful
time.sleep(2)
log(f"Scraping completed. Total results: {len(all_results)}")
return all_results
def save_to_database(results):
"""Save scraping results to database"""
if not results:
log("No results to save")
return 0
log(f"Saving {len(results)} results to database...")
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
saved_count = 0
for result in results:
try:
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)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
result.get('auction_id'),
result.get('title'),
result.get('auction_house'),
result.get('current_price'),
result.get('estimate_low'),
result.get('estimate_high'),
result.get('end_date'),
result.get('lot_number'),
result.get('url'),
result.get('search_term')
))
saved_count += 1
except Exception as e:
log(f"Error saving result: {str(e)}")
conn.commit()
conn.close()
log(f"✓ Saved {saved_count} new auction records")
return saved_count
def export_to_csv():
"""Export SQLite database to CSV for historical tracking"""
log("Exporting to CSV...")
try:
# Connect to database
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
# Check if auctions table exists
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='auctions'")
if not cursor.fetchone():
log("No auctions table found - creating initial CSV with headers")
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'
])
conn.close()
return True
# Get all auction records
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
""")
rows = cursor.fetchall()
# Check if CSV exists to determine if we need headers
file_exists = os.path.exists(CSV_PATH)
# Append to CSV (or create new)
mode = 'a' if file_exists else 'w'
with open(CSV_PATH, mode, newline='') as csvfile:
writer = csv.writer(csvfile)
# Write headers if new file
if not file_exists:
writer.writerow([
'scrape_date', 'auction_id', 'title', 'auction_house',
'current_price', 'estimate_low', 'estimate_high',
'end_date', 'lot_number', 'url', 'search_term'
])
# Write data with current date
scrape_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
for row in rows:
writer.writerow([scrape_date] + list(row))
log(f"✓ Exported {len(rows)} records to CSV")
conn.close()
return True
except Exception as e:
log(f"ERROR exporting to CSV: {str(e)}")
return False
def generate_summary():
"""Generate daily summary statistics"""
log("Generating summary...")
try:
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
# Get total auctions
cursor.execute("SELECT COUNT(*) FROM auctions")
total = cursor.fetchone()[0]
# Get auctions added today
today = datetime.now().strftime('%Y-%m-%d')
cursor.execute("""
SELECT COUNT(*) FROM auctions
WHERE date(created_at) = ?
""", (today,))
today_count = cursor.fetchone()[0]
log(f"Summary: {total} total auctions, {today_count} added today")
conn.close()
return {
'total': total,
'today': today_count
}
except Exception as e:
log(f"ERROR generating summary: {str(e)}")
return None
def main():
"""Main execution function"""
log("=" * 60)
log("LUXURY HANDBAG AUCTION SCRAPER - DAILY RUN")
log(f"Searching for {len(SEARCH_TERMS)} luxury handbag models")
log("=" * 60)
# Step 0: Initialize database
init_database()
# Step 1: Scrape auction sources
results = scrape_all_sources()
# Step 2: Save to database
saved_count = save_to_database(results)
# Step 3: Export to CSV
export_success = export_to_csv()
if not export_success:
log("CSV export failed")
sys.exit(1)
# Step 4: Generate summary
summary = generate_summary()
# Step 5: Report completion
log("=" * 60)
log("DAILY SCRAPE COMPLETED")
log(f"Brands searched: {len(SEARCH_TERMS)}")
log(f"Results found: {len(results)}")
log(f"New records saved: {saved_count}")
log(f"CSV file: {CSV_PATH}")
log(f"Database: {DB_PATH}")
log(f"Log file: {LOG_PATH}")
log("=" * 60)
# Note about LiveAuctioneers requiring JS
if len(results) == 0:
log("")
log("NOTE: LiveAuctioneers uses dynamic JavaScript loading.")
log("For production scraping, consider using Playwright or Selenium.")
log("Alternative: Use auction house APIs when available.")
log("")
sys.exit(0)
if __name__ == "__main__":
main()