← back to Lawyer Directory Builder

src/ingest/here_discover.ts

309 lines

/**
 * HERE Discover API enricher.
 *
 * For each existing law firm in the DB, query HERE's /v1/discover endpoint with
 * the firm name + city + lat/lng (when available) and category=legal-services,
 * then upsert phone / website / email rows from the returned record.
 *
 * Free tier: 30,000 transactions/month at platform.here.com (no card required).
 *  - 8,244 firms ≈ 27% of one month's quota
 *  - Polite cadence ~4 rps → ~35 min for the full pass
 *
 * Match policy: accept the top hit when its title fuzzy-matches our firm name.
 * "Fuzzy" = lowercased, suffix-stripped, then either substring containment or
 * Levenshtein distance ≤ 2 of normalized form.
 *
 * Env: HERE_API_KEY
 *
 * CLI:
 *   npx tsx src/ingest/here_discover.ts                # full pass
 *   npx tsx src/ingest/here_discover.ts --smoke        # first 5 firms only
 *   npx tsx src/ingest/here_discover.ts --limit 100    # cap at N firms
 *   npx tsx src/ingest/here_discover.ts --only-missing # skip firms that already have phone+website
 */
import 'dotenv/config';
import { pool, query, withTx } from '../db/pool.ts';

const SOURCE_NAME = 'HERE Discover API';
const DISCOVER_URL = 'https://discover.search.hereapi.com/v1/discover';

const API_KEY = process.env.HERE_API_KEY;
const RATE_DELAY_MS = Number(process.env.HERE_DELAY_MS || 250);
const USER_AGENT = process.env.USER_AGENT
  || 'LawyerDirectoryBuilder/0.1 (research; contact: steveabramsdesigns@gmail.com)';

type DiscoverItem = {
  title: string;
  id: string;
  position?: { lat: number; lng: number };
  address?: { label?: string; city?: string; postalCode?: string; state?: string; street?: string; houseNumber?: string };
  contacts?: Array<{ phone?: Array<{ value: string }>; www?: Array<{ value: string }>; email?: Array<{ value: string }> }>;
  categories?: Array<{ id: string; name: string; primary?: boolean }>;
  distance?: number;
};

type Firm = {
  id: number;
  name: string;
  address: string | null;
  city: string | null;
  zip: string | null;
  lat: number | null;
  lng: number | null;
  has_phone: boolean;
  has_website: boolean;
};

const argv = process.argv.slice(2);
const SMOKE = argv.includes('--smoke');
const ONLY_MISSING = argv.includes('--only-missing');
const LIMIT_IDX = argv.indexOf('--limit');
const LIMIT = LIMIT_IDX >= 0 ? Number(argv[LIMIT_IDX + 1]) : (SMOKE ? 5 : 0);

function clean(s: string | undefined | null) {
  if (!s) return null;
  const t = String(s).trim();
  return t.length === 0 ? null : t;
}

function normName(s: string): string {
  return s.toLowerCase()
    .replace(/[.,'"]/g, '')
    .replace(/\b(llp|llc|p\.?c\.?|pllc|pa|chartered|inc|ltd|esq|attorney at law|law office of|law offices of|law firm)\b/g, '')
    .replace(/&/g, 'and')
    .replace(/\s+/g, ' ')
    .trim();
}

// Quick distance-1 Levenshtein-ish containment test: accept if either string
// fully contains the other after normalization, or shares a 4+ word prefix.
function nameMatches(ours: string, theirs: string): boolean {
  const a = normName(ours);
  const b = normName(theirs);
  if (!a || !b) return false;
  if (a === b) return true;
  if (a.includes(b) || b.includes(a)) return true;
  const aw = a.split(' ').filter(Boolean);
  const bw = b.split(' ').filter(Boolean);
  if (aw.length >= 2 && bw.length >= 2) {
    const lead = Math.min(2, aw.length, bw.length);
    let shared = 0;
    for (let i = 0; i < lead; i++) if (aw[i] === bw[i]) shared++;
    if (shared >= 2) return true;
  }
  return false;
}

function normPhone(raw: string): string | null {
  const digits = raw.replace(/\D/g, '');
  if (digits.length === 11 && digits.startsWith('1')) {
    const d = digits.slice(1);
    return `(${d.slice(0,3)}) ${d.slice(3,6)}-${d.slice(6)}`;
  }
  if (digits.length === 10) {
    return `(${digits.slice(0,3)}) ${digits.slice(3,6)}-${digits.slice(6)}`;
  }
  return null;
}

async function ensureSource(): Promise<number> {
  await query(`
    INSERT INTO sources (source_name, source_type, base_url, terms_notes, allowed_method, rate_limit_rps)
    VALUES ($1, 'api', $2,
      'HERE Discover API. Free tier 30k transactions/month. ToS allows storage. Polite-use ~4 rps.',
      'api', 4.0)
    ON CONFLICT (source_name) DO NOTHING
  `, [SOURCE_NAME, DISCOVER_URL]);
  const r = await query<{ id: number }>(`SELECT id FROM sources WHERE source_name = $1`, [SOURCE_NAME]);
  return r.rows[0].id;
}

async function startJob(sourceId: number, label: string): Promise<number> {
  const r = await query<{ id: number }>(`
    INSERT INTO scrape_jobs (source_id, job_label, status, started_at)
    VALUES ($1, $2, 'running', NOW()) RETURNING id
  `, [sourceId, label]);
  return r.rows[0].id;
}

async function finishJob(jobId: number, fields: Record<string, unknown>) {
  const sets: string[] = [];
  const params: unknown[] = [];
  let i = 1;
  for (const [k, v] of Object.entries(fields)) { sets.push(`${k} = $${i++}`); params.push(v); }
  sets.push(`finished_at = NOW()`);
  params.push(jobId);
  await query(`UPDATE scrape_jobs SET ${sets.join(', ')} WHERE id = $${i}`, params);
}

async function loadFirms(): Promise<Firm[]> {
  const where = ONLY_MISSING
    ? `AND (
        o.id NOT IN (SELECT organization_id FROM phones)
        OR o.id NOT IN (SELECT organization_id FROM emails WHERE organization_id IS NOT NULL)
        OR o.website IS NULL
      )`
    : '';
  const lim = LIMIT > 0 ? `LIMIT ${LIMIT}` : '';
  const r = await query<Firm>(`
    SELECT
      o.id, o.name, o.address, o.city, o.zip, o.lat, o.lng,
      EXISTS(SELECT 1 FROM phones p WHERE p.organization_id = o.id) AS has_phone,
      o.website IS NOT NULL AS has_website
    FROM organizations o
    WHERE o.type = 'law_firm'
      AND o.address IS NOT NULL
      ${where}
    ORDER BY o.id
    ${lim}
  `);
  return r.rows;
}

async function fetchDiscover(firm: Firm): Promise<DiscoverItem | null> {
  const { fetch } = await import('undici');
  const params = new URLSearchParams({
    q: `${firm.name} ${firm.city || 'Los Angeles'}`,
    limit: '3',
    apiKey: API_KEY!,
  });
  if (firm.lat != null && firm.lng != null) {
    params.set('at', `${firm.lat},${firm.lng}`);
  } else {
    // LA County centroid as a fallback search bias.
    params.set('at', '34.0522,-118.2437');
  }
  // Restrict to legal-services category (HERE category id 600-6500).
  params.set('categoryFilter', '600-6500');
  const url = `${DISCOVER_URL}?${params.toString()}`;

  const r = await fetch(url, {
    headers: { 'User-Agent': USER_AGENT, Accept: 'application/json' },
    signal: AbortSignal.timeout(20000),
  });
  if (!r.ok) {
    if (r.status === 429) throw new Error('HERE_429_RATE_LIMIT');
    if (r.status === 401 || r.status === 403) throw new Error(`HERE_AUTH_${r.status}`);
    return null;
  }
  const j = await r.json() as { items?: DiscoverItem[] };
  if (!j.items || j.items.length === 0) return null;

  for (const item of j.items) {
    if (nameMatches(firm.name, item.title)) return item;
  }
  return null;
}

async function applyMatch(firm: Firm, item: DiscoverItem, sourceId: number): Promise<{ phones: number; emails: number; site: boolean }> {
  const sourceUrl = `https://wego.here.com/p/${item.id}`;
  const phones: string[] = [];
  const emails: string[] = [];
  let website: string | null = null;

  for (const c of item.contacts || []) {
    for (const p of c.phone || []) { const np = normPhone(p.value); if (np) phones.push(np); }
    for (const w of c.www || []) { if (!website) website = clean(w.value); }
    for (const e of c.email || []) { const ne = clean(e.value); if (ne) emails.push(ne.toLowerCase()); }
  }

  let phoneCount = 0, emailCount = 0;
  let siteSet = false;

  await withTx(async (client) => {
    if (website && !firm.has_website) {
      await client.query(
        `UPDATE organizations SET website = $1, updated_at = NOW() WHERE id = $2 AND website IS NULL`,
        [website, firm.id]);
      siteSet = true;
    }
    for (const p of phones) {
      const ins = await client.query(
        `INSERT INTO phones (organization_id, phone, phone_type, source_url, last_verified_at)
         VALUES ($1, $2, 'office', $3, NOW())
         ON CONFLICT DO NOTHING`,
        [firm.id, p, sourceUrl]);
      phoneCount += ins.rowCount || 0;
    }
    for (const e of emails) {
      const ins = await client.query(
        `INSERT INTO emails (organization_id, email, email_type, source_url, last_verified_at)
         VALUES ($1, $2, 'office', $3, NOW())
         ON CONFLICT DO NOTHING`,
        [firm.id, e, sourceUrl]);
      emailCount += ins.rowCount || 0;
    }
    // Provenance.
    await client.query(`
      INSERT INTO raw_records (source_id, source_url, entity_type, entity_id, raw_json, fetched_at, hash)
      VALUES ($1, $2, 'organization', $3, $4::jsonb, NOW(), encode(sha256($4::bytea), 'hex'))
      ON CONFLICT (source_id, hash) DO NOTHING
    `, [sourceId, sourceUrl, firm.id, JSON.stringify(item)]);
  });

  return { phones: phoneCount, emails: emailCount, site: siteSet };
}

async function main() {
  if (!API_KEY) {
    console.error('[here] HERE_API_KEY not set in .env — sign up at https://platform.here.com');
    process.exit(2);
  }

  const sourceId = await ensureSource();
  const label = SMOKE ? 'here:discover:smoke' : (LIMIT > 0 ? `here:discover:limit-${LIMIT}` : 'here:discover:la-firms');
  const jobId = await startJob(sourceId, label);

  const firms = await loadFirms();
  console.log(`[here] enriching ${firms.length} firms…`);

  let seen = 0, matched = 0, phonesNew = 0, emailsNew = 0, sitesNew = 0, errs = 0;
  try {
    for (const firm of firms) {
      seen++;
      try {
        const item = await fetchDiscover(firm);
        if (item) {
          matched++;
          const r = await applyMatch(firm, item, sourceId);
          phonesNew += r.phones;
          emailsNew += r.emails;
          if (r.site) sitesNew++;
        }
      } catch (e) {
        errs++;
        const msg = (e as Error).message;
        if (msg === 'HERE_429_RATE_LIMIT') {
          console.error(`[here] hit 429 — backing off 30s`);
          await new Promise(r => setTimeout(r, 30000));
        } else if (msg.startsWith('HERE_AUTH_')) {
          console.error(`[here] auth failure: ${msg} — aborting`);
          throw e;
        } else {
          console.error(`[here] firm=${firm.id} err: ${msg}`);
        }
      }
      if (seen % 100 === 0) console.log(`[here] ${seen}/${firms.length}  matched=${matched} phones+=${phonesNew} sites+=${sitesNew} emails+=${emailsNew}`);
      await new Promise(r => setTimeout(r, RATE_DELAY_MS));
    }
  } finally {
    await finishJob(jobId, {
      status: errs > firms.length / 2 ? 'failed' : 'completed',
      records_found: matched,
      records_inserted: phonesNew + emailsNew,
      records_updated: sitesNew,
      records_skipped: seen - matched,
      error_message: errs > 0 ? `${errs} per-firm errors` : null,
    });
  }

  console.log(`[here] done. seen=${seen} matched=${matched} phones+=${phonesNew} sites+=${sitesNew} emails+=${emailsNew} errs=${errs}`);
  await pool.end();
}

main().catch(async (err) => {
  console.error('[here] fatal:', err);
  try { await pool.end(); } catch {}
  process.exit(1);
});