scripts/loaders/philly/sdp_pssa_keystone.ts

// Load SDP PSSA + Keystone Zip → philly_school_year_metrics.
// Source: docs/cities/philly/sources.yaml entry `sdp_pssa_keystone_2024-25`.
//
// The Zip contains 4 CSVs (per findings.md F.6):
//   PSSA_Keystone_Schools_Acct_Suppressed.csv     ← canonical (O.8); population_cut="acct"
//   PSSA_Keystone_Schools_Actual_Suppressed.csv   ← all-tested cut;  population_cut="actual"
//   PSSA_Keystone_District_Acct_Suppressed.csv    ← district-level (not loaded into philly_school_year_metrics)
//   PSSA_Keystone_District_Actual_Suppressed.csv  ← district-level
//
// Per-row grain in the schools CSV: id_eos × testname × subject × grade ×
// category × group. The schema uses long-format columns (denom, bel/bas/prof/
// adv/profadv with num + score pairs) — see findings.md batch 2 inspection.
//
// Run:  npx tsx scripts/loaders/philly/sdp_pssa_keystone.ts

// yauzl-promise has no shipped types as of v3; declare ambient.
// @ts-expect-error — runtime-only import
import * as zip from "yauzl-promise";
import { parse } from "csv-parse/sync";
import {
  prisma,
  startDataLoad,
  finishDataLoad,
  readLatestProvenance,
  readLatestSourcePath,
  batched,
} from "./_lib";
import type { PhillySubgroup } from "@prisma/client";

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

// Map SDP CSV's group/category values → our PhillySubgroup enum.
const GROUP_MAP: Record<string, PhillySubgroup> = {
  "All Students": "ALL",
  "American Indian/Alaskan Native": "AMER_INDIAN_AK_NATIVE",
  "Asian": "ASIAN",
  "Native Hawaiian/Pacific Islander": "HAWAIIAN_PAC_ISL",
  "Black/African American": "BLACK",
  "Hispanic/Latino": "HISPANIC",
  "White": "WHITE",
  "Multi Racial/Other": "TWO_OR_MORE_RACES",
  "Economically Disadvantaged": "ECON_DISADV",
  "Not Economically Disadvantaged": "NOT_ECON_DISADV",
  "EL": "ELL",
  "Non-EL": "NOT_ELL",
  "Has IEP": "IEP",
  "Does Not Have IEP": "NOT_IEP",
  "Female": "FEMALE",
  "Male": "MALE",
  "Non-Binary": "NON_BINARY",
};

// Map (testname, subject, grade) → our metric_key.
// PSSA: grades 3-8 + "Grades 3-8" aggregate; subjects ELA, Math (Science g4/g8 too but v1 skips).
// Keystone: subjects Algebra I, Literature, Biology; grade typically "11" but file may use other markers.
function metricKey(testname: string, subject: string, grade: string): string | null {
  const t = testname.trim().toUpperCase();
  const s = subject.trim().toLowerCase();
  const g = grade.trim();

  if (t === "PSSA") {
    const subjKey =
      s === "ela" ? "ela" :
      s === "math" || s === "mathematics" ? "math" :
      null;
    if (!subjKey) return null;  // skip science for v1
    if (g === "Grades 3-8" || g.toLowerCase() === "all") {
      return `pssa_all_${subjKey}_proficiency`;
    }
    const gn = Number.parseInt(g, 10);
    if ([3, 4, 5, 6, 7, 8].includes(gn)) {
      return `pssa_grade${gn}_${subjKey}_proficiency`;
    }
    return null;
  }
  if (t === "KEYSTONE") {
    if (s.includes("algebra")) return "keystone_algebra_proficiency";
    if (s.includes("literature")) return "keystone_literature_proficiency";
    if (s.includes("biology")) return "keystone_biology_proficiency";
    return null;
  }
  return null;
}

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

type Row = Record<string, string>;

async function readZipEntry(zipPath: string, entryName: string): Promise<string> {
  const z = await zip.open(zipPath);
  try {
    for await (const entry of z) {
      if (entry.filename === entryName) {
        const stream = await entry.openReadStream();
        const chunks: Buffer[] = [];
        for await (const chunk of stream) chunks.push(chunk as Buffer);
        return Buffer.concat(chunks).toString("utf-8");
      }
    }
    throw new Error(`Entry not found in zip: ${entryName}`);
  } finally {
    await z.close();
  }
}

async function loadVariant(
  zipPath: string,
  entryName: string,
  populationCut: "acct" | "actual",
  ulcsSet: Set<string>,
  dataLoadId: string,
): Promise<{ facts: number; suppressed: number }> {
  console.log(`[load] reading ${entryName} (population_cut="${populationCut}")`);
  const text = await readZipEntry(zipPath, entryName);
  const rows = parse(text, { columns: true, skip_empty_lines: true, trim: true }) as Row[];
  console.log(`[load]   ${rows.length.toLocaleString()} CSV rows`);

  type Fact = {
    schoolUlcs: string;
    metricKey: string;
    subgroup: PhillySubgroup;
    value: number | null;
    denom: number | null;
    suppressed: boolean;
  };
  const facts: Fact[] = [];
  let phillyHits = 0;
  let unmapped = 0;

  for (const r of rows) {
    const idEos = r["id_eos"]?.trim();
    if (!idEos || !ulcsSet.has(idEos)) continue;
    phillyHits++;
    const mk = metricKey(r["testname"] ?? "", r["subject"] ?? "", r["grade"] ?? "");
    if (!mk) { unmapped++; continue; }
    const subgroup = GROUP_MAP[r["group"]?.trim() ?? ""];
    if (!subgroup) { unmapped++; continue; }
    const value = toNumber(r["profadv_score"]);
    const denom = toNumber(r["denom"]);
    facts.push({
      schoolUlcs: idEos,
      metricKey: mk,
      subgroup,
      value,
      denom,
      suppressed: value === null,
    });
  }
  console.log(`[load]   phillyHits=${phillyHits} unmappedRows=${unmapped} facts=${facts.length.toLocaleString()}`);

  let suppressed = 0;
  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: f.subgroup,
            populationCut,
          },
        },
        update: {
          value: f.value,
          denominator: f.denom,
          suppressed: f.suppressed,
          sourceLoadId: dataLoadId,
        },
        create: {
          schoolUlcs: f.schoolUlcs,
          year: YEAR_COVERED,
          metricKey: f.metricKey,
          subgroup: f.subgroup,
          populationCut,
          value: f.value,
          denominator: f.denom,
          suppressed: f.suppressed,
          sourceLoadId: dataLoadId,
        },
      });
      if (f.suppressed) suppressed++;
    }
  }, 8);
  return { facts: facts.length, suppressed };
}

async function main() {
  const zipPath = readLatestSourcePath(SOURCE_ID);
  const prov = readLatestProvenance(SOURCE_ID);
  if (!zipPath || !prov) {
    throw new Error(`No fetched source for ${SOURCE_ID}.`);
  }
  console.log(`[load] zip: ${zipPath}`);
  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 dataLoad = await startDataLoad({
    sourceId: SOURCE_ID,
    sourceUrl: prov.url,
    sha256: prov.sha256,
    bytes: prov.bytes,
    yearCovered: YEAR_COVERED,
    scriptPath: SCRIPT_PATH,
  });

  const acct = await loadVariant(zipPath, "PSSA_Keystone_Schools_Acct_Suppressed.csv", "acct", ulcsSet, dataLoad.id);
  const actual = await loadVariant(zipPath, "PSSA_Keystone_Schools_Actual_Suppressed.csv", "actual", ulcsSet, dataLoad.id);

  const totalFacts = acct.facts + actual.facts;
  const totalSuppressed = acct.suppressed + actual.suppressed;
  await finishDataLoad(dataLoad.id, {
    inserted: totalFacts - totalSuppressed,
    updated: 0,
    suppressed: totalSuppressed,
  });
  console.log(`[load] done — ${totalFacts.toLocaleString()} facts (${totalSuppressed.toLocaleString()} suppressed)`);
}

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