scripts/loaders/philly/sdp_enrollment_demographics.ts

// Load SDP Enrollment & Demographics CSV → philly_school_year_metrics.
// Source: docs/cities/philly/sources.yaml `sdp_enrollment_demographics_school_2024-25`.
//
// CSV grain: one row per (school × grade-level). For v1 we aggregate to
// school-level by summing counts and computing share = sum_n / sum_total.
// (Re-deriving share from counts is more accurate than averaging the per-grade
// pct columns, which would weight grades equally regardless of enrollment.)
//
// Stores both:
//   - "enrollment"  (subgroup=ALL, value = total students)
//   - "pct_<group>" (subgroup=ALL, value = aggregated share 0-100)
//
// Note: NYC stored demographics in a separate `school_year_demographics` table.
// We use `philly_school_year_metrics` with pct_* keys because (a) consistent
// shape simplifies the analysis pipeline, and (b) the residual-z engine reads
// covariates from the same long-format table it reads outcomes from.
//
// Run:  npx tsx scripts/loaders/philly/sdp_enrollment_demographics.ts

import * as fs from "node:fs";
import { parse } from "csv-parse/sync";
import {
  prisma,
  startDataLoad,
  finishDataLoad,
  readLatestProvenance,
  readLatestSourcePath,
  batched,
} from "./_lib";

const SOURCE_ID = "sdp_enrollment_demographics_school_2024-25";
const YEAR_COVERED = "2024-25";
const SCRIPT_PATH = "scripts/loaders/philly/sdp_enrollment_demographics.ts";

// CSV count-column → metric_key. The CSV has matched (count, pct) pairs
// (`black` + `blackpct`, etc.) — we re-derive pct from summed counts.
const COUNT_TO_METRIC: Record<string, string> = {
  ell: "pct_english_learner",
  iep: "pct_special_ed",
  female: "pct_female",
  male: "pct_male",
  indian: "pct_amer_indian",
  asian: "pct_asian",
  hawaiian: "pct_native_hawaiian",
  black: "pct_black",
  hispanic: "pct_hispanic",
  white: "pct_white",
  mult: "pct_two_or_more_races",
};

type Row = Record<string, string>;

function toNumber(v: unknown): number | null {
  if (v === null || v === undefined) return null;
  const s = String(v).trim();
  if (s === "" || s === "*") return null;
  const n = Number.parseFloat(s);
  return Number.isFinite(n) ? n : null;
}

async function main() {
  const filePath = readLatestSourcePath(SOURCE_ID);
  const prov = readLatestProvenance(SOURCE_ID);
  if (!filePath || !prov) {
    throw new Error(`No fetched source for ${SOURCE_ID}.`);
  }
  console.log(`[load] file: ${filePath}`);
  console.log(`[load] sha256: ${prov.sha256.slice(0, 12)}... (${prov.bytes.toLocaleString()} bytes)`);

  const phillySchools = await prisma.phillySchool.findMany({ select: { ulcsCode: true } });
  const ulcsSet = new Set(phillySchools.map((s) => s.ulcsCode));
  console.log(`[load] philly_schools universe: ${ulcsSet.size}`);

  const text = fs.readFileSync(filePath, "utf-8");
  const rows = parse(text, { columns: true, skip_empty_lines: true, trim: true }) as Row[];
  console.log(`[load] CSV rows: ${rows.length.toLocaleString()}`);

  // Aggregate per (ulcs, year): sum of all-grade counts + computed shares.
  type Agg = { allStudents: number; counts: Record<string, number>; cep_pct: number | null };
  const agg = new Map<string, Agg>();
  let phillyHits = 0;
  for (const r of rows) {
    const ulcs = r["ulcscode"]?.trim();
    if (!ulcs || !ulcsSet.has(ulcs)) continue;
    const all = toNumber(r["allstudents"]);
    if (all === null || all === 0) continue;
    phillyHits++;
    const key = ulcs;
    if (!agg.has(key)) {
      agg.set(key, { allStudents: 0, counts: {}, cep_pct: null });
    }
    const a = agg.get(key)!;
    a.allStudents += all;
    for (const [col] of Object.entries(COUNT_TO_METRIC)) {
      const n = toNumber(r[col]);
      if (n !== null) a.counts[col] = (a.counts[col] ?? 0) + n;
    }
    // CEP isn't grade-additive (it's a school-wide flag/share). Take the
    // first non-null we see per school.
    if (a.cep_pct === null) {
      a.cep_pct = toNumber(r["ceppct"]);
    }
  }
  console.log(`[load] aggregated ${agg.size} schools (${phillyHits} per-grade rows hit)`);

  const dataLoad = await startDataLoad({
    sourceId: SOURCE_ID,
    sourceUrl: prov.url,
    sha256: prov.sha256,
    bytes: prov.bytes,
    yearCovered: YEAR_COVERED,
    scriptPath: SCRIPT_PATH,
  });

  type Fact = { schoolUlcs: string; metricKey: string; value: number };
  const facts: Fact[] = [];
  for (const [ulcs, a] of agg) {
    if (a.allStudents <= 0) continue;
    facts.push({ schoolUlcs: ulcs, metricKey: "enrollment", value: a.allStudents });
    for (const [col, key] of Object.entries(COUNT_TO_METRIC)) {
      const n = a.counts[col];
      if (n === undefined) continue;
      const pct = (n / a.allStudents) * 100;
      facts.push({ schoolUlcs: ulcs, metricKey: key, value: pct });
    }
    if (a.cep_pct !== null) {
      facts.push({ schoolUlcs: ulcs, metricKey: "pct_econ_disadv", value: a.cep_pct });
    }
  }
  console.log(`[load] total facts: ${facts.length.toLocaleString()}`);

  await batched(facts, 200, async (batch) => {
    for (const f of batch) {
      await prisma.phillySchoolYearMetric.upsert({
        where: {
          schoolUlcs_year_metricKey_subgroup_populationCut: {
            schoolUlcs: f.schoolUlcs,
            year: YEAR_COVERED,
            metricKey: f.metricKey,
            subgroup: "ALL",
            populationCut: "n/a",
          },
        },
        update: { value: f.value, suppressed: false, sourceLoadId: dataLoad.id },
        create: {
          schoolUlcs: f.schoolUlcs,
          year: YEAR_COVERED,
          metricKey: f.metricKey,
          subgroup: "ALL",
          populationCut: "n/a",
          value: f.value,
          suppressed: false,
          sourceLoadId: dataLoad.id,
        },
      });
    }
  }, 8);

  await finishDataLoad(dataLoad.id, { inserted: facts.length, updated: 0, suppressed: 0 });
  console.log(`[load] done — ${facts.length.toLocaleString()} demographic facts upserted`);
}

main()
  .catch((e) => { console.error(e); process.exit(1); })
  .finally(() => prisma.$disconnect());