← back to Watches
scripts/omega-daily-scraper.py
377 lines
#!/usr/bin/env python3
"""
Omega Official Website Daily Price Scraper
Scrapes ALL products from omegawatches.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" / "omega-official-prices.json"
LOG_DIR = PROJECT_DIR / "logs"
# Omega website URLs - all collections
OMEGA_COLLECTIONS = [
# Speedmaster
{"name": "Speedmaster Moonwatch Professional", "url": "https://www.omegawatches.com/en-us/watches/speedmaster/moonwatch-professional/catalog"},
{"name": "Speedmaster Moonwatch", "url": "https://www.omegawatches.com/en-us/watches/speedmaster/moonwatch/catalog"},
{"name": "Speedmaster Heritage", "url": "https://www.omegawatches.com/en-us/watches/speedmaster/heritage-models/catalog"},
{"name": "Speedmaster 57", "url": "https://www.omegawatches.com/en-us/watches/speedmaster/speedmaster-57/catalog"},
{"name": "Speedmaster Super Racing", "url": "https://www.omegawatches.com/en-us/watches/speedmaster/super-racing/catalog"},
{"name": "Speedmaster Chrono Chime", "url": "https://www.omegawatches.com/en-us/watches/speedmaster/chrono-chime/catalog"},
{"name": "Speedmaster Dark Side of the Moon", "url": "https://www.omegawatches.com/en-us/watches/speedmaster/dark-side-of-the-moon/catalog"},
{"name": "Speedmaster Two Counters", "url": "https://www.omegawatches.com/en-us/watches/speedmaster/two-counters/catalog"},
# Seamaster
{"name": "Seamaster Diver 300M", "url": "https://www.omegawatches.com/en-us/watches/seamaster/diver-300m/catalog"},
{"name": "Seamaster Planet Ocean 600M", "url": "https://www.omegawatches.com/en-us/watches/seamaster/planet-ocean-600m/catalog"},
{"name": "Seamaster Aqua Terra", "url": "https://www.omegawatches.com/en-us/watches/seamaster/aqua-terra-150m/catalog"},
{"name": "Seamaster 300", "url": "https://www.omegawatches.com/en-us/watches/seamaster/seamaster-300/catalog"},
{"name": "Seamaster Ultra Deep", "url": "https://www.omegawatches.com/en-us/watches/seamaster/ultra-deep/catalog"},
{"name": "Seamaster Planet Ocean Ultra Deep", "url": "https://www.omegawatches.com/en-us/watches/seamaster/planet-ocean-ultra-deep/catalog"},
# Constellation
{"name": "Constellation", "url": "https://www.omegawatches.com/en-us/watches/constellation/constellation/catalog"},
{"name": "Constellation Globemaster", "url": "https://www.omegawatches.com/en-us/watches/constellation/globemaster/catalog"},
# De Ville
{"name": "De Ville Prestige", "url": "https://www.omegawatches.com/en-us/watches/de-ville/prestige/catalog"},
{"name": "De Ville Tresor", "url": "https://www.omegawatches.com/en-us/watches/de-ville/tresor/catalog"},
{"name": "De Ville Ladymatic", "url": "https://www.omegawatches.com/en-us/watches/de-ville/ladymatic/catalog"},
{"name": "De Ville Tourbillon", "url": "https://www.omegawatches.com/en-us/watches/de-ville/tourbillon/catalog"},
]
# Alternative: scrape main catalog pages
OMEGA_MAIN_CATALOGS = [
"https://www.omegawatches.com/en-us/watches/speedmaster/catalog",
"https://www.omegawatches.com/en-us/watches/seamaster/catalog",
"https://www.omegawatches.com/en-us/watches/constellation/catalog",
"https://www.omegawatches.com/en-us/watches/de-ville/catalog",
]
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": "omegawatches.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 scrape_catalog_page(url, collection_name):
"""Scrape a catalog page for watch prices."""
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')
# Find product cards - Omega uses various class patterns
product_cards = soup.find_all(['div', 'article', 'li'], class_=lambda x: x and ('product' in x.lower() or 'watch' in x.lower() or 'item' in x.lower()))
if not product_cards:
# Try alternative selectors
product_cards = soup.find_all('a', href=lambda x: x and '/watches/' in x and '/p-' in x)
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 number
ref_elem = card.find(['span', 'div', 'p'], class_=lambda x: x and ('ref' in x.lower() or 'sku' in x.lower()))
ref = ref_elem.get_text(strip=True) if ref_elem else None
# Try to get ref from URL
if not ref:
link = card.find('a', href=True)
if link and '/p-' in link['href']:
ref_match = re.search(r'/p-([a-z0-9.-]+)', link['href'], re.I)
if ref_match:
ref = ref_match.group(1)
# 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.omegawatches.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 to extract from 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'),
"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'),
"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_product_page(url):
"""Scrape individual product page for detailed info."""
try:
resp = requests.get(url, headers=HEADERS, timeout=REQUEST_TIMEOUT)
resp.raise_for_status()
soup = BeautifulSoup(resp.text, 'html.parser')
# Extract from JSON-LD
scripts = soup.find_all('script', type='application/ld+json')
for script in scripts:
try:
data = json.loads(script.string)
if isinstance(data, dict) and data.get('@type') == 'Product':
price = data.get('offers', {}).get('price')
return {
"name": data.get('name'),
"reference": data.get('sku'),
"price": int(float(price)) if price else None,
"description": data.get('description'),
"url": url
}
except:
continue
# Fallback to HTML parsing
name = soup.find('h1')
name = name.get_text(strip=True) if name else None
price_elem = soup.find(class_=lambda x: x and 'price' in x.lower())
price = extract_price(price_elem.get_text() if price_elem else '')
ref_elem = soup.find(string=re.compile(r'\d{3}\.\d{2}\.\d{2}\.\d{2}\.\d{2}\.\d{3}'))
ref = ref_elem.strip() if ref_elem else None
return {
"name": name,
"reference": ref,
"price": price,
"url": url
}
except Exception as e:
print(f" Error scraping product: {e}", file=sys.stderr)
return None
def scrape_all_products():
"""Scrape all Omega products."""
print(f"\n{'='*60}", file=sys.stderr)
print(f"OMEGA 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 main catalog pages
for catalog_url in OMEGA_MAIN_CATALOGS:
collection_name = catalog_url.split('/watches/')[1].split('/')[0].title()
products = scrape_catalog_page(catalog_url, collection_name)
all_products.extend(products)
time.sleep(REQUEST_DELAY)
# Also scrape sub-collections for more detail
for collection in OMEGA_COLLECTIONS:
products = scrape_catalog_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:
# Update today's scrape
idx = data["scrapes"].index(existing_today)
data["scrapes"][idx] = snapshot
print(f"\nUpdated today's scrape ({today})", file=sys.stderr)
else:
# Add new scrape
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()