`src/ingest/osm_name_search.ts`

228 lines
/**
 * OSM Overpass — broader pass: any node/way in LA County with a name suggesting
 * legal services, even if NOT tagged office=lawyer. Catches firms mappers tagged
 * generically (office, shop, building) plus mis-tagged ones.
 *
 * Filters out obvious false positives by requiring at least one of:
 *   • a "law"/"attorney"/"legal"/"counsel" word in the name,
 *   • AND not tagged as a clearly non-firm thing (school, restaurant, etc.).
 */
import 'dotenv/config';
import crypto from 'node:crypto';
import { fetch } from 'undici';
import { pool, query, withTx } from '../db/pool.ts';

const SOURCE_NAME = 'OSM Overpass — name match';
const OVERPASS = 'https://overpass-api.de/api/interpreter';
const USER_AGENT = process.env.USER_AGENT || 'LawyerDirectoryBuilder/0.1 (research; contact: steveabramsdesigns@gmail.com)';

// Match firm-name vocabulary; exclude schools/restaurants/courts/etc.
const QL = `
[out:json][timeout:120];
area["wikidata"="Q104994"]->.la;
(
  nwr["name"~"\\\\b(law offices?|law firm|law group|attorneys?( at law)?|counsel(o?r)?|legal( services| aid)?|esquire|, esq)\\\\b",i](area.la);
);
(._; - nwr["amenity"~"^(school|restaurant|cafe|bar|pub|fast_food|college|university|library|courthouse|police|hospital|clinic)$"];);
out center tags 600;
`.trim();

const EXCLUDE_AMENITY = new Set([
  'school', 'restaurant', 'cafe', 'bar', 'pub', 'fast_food', 'college', 'university',
  'library', 'courthouse', 'police', 'hospital', 'clinic', 'pharmacy', 'fuel', 'bank'
]);

interface Element {
  type: 'node' | 'way' | 'relation';
  id: number;
  lat?: number;
  lon?: number;
  center?: { lat: number; lon: number };
  tags?: Record<string, string>;
}

const clean = (s: string | undefined | null) => s ? String(s).trim() || null : null;
const normAddress = (s: string | null) => s
  ? s.toLowerCase().replace(/[.,]/g, ' ').replace(/\b(suite|ste|unit|apt|#)\b/g, '').replace(/\s+/g, ' ').trim()
  : null;

function buildAddress(t: Record<string, string>) {
  const street = [t['addr:housenumber'], t['addr:street']].filter(Boolean).join(' ');
  const unit = t['addr:unit'] ? ` Suite ${t['addr:unit']}` : '';
  const line1 = clean(street + unit);
  const city = clean(t['addr:city']);
  const state = clean(t['addr:state']) || 'CA';
  const zip = clean(t['addr:postcode']);
  const full = (line1 || city) ? [line1, city, state, zip].filter(Boolean).join(', ') : null;
  return { line1, city, state, zip, full };
}

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, $3, 'api', 0.10)
    ON CONFLICT (source_name) DO NOTHING
  `, [SOURCE_NAME, OVERPASS, 'OSM ODbL. Broader name-based query for legal-services entities.']);
  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) {
  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 fetchOverpass(): Promise<Element[]> {
  const r = await fetch(OVERPASS, {
    method: 'POST',
    headers: {
      'User-Agent': USER_AGENT,
      'Content-Type': 'application/x-www-form-urlencoded',
      Accept: 'application/json',
    },
    body: `data=${encodeURIComponent(QL)}`,
    signal: AbortSignal.timeout(180000),
  });
  if (!r.ok) throw new Error(`Overpass ${r.status}: ${(await r.text()).slice(0, 300)}`);
  const j = await r.json() as { elements: Element[] };
  return j.elements;
}

async function upsert(el: Element, sourceId: number) {
  const t = el.tags || {};
  const name = clean(t.name);
  if (!name) return null;
  if (t.amenity && EXCLUDE_AMENITY.has(t.amenity)) return null;

  const lat = el.lat ?? el.center?.lat ?? null;
  const lng = el.lon ?? el.center?.lon ?? null;
  const addr = buildAddress(t);
  const phone = clean(t.phone) || clean(t['contact:phone']);
  const website = clean(t.website) || clean(t['contact:website']);
  const sourceUrl = `https://www.openstreetmap.org/${el.type}/${el.id}`;

  return await withTx(async (client) => {
    const addressNorm = normAddress(addr.full);
    let orgId: number | null = null;
    if (website) {
      const r = await client.query<{ id: number }>(
        `SELECT id FROM organizations WHERE LOWER(website) = LOWER($1) LIMIT 1`, [website]);
      if (r.rowCount) orgId = r.rows[0].id;
    }
    if (!orgId && addressNorm) {
      const r = await client.query<{ id: number }>(
        `SELECT id FROM organizations WHERE address_norm = $1 AND LOWER(name) = LOWER($2) LIMIT 1`,
        [addressNorm, name]);
      if (r.rowCount) orgId = r.rows[0].id;
    }
    if (!orgId && addr.city) {
      const r = await client.query<{ id: number }>(
        `SELECT id FROM organizations WHERE LOWER(name) = LOWER($1) AND LOWER(city) = LOWER($2) LIMIT 1`,
        [name, addr.city]);
      if (r.rowCount) orgId = r.rows[0].id;
    }

    if (orgId) {
      // Enrich missing fields. OSM has phone/website that LA City lacks.
      await client.query(`
        UPDATE organizations
        SET address = COALESCE(address, $2),
            address_norm = COALESCE(address_norm, $3),
            city = COALESCE(city, $4),
            neighborhood = COALESCE(neighborhood, $4),
            state = COALESCE(state, $5),
            zip = COALESCE(zip, $6),
            lat = COALESCE(lat, $7::double precision),
            lng = COALESCE(lng, $8::double precision),
            geocoded_at = COALESCE(geocoded_at, CASE WHEN $7::double precision IS NOT NULL THEN NOW() END),
            phone = COALESCE(phone, $9),
            website = COALESCE(website, $10),
            source_url = COALESCE(source_url, $11),
            updated_at = NOW()
        WHERE id = $1
      `, [orgId, addr.full, addressNorm, addr.city, addr.state, addr.zip, lat, lng, phone, website, sourceUrl]);
    } else {
      const r = await client.query<{ id: number }>(`
        INSERT INTO organizations (
          name, type, address, address_norm, city, neighborhood, state, county, zip,
          lat, lng, geocoded_at, phone, website, source_url
        ) VALUES (
          $1,'law_firm',$2,$3,$4,$4,$5,'Los Angeles',$6,
          $7::double precision, $8::double precision,
          CASE WHEN $7::double precision IS NOT NULL THEN NOW() END,
          $9, $10, $11
        ) RETURNING id
      `, [name, addr.full, addressNorm, addr.city, addr.state, addr.zip, lat, lng, phone, website, sourceUrl]);
      orgId = r.rows[0].id;
    }

    // Phones / emails as separate records (idempotent check)
    if (phone) {
      const exists = await client.query(`SELECT 1 FROM phones WHERE organization_id=$1 AND phone=$2 LIMIT 1`, [orgId, phone]);
      if (exists.rowCount === 0) {
        await client.query(`INSERT INTO phones (organization_id, phone, phone_type, source_url, last_verified_at)
                            VALUES ($1,$2,'office',$3,NOW())`, [orgId, phone, sourceUrl]);
      }
    }

    const rawJson = JSON.stringify({ ...el, tags: t });
    const hash = crypto.createHash('sha256').update(rawJson + '|osm-name|' + el.id + '|' + orgId).digest('hex');
    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(),$5)
      ON CONFLICT (source_id, hash) DO NOTHING
    `, [sourceId, sourceUrl, orgId, rawJson, hash]);

    return orgId;
  });
}

async function main() {
  console.log('[osm-name] querying broader name-based law-firm OSM…');
  const sourceId = await ensureSource();
  const jobId = await startJob(sourceId, 'osm:name-match:la-county');

  let elements: Element[] = [];
  try {
    elements = await fetchOverpass();
  } catch (e) {
    await finishJob(jobId, { status: 'failed', error_message: (e as Error).message });
    throw e;
  }
  console.log(`[osm-name] received ${elements.length} elements`);

  let inserted = 0, skipped = 0;
  for (const el of elements) {
    try {
      const id = await upsert(el, sourceId);
      if (id) inserted++; else skipped++;
    } catch (e) {
      console.error(`[osm-name] err ${el.type}/${el.id}: ${(e as Error).message}`);
      skipped++;
    }
  }

  await finishJob(jobId, { status: 'completed', records_found: elements.length, records_inserted: inserted, records_skipped: skipped });
  console.log(`[osm-name] done. seen=${elements.length} kept=${inserted} skipped=${skipped}`);
  await pool.end();
}

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