← back to Handbag Auth Nextjs
scripts/scrape-auction-data-secure.py
464 lines
#!/usr/bin/env python3
"""
Secure Daily Auction Data Scraper
Enhanced with security measures for production use
"""
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 hmac
import re
from urllib.parse import urlparse, quote
import logging
from logging.handlers import RotatingFileHandler
import random
# Security: Secure 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-secure.log')
# Ensure data directory exists with secure permissions
os.makedirs(DATA_DIR, exist_ok=True)
os.chmod(DATA_DIR, 0o750)
# Security: Setup secure logging
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
# Rotate logs to prevent disk filling attacks
handler = RotatingFileHandler(
LOG_PATH,
maxBytes=10*1024*1024, # 10MB
backupCount=5
)
formatter = logging.Formatter(
'%(asctime)s - %(levelname)s - %(funcName)s - %(message)s'
)
handler.setFormatter(formatter)
logger.addHandler(handler)
# Security: Whitelisted domains for scraping
ALLOWED_DOMAINS = [
'liveauctioneers.com',
'christies.com',
'sothebys.com',
'bonhams.com',
'phillips.com'
]
# Security: User agents rotation
USER_AGENTS = [
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
'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',
'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
]
# Top luxury handbag brands (sanitized)
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"
]
class SecureScraper:
"""Secure web scraper with validation and sanitization"""
def __init__(self):
self.session = requests.Session()
self.session.headers.update({
'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',
'DNT': '1',
'Connection': 'keep-alive',
'Upgrade-Insecure-Requests': '1'
})
def validate_url(self, url):
"""Security: Validate and sanitize URLs"""
try:
parsed = urlparse(url)
# Check for valid scheme
if parsed.scheme not in ['http', 'https']:
logger.warning(f"Invalid URL scheme: {parsed.scheme}")
return None
# Check against whitelist
domain = parsed.netloc.replace('www.', '')
if not any(allowed in domain for allowed in ALLOWED_DOMAINS):
logger.warning(f"Domain not in whitelist: {domain}")
return None
# Prevent SSRF attacks
if parsed.hostname in ['localhost', '127.0.0.1', '0.0.0.0']:
logger.warning(f"Blocked local URL: {url}")
return None
return url
except Exception as e:
logger.error(f"URL validation error: {str(e)}")
return None
def sanitize_text(self, text):
"""Security: Sanitize text to prevent injection attacks"""
if not text:
return ""
# Convert to string and strip
text = str(text).strip()
# Remove control characters
text = ''.join(char for char in text if ord(char) >= 32)
# Remove potentially dangerous characters
text = re.sub(r'[<>\"\'&;`]', '', text)
# Limit length
if len(text) > 1000:
text = text[:1000]
return text
def sanitize_number(self, value, min_val=0, max_val=1000000):
"""Security: Sanitize numeric values"""
try:
num = float(value)
if num < min_val or num > max_val:
return None
return num
except (TypeError, ValueError):
return None
def fetch_page(self, url, timeout=10):
"""Security: Fetch page with timeout and error handling"""
validated_url = self.validate_url(url)
if not validated_url:
return None
try:
# Random user agent
self.session.headers['User-Agent'] = random.choice(USER_AGENTS)
response = self.session.get(
validated_url,
timeout=timeout,
allow_redirects=False, # Prevent redirect attacks
verify=True # Verify SSL certificates
)
# Check content type
content_type = response.headers.get('content-type', '')
if 'text/html' not in content_type:
logger.warning(f"Unexpected content type: {content_type}")
return None
# Size limit (10MB)
if len(response.content) > 10 * 1024 * 1024:
logger.warning(f"Response too large: {len(response.content)}")
return None
response.raise_for_status()
return response.content
except requests.RequestException as e:
logger.error(f"Request error: {str(e)}")
return None
def init_secure_database():
"""Initialize SQLite database with security measures"""
logger.info("Initializing secure database...")
try:
conn = sqlite3.connect(DB_PATH)
conn.execute("PRAGMA foreign_keys = ON")
conn.execute("PRAGMA journal_mode = WAL")
cursor = conn.cursor()
# Create table with constraints
cursor.execute("""
CREATE TABLE IF NOT EXISTS auctions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
auction_id TEXT UNIQUE NOT NULL CHECK(length(auction_id) <= 100),
title TEXT NOT NULL CHECK(length(title) <= 500),
auction_house TEXT CHECK(length(auction_house) <= 200),
current_price REAL CHECK(current_price >= 0 AND current_price <= 10000000),
estimate_low REAL CHECK(estimate_low >= 0 AND estimate_low <= 10000000),
estimate_high REAL CHECK(estimate_high >= 0 AND estimate_high <= 10000000),
end_date TEXT CHECK(length(end_date) <= 50),
lot_number TEXT CHECK(length(lot_number) <= 50),
url TEXT CHECK(length(url) <= 500),
search_term TEXT CHECK(length(search_term) <= 100),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
data_hash TEXT,
CHECK(estimate_high >= estimate_low OR estimate_high IS NULL OR estimate_low IS NULL)
)
""")
# Create indexes for performance
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_auction_id ON auctions(auction_id)
""")
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_created_at ON auctions(created_at)
""")
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_search_term ON auctions(search_term)
""")
conn.commit()
conn.close()
# Set secure file permissions
os.chmod(DB_PATH, 0o640)
logger.info("Database initialized successfully")
except sqlite3.Error as e:
logger.error(f"Database initialization error: {str(e)}")
sys.exit(1)
def generate_auction_hash(data):
"""Generate hash for auction data to detect duplicates"""
hash_string = f"{data.get('auction_id')}_{data.get('title')}_{data.get('current_price')}"
return hashlib.sha256(hash_string.encode()).hexdigest()
def scrape_liveauctioneers_secure(search_term):
"""Securely scrape LiveAuctioneers with validation"""
scraper = SecureScraper()
logger.info(f"Securely scraping for: {search_term}")
try:
# Sanitize search term
safe_search = quote(scraper.sanitize_text(search_term))
# Build URL with parameters
base_url = f"https://www.liveauctioneers.com/search/?keyword={safe_search}&sort=-relevance&status=archive"
# Fetch page content
content = scraper.fetch_page(base_url)
if not content:
logger.warning(f"Failed to fetch content for {search_term}")
return []
soup = BeautifulSoup(content, 'html.parser')
# Extract auction data (placeholder - actual parsing would go here)
# This is where you would implement the actual parsing logic
# For security, all extracted data must be sanitized
results = []
# Example structure (would be filled with actual parsing):
# items = soup.find_all('div', class_='auction-item')
# for item in items:
# auction_data = {
# 'auction_id': scraper.sanitize_text(item.get('data-id')),
# 'title': scraper.sanitize_text(item.find('h3').text),
# 'auction_house': scraper.sanitize_text(item.find('.auction-house').text),
# 'current_price': scraper.sanitize_number(item.find('.price').text),
# # ... more fields
# }
# results.append(auction_data)
logger.info(f"Found {len(results)} items for {search_term}")
return results
except Exception as e:
logger.error(f"Scraping error for {search_term}: {str(e)}")
return []
def save_to_database_secure(results):
"""Securely save results to database with validation"""
if not results:
logger.info("No results to save")
return 0
logger.info(f"Saving {len(results)} results to database...")
conn = None
try:
conn = sqlite3.connect(DB_PATH)
conn.execute("PRAGMA foreign_keys = ON")
cursor = conn.cursor()
saved_count = 0
for result in results:
try:
# Generate hash for duplicate detection
data_hash = generate_auction_hash(result)
# Use parameterized queries to prevent SQL injection
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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
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'),
data_hash
))
if cursor.rowcount > 0:
saved_count += 1
except sqlite3.IntegrityError as e:
logger.warning(f"Integrity error (likely duplicate): {str(e)}")
except Exception as e:
logger.error(f"Error saving result: {str(e)}")
conn.commit()
logger.info(f"Successfully saved {saved_count} new records")
return saved_count
except sqlite3.Error as e:
logger.error(f"Database error: {str(e)}")
if conn:
conn.rollback()
return 0
finally:
if conn:
conn.close()
def export_to_csv_secure():
"""Securely export database to CSV"""
logger.info("Exporting to CSV...")
try:
conn = sqlite3.connect(DB_PATH)
conn.execute("PRAGMA foreign_keys = ON")
cursor = conn.cursor()
# Check table exists
cursor.execute("""
SELECT name FROM sqlite_master
WHERE type='table' AND name='auctions'
""")
if not cursor.fetchone():
logger.warning("No auctions table found")
conn.close()
return False
# Get auction records with limit
cursor.execute("""
SELECT auction_id, title, auction_house, current_price,
estimate_low, estimate_high, end_date, lot_number,
url, search_term, created_at
FROM auctions
ORDER BY created_at DESC
LIMIT 10000
""")
rows = cursor.fetchall()
# Write to CSV with proper escaping
with open(CSV_PATH, 'w', newline='', encoding='utf-8') as csvfile:
writer = csv.writer(csvfile, quoting=csv.QUOTE_MINIMAL)
# Write headers
writer.writerow([
'auction_id', 'title', 'auction_house', 'current_price',
'estimate_low', 'estimate_high', 'end_date', 'lot_number',
'url', 'search_term', 'created_at'
])
# Write data
for row in rows:
writer.writerow(row)
# Set secure file permissions
os.chmod(CSV_PATH, 0o640)
logger.info(f"Exported {len(rows)} records to CSV")
conn.close()
return True
except Exception as e:
logger.error(f"CSV export error: {str(e)}")
return False
def main():
"""Main execution with security measures"""
logger.info("=" * 60)
logger.info("SECURE LUXURY HANDBAG AUCTION SCRAPER")
logger.info(f"Started at: {datetime.now()}")
logger.info("=" * 60)
try:
# Initialize secure database
init_secure_database()
# Scrape with rate limiting
all_results = []
for term in SEARCH_TERMS:
results = scrape_liveauctioneers_secure(term)
all_results.extend(results)
# Rate limiting - random delay
delay = random.uniform(2, 5)
time.sleep(delay)
# Save to database
saved_count = save_to_database_secure(all_results)
# Export to CSV
export_success = export_to_csv_secure()
# Log summary
logger.info("=" * 60)
logger.info("SCRAPING COMPLETED")
logger.info(f"Brands searched: {len(SEARCH_TERMS)}")
logger.info(f"Results found: {len(all_results)}")
logger.info(f"New records saved: {saved_count}")
logger.info(f"CSV export: {'Success' if export_success else 'Failed'}")
logger.info("=" * 60)
sys.exit(0 if export_success else 1)
except Exception as e:
logger.error(f"Fatal error: {str(e)}")
sys.exit(1)
if __name__ == "__main__":
main()