`src/ingest/parcels/csv.ts`

59 lines
/**
 * Streaming CSV parser — async generator of string[] records. Handles quoted
 * fields (embedded commas/newlines/"" escapes), CRLF, and never buffers the
 * whole file (PLUTO ~165MB, King RPSale ~500MB unzipped).
 */
import { createReadStream } from 'node:fs';

export async function* csvRecords(path: string, encoding: BufferEncoding = 'utf8'): AsyncGenerator<string[]> {
  const stream = createReadStream(path, { encoding, highWaterMark: 1 << 20 });
  let field = '';
  let record: string[] = [];
  let inQuotes = false;
  let sawQuote = false; // current field was quoted (so '' stays distinguishable)
  let prevCR = false;

  for await (const chunk of stream) {
    const s = chunk as string;
    for (let i = 0; i < s.length; i++) {
      const c = s[i];
      if (inQuotes) {
        if (c === '"') {
          if (s[i + 1] === '"') { field += '"'; i++; }
          else inQuotes = false;
        } else field += c;
        continue;
      }
      if (c === '"') { inQuotes = true; sawQuote = true; continue; }
      if (c === ',') { record.push(field); field = ''; sawQuote = false; prevCR = false; continue; }
      if (c === '\r') { prevCR = true; continue; }
      if (c === '\n') {
        record.push(field); field = ''; sawQuote = false; prevCR = false;
        if (record.length > 1 || record[0] !== '') yield record;
        record = [];
        continue;
      }
      if (prevCR) { field += '\r'; prevCR = false; } // lone CR inside a field
      field += c;
    }
  }
  if (field !== '' || sawQuote || record.length) {
    record.push(field);
    if (record.length > 1 || record[0] !== '') yield record;
  }
}

/** Header-mapped variant: yields Record<lowercased header, value>. */
export async function* csvObjects(path: string, encoding: BufferEncoding = 'utf8'): AsyncGenerator<Record<string, string>> {
  let headers: string[] | null = null;
  for await (const rec of csvRecords(path, encoding)) {
    if (!headers) {
      headers = rec.map(h => h.replace(/^/, '').trim().toLowerCase());
      continue;
    }
    const obj: Record<string, string> = {};
    for (let i = 0; i < headers.length; i++) obj[headers[i]] = (rec[i] ?? '').trim();
    yield obj;
  }
}