`src/ingest/osm_overpass.ts`

276 lines
/**
 * OSM Overpass importer — pulls law-firm OFFICES in LA County.
 * Free, no key, no captcha. Tag filters: office=lawyer + amenity=lawyer (legacy).
 *
 * The data is the ground truth for "where the firms physically are":
 * each result has lat/lng, addr fields, and the firm name as tagged by mappers.
 */
import 'dotenv/config';
import crypto from 'node:crypto';
import { fetchCompliant } from '../lib/compliance.ts';
import { pool, query, withTx } from '../db/pool.ts';

const OVERPASS = process.env.OVERPASS_URL || 'https://overpass-api.de/api/interpreter';
const SOURCE_NAME = 'OpenStreetMap Overpass API';

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

const QL = `
[out:json][timeout:90];
// LA County — wikidata Q104994
area["wikidata"="Q104994"]->.la;
(
  nwr["office"="lawyer"](area.la);
  nwr["amenity"="lawyer"](area.la);
);
out center tags;
`.trim();

function sha256(s: string) {
  return crypto.createHash('sha256').update(s).digest('hex');
}

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

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']);
  // Only build a "full" address if we actually have a street line OR a city.
  // Otherwise the field would be a useless "CA, 90210" or just "CA".
  const full = (line1 || city)
    ? [line1, city, state, zip].filter(Boolean).join(', ')
    : null;
  return { line1, city, state, zip, full };
}

function normAddress(s: string | null) {
  if (!s) return null;
  return s.toLowerCase()
    .replace(/[.,]/g, ' ')
    .replace(/\b(suite|ste|unit|apt|#)\b/g, '')
    .replace(/\s+/g, ' ')
    .trim();
}

function neighborhoodOf(t: Record<string, string>) {
  // Use addr:suburb or addr:city for neighborhood. Beverly Hills, Century City, etc.
  return clean(t['addr:suburb']) || clean(t['addr:district']) || clean(t['addr:city']);
}

function firmSizeBand(name: string | null): string | null {
  if (!name) return null;
  const n = name.toLowerCase();
  // Heuristic — rough; refined later by attorney_count.
  if (/\b(llp|llc|p\.c\.|pc|pllc|chartered)\b/.test(n)) return 'medium';
  if (/&|and associates|group/.test(n)) return 'small';
  return 'solo';
}

async function getSourceId(): Promise<number> {
  const r = await query<{ id: number }>('SELECT id FROM sources WHERE source_name = $1', [SOURCE_NAME]);
  if (r.rowCount === 0) throw new Error(`Source not seeded: ${SOURCE_NAME}`);
  return r.rows[0].id;
}

async function startJob(sourceId: number, label: string) {
  // The lawyer DB schema uses scrape_jobs without job_label column? Let me match the actual schema.
  // It does have job_label per migration 001.
  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<OverpassElement[]> {
  // Overpass API is an explicitly programmatic endpoint per the OSM usage policy
  // (https://operations.osmfoundation.org/policies/api/). Their robots.txt blocks
  // generic crawlers but the API is intended for client consumption with a meaningful
  // User-Agent. We respect their RATE policy (max ~10k req/day, polite-use) instead.
  const { fetch } = await import('undici');
  const body = `data=${encodeURIComponent(QL)}`;
  const r = await fetch(OVERPASS, {
    method: 'POST',
    headers: {
      'User-Agent': process.env.USER_AGENT || 'LawyerDirectoryBuilder/0.1 (research; contact: steveabramsdesigns@gmail.com)',
      'Content-Type': 'application/x-www-form-urlencoded',
      Accept: 'application/json',
    },
    body,
    signal: AbortSignal.timeout(180000),
  });
  if (!r.ok) {
    const txt = await r.text();
    throw new Error(`Overpass ${r.status}: ${txt.slice(0, 400)}`);
  }
  const json = await r.json() as { elements: OverpassElement[] };
  return json.elements;
}

async function upsert(el: OverpassElement, sourceId: number) {
  const tags = el.tags || {};
  const name = clean(tags.name);
  if (!name) return null;

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

  return await withTx(async (client) => {
    // Upsert organization, dedup by (osm_type, osm_id) via raw_records or by name+address_norm.
    const addressNorm = normAddress(addr.full);

    // Try to find existing by website (most stable) or name+address_norm
    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 && r.rowCount > 0) 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 && r.rowCount > 0) orgId = r.rows[0].id;
    }

    if (orgId) {
      await client.query(`
        UPDATE organizations
        SET name = $2,
            address = COALESCE($3, address),
            city = COALESCE($4, city),
            state = COALESCE($5, state),
            zip = COALESCE($6, zip),
            county = COALESCE($7, county),
            neighborhood = COALESCE($8, neighborhood),
            lat = COALESCE($9, lat),
            lng = COALESCE($10, lng),
            geocoded_at = COALESCE(geocoded_at, NOW()),
            phone = COALESCE($11, phone),
            website = COALESCE($12, website),
            address_norm = COALESCE($13, address_norm),
            firm_size_band = COALESCE(firm_size_band, $14),
            source_url = COALESCE(source_url, $15),
            updated_at = NOW()
        WHERE id = $1
      `, [orgId, name, addr.full, addr.city, addr.state, addr.zip, 'Los Angeles',
          neighborhood, lat, lng, phone, website, addressNorm,
          firmSizeBand(name), sourceUrl]);
    } else {
      const r = await client.query<{ id: number }>(`
        INSERT INTO organizations (
          name, type, address, city, state, zip, county, neighborhood,
          lat, lng, geocoded_at, phone, website, address_norm,
          firm_size_band, source_url
        ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,NOW(),$11,$12,$13,$14,$15)
        RETURNING id
      `, [name, 'law_firm', addr.full, addr.city, addr.state, addr.zip, 'Los Angeles',
          neighborhood, lat, lng, phone, website, addressNorm,
          firmSizeBand(name), sourceUrl]);
      orgId = r.rows[0].id;
    }

    if (email) {
      await client.query(`
        INSERT INTO emails (organization_id, email, email_type, source_url, last_verified_at)
        VALUES ($1,$2,'office',$3,NOW())
        `, [orgId, email, sourceUrl]);
    }
    if (phone) {
      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
      `, [orgId, phone, sourceUrl]);
    }

    // Stash raw element for provenance.
    const rawJson = JSON.stringify({ ...el, tags });
    const hash = sha256(rawJson + '|organization|' + orgId);
    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] querying Overpass for office=lawyer in LA County (Q104994)…');
  const sourceId = await getSourceId();
  const jobId = await startJob(sourceId, 'osm:overpass:la-county');

  let elements: OverpassElement[] = [];
  try {
    elements = await fetchOverpass();
  } catch (e) {
    await finishJob(jobId, { status: 'failed', error_message: (e as Error).message });
    throw e;
  }

  console.log(`[osm] 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] upsert 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] done. seen=${elements.length} kept=${inserted} skipped=${skipped}`);
  await pool.end();
}

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