`scrapers/triage-candidates.js`

181 lines
#!/usr/bin/env node
'use strict';
// Tick 2 — Classify sod_email_candidates rows still at status='new' via local
// Ollama (qwen3:14b). Each row gets exactly one of:
//   is_designer  — individual interior/residential/commercial designer
//   is_firm      — design firm / studio / agency
//   is_vendor    — manufacturer, supplier, fabric/wallcovering brand, rep
//   is_press     — press / editor / publication
//   is_noise     — personal, unrelated, transactional, family, friends
//
// Why local-only — Steve's standing rule (memory: feedback_never_anthropic_api_key)
// + this is a YOLO loop expected to be unmetered.
//
// Usage:
//   node scrapers/triage-candidates.js                # all status='new'
//   node scrapers/triage-candidates.js --limit=20     # cap for safety
//   node scrapers/triage-candidates.js --redo=is_designer  # re-triage existing
//   node scrapers/triage-candidates.js --dry          # don't write

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

const args = process.argv.slice(2).reduce((m, a) => {
  const [k, v] = a.replace(/^--/, '').split('=');
  m[k] = v ?? true; return m;
}, {});
const LIMIT = parseInt(args.limit || '500', 10);
const REDO  = args.redo || null;
const DRY   = !!args.dry;
// Default: gemma3:12b. We tried qwen3:14b first but its <think> mode swallows
// the entire 200-token output before producing any JSON, even with format:json.
// gemma3:12b is fast (~0.5s/row) and stable in JSON-mode.
const MODEL = args.model || 'gemma3:12b';
const OLLAMA = process.env.OLLAMA_HOST || 'http://127.0.0.1:11434';

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

const VALID = new Set(['is_designer','is_firm','is_vendor','is_press','is_noise']);

const SYSTEM = `You classify the role of an email contact from Steve Abrams' inbox.
Steve runs Designer Wallcoverings (DW), a luxury wallpaper/fabric brand. He
receives email from interior designers (his customers + target audience),
design firms (often as a unit), manufacturer vendors (suppliers, brand reps),
press (magazine editors, journalists, design-publication staff), and noise
(family, friends, banks, SaaS notifications, unrelated personal).

Return one of these labels exactly:
  is_designer   - an individual interior/residential/commercial designer
  is_firm       - a design firm/studio/agency (often via info@firm.com)
  is_vendor     - manufacturer, fabric/wallcovering brand, rep, supplier
  is_press      - press, editor, publication, journalist
  is_noise      - personal, unrelated, transactional, family, accountant

Also extract a confidence (low/medium/high) and an optional short reason.
Output STRICT JSON only — no preamble, no markdown fences:
  {"label":"is_designer","confidence":"high","reason":"..."}
`;

function buildUserPrompt(c) {
  const parts = [];
  parts.push(`Email: ${c.email}`);
  if (c.display_name) parts.push(`Display name: ${c.display_name}`);
  parts.push(`Domain: ${c.domain || c.email.split('@')[1]}`);
  if (c.notes) parts.push(`Notes / subject snippet: ${c.notes.slice(0, 240)}`);
  parts.push(`Thread count with Steve: ${c.thread_count}`);
  parts.push(`Replies from Steve: ${c.steve_replied}`);
  if (c.signal_score != null) parts.push(`Reply-signal score: ${Number(c.signal_score).toFixed(2)}`);
  return parts.join('\n');
}

async function ask(systemPrompt, userPrompt) {
  const res = await fetch(`${OLLAMA}/api/chat`, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({
      model: MODEL,
      stream: false,
      format: 'json',
      messages: [
        { role: 'system', content: systemPrompt },
        { role: 'user',   content: userPrompt },
      ],
      options: { temperature: 0.1, num_predict: 200 },
    }),
  });
  if (!res.ok) throw new Error(`ollama ${res.status}: ${(await res.text()).slice(0,200)}`);
  const j = await res.json();
  const raw = (j.message && j.message.content) || '';
  // qwen3 sometimes wraps in <think>…</think> blocks before the JSON
  const stripped = raw.replace(/<think>[\s\S]*?<\/think>/gi, '').trim();
  // First {...} block
  const m = stripped.match(/\{[\s\S]*\}/);
  if (!m) throw new Error('no JSON object in response: ' + stripped.slice(0,160));
  return JSON.parse(m[0]);
}

async function main() {
  const where = REDO ? `status = $1` : `status = 'new'`;
  const params = REDO ? [REDO, LIMIT] : [LIMIT];
  const limitParam = REDO ? '$2' : '$1';
  const r = await pool.query(
    `SELECT id, email, display_name, domain, notes, thread_count, steve_replied, signal_score
       FROM sod_email_candidates
      WHERE ${where}
      ORDER BY signal_score DESC NULLS LAST, id
      LIMIT ${limitParam}`,
    params
  );
  console.log(`[triage] ${r.rows.length} candidate(s)  model=${MODEL}  dry=${DRY}`);
  let runId = null;
  if (!DRY) {
    const ir = await pool.query(
      `INSERT INTO sod_ingest_runs (run_kind, source, status) VALUES ('llm-triage', $1, 'running') RETURNING id`,
      [MODEL]
    );
    runId = ir.rows[0].id;
  }

  const tally = { is_designer:0, is_firm:0, is_vendor:0, is_press:0, is_noise:0, error:0 };
  let promoteList = [];

  for (const c of r.rows) {
    try {
      // Cheap deterministic shortcuts before paying for an LLM call.
      if (/^customer\d+@example\.com$/i.test(c.email)) {
        const label = 'is_noise';
        tally[label]++;
        if (!DRY) await pool.query(
          `UPDATE sod_email_candidates SET status=$1, notes=CONCAT_WS(' || ', NULLIF(notes,''), 'triage:is_noise(high) — masked-customer placeholder, not a real contact'), updated_at=NOW() WHERE id=$2`,
          [label, c.id]
        );
        console.log(`  ${c.email.padEnd(40)} → ${label} (skip-LLM)`);
        continue;
      }
      const out = await ask(SYSTEM, buildUserPrompt(c));
      const label = (out.label || '').trim();
      if (!VALID.has(label)) throw new Error('invalid label: ' + label);
      tally[label] = (tally[label] || 0) + 1;
      const noteFrag = `triage:${label}(${out.confidence||'?'})${out.reason ? ' — ' + out.reason.slice(0,120) : ''}`;
      console.log(`  ${c.email.padEnd(40)} → ${label.padEnd(12)} ${out.confidence||''} · ${out.reason||''}`);
      if (!DRY) {
        await pool.query(
          `UPDATE sod_email_candidates
              SET status = $1,
                  notes  = CONCAT_WS(' || ', NULLIF(notes,''), $2::text),
                  updated_at = NOW()
            WHERE id = $3`,
          [label, noteFrag, c.id]
        );
      }
      if ((label === 'is_designer' || label === 'is_firm') && (out.confidence === 'high' || out.confidence === 'medium')) {
        promoteList.push(c.email);
      }
    } catch (e) {
      tally.error++;
      console.error(`  ${c.email}: ${e.message}`);
    }
  }

  if (runId) {
    await pool.query(
      `UPDATE sod_ingest_runs SET status='ok', finished_at=NOW(), records_in=$1, records_out=$2, notes=$3 WHERE id=$4`,
      [r.rows.length, r.rows.length - tally.error, JSON.stringify(tally), runId]
    );
  }
  console.log(`[triage] done. tally=${JSON.stringify(tally)}  to-promote=${promoteList.length}`);
  await pool.end();

  // Persist promote list so the wrapper can spawn promote-candidates.js
  if (promoteList.length && !DRY) {
    const fs = require('fs');
    fs.writeFileSync('/tmp/sod-triage-promote.txt', promoteList.join('\n') + '\n');
    console.log(`[triage] wrote /tmp/sod-triage-promote.txt with ${promoteList.length} emails`);
  }
}

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