← back to Watches
scripts/wayback-price-scraper.py
309 lines
#!/usr/bin/env python3
"""
Wayback Machine Price History Scraper
Fetches historical Omega watch prices from archived Chrono24 pages.
Uses Internet Archive's CDX API to find snapshots from the past 2 years.
"""
import json
import sys
import argparse
import time
import re
from datetime import datetime, timedelta
from html.parser import HTMLParser
try:
import requests
except ImportError:
print("Installing requests...", file=sys.stderr)
import subprocess
subprocess.check_call([sys.executable, "-m", "pip", "install", "requests", "-q"])
import requests
class PriceExtractor(HTMLParser):
"""Extract price from archived Chrono24 HTML pages."""
def __init__(self):
super().__init__()
self.prices = []
self.in_price = False
self.current_data = ""
def handle_starttag(self, tag, attrs):
attrs_dict = dict(attrs)
# Look for price elements - Chrono24 uses various price classes
class_name = attrs_dict.get('class', '')
if 'price' in class_name.lower() or 'amount' in class_name.lower():
self.in_price = True
def handle_endtag(self, tag):
if self.in_price:
self.in_price = False
if self.current_data:
# Try to extract price from collected data
price = self.extract_price(self.current_data)
if price:
self.prices.append(price)
self.current_data = ""
def handle_data(self, data):
if self.in_price:
self.current_data += data
# Also look for price patterns in regular content
price = self.extract_price(data)
if price and price > 1000 and price < 500000:
self.prices.append(price)
def extract_price(self, text):
"""Extract USD price from text."""
# Match patterns like $12,345 or USD 12,345 or 12,345 USD
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, text)
if match:
try:
price_str = match.group(1).replace(',', '')
return int(float(price_str))
except (ValueError, AttributeError):
continue
return None
def fetch_cdx_data(base_url, from_year=2024, to_year=2026, limit=100):
"""Fetch CDX index data from Wayback Machine."""
cdx_url = "https://web.archive.org/cdx/search/cdx"
params = {
"url": base_url,
"matchType": "prefix",
"from": str(from_year),
"to": str(to_year),
"output": "json",
"filter": "statuscode:200",
"limit": str(limit)
}
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
}
try:
resp = requests.get(cdx_url, params=params, headers=headers, timeout=60)
resp.raise_for_status()
data = resp.json()
# First row is headers, rest is data
if len(data) > 1:
return data[1:] # Skip header row
return []
except Exception as e:
print(f"Error fetching CDX data: {e}", file=sys.stderr)
return []
def fetch_archived_page(archive_url):
"""Fetch content from an archived page."""
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
}
try:
resp = requests.get(archive_url, headers=headers, timeout=60)
resp.raise_for_status()
return resp.text
except Exception as e:
print(f"Error fetching archived page: {e}", file=sys.stderr)
return None
def extract_price_from_html(html_content):
"""Extract price from HTML content."""
if not html_content:
return None
# Method 1: Use HTML parser
parser = PriceExtractor()
try:
parser.feed(html_content)
if parser.prices:
# Return most common price (filter outliers)
valid_prices = [p for p in parser.prices if 1000 < p < 500000]
if valid_prices:
return max(set(valid_prices), key=valid_prices.count)
except Exception:
pass
# Method 2: Regex fallback for Chrono24 specific patterns
# Look for price in structured data or specific divs
patterns = [
r'"price"\s*:\s*"?\$?([\d,]+)"?',
r'data-price="([\d,]+)"',
r'class="[^"]*price[^"]*"[^>]*>([\s\S]*?\$[\d,]+)',
r'<span[^>]*class="[^"]*price[^"]*"[^>]*>\s*\$?([\d,]+)',
]
for pattern in patterns:
matches = re.findall(pattern, html_content, re.IGNORECASE)
for match in matches:
try:
price_str = re.sub(r'[^\d]', '', str(match))
price = int(price_str)
if 1000 < price < 500000:
return price
except (ValueError, TypeError):
continue
return None
def get_omega_search_urls():
"""Get Chrono24 search URLs for various Omega models."""
models = {
'speedmaster-moonwatch': 'chrono24.com/omega/speedmaster-moonwatch',
'speedmaster-professional': 'chrono24.com/omega/speedmaster-professional',
'speedmaster': 'chrono24.com/omega/speedmaster',
'seamaster-300': 'chrono24.com/omega/seamaster-300',
'seamaster-diver-300': 'chrono24.com/omega/seamaster-diver-300m',
'seamaster-planet-ocean': 'chrono24.com/omega/seamaster-planet-ocean',
'seamaster-aqua-terra': 'chrono24.com/omega/seamaster-aqua-terra',
'constellation': 'chrono24.com/omega/constellation',
'de-ville': 'chrono24.com/omega/de-ville',
}
return models
def scrape_historical_prices(model_key, limit=50, delay=1.0):
"""
Scrape historical prices for a specific Omega model.
Args:
model_key: Key from get_omega_search_urls()
limit: Max number of archived pages to fetch
delay: Delay between requests (be respectful to archive.org)
Returns:
dict with price history data
"""
models = get_omega_search_urls()
if model_key not in models:
return {'error': f'Unknown model: {model_key}', 'prices': []}
base_url = models[model_key]
print(f"Fetching CDX data for {base_url}...", file=sys.stderr)
# Get archived page list
cdx_data = fetch_cdx_data(base_url, from_year=2024, to_year=2026, limit=limit)
if not cdx_data:
return {
'success': False,
'error': 'No archived pages found',
'model': model_key,
'prices': []
}
print(f"Found {len(cdx_data)} archived pages", file=sys.stderr)
prices_by_date = {}
# Sample pages (don't fetch all to avoid overloading)
sample_size = min(20, len(cdx_data))
sampled = cdx_data[::max(1, len(cdx_data) // sample_size)][:sample_size]
for i, entry in enumerate(sampled):
try:
# CDX format: urlkey, timestamp, original, mimetype, statuscode, digest, length
timestamp = entry[1]
original_url = entry[2]
# Parse timestamp to date
date_str = datetime.strptime(timestamp[:8], '%Y%m%d').strftime('%Y-%m-%d')
# Build archive URL
archive_url = f"https://web.archive.org/web/{timestamp}/{original_url}"
print(f" [{i+1}/{len(sampled)}] Fetching {date_str}...", file=sys.stderr)
# Fetch and parse page
html = fetch_archived_page(archive_url)
if html:
price = extract_price_from_html(html)
if price:
if date_str not in prices_by_date:
prices_by_date[date_str] = []
prices_by_date[date_str].append(price)
print(f" Found price: ${price:,}", file=sys.stderr)
# Be respectful - add delay
time.sleep(delay)
except Exception as e:
print(f" Error processing entry: {e}", file=sys.stderr)
continue
# Aggregate prices by date
price_history = []
for date_str, prices in sorted(prices_by_date.items()):
avg_price = int(sum(prices) / len(prices))
price_history.append({
'date': date_str,
'avgPrice': avg_price,
'minPrice': min(prices),
'maxPrice': max(prices),
'sampleCount': len(prices)
})
# Calculate statistics
all_prices = [p for prices in prices_by_date.values() for p in prices]
return {
'success': True,
'model': model_key,
'pagesFound': len(cdx_data),
'pagesScraped': len(sampled),
'priceHistory': price_history,
'statistics': {
'avgPrice': int(sum(all_prices) / len(all_prices)) if all_prices else None,
'minPrice': min(all_prices) if all_prices else None,
'maxPrice': max(all_prices) if all_prices else None,
'totalPricesFound': len(all_prices)
},
'source': 'Wayback Machine (archive.org)',
'timestamp': int(time.time() * 1000),
'fetchedAt': datetime.now().isoformat()
}
def main():
parser = argparse.ArgumentParser(description='Scrape historical Omega prices from Wayback Machine')
parser.add_argument('--model', default='speedmaster',
help='Model to search (speedmaster, seamaster-300, etc.)')
parser.add_argument('--limit', type=int, default=50,
help='Max archived pages to search')
parser.add_argument('--delay', type=float, default=1.0,
help='Delay between requests in seconds')
parser.add_argument('--list-models', action='store_true',
help='List available model keys')
args = parser.parse_args()
if args.list_models:
models = get_omega_search_urls()
print(json.dumps({'models': list(models.keys())}, indent=2))
return
result = scrape_historical_prices(
model_key=args.model,
limit=args.limit,
delay=args.delay
)
print(json.dumps(result, indent=2))
if __name__ == '__main__':
main()