← back to Watches
scripts/rolex-daily-scraper.py
320 lines
#!/usr/bin/env python3
"""
Rolex Official Website Daily Price Scraper
Scrapes ALL products from rolex.com and logs prices daily.
Designed to run as a cron job at 6am.
"""
import json
import sys
import os
import time
import re
from datetime import datetime
from pathlib import Path
try:
import requests
from bs4 import BeautifulSoup
except ImportError:
print("Installing dependencies...", file=sys.stderr)
import subprocess
subprocess.check_call([sys.executable, "-m", "pip", "install", "requests", "beautifulsoup4", "-q"])
import requests
from bs4 import BeautifulSoup
# Configuration
SCRIPT_DIR = Path(__file__).parent.absolute()
PROJECT_DIR = SCRIPT_DIR.parent
DATA_FILE = PROJECT_DIR / "data" / "rolex-official-prices.json"
LOG_DIR = PROJECT_DIR / "logs"
# Rolex website URLs - all collections
ROLEX_COLLECTIONS = [
# Professional Watches
{"name": "Submariner", "url": "https://www.rolex.com/en-us/watches/submariner.html"},
{"name": "Cosmograph Daytona", "url": "https://www.rolex.com/en-us/watches/cosmograph-daytona.html"},
{"name": "GMT-Master II", "url": "https://www.rolex.com/en-us/watches/gmt-master-ii.html"},
{"name": "Explorer", "url": "https://www.rolex.com/en-us/watches/explorer.html"},
{"name": "Sea-Dweller", "url": "https://www.rolex.com/en-us/watches/sea-dweller.html"},
{"name": "Deepsea", "url": "https://www.rolex.com/en-us/watches/deepsea.html"},
{"name": "Yacht-Master", "url": "https://www.rolex.com/en-us/watches/yacht-master.html"},
{"name": "Air-King", "url": "https://www.rolex.com/en-us/watches/air-king.html"},
# Classic Watches
{"name": "Datejust", "url": "https://www.rolex.com/en-us/watches/datejust.html"},
{"name": "Day-Date", "url": "https://www.rolex.com/en-us/watches/day-date.html"},
{"name": "Sky-Dweller", "url": "https://www.rolex.com/en-us/watches/sky-dweller.html"},
{"name": "Oyster Perpetual", "url": "https://www.rolex.com/en-us/watches/oyster-perpetual.html"},
{"name": "Cellini", "url": "https://www.rolex.com/en-us/watches/cellini.html"},
{"name": "1908", "url": "https://www.rolex.com/en-us/watches/1908.html"},
]
# Alternative catalog URLs
ROLEX_CATALOG_URLS = [
"https://www.rolex.com/en-us/watches/configure.html",
"https://www.rolex.com/en-us/watches.html",
]
HEADERS = {
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8',
'Accept-Language': 'en-US,en;q=0.9',
'Accept-Encoding': 'gzip, deflate, br',
'Connection': 'keep-alive',
'Cache-Control': 'max-age=0',
'Sec-Fetch-Dest': 'document',
'Sec-Fetch-Mode': 'navigate',
'Sec-Fetch-Site': 'none',
'Sec-Fetch-User': '?1',
'Upgrade-Insecure-Requests': '1',
}
REQUEST_DELAY = 3 # seconds between requests
REQUEST_TIMEOUT = 60 # seconds
def load_existing_data():
"""Load existing price data."""
if DATA_FILE.exists():
with open(DATA_FILE) as f:
return json.load(f)
return {
"metadata": {
"source": "rolex.com",
"created": datetime.now().isoformat(),
"lastUpdated": None,
"totalScrapes": 0
},
"scrapes": []
}
def save_data(data):
"""Save price data."""
data["metadata"]["lastUpdated"] = datetime.now().isoformat()
data["metadata"]["totalScrapes"] = len(data["scrapes"])
# Ensure data directory exists
DATA_FILE.parent.mkdir(parents=True, exist_ok=True)
with open(DATA_FILE, 'w') as f:
json.dump(data, f, indent=2)
print(f"Saved to {DATA_FILE}", file=sys.stderr)
def extract_price(text):
"""Extract USD price from text."""
if not text:
return None
# Match $X,XXX or $XX,XXX patterns
patterns = [
r'\$\s*([\d,]+(?:\.\d{2})?)',
r'USD\s*([\d,]+(?:\.\d{2})?)',
r'([\d,]+(?:\.\d{2})?)\s*USD',
]
for pattern in patterns:
match = re.search(pattern, str(text))
if match:
try:
price_str = match.group(1).replace(',', '')
return int(float(price_str))
except (ValueError, TypeError):
continue
return None
def extract_reference(text):
"""Extract Rolex reference number from text or URL."""
if not text:
return None
# Rolex reference patterns: m126610ln-0001, 126610LN, etc.
patterns = [
r'm?(\d{6}[a-z]{2,4})-?\d{4}', # m126610ln-0001
r'(\d{6}[A-Z]{2,4})', # 126610LN
r'ref[:\s]*(\d{6})', # ref: 126610
]
for pattern in patterns:
match = re.search(pattern, str(text), re.I)
if match:
return match.group(1).upper()
return None
def scrape_collection_page(url, collection_name):
"""Scrape a collection page for watch data."""
products = []
try:
print(f" Scraping {collection_name}...", file=sys.stderr)
resp = requests.get(url, headers=HEADERS, timeout=REQUEST_TIMEOUT)
resp.raise_for_status()
soup = BeautifulSoup(resp.text, 'html.parser')
# Try to find product cards
product_cards = soup.find_all(['div', 'article', 'a'], class_=lambda x: x and ('product' in x.lower() or 'watch' in x.lower() or 'card' in x.lower()))
for card in product_cards:
try:
# Extract product name
name_elem = card.find(['h2', 'h3', 'h4', 'span', 'div'], class_=lambda x: x and ('name' in x.lower() or 'title' in x.lower()))
name = name_elem.get_text(strip=True) if name_elem else None
# Extract reference
ref = extract_reference(card.get_text())
if not ref:
link = card.find('a', href=True)
if link:
ref = extract_reference(link['href'])
# Extract price
price_elem = card.find(['span', 'div', 'p'], class_=lambda x: x and 'price' in x.lower())
price_text = price_elem.get_text(strip=True) if price_elem else card.get_text()
price = extract_price(price_text)
# Get product URL
link = card.find('a', href=True)
product_url = link['href'] if link else None
if product_url and not product_url.startswith('http'):
product_url = f"https://www.rolex.com{product_url}"
if name or ref or price:
products.append({
"collection": collection_name,
"name": name,
"reference": ref,
"price": price,
"url": product_url
})
except Exception as e:
continue
# Also try JSON-LD structured data
scripts = soup.find_all('script', type='application/ld+json')
for script in scripts:
try:
data = json.loads(script.string)
if isinstance(data, list):
for item in data:
if item.get('@type') == 'Product':
products.append({
"collection": collection_name,
"name": item.get('name'),
"reference": item.get('sku') or extract_reference(item.get('url', '')),
"price": extract_price(str(item.get('offers', {}).get('price'))),
"url": item.get('url')
})
elif data.get('@type') == 'Product':
products.append({
"collection": collection_name,
"name": data.get('name'),
"reference": data.get('sku') or extract_reference(data.get('url', '')),
"price": extract_price(str(data.get('offers', {}).get('price'))),
"url": data.get('url')
})
except (json.JSONDecodeError, AttributeError):
continue
print(f" Found {len(products)} products", file=sys.stderr)
except Exception as e:
print(f" Error scraping {collection_name}: {e}", file=sys.stderr)
return products
def scrape_all_products():
"""Scrape all Rolex products."""
print(f"\n{'='*60}", file=sys.stderr)
print(f"ROLEX OFFICIAL PRICE SCRAPER", file=sys.stderr)
print(f"Date: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", file=sys.stderr)
print(f"{'='*60}\n", file=sys.stderr)
all_products = []
# Scrape each collection
for collection in ROLEX_COLLECTIONS:
products = scrape_collection_page(collection["url"], collection["name"])
all_products.extend(products)
time.sleep(REQUEST_DELAY)
# Deduplicate by reference number
seen_refs = set()
unique_products = []
for p in all_products:
ref = p.get('reference') or p.get('name')
if ref and ref not in seen_refs:
seen_refs.add(ref)
unique_products.append(p)
# Filter out products without prices
products_with_prices = [p for p in unique_products if p.get('price')]
print(f"\nTotal unique products: {len(unique_products)}", file=sys.stderr)
print(f"Products with prices: {len(products_with_prices)}", file=sys.stderr)
return unique_products
def run_daily_scrape():
"""Run the daily scrape and save results."""
# Load existing data
data = load_existing_data()
# Scrape all products
products = scrape_all_products()
# Create today's snapshot
today = datetime.now().strftime('%Y-%m-%d')
snapshot = {
"date": today,
"timestamp": datetime.now().isoformat(),
"productsScraped": len(products),
"productsWithPrices": len([p for p in products if p.get('price')]),
"products": products
}
# Check if we already have a scrape for today
existing_today = next((s for s in data["scrapes"] if s["date"] == today), None)
if existing_today:
idx = data["scrapes"].index(existing_today)
data["scrapes"][idx] = snapshot
print(f"\nUpdated today's scrape ({today})", file=sys.stderr)
else:
data["scrapes"].append(snapshot)
print(f"\nAdded new scrape ({today})", file=sys.stderr)
# Save data
save_data(data)
# Print summary
print(f"\n{'='*60}", file=sys.stderr)
print(f"SCRAPE COMPLETE", file=sys.stderr)
print(f"Date: {today}", file=sys.stderr)
print(f"Products scraped: {len(products)}", file=sys.stderr)
print(f"Products with prices: {len([p for p in products if p.get('price')])}", file=sys.stderr)
print(f"Total scrapes in database: {len(data['scrapes'])}", file=sys.stderr)
print(f"{'='*60}\n", file=sys.stderr)
# Output JSON summary to stdout
summary = {
"success": True,
"date": today,
"productsScraped": len(products),
"productsWithPrices": len([p for p in products if p.get('price')]),
"totalScrapes": len(data["scrapes"])
}
print(json.dumps(summary, indent=2))
return data
if __name__ == '__main__':
run_daily_scrape()