// Miami loader common module. Parallel to scripts/loaders/philly/_lib.ts but // writes to the miami_* tables. Decisions: docs/cities/miami/00_overview.md. // // Provenance records (sha256 + bytes) come from the discovered file in // data/cities/miami/sources///provenance.json — written by // pipeline_miami/discover.py — so loaders re-use them rather than re-hashing. import "dotenv/config"; import { config as loadEnv } from "dotenv"; import path from "node:path"; import fs from "node:fs"; import { execSync } from "node:child_process"; loadEnv({ path: path.join(process.cwd(), ".env.local") }); // Same Postgres binding as the NYC / Philly loaders — DIRECT_URL bypasses RLS. const direct = process.env.DIRECT_URL ?? process.env.POSTGRES_URL_NON_POOLING; if (direct) { process.env.DATABASE_URL = direct; } import { prisma } from "@/prisma/client"; export { prisma }; export const SOURCES_ROOT = path.join(process.cwd(), "data", "cities", "miami", "sources"); export function gitSha(): string | null { try { return execSync("git rev-parse HEAD", { stdio: ["ignore", "pipe", "ignore"] }) .toString() .trim(); } catch { return null; } } export type Provenance = { source_id: string; url: string; retrieved_at: string; sha256: string; bytes: number; publisher?: string; format?: string; fetched_from?: string; notes?: string; }; function latestDatedDir(sourceId: string): string | null { const dir = path.join(SOURCES_ROOT, sourceId); if (!fs.existsSync(dir)) return null; const dated = fs .readdirSync(dir) .filter((n) => /^\d{4}-\d{2}-\d{2}$/.test(n)) .sort() .reverse(); for (const d of dated) { if (fs.existsSync(path.join(dir, d, "provenance.json"))) return path.join(dir, d); } return null; } /** Latest provenance.json for a source_id, or null if never fetched. */ export function readLatestProvenance(sourceId: string): Provenance | null { const d = latestDatedDir(sourceId); if (!d) return null; return JSON.parse(fs.readFileSync(path.join(d, "provenance.json"), "utf-8")) as Provenance; } /** Path to the latest downloaded source file living beside provenance.json. */ export function readLatestSourcePath(sourceId: string): string | null { const d = latestDatedDir(sourceId); if (!d) return null; const entries = fs.readdirSync(d).filter((n) => n !== "provenance.json").sort(); if (entries.length === 0) return null; return path.join(d, entries[0]); } /** Both at once, throwing the standard "run discover.py" message. */ export function requireSource(sourceId: string): { filePath: string; prov: Provenance } { const filePath = readLatestSourcePath(sourceId); const prov = readLatestProvenance(sourceId); if (!filePath || !prov) { throw new Error( `No fetched source for ${sourceId}. Run: python -m pipeline_miami.discover --source ${sourceId}`, ); } console.log(`[load] file: ${filePath}`); console.log(`[load] sha256: ${prov.sha256.slice(0, 12)}... (${prov.bytes.toLocaleString()} bytes)`); return { filePath, prov }; } // Metadata and fact candidates stay in memory until one atomic replacement commits. // Non-fact loaders stage their writes with stageLoadOperation in the same transaction. import { createDataLoader, type LoadOptions } from "./atomic-load"; export type { Fact } from "./atomic-load"; const dataLoader = createDataLoader(prisma); export const startDataLoad = (opts: Omit) => dataLoader.startDataLoad({ ...opts, gitSha: gitSha() }); export const { finishDataLoad, upsertFacts, stageLoadOperation } = dataLoader; /** Run fn against batches of items with bounded concurrency (same as NYC/Philly). */ export async function batched( items: T[], batchSize: number, fn: (batch: T[], batchIndex: number) => Promise, concurrency = 4, ): Promise { const batches: T[][] = []; for (let i = 0; i < items.length; i += batchSize) { batches.push(items.slice(i, i + batchSize)); } let next = 0; async function worker() { while (next < batches.length) { const idx = next++; await fn(batches[idx], idx); } } await Promise.all(Array.from({ length: concurrency }, worker)); } // ── Florida-specific normalizers ──────────────────────────────────────────── /** * FLDOE school numbers are zero-padded text in some files ("0041") and bare * integers in others (41). Normalize to 4-digit text (findings.md F.1). * Returns null for blanks / non-numeric tokens (district total rows, "*"). */ export function padSchoolNumber(v: unknown): string | null { if (v === null || v === undefined) return null; const s = String(v).trim(); if (s === "") return null; const n = s.replace(/\.0+$/, ""); if (!/^\d{1,4}$/.test(n)) return null; return n.padStart(4, "0"); } /** "2425" → "2024-25"; "2024-25" passes through; "25" (spring) → "2024-25". */ export function schoolYearFromSuffix(v: string): string { const s = String(v).trim(); if (/^\d{4}-\d{2}$/.test(s)) return s; if (/^\d{4}$/.test(s)) { const a = 2000 + Number(s.slice(0, 2)); return `${a}-${s.slice(2)}`; } if (/^\d{2}$/.test(s)) { const end = 2000 + Number(s); return `${end - 1}-${s}`; } throw new Error(`cannot derive a school year from "${v}"`); } /** "2024-25" + (−1) → "2023-24". */ export function shiftSchoolYear(year: string, offset: number): string { const m = /^(\d{4})-(\d{2})$/.exec(year); if (!m) throw new Error(`bad school year: ${year}`); const start = Number(m[1]) + offset; return `${start}-${String((start + 1) % 100).padStart(2, "0")}`; } /** * FLDOE suppression tokens vary by file and year: "*" (n<10), "**" (older * membership files), blank (School Grades components), "i.s.", "n/a" * (findings.md F.2). parseCell returns the number, or null with a * `suppressed` flag when the cell carried a token rather than being empty. */ export const SUPPRESSION_TOKENS = new Set(["*", "**", "***", "i.s.", "n.a.", "n/a", "na", "s", "is", "--", "†", "‡"]); export function parseCell(v: unknown): { value: number | null; suppressed: boolean; token: string | null } { if (v === null || v === undefined) return { value: null, suppressed: false, token: null }; const raw = String(v).trim(); if (raw === "") return { value: null, suppressed: false, token: null }; const cleaned = raw.replace(/[,%$]/g, ""); const n = Number.parseFloat(cleaned); if (Number.isFinite(n) && /^-?\d*\.?\d+$/.test(cleaned)) return { value: n, suppressed: false, token: null }; return { value: null, suppressed: SUPPRESSION_TOKENS.has(raw.toLowerCase()), token: raw }; } /** Census of non-numeric tokens per column, for the load notes and the QA. */ export class TokenCensus { private counts = new Map(); add(token: string | null, column: string) { if (token === null) return; const k = `${column}::${token}`; this.counts.set(k, (this.counts.get(k) ?? 0) + 1); } summary(): Record { return Object.fromEntries([...this.counts.entries()].sort((a, b) => b[1] - a[1])); } toString(): string { const s = this.summary(); const keys = Object.keys(s); if (keys.length === 0) return "no non-numeric tokens"; return keys.slice(0, 12).map((k) => `${k}=${s[k]}`).join(", ") + (keys.length > 12 ? ` (+${keys.length - 12} more)` : ""); } } /** Standard entry point wrapper: run, print errors, disconnect. */ export function runMain(main: () => Promise) { main() .catch((e) => { console.error(e); process.exit(1); }) .finally(() => prisma.$disconnect()); } // ── Shared fact writer ────────────────────────────────────────────────────── /** The set of school numbers in miami_schools (the universe every loader filters to). */ export async function schoolUniverse(): Promise> { const rows = await prisma.miamiSchool.findMany({ select: { schoolNumber: true } }); return new Set(rows.map((r) => r.schoolNumber)); } /** Read an Excel sheet as row arrays and locate the header row by a cell value. */ export function sheetRows(filePath: string, sheetName?: string): { rows: unknown[][]; sheetNames: string[] } { // Lazy import keeps _lib usable from scripts that never touch Excel. // eslint-disable-next-line @typescript-eslint/no-require-imports const XLSX = require("xlsx") as typeof import("xlsx"); const wb = XLSX.readFile(filePath, { cellDates: false }); const name = sheetName ?? wb.SheetNames[0]; const sheet = wb.Sheets[name]; if (!sheet) throw new Error(`sheet "${name}" not in ${filePath} (have: ${wb.SheetNames.join(", ")})`); const rows = XLSX.utils.sheet_to_json(sheet, { header: 1, defval: null, raw: true }); return { rows, sheetNames: wb.SheetNames }; } export function findHeaderRow(rows: unknown[][], firstCell: string, maxScan = 12): number { for (let i = 0; i < Math.min(maxScan, rows.length); i++) { const c = rows[i]?.[0]; if (c !== null && c !== undefined && String(c).trim().toLowerCase() === firstCell.toLowerCase()) return i; } throw new Error(`header row starting with "${firstCell}" not found in first ${maxScan} rows`); }