scripts/loaders/philly/pses.ts

// Load PSES topic+subtopic scores → philly_school_year_metrics.
// Source: pses_topic_subtopic_scores_all_years.
//
// Architecture: the Zip ships two xlsx files. The subtopic file has
// per-respondent breakdowns (column "survey" = Parent/Teacher/Student/Support),
// while the topic file aggregates across respondents. For O.6 — teacher +
// student first-class — we need per-respondent topic-level rollups, which we
// compute by averaging the subtopic scores within each (school, year, topic,
// respondent).
//
// Suppression codes in score_display: "n.a." (no data) and "i.s." (insufficient
// sample). Both become value=null, suppressed=true.
//
// v1 scope per O.6: teacher + student rollups. Parent + Support are ingested
// too (extra rows in DB are cheap) but no metric_definition is registered for
// them, so they'd FK-fail; we skip them at insert time.
//
// Run:  npx tsx scripts/loaders/philly/pses.ts

// @ts-expect-error — runtime-only import (no shipped types in yauzl-promise v4)
import * as zip from "yauzl-promise";
import * as XLSX from "xlsx";
import {
  prisma,
  startDataLoad,
  finishDataLoad,
  readLatestProvenance,
  readLatestSourcePath,
  batched,
} from "./_lib";

const SOURCE_ID = "pses_topic_subtopic_scores_all_years";
const SCRIPT_PATH = "scripts/loaders/philly/pses.ts";

// Topic name → metric_key suffix
const TOPIC_MAP: Record<string, string> = {
  "School Climate": "school_climate",
  "Instructional Environment": "instructional_environment",
  "School Leadership": "school_leadership",
  "Professional Capacity": "professional_capacity",
  "Family Engagement": "family_engagement",
  "Diversity, Equity, and Inclusion": "dei",
};

// O.6: v1 first-class respondents
const RESPONDENT_MAP: Record<string, string> = {
  Teacher: "teacher",
  Student: "student",
  // Parent and Support are ingested too (Zip contains them), but no
  // metric_definition exists for them, so we'll skip-write at insert.
  Parent: "parent",
  Support: "support",
};

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

function normalizeYear(year: string): string {
  // "2019-2020" → "2019-20"
  const m = year.match(/^(\d{4})-\d{2}(\d{2})$/);
  return m ? `${m[1]}-${m[2]}` : year;
}

type Row = Record<string, unknown>;

async function readSubtopicSheet(zipPath: string): Promise<Row[]> {
  const z = await zip.open(zipPath);
  try {
    for await (const entry of z) {
      if (entry.filename === "open_data_subtopic_scores.xlsx") {
        const stream = await entry.openReadStream();
        const chunks: Buffer[] = [];
        for await (const chunk of stream) chunks.push(chunk as Buffer);
        const wb = XLSX.read(Buffer.concat(chunks), { type: "buffer", cellDates: false });
        const sheet = wb.Sheets["School"];
        return XLSX.utils.sheet_to_json<Row>(sheet, { defval: null });
      }
    }
    throw new Error("subtopic xlsx not found in Zip");
  } finally {
    await z.close();
  }
}

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}`);

  console.log(`[load] reading open_data_subtopic_scores.xlsx (School sheet)`);
  const rows = await readSubtopicSheet(zipPath);
  console.log(`[load]   ${rows.length.toLocaleString()} subtopic rows`);

  // Aggregate per (ulcs, year, topic, respondent) → mean of subtopic scores
  type AggKey = string;
  type Agg = { sum: number; n: number };
  const agg = new Map<AggKey, Agg>();
  let parseSkipped = 0;
  for (const r of rows) {
    const ulcs = String(r["ulcs_code"] ?? "").trim();
    if (!ulcs || !ulcsSet.has(ulcs)) continue;
    const year = normalizeYear(String(r["year_academic"] ?? ""));
    const respondentKey = RESPONDENT_MAP[String(r["survey"] ?? "")];
    if (!respondentKey) continue;
    const topicKey = TOPIC_MAP[String(r["topic"] ?? "")];
    if (!topicKey) continue;
    const v = toNumber(r["score_display"]);
    if (v === null) { parseSkipped++; continue; }
    const k = `${ulcs}|${year}|${topicKey}|${respondentKey}`;
    if (!agg.has(k)) agg.set(k, { sum: 0, n: 0 });
    const a = agg.get(k)!;
    a.sum += v;
    a.n += 1;
  }
  console.log(`[load]   aggregated ${agg.size.toLocaleString()} (school × year × topic × respondent) cells`);
  console.log(`[load]   subtopic rows with no parseable score: ${parseSkipped.toLocaleString()}`);

  const dataLoad = await startDataLoad({
    sourceId: SOURCE_ID, sourceUrl: prov.url, sha256: prov.sha256, bytes: prov.bytes,
    yearCovered: "2019-20..2024-25", scriptPath: SCRIPT_PATH,
  });

  type Fact = {
    schoolUlcs: string;
    year: string;
    metricKey: string;
    value: number;
  };
  const facts: Fact[] = [];
  let skippedNoMetric = 0;
  for (const [k, a] of agg) {
    const [ulcs, year, topic, respondent] = k.split("|");
    // Per O.6, only teacher + student have registered metric_definitions
    if (respondent !== "teacher" && respondent !== "student") {
      skippedNoMetric++;
      continue;
    }
    const metricKey = `survey_${respondent}_${topic}`;
    // PSES scores are on a 0-10 display scale; store as PERCENT-ish but
    // multiplied by 10 so the display unit aligns with the metric_definitions
    // unit=PERCENT (the values are bounded 0-100 after ×10, except DEI which
    // can show as fractional). For v1 we store raw mean as-is; analysis can
    // re-scale if needed.
    facts.push({
      schoolUlcs: ulcs,
      year,
      metricKey,
      value: a.sum / a.n,
    });
  }
  console.log(`[load]   facts to upsert: ${facts.length.toLocaleString()}`);
  console.log(`[load]   skipped (parent/support — no metric_definition): ${skippedNoMetric.toLocaleString()}`);

  let inserted = 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: f.year,
            metricKey: f.metricKey,
            subgroup: "ALL",
            populationCut: "n/a",
          },
        },
        update: { value: f.value, suppressed: false, sourceLoadId: dataLoad.id },
        create: {
          schoolUlcs: f.schoolUlcs,
          year: f.year,
          metricKey: f.metricKey,
          subgroup: "ALL",
          populationCut: "n/a",
          value: f.value,
          suppressed: false,
          sourceLoadId: dataLoad.id,
        },
      });
      inserted++;
    }
  }, 8);

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

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