`collectors/base-collector.mjs`

238 lines
/**
 * Base Collector — Abstract class for all data source collectors
 * Omega Watches 2.0 Enterprise Pipeline
 */
import pg from 'pg';
const { Pool } = pg;

const pool = new Pool({
  host: process.env.DB_HOST || '127.0.0.1',
  port: parseInt(process.env.DB_PORT || '5432'),
  database: process.env.DB_NAME || 'omega_watches_v2',
  user: process.env.DB_USER || 'omega_admin',
  password: process.env.DB_PASSWORD || 'OmegaWatch2024Pass',
  max: 5,
});

export class BaseCollector {
  constructor(sourceName) {
    this.sourceName = sourceName;
    this.source = null;
    this.parser = null;
    this.runId = null;
    this.stats = {
      fetched: 0,
      parsed: 0,
      inserted: 0,
      updated: 0,
      quarantined: 0,
      deduplicated: 0,
    };
    this.startTime = null;
  }

  async initialize() {
    // Load source config
    const { rows } = await pool.query(
      'SELECT * FROM data_source WHERE name = $1',
      [this.sourceName]
    );
    if (!rows[0]) throw new Error(`Source not found: ${this.sourceName}`);
    this.source = rows[0];

    // Check ToS compliance
    if (this.source.tos_reviewed && this.source.tos_allows_collection === false) {
      throw new Error(`ToS prohibits automated collection for ${this.sourceName}`);
    }

    // Load current parser
    const { rows: parsers } = await pool.query(
      'SELECT * FROM parser_version WHERE source_id = $1 AND is_current = true',
      [this.source.id]
    );
    this.parser = parsers[0] || { version: '1.0.0', selectors_json: {} };

    // Create collector run record
    const { rows: runs } = await pool.query(
      `INSERT INTO collector_run (source_id, job_type, status, parser_version, started_at)
       VALUES ($1, $2, 'running', $3, NOW()) RETURNING run_id`,
      [this.source.id, this.getJobType(), this.parser.version]
    );
    this.runId = runs[0].run_id;
    this.startTime = Date.now();

    console.log(`[${this.sourceName}] Collector initialized (run #${this.runId}, parser v${this.parser.version})`);
  }

  getJobType() { return 'incremental'; }

  // Rate limiting
  async rateLimit() {
    const rpm = this.source?.rate_limit_rpm || 10;
    const delayMs = (60 / rpm) * 1000;
    await new Promise(r => setTimeout(r, delayMs));
  }

  // Insert market event with dedup check
  async insertMarketEvent(event) {
    try {
      // Hard dedup: (source_name, source_listing_id)
      const { rows: existing } = await pool.query(
        'SELECT event_id FROM market_event WHERE source_name = $1 AND source_listing_id = $2',
        [event.source_name, event.source_listing_id]
      );

      if (existing.length > 0) {
        this.stats.deduplicated++;
        return { action: 'deduplicated', event_id: existing[0].event_id };
      }

      // Validate and quarantine if needed
      const quarantineReason = this.validate(event);
      if (quarantineReason) {
        event.is_quarantined = true;
        event.quarantine_reason = quarantineReason;
        this.stats.quarantined++;
      }

      const { rows } = await pool.query(
        `INSERT INTO market_event (
          source_name, source_listing_id, event_type, reference_number_raw,
          serial_number, year_text, case_material, movement,
          condition_raw, condition_normalized, provenance_raw,
          sale_date, location_text, seller_type,
          price_amount, currency, hammer_price, buyers_premium, total_to_buyer, fee_semantics,
          images_json, source_url, page_hash, parser_version,
          is_quarantined, quarantine_reason
        ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26)
        RETURNING event_id`,
        [
          event.source_name, event.source_listing_id, event.event_type,
          event.reference_number_raw, event.serial_number, event.year_text,
          event.case_material, event.movement, event.condition_raw,
          event.condition_normalized || 'unknown', event.provenance_raw,
          event.sale_date, event.location_text, event.seller_type || 'unknown',
          event.price_amount, event.currency || 'USD',
          event.hammer_price, event.buyers_premium, event.total_to_buyer,
          event.fee_semantics || 'unknown',
          event.images_json ? JSON.stringify(event.images_json) : null,
          event.source_url, event.page_hash, this.parser.version,
          event.is_quarantined || false, event.quarantine_reason
        ]
      );

      this.stats.inserted++;

      // Create quarantine record if quarantined
      if (event.is_quarantined) {
        await pool.query(
          `INSERT INTO quarantine_record (event_id, reason, severity, details)
           VALUES ($1, $2, $3, $4)`,
          [rows[0].event_id, quarantineReason, this.classifySeverity(quarantineReason), JSON.stringify({ source: this.sourceName })]
        );
      }

      return { action: 'inserted', event_id: rows[0].event_id };
    } catch (err) {
      if (err.code === '23505') { // unique violation
        this.stats.deduplicated++;
        return { action: 'deduplicated' };
      }
      throw err;
    }
  }

  // Insert MSRP snapshot
  async insertMSRPSnapshot(snapshot) {
    try {
      await pool.query(
        `INSERT INTO msrp_snapshot (reference_id, region_code, currency, msrp_amount, captured_date, source_url, page_hash, parser_version)
         VALUES ($1,$2,$3,$4,$5,$6,$7,$8)
         ON CONFLICT (reference_id, region_code, captured_date) DO NOTHING`,
        [
          snapshot.reference_id, snapshot.region_code, snapshot.currency,
          snapshot.msrp_amount, snapshot.captured_date, snapshot.source_url,
          snapshot.page_hash, this.parser.version
        ]
      );
      this.stats.inserted++;
    } catch (err) {
      console.error(`[${this.sourceName}] MSRP insert error:`, err.message);
    }
  }

  // Validation — override in subclasses for source-specific rules
  validate(event) {
    if (!event.price_amount && !event.hammer_price) return 'missing_price';
    if (event.price_amount < 0) return 'negative_price';
    if (event.price_amount > 50000000) return 'implausible_price';
    if (!event.source_listing_id) return 'missing_listing_id';
    return null; // no quarantine needed
  }

  classifySeverity(reason) {
    const critical = ['implausible_price'];
    const high = ['missing_price', 'negative_price'];
    if (critical.includes(reason)) return 'critical';
    if (high.includes(reason)) return 'high';
    return 'medium';
  }

  // Archive raw HTML/PDF snapshot
  async archiveSnapshot(url, content, contentType = 'text/html') {
    const crypto = await import('crypto');
    const contentHash = crypto.createHash('sha256').update(content).digest('hex');

    try {
      await pool.query(
        `INSERT INTO raw_snapshot (source_name, url, content_hash, content_type, content_size_bytes, parser_version)
         VALUES ($1,$2,$3,$4,$5,$6)
         ON CONFLICT DO NOTHING`,
        [this.sourceName, url, contentHash, contentType, Buffer.byteLength(content), this.parser.version]
      );
    } catch (err) {
      console.warn(`[${this.sourceName}] Snapshot archive warning:`, err.message);
    }
    return contentHash;
  }

  // Finalize run
  async finalize(status = 'completed', errorMessage = null) {
    const durationMs = Date.now() - this.startTime;
    await pool.query(
      `UPDATE collector_run SET
        status = $1, completed_at = NOW(), duration_ms = $2,
        records_fetched = $3, records_parsed = $4, records_inserted = $5,
        records_updated = $6, records_quarantined = $7, records_deduplicated = $8,
        error_message = $9
       WHERE run_id = $10`,
      [
        status, durationMs,
        this.stats.fetched, this.stats.parsed, this.stats.inserted,
        this.stats.updated, this.stats.quarantined, this.stats.deduplicated,
        errorMessage, this.runId
      ]
    );
    console.log(`[${this.sourceName}] Run #${this.runId} ${status} in ${durationMs}ms:`, this.stats);
  }

  // Override in subclass
  async collect() {
    throw new Error('collect() must be implemented by subclass');
  }

  // Main entry point
  async run() {
    try {
      await this.initialize();
      await this.collect();
      await this.finalize('completed');
    } catch (err) {
      console.error(`[${this.sourceName}] Collection failed:`, err.message);
      if (this.runId) await this.finalize('failed', err.message);
    }
  }
}

export { pool };