scripts/loaders/miami/_lib.ts

// 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/<source_id>/<date>/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<LoadOptions, "gitSha">) => 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<T>(
  items: T[],
  batchSize: number,
  fn: (batch: T[], batchIndex: number) => Promise<void>,
  concurrency = 4,
): Promise<void> {
  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<string, number>();
  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<string, number> {
    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<void>) {
  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<Set<string>> {
  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<unknown[]>(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`);
}