`src/ingest/zillow_research.ts`

161 lines
/**
 * Zillow Research ingest — ZHVI + ZORI, county + metro, last 36 monthly columns.
 * Wide CSV: one row per region, one column per month (month-end headers,
 * normalized to first-of-month on store). County crosswalk = StateCodeFIPS+MunicipalCodeFIPS
 * -> region.fips; metro crosswalk = overrides -> exact CBSA-name -> primary-city prefix.
 * Unmatched metros are recorded + skipped, never guessed.
 * Run with --cached to reuse data/downloads copies during dev.
 */
import { openRun, closeRun, download } from './run.ts';
import {
  loadRegionMaps, loadOverrides, buildMetroMatcher, UnmatchedCollector,
  upsertMetricRows, upsertSourceRegionMap, firstOfMonth,
  type MetricRow, type RegionMaps,
} from './crosswalk.ts';
import { readFileSync } from 'node:fs';
import { pool } from '../../db/pool.ts';

const BASE = 'https://files.zillowstatic.com/research/public_csvs';
const FILES = [
  { url: `${BASE}/zhvi/County_zhvi_uc_sfrcondo_tier_0.33_0.67_sm_sa_month.csv`, file: 'zillow_county_zhvi.csv', metric: 'zhvi', level: 'county' as const },
  { url: `${BASE}/zhvi/Metro_zhvi_uc_sfrcondo_tier_0.33_0.67_sm_sa_month.csv`, file: 'zillow_metro_zhvi.csv', metric: 'zhvi', level: 'metro' as const },
  { url: `${BASE}/zori/County_zori_uc_sfrcondomfr_sm_sa_month.csv`, file: 'zillow_county_zori.csv', metric: 'zori', level: 'county' as const },
  { url: `${BASE}/zori/Metro_zori_uc_sfrcondomfr_sm_sa_month.csv`, file: 'zillow_metro_zori.csv', metric: 'zori', level: 'metro' as const },
];
const MONTHS = 36;

function parseCsvLine(line: string): string[] {
  const out: string[] = [];
  let field = '';
  let inQ = false;
  for (let i = 0; i < line.length; i++) {
    const c = line[i];
    if (inQ) {
      if (c === '"') {
        if (line[i + 1] === '"') { field += '"'; i++; }
        else inQ = false;
      } else field += c;
    } else if (c === '"') inQ = true;
    else if (c === ',') { out.push(field); field = ''; }
    else field += c;
  }
  out.push(field);
  return out;
}

interface FileResult { rows: MetricRow[]; maps: { source: string; sourceRegionId: string; regionId: number; matchMethod: string }[]; parsed: number; skipped: number; }

function processFile(
  path: string, metric: string, level: 'county' | 'metro',
  maps: RegionMaps, matchMetro: ReturnType<typeof buildMetroMatcher>,
  overrides: Record<string, string>, unmatched: UnmatchedCollector,
): FileResult {
  const text = readFileSync(path, 'utf8');
  const lines = text.split('\n');
  const header = parseCsvLine(lines[0].replace(/\r$/, ''));
  const col = (name: string) => header.indexOf(name);
  const dateCols: { idx: number; period: string }[] = [];
  header.forEach((h, i) => { if (/^\d{4}-\d{2}-\d{2}$/.test(h)) dateCols.push({ idx: i, period: firstOfMonth(h) }); });
  const useCols = dateCols.slice(-MONTHS);
  if (useCols.length === 0) throw new Error(`no date columns found in ${path}`);

  const iRegionId = col('RegionID');
  const iRegionName = col('RegionName');
  const iStateFips = col('StateCodeFIPS');
  const iMuniFips = col('MunicipalCodeFIPS');

  const res: FileResult = { rows: [], maps: [], parsed: 0, skipped: 0 };
  for (let li = 1; li < lines.length; li++) {
    const line = lines[li].replace(/\r$/, '');
    if (!line) continue;
    const f = parseCsvLine(line);
    const regionName = f[iRegionName];
    let regionId: number | null = null;
    let method = '';
    if (level === 'county') {
      const fips = f[iStateFips].padStart(2, '0') + f[iMuniFips].padStart(3, '0');
      const region = maps.byFips.get(fips);
      if (region) { regionId = region.id; method = 'fips'; }
    } else {
      const ovr = overrides[regionName];
      if (ovr) {
        const region = maps.byCanonicalKey.get(ovr);
        if (region) { regionId = region.id; method = 'override'; }
      }
      if (regionId === null && regionName !== 'United States') {
        const m = matchMetro(regionName);
        if (m) { regionId = m.region.id; method = m.method; }
      }
    }
    if (regionId === null) {
      res.skipped++;
      if (level === 'metro' && regionName !== 'United States') {
        unmatched.add('zillow_metro', { regionId: f[iRegionId], regionName });
      } else if (level === 'county') {
        unmatched.add('zillow_county', { regionId: f[iRegionId], regionName, fips: f[iStateFips] + '/' + f[iMuniFips] });
      }
      continue;
    }
    res.maps.push({ source: 'zillow', sourceRegionId: f[iRegionId], regionId, matchMethod: method });
    for (const { idx, period } of useCols) {
      const v = f[idx];
      if (v === '' || v === undefined) continue;
      const num = Number(v);
      if (!Number.isFinite(num)) continue;
      res.rows.push({ regionId, source: 'zillow', metric, period, value: num });
    }
    res.parsed++;
  }
  return res;
}

async function main() {
  const cached = process.argv.includes('--cached');
  const runId = await openRun('zillow', BASE);
  try {
    const maps = await loadRegionMaps();
    const matchMetro = buildMetroMatcher(maps);
    const overrides = loadOverrides('zillow');
    const unmatched = new UnmatchedCollector();

    let totalUpserted = 0;
    let totalSkipped = 0;
    let totalParsed = 0;
    const hashes: string[] = [];
    for (const spec of FILES) {
      console.log(`[zillow] downloading ${spec.file}${cached ? ' (cached ok)' : ''}`);
      const { path, sha256 } = await download(spec.url, spec.file, { reuseCached: cached });
      hashes.push(`${spec.file}:${sha256.slice(0, 12)}`);
      const r = processFile(path, spec.metric, spec.level, maps, matchMetro, overrides, unmatched);
      console.log(`[zillow] ${spec.file}: ${r.parsed} regions matched, ${r.skipped} skipped, ${r.rows.length} metric rows`);
      await upsertSourceRegionMap(r.maps);
      totalUpserted += await upsertMetricRows(r.rows, runId);
      totalSkipped += r.skipped;
      totalParsed += r.parsed;
    }
    unmatched.flush();
    const unmatchedMetro = unmatched.count('zillow_metro');
    const unmatchedCounty = unmatched.count('zillow_county');
    console.log(`[zillow] TOTAL upserted=${totalUpserted} regions=${totalParsed} unmatched: county=${unmatchedCounty} metro=${unmatchedMetro}`);

    if (totalUpserted === 0) {
      await closeRun(runId, 'failed', { notes: '0 rows parsed — format drift?' });
      process.exitCode = 1;
    } else {
      await closeRun(runId, 'ok', {
        upserted: totalUpserted, skipped: totalSkipped,
        notes: `regions=${totalParsed} unmatched_county=${unmatchedCounty} unmatched_metro=${unmatchedMetro}`,
        fileHash: hashes.join(' '),
      });
    }
  } catch (e) {
    console.error('[zillow] FAILED', e);
    await closeRun(runId, 'failed', { notes: String((e as Error).message || e) });
    process.exitCode = 1;
  } finally {
    await pool.end();
  }
}

main();