`src/ingest/places/seed.ts`

213 lines
/**
 * M-P1 — Google Places DISCOVERY seed (feeds the free firm pipeline).
 *
 * Places is used as a SEED/pointer, not a stored dataset:
 *   pick stalest metro → official Places API (v1 searchText) → for each hit,
 *   persist ONLY place_id (the one cacheable field) + create/link a `firm` row
 *   (source='google_places', source_id=place_id) carrying its website → the
 *   existing discover:firms / crawl:firms jobs then enrich firm_site from the
 *   firm's OWN public site (first-party, storable). Idempotent: place_id dedupe.
 *
 * COST GUARD — this is the only metered source in the app. A monthly hard cap
 * (places_quota) STOPS all calls before we cross the Google free-tier line, so
 * steady-state spend is $0. Every call is logged to the cost-tracker ledger.
 * Runs DRY by default; set PLACES_SEED_LIVE=1 to make real calls.
 *
 *   npm run ingest:places                 # stalest metro, dry-run unless PLACES_SEED_LIVE=1
 *   npm run ingest:places -- --metro=31080 --live
 *
 * Env: GOOGLE_PLACES_API_KEY (required for live), PLACES_MONTHLY_CAP=1500,
 *      PLACES_SEED_LIVE=0, PLACES_QUERIES=<comma templates>.
 */
import 'dotenv/config';
import { appendFileSync, mkdirSync } from 'node:fs';
import { join } from 'node:path';
import { homedir } from 'node:os';
import { pool, query } from '../../../db/pool.ts';
import { openRun, closeRun } from '../run.ts';

const SOURCE = 'google_places';
const API = 'https://places.googleapis.com/v1/places:searchText';
const KEY = process.env.GOOGLE_PLACES_API_KEY || '';
const LIVE = process.env.PLACES_SEED_LIVE === '1' || process.argv.includes('--live');
const CAP = Number(process.env.PLACES_MONTHLY_CAP || 1500);
const SEED_TARGET = Number(process.env.PLACES_SEED_TARGET || 60); // seeds/metro before it's "covered" and we move on
// Text-Search (Pro SKU) list price if we ever exceeded the free tier — for the ledger only.
const RATE_PER_CALL = 0.032;

const QUERY_TEMPLATES = (process.env.PLACES_QUERIES ||
  'real estate agency in {m};commercial real estate brokerage in {m};apartment leasing office in {m}')
  .split(';').map(s => s.trim()).filter(Boolean);

const FIELD_MASK = 'places.id,places.displayName,places.websiteUri,places.formattedAddress,places.location';

function ym(): string {
  // Date.now/new Date() are fine at runtime here (not a workflow script).
  const d = new Date();
  return `${d.getUTCFullYear()}-${String(d.getUTCMonth() + 1).padStart(2, '0')}`;
}

function logCost(calls: number, note: string): void {
  try {
    const dir = join(homedir(), '.claude');
    mkdirSync(dir, { recursive: true });
    const entry = {
      ts: new Date().toISOString(), skill: 'usre-places-seed', provider: 'google_places',
      units: calls, unit: 'searchText_call', rate: RATE_PER_CALL,
      cost: 0, list_cost_if_billed: +(calls * RATE_PER_CALL).toFixed(4),
      note: `${note} (free-tier, capped ${CAP}/mo)`,
    };
    appendFileSync(join(dir, 'cost-ledger.jsonl'), JSON.stringify(entry) + '\n');
  } catch { /* ledger is best-effort */ }
}

async function callsUsed(): Promise<number> {
  const r = await query<{ calls_used: number }>(
    `SELECT calls_used FROM places_quota WHERE year_month = $1`, [ym()]);
  return r.rows[0]?.calls_used ?? 0;
}
async function bumpQuota(n: number): Promise<void> {
  await query(
    `INSERT INTO places_quota (year_month, calls_used) VALUES ($1, $2)
     ON CONFLICT (year_month) DO UPDATE
       SET calls_used = places_quota.calls_used + $2, updated_at = NOW()`,
    [ym(), n]);
}

/** Stalest metro: fewest seeds first, then least-recently seeded. */
async function pickMetro(): Promise<{ id: number; name: string; key: string } | null> {
  const argIdx = process.argv.findIndex(a => a.startsWith('--metro='));
  if (argIdx > -1) {
    const cbsa = process.argv[argIdx].split('=')[1];
    const r = await query<{ id: number; name: string; canonical_key: string }>(
      `SELECT id, name, canonical_key FROM region WHERE region_type='metro' AND cbsa_code=$1 LIMIT 1`, [cbsa]);
    if (r.rows.length) return { id: r.rows[0].id, name: r.rows[0].name, key: r.rows[0].canonical_key };
  }
  // Geographic sweep: state is embedded in the metro name ("…, CA Metro Area",
  // "…, OR-WA Metro Area") since region.state_code is null. Tier west→east and
  // seed UNDER-COVERED metros (seeds < SEED_TARGET) lowest-tier-first; once every
  // West Coast metro hits the target the sweep advances east on its own. When all
  // are covered it falls back to re-seeding the stalest (keeps data fresh).
  const r = await query<{ id: number; name: string; canonical_key: string }>(
    `WITH m AS (
       SELECT r.id, r.name, r.canonical_key,
         COALESCE((SELECT COUNT(*) FROM places_seed s WHERE s.region_id = r.id),0) AS seeds,
         (SELECT MAX(last_seen) FROM places_seed s WHERE s.region_id = r.id) AS ls,
         substring(r.name from ',\\s+([A-Z]{2}(-[A-Z]{2})*)\\s') AS st
       FROM region r WHERE r.region_type = 'metro')
     SELECT id, name, canonical_key FROM m
      ORDER BY
        CASE WHEN seeds < $1 THEN
          CASE WHEN st ~ '(^|-)(CA|OR|WA)($|-)'    THEN 0   -- West Coast
               WHEN st ~ '(^|-)(NV|AZ|ID|UT)($|-)' THEN 1   -- Interior West
               WHEN st ~ '(^|-)(MT|WY|CO|NM)($|-)' THEN 2   -- Mountain
               ELSE 5 END                                    -- rest of US
        ELSE 999 END ASC,
        CASE WHEN r.name ILIKE '%Micro Area%' THEN 1 ELSE 0 END ASC,  -- big Metro Areas before small Micro Areas
        seeds ASC, ls ASC NULLS FIRST, name ASC
      LIMIT 1`, [SEED_TARGET]);
  return r.rows.length ? { id: r.rows[0].id, name: r.rows[0].name, key: r.rows[0].canonical_key } : null;
}

/** "Asheville, NC Metro Area" -> "Asheville, NC" for natural Places queries. */
function placeLabel(name: string): string {
  return name.replace(/\s+(Metro(politan)?( Statistical)?( Area)?|Micro(politan)?( Area)?)\s*$/i, '').trim();
}

function hostOf(url: string): string | null {
  try { return new URL(url).host.replace(/^www\./, '').toLowerCase(); } catch { return null; }
}

async function searchText(q: string): Promise<Array<{ id: string; name: string; website?: string; addr?: string }>> {
  const res = await fetch(API, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json', 'X-Goog-Api-Key': KEY, 'X-Goog-FieldMask': FIELD_MASK },
    body: JSON.stringify({ textQuery: q, maxResultCount: 20 }),
  });
  if (!res.ok) throw new Error(`places searchText ${res.status}: ${(await res.text()).slice(0, 200)}`);
  const j: any = await res.json();
  return (j.places || []).map((p: any) => ({
    id: p.id, name: p.displayName?.text || p.id, website: p.websiteUri, addr: p.formattedAddress,
  }));
}

/** Persist place_id + create/link a firm (source-of-record data comes later from the crawl). */
async function linkSeed(placeId: string, q: string, regionId: number, name: string, website?: string): Promise<'new' | 'seen'> {
  const existed = await query(`SELECT 1 FROM places_seed WHERE place_id=$1`, [placeId]);
  const host = website ? hostOf(website) : null;
  if (existed.rows.length) {
    await query(`UPDATE places_seed SET last_seen=NOW() WHERE place_id=$1`, [placeId]);
    return 'seen';
  }
  let firmId: number | null = null;
  if (website) {
    const f = await query<{ id: number }>(
      `INSERT INTO firm (name, normalized_name, website, source, source_id, region_id)
       VALUES ($1,$2,$3,$4,$5,$6)
       ON CONFLICT (source, source_id) DO UPDATE SET website = COALESCE(EXCLUDED.website, firm.website)
       RETURNING id`,
      [name, name.toLowerCase().replace(/[^a-z0-9]+/g, ' ').trim(), website, SOURCE, placeId, regionId]);
    firmId = f.rows[0].id;
    // Seed firm_site directly so crawl:firms enriches from the firm's OWN site
    // (first-party, storable) — skips the redundant discover:firms web search.
    // crawl_status left NULL: that IS the crawler's work-queue signal.
    await query(
      `INSERT INTO firm_site (firm_id, url, discovery_method)
       VALUES ($1,$2,'google_places')
       ON CONFLICT (firm_id) DO NOTHING`,
      [firmId, website]);
  }
  await query(
    `INSERT INTO places_seed (place_id, query, region_id, website_host, firm_id, status)
     VALUES ($1,$2,$3,$4,$5,$6)`,
    [placeId, q, regionId, host, firmId, website ? 'linked' : 'no_website']);
  return 'new';
}

async function main() {
  const metro = await pickMetro();
  if (!metro) { console.log('[places] no metro regions seeded yet — run seed:regions first'); await pool.end(); return; }

  const used = await callsUsed();
  const budget = Math.max(0, CAP - used);
  console.log(`[places] metro=${metro.name} (${metro.key})  quota ${used}/${CAP} used, ${budget} left, live=${LIVE}`);

  if (!LIVE) {
    console.log('[places] DRY-RUN (set PLACES_SEED_LIVE=1 to make real calls). Would query:');
    for (const t of QUERY_TEMPLATES) console.log(`   • "${t.replace('{m}', placeLabel(metro.name))}"`);
    console.log(`[places] est cost: $0 (free-tier; ${QUERY_TEMPLATES.length} calls, would be $${(QUERY_TEMPLATES.length * RATE_PER_CALL).toFixed(3)} if billed)`);
    await pool.end();
    return;
  }
  if (!KEY) { console.error('[places] PLACES_SEED_LIVE=1 but GOOGLE_PLACES_API_KEY unset'); process.exit(2); }
  if (budget <= 0) { console.log(`[places] monthly cap ${CAP} reached — HARD STOP, $0 spent`); await pool.end(); return; }

  const runId = await openRun(SOURCE, `${metro.key}`);
  let calls = 0, fresh = 0, seen = 0, linked = 0;
  try {
    for (const t of QUERY_TEMPLATES) {
      if (calls >= budget) { console.log('[places] cap reached mid-run — stopping'); break; }
      const q = t.replace('{m}', placeLabel(metro.name));
      const hits = await searchText(q);
      calls++; await bumpQuota(1); logCost(1, `searchText "${q}"`);
      for (const h of hits) {
        const r = await linkSeed(h.id, q, metro.id, h.name, h.website);
        if (r === 'new') { fresh++; if (h.website) linked++; } else seen++;
      }
      console.log(`[places]   "${q}" -> ${hits.length} hits`);
      await new Promise(r => setTimeout(r, 400)); // polite gap
    }
    await closeRun(runId, 'ok', { upserted: fresh, skipped: seen,
      notes: `${calls} calls, ${fresh} new (${linked} w/ website), ${seen} seen; metro=${metro.name}` });
    console.log(`[places] done: ${calls} calls, ${fresh} new firms (${linked} w/ site), ${seen} already seen. $0 (free-tier).`);
  } catch (e: any) {
    await closeRun(runId, 'failed', { notes: e.message });
    console.error('[places] FAILED:', e.message);
    await pool.end();
    process.exit(1);
  }
  await pool.end();
}

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