`collectors/omega-official/collector.mjs`

214 lines
/**
 * Omega Official MSRP Collector
 *
 * Fetches current retail prices from omegawatches.com for all
 * references in our database. Writes to the msrp_snapshot ledger.
 *
 * Data flow:
 *   1. Load all watch_reference rows
 *   2. For each reference, build the Omega product page URL
 *   3. Fetch the page and extract the retail price
 *   4. Insert into msrp_snapshot (dedup: reference_id + region + date)
 *   5. Track stats in collector_run
 *
 * The Omega website uses the pattern:
 *   https://www.omegawatches.com/watch-omega-{collection}-{ref}
 * But since URLs vary, we also use their product API endpoint.
 */
import { BaseCollector, pool } from '../base-collector.mjs';
import https from 'https';
import crypto from 'crypto';

const OMEGA_BASE = 'https://www.omegawatches.com';
const USER_AGENT = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36';

export default class OmegaOfficialCollector extends BaseCollector {
  constructor() {
    super('omega_official');
    this.references = [];
  }

  getJobType() { return 'daily_msrp'; }

  /**
   * HTTP GET with TLS settings matching a regular browser
   */
  fetch(url) {
    return new Promise((resolve, reject) => {
      const req = https.get(url, {
        headers: {
          'User-Agent': USER_AGENT,
          'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
          'Accept-Language': 'en-US,en;q=0.9',
        },
        timeout: 30000,
      }, (res) => {
        if (res.statusCode >= 300 && res.statusCode < 400 && res.headers.location) {
          return this.fetch(res.headers.location).then(resolve).catch(reject);
        }
        let body = '';
        res.on('data', chunk => body += chunk);
        res.on('end', () => resolve({ statusCode: res.statusCode, body, headers: res.headers }));
      });
      req.on('error', reject);
      req.on('timeout', () => { req.destroy(); reject(new Error('Request timeout')); });
    });
  }

  /**
   * Extract price from Omega product page HTML
   * Omega uses structured data (JSON-LD) and visible price elements
   */
  extractPrice(html) {
    // Try JSON-LD structured data first
    const ldMatch = html.match(/<script[^>]*type="application\/ld\+json"[^>]*>([\s\S]*?)<\/script>/gi);
    if (ldMatch) {
      for (const block of ldMatch) {
        try {
          const jsonStr = block.replace(/<\/?script[^>]*>/gi, '');
          const data = JSON.parse(jsonStr);
          const product = Array.isArray(data) ? data.find(d => d['@type'] === 'Product') : data;
          if (product?.offers?.price) {
            return {
              amount: parseFloat(product.offers.price),
              currency: product.offers.priceCurrency || 'USD',
            };
          }
          if (product?.offers?.[0]?.price) {
            return {
              amount: parseFloat(product.offers[0].price),
              currency: product.offers[0].priceCurrency || 'USD',
            };
          }
        } catch {}
      }
    }

    // Fallback: Extract from visible price elements
    // Omega typically renders: <span class="price">$6,350</span> or similar
    const pricePatterns = [
      /class="[^"]*price[^"]*"[^>]*>\s*\$?\s*([\d,]+(?:\.\d{2})?)\s*</i,
      /data-price="([\d.]+)"/i,
      /itemprop="price"\s+content="([\d.]+)"/i,
      />\s*(?:USD|US\$|\$)\s*([\d,]+(?:\.\d{2})?)\s*</i,
    ];

    for (const pat of pricePatterns) {
      const m = html.match(pat);
      if (m) {
        const amount = parseFloat(m[1].replace(/,/g, ''));
        if (amount > 100 && amount < 5000000) {
          return { amount, currency: 'USD' };
        }
      }
    }

    return null;
  }

  /**
   * Build search URL for reference number
   * Omega's search endpoint returns product pages
   */
  buildSearchUrl(referenceNumber) {
    // Omega's internal search API
    return `${OMEGA_BASE}/en-us/search?q=${encodeURIComponent(referenceNumber)}`;
  }

  /**
   * Build direct product URL from reference number
   * Omega format: reference dots removed, lowercased
   * e.g., 310.30.42.50.01.001 → /en-us/watch/omega-speedmaster/31030425001001
   */
  buildProductUrl(referenceNumber) {
    const refClean = referenceNumber.replace(/\./g, '');
    return `${OMEGA_BASE}/en-us/watch/omega-watch/${refClean}`;
  }

  async collect() {
    // Load all references from our database
    const { rows } = await pool.query(
      'SELECT id, reference_number, collection, model_name FROM watch_reference ORDER BY collection, reference_number'
    );
    this.references = rows;
    this.stats.fetched = rows.length;

    console.log(`[omega_official] Collecting MSRP for ${rows.length} references`);

    const today = new Date().toISOString().split('T')[0];
    let successCount = 0;
    let skipCount = 0;

    for (const ref of rows) {
      try {
        await this.rateLimit();

        // Check if we already have today's snapshot
        const { rows: existing } = await pool.query(
          `SELECT msrp_id FROM msrp_snapshot
           WHERE reference_id = $1 AND region_code = 'US' AND captured_date = $2`,
          [ref.id, today]
        );

        if (existing.length > 0) {
          this.stats.deduplicated++;
          skipCount++;
          continue;
        }

        // Try direct product URL first
        const url = this.buildProductUrl(ref.reference_number);
        let response;
        try {
          response = await this.fetch(url);
        } catch (fetchErr) {
          console.warn(`[omega_official] Fetch error for ${ref.reference_number}: ${fetchErr.message}`);
          continue;
        }

        this.stats.parsed++;

        if (response.statusCode === 200 && response.body.length > 1000) {
          const pageHash = crypto.createHash('sha256').update(response.body).digest('hex');
          const price = this.extractPrice(response.body);

          if (price && price.amount > 0) {
            await this.insertMSRPSnapshot({
              reference_id: ref.id,
              region_code: 'US',
              currency: price.currency === 'USD' ? 'USD' : price.currency,
              msrp_amount: price.amount,
              captured_date: today,
              source_url: url,
              page_hash: pageHash,
            });
            successCount++;
            console.log(`  ${ref.reference_number}: ${price.currency} ${price.amount.toLocaleString()}`);
          } else {
            // Page loaded but no price found — maybe discontinued
            console.log(`  ${ref.reference_number}: No price found (page ${response.statusCode})`);
          }

          // Archive the raw page
          await this.archiveSnapshot(url, response.body);
        } else if (response.statusCode === 404) {
          console.log(`  ${ref.reference_number}: 404 — reference not on Omega site`);
        } else {
          console.log(`  ${ref.reference_number}: HTTP ${response.statusCode}`);
        }
      } catch (err) {
        console.error(`  ${ref.reference_number}: Error — ${err.message}`);
      }
    }

    console.log(`[omega_official] Complete: ${successCount} prices captured, ${skipCount} already today, ${this.stats.quarantined} quarantined`);
  }

  validate(event) {
    // MSRP-specific validation (for any events we insert)
    if (event.price_amount && event.price_amount < 500) return 'msrp_too_low';
    if (event.price_amount && event.price_amount > 2000000) return 'msrp_too_high';
    return null;
  }
}