`src/ingest/osm_overpass.ts`

213 lines
/**
 * OpenStreetMap Overpass ingest — Ventura Blvd corridor.
 *
 * Free, no API key. Hits the public Overpass API at overpass-api.de.
 * Be polite: one big query rather than 27 small ones.
 *
 * Strategy:
 *   1. Bbox the entire 17-mi corridor (~34.130..34.165 lat × -118.620..-118.370 lng).
 *   2. Pull every node + way that has an `addr:street` tag matching `Ventura B*`
 *      AND at least one business-y tag (shop, amenity, office, craft, tourism, healthcare, leisure).
 *   3. UPSERT into businesses(source='osm', source_id=node-/way-/relation-id).
 *
 * Coverage: OSM has ~40-60% of Ventura Blvd storefronts depending on which
 * stretch (Studio City and Sherman Oaks better-mapped than Calabasas).
 * We layer this with HERE/Foursquare later — UPSERT keys are (source, source_id)
 * so each provider has its own row; dedupe across providers is a separate pass.
 *
 * Run: npm run ingest:overpass
 */
import 'dotenv/config';
import { pool, query } from '../../db/pool.ts';

const OVERPASS_URL = 'https://overpass-api.de/api/interpreter';
const BBOX = {
  south: parseFloat(process.env.VENTURA_BBOX_SOUTH || '34.130'),
  west:  parseFloat(process.env.VENTURA_BBOX_WEST  || '-118.620'),
  north: parseFloat(process.env.VENTURA_BBOX_NORTH || '34.165'),
  east:  parseFloat(process.env.VENTURA_BBOX_EAST  || '-118.370'),
};

// Overpass QL — businesses with a Ventura Blvd address.
// We require BOTH: (a) some kind of business tag, (b) addr:street matches.
const QUERY = `
[out:json][timeout:90];
(
  node["addr:street"~"Ventura B(l)?(v)?d|Ventura Boulevard",i]
      [~"^(shop|amenity|office|craft|tourism|healthcare|leisure)$"~".",i]
      (${BBOX.south},${BBOX.west},${BBOX.north},${BBOX.east});
  way["addr:street"~"Ventura B(l)?(v)?d|Ventura Boulevard",i]
      [~"^(shop|amenity|office|craft|tourism|healthcare|leisure)$"~".",i]
      (${BBOX.south},${BBOX.west},${BBOX.north},${BBOX.east});
);
out tags center;
`.trim();

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

const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

function topCategory(tags: Record<string, string>): { name: string | null; raw: string | null } {
  // Order matters — pick the most specific business tag first.
  for (const k of ['shop', 'amenity', 'craft', 'office', 'healthcare', 'tourism', 'leisure']) {
    if (tags[k]) {
      const friendly = `${k}: ${tags[k].replace(/_/g, ' ')}`;
      return { name: friendly, raw: `${k}=${tags[k]}` };
    }
  }
  return { name: null, raw: null };
}

function isVenturaBlvd(s?: string): boolean {
  if (!s) return false;
  return /\bventura\s+(b(l)?(v)?d|boulevard)\b/i.test(s);
}

function normalizeStreet(s?: string): string | null {
  if (!s) return null;
  return s.toUpperCase().replace(/\s+/g, ' ').trim();
}

async function upsertElement(el: OsmElement): Promise<'new' | 'updated' | 'skipped'> {
  const t = el.tags || {};
  if (!t.name) return 'skipped';        // anonymous nodes are noise
  const cat = topCategory(t);
  if (!cat.name) return 'skipped';      // no business tag — shouldn't happen given our query

  const lat = el.lat ?? el.center?.lat ?? null;
  const lng = el.lon ?? el.center?.lon ?? null;
  const street = t['addr:street'] || null;
  const houseNum = t['addr:housenumber'] || null;
  const block = houseNum ? Math.floor(parseInt(houseNum, 10) / 100) * 100 || null : null;
  const onCorridor = isVenturaBlvd(street);

  const r = await query<{ id: number; first_seen_at: string }>(`
    INSERT INTO businesses (
      source, source_id, name, category, category_raw,
      address, street_number, street_name, city, zip, lat, lng,
      phone, website, on_corridor, corridor_block, raw, last_seen_at
    ) VALUES (
      'osm', $1, $2, $3, $4,
      $5, $6, $7, $8, $9, $10, $11,
      $12, $13, $14, $15, $16, NOW()
    )
    ON CONFLICT (source, source_id) DO UPDATE SET
      name           = EXCLUDED.name,
      category       = EXCLUDED.category,
      category_raw   = EXCLUDED.category_raw,
      address        = EXCLUDED.address,
      street_number  = EXCLUDED.street_number,
      street_name    = EXCLUDED.street_name,
      city           = COALESCE(EXCLUDED.city, businesses.city),
      zip            = COALESCE(EXCLUDED.zip, businesses.zip),
      lat            = EXCLUDED.lat,
      lng            = EXCLUDED.lng,
      phone          = COALESCE(EXCLUDED.phone, businesses.phone),
      website        = COALESCE(EXCLUDED.website, businesses.website),
      on_corridor    = EXCLUDED.on_corridor OR businesses.on_corridor,
      corridor_block = COALESCE(EXCLUDED.corridor_block, businesses.corridor_block),
      raw            = EXCLUDED.raw,
      last_seen_at   = NOW()
    RETURNING id, first_seen_at
  `, [
    `${el.type}/${el.id}`,
    t.name,
    cat.name,
    cat.raw,
    [houseNum, street, t['addr:city'], t['addr:postcode']].filter(Boolean).join(' '),
    houseNum,
    normalizeStreet(street),
    t['addr:city'] || null,
    t['addr:postcode'] || null,
    lat,
    lng,
    t['phone'] || t['contact:phone'] || null,
    t['website'] || t['contact:website'] || t['url'] || null,
    onCorridor,
    block,
    JSON.stringify(el),
  ]);
  return Date.parse(r.rows[0].first_seen_at) > Date.now() - 5_000 ? 'new' : 'updated';
}

async function main() {
  const run = await query<{ id: number }>(
    `INSERT INTO ingest_runs (source, notes) VALUES ('osm', $1) RETURNING id`,
    [`Overpass — bbox ${BBOX.south},${BBOX.west} → ${BBOX.north},${BBOX.east} · addr:street ~ Ventura B*`],
  );
  const runId = run.rows[0].id;
  console.log(`[osm] ingest_run #${runId} starting · single Overpass query`);
  console.log(`[osm] bbox ${BBOX.south},${BBOX.west} → ${BBOX.north},${BBOX.east}`);

  let inTotal = 0, newTotal = 0, updTotal = 0, skippedTotal = 0, errMessage: string | null = null;

  try {
    process.stdout.write('[osm] firing Overpass query (up to 90s timeout) ... ');
    const r = await fetch(OVERPASS_URL, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/x-www-form-urlencoded',
        'User-Agent': 'ventura-corridor/0.1 (local research; contact via project repo)',
      },
      body: 'data=' + encodeURIComponent(QUERY),
    });
    if (!r.ok) {
      const txt = await r.text().catch(() => '');
      throw new Error(`Overpass ${r.status} ${r.statusText}: ${txt.slice(0, 300)}`);
    }
    const j = await r.json() as { elements?: OsmElement[] };
    const elements = j.elements || [];
    console.log(`got ${elements.length} elements`);
    inTotal = elements.length;

    let i = 0;
    for (const el of elements) {
      i++;
      if (i % 25 === 0) process.stdout.write(`  · processed ${i}/${elements.length}\n`);
      const v = await upsertElement(el);
      if (v === 'new') newTotal++;
      else if (v === 'updated') updTotal++;
      else skippedTotal++;
    }
  } catch (e: any) {
    console.log(`ERROR: ${e.message}`);
    errMessage = e.message;
  }

  await query(
    `UPDATE ingest_runs SET finished_at=NOW(), rows_in=$1, rows_new=$2, rows_updated=$3, error_message=$4 WHERE id=$5`,
    [inTotal, newTotal, updTotal, errMessage, runId],
  );

  // Summary
  const corridor = await query<{ n: string }>(`SELECT COUNT(*)::text AS n FROM businesses WHERE on_corridor`);
  const withSite = await query<{ n: string }>(`SELECT COUNT(*)::text AS n FROM businesses WHERE on_corridor AND website IS NOT NULL`);
  const cities = await query<{ city: string; n: string }>(`
    SELECT COALESCE(city, '(none)') AS city, COUNT(*)::text AS n
    FROM businesses WHERE on_corridor
    GROUP BY city ORDER BY COUNT(*) DESC LIMIT 10
  `);

  console.log('');
  console.log(`[osm] done. fetched ${inTotal} · new ${newTotal} · updated ${updTotal} · skipped ${skippedTotal}`);
  console.log(`[osm] businesses on Ventura Blvd: ${corridor.rows[0].n}  (with website: ${withSite.rows[0].n})`);
  if (cities.rows.length) {
    console.log('[osm] by city:');
    for (const c of cities.rows) console.log(`        ${c.city.padEnd(20)} ${c.n}`);
  }

  await pool.end();
}

main().catch((e) => {
  console.error('[osm]', e);
  process.exit(1);
});