`scrapers/shopify-recon.js`

192 lines
#!/usr/bin/env node
'use strict';
// Pull designer-adjacent contacts from Steve's Shopify-side databases into
// the StarsOfDesign directory. Three sources, three signal tiers:
//
//   1. dw_unified.trade_signups
//        — verified DW trade-program signups (architect/interior_designer/stager).
//        — Highest signal: every row is a self-identified design pro.
//        — Lands in sod_email_candidates with status='is_designer' (auto-promoted)
//          and signal_score=0.95; promote-candidates.js will then create the
//          sod_designers + sod_firms rows.
//   2. dw_unified.digital_sample_orders
//        — paid DW customers who ordered digital samples. Could be designers,
//          could be DIY homeowners. Mid signal (0.70).
//        — Land in sod_email_candidates with status='new', awaiting triage.
//   3. designer_representatives.designer_reps (separate DB)
//        — manufacturer rep agencies that already have name+agency+territories.
//        — Don't go into sod_email_candidates (they're not candidate-stage —
//          they're vetted). Direct-insert into sod_designers + sod_firms with
//          a 'rep' role tag.
//
// Idempotent. Re-running merges via ON CONFLICT.

require('dotenv').config();
const { Pool } = require('pg');

const dw = new Pool({
  connectionString: process.env.DATABASE_URL || 'postgresql:///dw_unified?host=/tmp&user=stevestudio2',
  max: 4,
});
const reps = new Pool({
  connectionString: 'postgresql:///designer_representatives?host=/tmp&user=stevestudio2',
  max: 2,
});

function slugify(s) {
  return String(s || '').toLowerCase().replace(/[^a-z0-9]+/g, '-').replace(/^-+|-+$/g, '').slice(0, 100);
}

async function runStart(kind, source) {
  const r = await dw.query(
    `INSERT INTO sod_ingest_runs (run_kind, source, status) VALUES ($1, $2, 'running') RETURNING id`,
    [kind, source]
  );
  return r.rows[0].id;
}
async function runDone(id, ok, inN, note) {
  await dw.query(
    `UPDATE sod_ingest_runs SET status=$1, finished_at=NOW(), records_in=$2, records_out=$3, notes=$4 WHERE id=$5`,
    [ok ? 'ok' : 'err', inN, ok ? inN : 0, note || null, id]
  );
}

// ── Source 1: trade_signups → sod_email_candidates (auto-promoted to designer)
async function importTradeSignups() {
  const runId = await runStart('shopify-trade-signups', 'dw_unified.trade_signups');
  const r = await dw.query(`SELECT name, email, company, profession, license, status, created_at FROM trade_signups`);
  let ok = 0;
  for (const t of r.rows) {
    const email = String(t.email || '').toLowerCase().trim();
    if (!email.includes('@')) continue;
    const domain = email.split('@')[1];
    const note = `dw_trade_signup · ${t.profession || 'unknown'} · ${t.status || 'approved'}${t.license ? ' · lic ' + t.license : ''}${t.company ? ' · ' + t.company : ''}`;
    await dw.query(`
      INSERT INTO sod_email_candidates
        (email, display_name, domain, first_seen, last_seen, thread_count, steve_replied,
         signal_score, status, notes)
      VALUES ($1,$2,$3,$4,$4,0,0,0.95,'is_designer',$5)
      ON CONFLICT (email) DO UPDATE SET
        display_name = COALESCE(EXCLUDED.display_name, sod_email_candidates.display_name),
        signal_score = GREATEST(sod_email_candidates.signal_score, EXCLUDED.signal_score),
        status       = CASE WHEN sod_email_candidates.status = 'new' THEN 'is_designer' ELSE sod_email_candidates.status END,
        notes        = CONCAT_WS(' || ', NULLIF(sod_email_candidates.notes,''), EXCLUDED.notes),
        updated_at   = NOW()
    `, [email, t.name, domain, t.created_at, note]);
    ok++;
  }
  await runDone(runId, true, ok, `trade signups: ${ok}/${r.rows.length}`);
  console.log(`[trade_signups] ${ok}/${r.rows.length} rows`);
}

// ── Source 2: digital_sample_orders → sod_email_candidates (mid signal)
async function importDigitalSampleOrders() {
  const runId = await runStart('shopify-digital-samples', 'dw_unified.digital_sample_orders');
  const r = await dw.query(`
    SELECT customer_email, MAX(customer_name) AS customer_name,
           MIN(detected_at) AS first_seen, MAX(detected_at) AS last_seen,
           count(*) AS order_count,
           array_agg(DISTINCT unnest_skus) FILTER (WHERE unnest_skus IS NOT NULL) AS skus
      FROM digital_sample_orders, LATERAL unnest(COALESCE(dig_skus,'{}')) AS unnest_skus
     WHERE customer_email IS NOT NULL AND customer_email <> ''
     GROUP BY customer_email`);
  let ok = 0;
  for (const c of r.rows) {
    const email = String(c.customer_email || '').toLowerCase().trim();
    if (!email.includes('@')) continue;
    const domain = email.split('@')[1];
    const note = `dw_sample_orders · ${c.order_count} order(s)${c.skus ? ' · ' + c.skus.slice(0,5).join(',') : ''}`;
    await dw.query(`
      INSERT INTO sod_email_candidates
        (email, display_name, domain, first_seen, last_seen, thread_count, steve_replied,
         signal_score, status, notes)
      VALUES ($1,$2,$3,$4,$5,$6,0,0.70,'new',$7)
      ON CONFLICT (email) DO UPDATE SET
        display_name  = COALESCE(EXCLUDED.display_name, sod_email_candidates.display_name),
        thread_count  = sod_email_candidates.thread_count + EXCLUDED.thread_count,
        first_seen    = LEAST(sod_email_candidates.first_seen, EXCLUDED.first_seen),
        last_seen     = GREATEST(sod_email_candidates.last_seen, EXCLUDED.last_seen),
        signal_score  = GREATEST(sod_email_candidates.signal_score, EXCLUDED.signal_score),
        notes         = CONCAT_WS(' || ', NULLIF(sod_email_candidates.notes,''), EXCLUDED.notes),
        updated_at    = NOW()
    `, [email, c.customer_name, domain, c.first_seen, c.last_seen, c.order_count, note]);
    ok++;
  }
  await runDone(runId, true, ok, `digital sample orders: ${ok}/${r.rows.length}`);
  console.log(`[digital_sample_orders] ${ok}/${r.rows.length} unique customers`);
}

// ── Source 3: designer_reps (separate DB) → sod_designers + sod_firms (direct curated rows)
async function importDesignerReps() {
  const runId = await runStart('shopify-designer-reps', 'designer_representatives.designer_reps');
  const r = await reps.query(`
    SELECT slug, name, agency, city, region, email, phone, website_url, bio,
           territories, lines, vertical, is_claimed
      FROM designer_reps
     WHERE is_published = true`);
  let ok = 0, firmsCreated = 0, designersCreated = 0;
  for (const rep of r.rows) {
    // Firm/agency first
    let firmId = null;
    if (rep.agency) {
      const firmSlug = slugify(rep.agency);
      const f = await dw.query(`
        INSERT INTO sod_firms (slug, name, city, state_or_region, primary_email, website, bio, styles)
        VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
        ON CONFLICT (slug) DO UPDATE SET
          primary_email = COALESCE(sod_firms.primary_email, EXCLUDED.primary_email),
          website       = COALESCE(sod_firms.website, EXCLUDED.website),
          bio           = COALESCE(sod_firms.bio, EXCLUDED.bio),
          updated_at    = NOW()
        RETURNING id, (xmax = 0) AS inserted`,
        [firmSlug, rep.agency, rep.city, rep.region, rep.email, rep.website_url, rep.bio, rep.lines || []]);
      firmId = f.rows[0].id;
      if (f.rows[0].inserted) firmsCreated++;
    }
    // Designer (treat rep as a designer-adjacent contact with role='rep')
    if (rep.name) {
      const designerSlug = slugify(rep.name) + '-rep';
      const d = await dw.query(`
        INSERT INTO sod_designers (slug, full_name, primary_email, city, state_or_region, role, bio, styles, headshot_source)
        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, 'made-with-ai')
        ON CONFLICT (slug) DO UPDATE SET
          primary_email   = COALESCE(sod_designers.primary_email, EXCLUDED.primary_email),
          city            = COALESCE(sod_designers.city, EXCLUDED.city),
          state_or_region = COALESCE(sod_designers.state_or_region, EXCLUDED.state_or_region),
          role            = COALESCE(sod_designers.role, EXCLUDED.role),
          bio             = COALESCE(sod_designers.bio, EXCLUDED.bio),
          updated_at      = NOW()
        RETURNING id, (xmax = 0) AS inserted`,
        [designerSlug, rep.name, rep.email, rep.city, rep.region, 'Manufacturer Rep', rep.bio, rep.lines || []]);
      const designerId = d.rows[0].id;
      if (d.rows[0].inserted) designersCreated++;
      if (firmId) {
        await dw.query(`
          INSERT INTO sod_designer_firm (designer_id, firm_id, title, is_current, source)
          VALUES ($1, $2, 'Manufacturer Rep', true, 'designer-reps-import')
          ON CONFLICT DO NOTHING`, [designerId, firmId]);
      }
      // Add website + email as sod_links if present
      if (rep.website_url) {
        await dw.query(`
          INSERT INTO sod_links (designer_id, firm_id, url, kind, added_by)
          VALUES ($1, $2, $3, 'firm-site', 'designer-reps-import')
          ON CONFLICT DO NOTHING`, [designerId, firmId, rep.website_url]);
      }
    }
    ok++;
  }
  await runDone(runId, true, ok, `reps: ${ok}/${r.rows.length} · firms+${firmsCreated} · designers+${designersCreated}`);
  console.log(`[designer_reps] ${ok}/${r.rows.length} reps · firms+${firmsCreated} · designers+${designersCreated}`);
}

async function main() {
  await importTradeSignups();
  await importDigitalSampleOrders();
  await importDesignerReps();
  await dw.end();
  await reps.end();
}

main().catch(e => { console.error('fatal:', e); process.exit(1); });