`scripts/bulk-draft-interviews.js`

101 lines
#!/usr/bin/env node
// Bulk AI-draft every business that doesn't yet have an interview.
//
// Discovers alive Ollama hosts (MS1 + Mac2) at startup, then runs ONE worker
// per alive host — each worker pulls from a shared queue and pins its draft
// requests to its assigned host. If MS1 is wedged we get serial Mac2-only
// throughput; if both alive we get ~2× throughput automatically.
//
//   node scripts/bulk-draft-interviews.js             # all unfilled, all alive hosts
//   LIMIT=10 node scripts/bulk-draft-interviews.js    # cap to first 10
//   DRY=1 node scripts/bulk-draft-interviews.js       # plan only, no LLM/DB
//   HOSTS=http://127.0.0.1:11434 node ...             # force a specific host

import 'dotenv/config';
import { many, query } from '../src/lib/db.js';
import { draftInterview, aliveHosts, KNOWN_HOSTS } from '../src/lib/ollama-draft.js';

const LIMIT = process.env.LIMIT ? parseInt(process.env.LIMIT, 10) : null;
const DRY   = process.env.DRY === '1';
const FORCED_HOSTS = process.env.HOSTS ? process.env.HOSTS.split(',').map(s => s.trim()) : null;

function ts() { return new Date().toISOString().replace('T', ' ').slice(0, 19); }
function log(msg) { console.log(`[${ts()}] ${msg}`); }

async function main() {
  const todo = await many(`
    SELECT b.id, b.slug, b.name, b.neighborhood, b.category, b.address,
           b.years_in_business, b.owner_name, b.website, b.about_text, b.source_data_json
      FROM businesses b
      LEFT JOIN LATERAL (
        SELECT 1 FROM business_interviews i
         WHERE i.business_id = b.id
           AND jsonb_typeof(i.qa_json) = 'object' AND i.qa_json <> '{}'::jsonb
         LIMIT 1
      ) hasInt ON true
     WHERE hasInt IS NULL
     ORDER BY b.ranking_score DESC NULLS LAST, b.id
     ${LIMIT ? `LIMIT ${LIMIT}` : ''}
  `);
  log(`${todo.length} businesses need a draft${DRY ? ' (DRY RUN)' : ''}`);
  if (DRY) {
    todo.slice(0, 20).forEach(b => log(`  • ${b.slug} (${b.neighborhood || '—'} · ${b.category || '—'})`));
    return;
  }

  // Discover alive hosts. If user forced a list, trust them.
  let hosts = FORCED_HOSTS || await aliveHosts();
  if (!hosts.length) {
    log(`✗ No Ollama host responsive. Tried: ${KNOWN_HOSTS.join(', ')}`);
    log(`  MS1 fix: ssh stevestudio@192.168.1.133 'pkill ollama && nohup ollama serve >/tmp/ollama.log 2>&1 &'`);
    process.exit(1);
  }
  log(`alive hosts (${hosts.length}): ${hosts.join(', ')}`);

  // Shared queue + counters. Workers race for items via shift().
  const queue = todo.slice();
  let ok = 0, fail = 0, fields = 0, msTotal = 0;
  const startAt = Date.now();
  const total = queue.length;

  async function worker(host) {
    while (queue.length) {
      const b = queue.shift();
      if (!b) break;
      const idx = total - queue.length;
      const head = `[${idx}/${total} · ${host.replace(/.*\/\//, '')}] ${b.slug}`;
      try {
        const t0 = Date.now();
        const result = await draftInterview(b, { host });
        const ms = Date.now() - t0;
        const fc = Object.keys(result.qa).length;
        if (!fc) throw new Error('LLM returned 0 fields');
        await query(
          `INSERT INTO business_interviews (business_id, qa_json, recorded_by) VALUES ($1, $2, $3)`,
          [b.id, JSON.stringify(result.qa), `AI Draft (${result.model})`]
        );
        ok += 1; fields += fc; msTotal += ms;
        log(`${head} ✓ ${fc} fields, ${ms}ms`);
      } catch (e) {
        fail += 1;
        const msg = String(e.message || e).slice(0, 200);
        log(`${head} ✗ ${msg}`);
        // If the host wedges mid-run (request times out / aborts), this worker
        // exits — the other worker keeps draining the queue.
        if (msg.includes('aborted') || msg.includes('No Ollama host')) {
          log(`  [${host}] worker exiting — host wedged`);
          return;
        }
      }
    }
  }

  await Promise.all(hosts.map(worker));

  const totalMs = Date.now() - startAt;
  log(`DONE · ok=${ok} fail=${fail} fields=${fields} avg=${ok ? Math.round(msTotal / ok) : 0}ms wall=${Math.round(totalMs / 1000)}s`);
  process.exit(fail > ok ? 1 : 0);
}

main().catch(e => { console.error(e); process.exit(2); });