scripts/loaders/philly/sdp_attendance_and_suspensions.ts

// Load SDP attendance (≥90%, ADA) + out-of-school suspensions distribution
// → philly_school_year_metrics.
//
// Three sources, similar shape:
//   sdp_attendance_90_school   → attendance_rate_above_90
//   sdp_attendance_ada_school  → average_daily_attendance
//   sdp_suspensions_school     → derives oss_pct_any / multiple / chronic (F.10)
//
// Run:  npx tsx scripts/loaders/philly/sdp_attendance_and_suspensions.ts

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

const SCRIPT_PATH = "scripts/loaders/philly/sdp_attendance_and_suspensions.ts";

// SDP's "Group" values for these CSVs differ slightly from the grad CSV — they
// use straight strings rather than the racial taxonomy. Build a flexible map.
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",
};

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

type Row = Record<string, string>;

async function readCsv(sourceId: string): Promise<{ rows: Row[]; prov: ReturnType<typeof readLatestProvenance> }> {
  const filePath = readLatestSourcePath(sourceId);
  const prov = readLatestProvenance(sourceId);
  if (!filePath || !prov) throw new Error(`No fetched source for ${sourceId}.`);
  const rows = parse(fs.readFileSync(filePath, "utf-8"), {
    columns: true,
    skip_empty_lines: true,
    trim: true,
  }) as Row[];
  return { rows, prov };
}

type Fact = {
  schoolUlcs: string;
  year: string;
  metricKey: string;
  subgroup: PhillySubgroup;
  value: number | null;
  denom: number | null;
  suppressed: boolean;
};

async function upsertFacts(facts: Fact[], dataLoadId: string): Promise<{ inserted: number; suppressed: number }> {
  let inserted = 0;
  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: f.year,
            metricKey: f.metricKey,
            subgroup: f.subgroup,
            populationCut: "n/a",
          },
        },
        update: { value: f.value, denominator: f.denom, suppressed: f.suppressed, sourceLoadId: dataLoadId },
        create: {
          schoolUlcs: f.schoolUlcs,
          year: f.year,
          metricKey: f.metricKey,
          subgroup: f.subgroup,
          populationCut: "n/a",
          value: f.value,
          denominator: f.denom,
          suppressed: f.suppressed,
          sourceLoadId: dataLoadId,
        },
      });
      inserted++;
      if (f.suppressed) suppressed++;
    }
  }, 8);
  return { inserted, suppressed };
}

async function loadAttendance(
  sourceId: string,
  metricKey: string,
  valueCol: string,
  denomCol: string,
  ulcsSet: Set<string>,
): Promise<void> {
  console.log(`\n[load] ${sourceId} → ${metricKey}`);
  const { rows, prov } = await readCsv(sourceId);
  console.log(`[load]   ${rows.length.toLocaleString()} CSV rows`);
  const dataLoad = await startDataLoad({
    sourceId, sourceUrl: prov!.url, sha256: prov!.sha256, bytes: prov!.bytes,
    scriptPath: SCRIPT_PATH,
  });
  const facts: Fact[] = [];
  for (const r of rows) {
    const ulcs = r["ULCS Code"]?.trim();
    if (!ulcs || !ulcsSet.has(ulcs)) continue;
    const year = (r["School Year"] ?? "").trim();
    if (!/^\d{4}-\d{4}$/.test(year)) continue;
    const yearCanon = `${year.slice(0, 4)}-${year.slice(7, 9)}`;
    const subgroup = GROUP_MAP[r["Group"]?.trim() ?? ""];
    if (!subgroup) continue;
    facts.push({
      schoolUlcs: ulcs,
      year: yearCanon,
      metricKey,
      subgroup,
      value: toNumber(r[valueCol]),
      denom: toNumber(r[denomCol]),
      suppressed: toNumber(r[valueCol]) === null,
    });
  }
  console.log(`[load]   facts: ${facts.length.toLocaleString()}`);
  const { inserted, suppressed } = await upsertFacts(facts, dataLoad.id);
  await finishDataLoad(dataLoad.id, { inserted: inserted - suppressed, updated: 0, suppressed });
  console.log(`[load]   done — ${inserted.toLocaleString()} (${suppressed.toLocaleString()} suppressed)`);
}

async function loadAdaSimple(ulcsSet: Set<string>): Promise<void> {
  const SOURCE_ID = "sdp_attendance_ada_school";
  console.log(`\n[load] ${SOURCE_ID} → average_daily_attendance (school-wide, ALL subgroup only)`);
  const { rows, prov } = await readCsv(SOURCE_ID);
  console.log(`[load]   ${rows.length.toLocaleString()} CSV rows`);
  const dataLoad = await startDataLoad({
    sourceId: SOURCE_ID, sourceUrl: prov!.url, sha256: prov!.sha256, bytes: prov!.bytes,
    scriptPath: SCRIPT_PATH,
  });
  const facts: Fact[] = [];
  for (const r of rows) {
    const ulcs = r["ULCS Code"]?.trim();
    if (!ulcs || !ulcsSet.has(ulcs)) continue;
    const year = (r["School Year"] ?? "").trim();
    if (!/^\d{4}-\d{4}$/.test(year)) continue;
    const yearCanon = `${year.slice(0, 4)}-${year.slice(7, 9)}`;
    const value = toNumber(r["Average Daily Attendance (YTD)"]);
    facts.push({
      schoolUlcs: ulcs,
      year: yearCanon,
      metricKey: "average_daily_attendance",
      subgroup: "ALL",
      value,
      denom: null,
      suppressed: value === null,
    });
  }
  console.log(`[load]   facts: ${facts.length.toLocaleString()}`);
  const { inserted, suppressed } = await upsertFacts(facts, dataLoad.id);
  await finishDataLoad(dataLoad.id, { inserted: inserted - suppressed, updated: 0, suppressed });
  console.log(`[load]   done — ${inserted.toLocaleString()} (${suppressed.toLocaleString()} suppressed)`);
}

async function loadSuspensions(ulcsSet: Set<string>): Promise<void> {
  const SOURCE_ID = "sdp_suspensions_school";
  console.log(`\n[load] ${SOURCE_ID} → oss_pct_{any,multiple,chronic}`);
  const { rows, prov } = await readCsv(SOURCE_ID);
  console.log(`[load]   ${rows.length.toLocaleString()} CSV rows`);
  const dataLoad = await startDataLoad({
    sourceId: SOURCE_ID, sourceUrl: prov!.url, sha256: prov!.sha256, bytes: prov!.bytes,
    scriptPath: SCRIPT_PATH,
  });
  const facts: Fact[] = [];
  for (const r of rows) {
    const ulcs = r["ULCS Code"]?.trim();
    if (!ulcs || !ulcsSet.has(ulcs)) continue;
    const year = (r["School Year"] ?? "").trim();
    if (!/^\d{4}-\d{4}$/.test(year)) continue;
    const yearCanon = `${year.slice(0, 4)}-${year.slice(7, 9)}`;
    const subgroup = GROUP_MAP[r["Group"]?.trim() ?? ""];
    if (!subgroup) continue;
    const denom = toNumber(r["Total Students (Yearly)"]);
    const pctZero = toNumber(r["% with Zero OS Suspensions (Yearly)"]);
    const pctTwo = toNumber(r["% with 2 OS Suspensions (Yearly)"]);
    const pctThree = toNumber(r["% with 3 OS Suspensions (Yearly)"]);
    // We don't have a published "% with 4+" column directly — derive as
    // residual: 100 − (sum of 0+1+2+3 percents)
    const pctOne = toNumber(r["% with 1 OS Suspension (Yearly)"]);
    const four_plus =
      pctZero !== null && pctOne !== null && pctTwo !== null && pctThree !== null
        ? Math.max(0, 100 - (pctZero + pctOne + pctTwo + pctThree))
        : null;

    // Derived metrics
    const pctAny = pctZero === null ? null : 100 - pctZero;
    const pctMultiple =
      pctTwo !== null && pctThree !== null && four_plus !== null
        ? pctTwo + pctThree + four_plus
        : null;
    const pctChronic = four_plus;

    for (const [key, value] of [
      ["oss_pct_any", pctAny],
      ["oss_pct_multiple", pctMultiple],
      ["oss_pct_chronic", pctChronic],
    ] as const) {
      facts.push({
        schoolUlcs: ulcs,
        year: yearCanon,
        metricKey: key,
        subgroup,
        value,
        denom,
        suppressed: value === null,
      });
    }
  }
  console.log(`[load]   facts: ${facts.length.toLocaleString()}`);
  const { inserted, suppressed } = await upsertFacts(facts, dataLoad.id);
  await finishDataLoad(dataLoad.id, { inserted: inserted - suppressed, updated: 0, suppressed });
  console.log(`[load]   done — ${inserted.toLocaleString()} (${suppressed.toLocaleString()} suppressed)`);
}

async function main() {
  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}`);

  await loadAttendance(
    "sdp_attendance_90_school",
    "attendance_rate_above_90",
    "% with 90%+ Attendance (Yearly)",
    "Total Students (Yearly)",
    ulcsSet,
  );
  // ADA file is school-wide only (no subgroups), 5 columns total.
  // Custom loader since loadAttendance assumes Group + denominator columns.
  await loadAdaSimple(ulcsSet);
  await loadSuspensions(ulcsSet);
}

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