#!/usr/bin/env npx tsx
/**
 * vision_runner.ts — Gleis B: Vision-LLM-Extraktion für Buyer Information Sheets
 *
 * Liest out/buyers.json (Ergebnis von Gleis A / batch_runner.ts), nimmt alle
 * Einträge mit is_buyer_sheet === false, rendert die PDF-Seiten via pdftoppm
 * (poppler-utils) zu PNGs und schickt sie an einen llama-server
 * (OpenAI-kompatibel, Qwen3.6 + mmproj).
 *
 * Prinzip: Das VLM TRANSKRIBIERT nur (Rohwerte, verbatim). Sämtliche
 * Normalisierung (Leer-Marker, State, down_payment, Datum) passiert
 * deterministisch hier in TypeScript — identische Regeln wie Gleis A.
 *
 * Aufruf (AMD/Vulkan, inhouse):
 *   npx tsx vision_runner.ts \
 *     --input out/buyers.json \
 *     --pdf-root "/mnt/bizmatch-nas/AA Buyers NDA's/Buyers NDA's A-Z" \
 *     --api http://localhost:8000/v1 --limit 5
 *
 * Aufruf (NVIDIA/CUDA, EC2):
 *   npx tsx vision_runner.ts --input out/buyers.json \
 *     --pdf-root ~/data --api http://localhost:8000/v1 --limit 5
 *
 * Voraussetzungen: Node >= 18 (fetch), poppler-utils (pdftoppm) installiert.
 *
 * Outputs:
 *   out/buyers_vision.json  — nur die Vision-Ergebnisse (Resume-Datei)
 *   out/buyers_merged.json  — Gleis A + Gleis B zusammengeführt
 */

import { execFile } from "node:child_process";
import { promisify } from "node:util";
import * as fsp from "node:fs/promises";
import * as fs from "node:fs";
import * as path from "node:path";
import * as os from "node:os";

const execFileP = promisify(execFile);

// ---------------------------------------------------------------------------
// CLI
// ---------------------------------------------------------------------------

interface Args {
  input: string;
  pdfRoot: string;
  api: string;
  limit: number;
  only: string | null;
  dpi: number;
  maxPages: number;
  force: boolean;
  outDir: string;
  timeoutMs: number;
}

function parseArgs(): Args {
  const a = process.argv.slice(2);
  const get = (flag: string, def: string | null = null): string | null => {
    const i = a.indexOf(flag);
    return i >= 0 && a[i + 1] !== undefined ? a[i + 1] : def;
  };
  const input = get("--input", "out/buyers.json")!;
  const pdfRoot = get("--pdf-root");
  const api = (get("--api", "http://localhost:8000/v1") || "").replace(/\/+$/, "");
  if (!pdfRoot) {
    console.error("Fehler: --pdf-root <Verzeichnis> ist erforderlich.");
    process.exit(1);
  }
  return {
    input,
    pdfRoot: pdfRoot.replace(/^~(?=\/)/, os.homedir()),
    api,
    limit: parseInt(get("--limit", "0")!, 10) || 0,
    only: get("--only"),
    dpi: parseInt(get("--dpi", "150")!, 10) || 150,
    maxPages: parseInt(get("--max-pages", "8")!, 10) || 8,
    force: a.includes("--force"),
    outDir: get("--out-dir", path.dirname(input))!,
    timeoutMs: (parseInt(get("--timeout", "180")!, 10) || 180) * 1000,
  };
}

// ---------------------------------------------------------------------------
// Typen
// ---------------------------------------------------------------------------

interface BuyerRecord {
  file_name: string;
  is_buyer_sheet: boolean;
  [k: string]: unknown;
}

/** Rohantwort des VLM — alles verbatim, Normalisierung erfolgt in TS. */
interface VisionRaw {
  doc_type: "buyer_sheet" | "ca_only" | "other";
  info_sheet_count: number;
  info_page: number | null;
  ca_page: number | null;
  name_company: string | null;
  prospective_buyer: string | null;
  company: string | null;
  phone: string | null;
  cell: string | null;
  email: string | null;
  address: string | null;
  state: string | null;
  how_did_you_hear: string | null;
  interested_in_updates: string | null;
  types_of_business_raw: string | null;
  background_experience: string | null;
  total_purchase_price: string | null;
  down_payment_raw: string | null;
  date_of_introduction_raw: string | null;
}

// ---------------------------------------------------------------------------
// Normalisierung — identisch zu Gleis A halten!
// (Falls BuyerSheetParser.ts diese Funktionen exportiert, stattdessen
//  importieren, damit beide Gleise garantiert dieselbe Logik nutzen.)
// ---------------------------------------------------------------------------

const NULL_MARKERS = /^(n|na|n\/a|none|nil|x|-+|\.+)$/i;

function cleanStr(v: string | null | undefined): string | null {
  if (v == null) return null;
  const t = String(v).replace(/\s+/g, " ").trim();
  if (!t || NULL_MARKERS.test(t)) return null;
  return t;
}

const STATE_MAP: Record<string, string> = {
  alabama: "AL", alaska: "AK", arizona: "AZ", arkansas: "AR", california: "CA",
  colorado: "CO", connecticut: "CT", delaware: "DE", florida: "FL", georgia: "GA",
  hawaii: "HI", idaho: "ID", illinois: "IL", indiana: "IN", iowa: "IA",
  kansas: "KS", kentucky: "KY", louisiana: "LA", maine: "ME", maryland: "MD",
  massachusetts: "MA", michigan: "MI", minnesota: "MN", mississippi: "MS",
  missouri: "MO", montana: "MT", nebraska: "NE", nevada: "NV",
  "new hampshire": "NH", "new jersey": "NJ", "new mexico": "NM",
  "new york": "NY", "north carolina": "NC", "north dakota": "ND", ohio: "OH",
  oklahoma: "OK", oregon: "OR", pennsylvania: "PA", "rhode island": "RI",
  "south carolina": "SC", "south dakota": "SD", tennessee: "TN", texas: "TX",
  utah: "UT", vermont: "VT", virginia: "VA", washington: "WA",
  "west virginia": "WV", wisconsin: "WI", wyoming: "WY",
  "district of columbia": "DC", "washington dc": "DC",
};
const STATE_CODES = new Set(Object.values(STATE_MAP));

function normState(v: string | null): string | null {
  const c = cleanStr(v);
  if (!c) return null;
  const up = c.toUpperCase().replace(/\./g, "");
  if (up.length === 2 && STATE_CODES.has(up)) return up;
  const full = STATE_MAP[c.toLowerCase().replace(/\./g, "")];
  return full ?? c; // unbekannt: bereinigt durchreichen, nicht raten
}

/** "$350,000" → "350000"; "1.5M" → "1500000"; sonst null + raw. */
function normDownPayment(raw: string | null): { value: string | null; raw: string | null } {
  const c = cleanStr(raw);
  if (!c) return { value: null, raw: null };
  const m = c.match(/^\$?\s*(\d[\d,]*(?:\.\d+)?)\s*([kKmM])?\s*$/);
  if (!m) return { value: null, raw: c };
  let num = parseFloat(m[1].replace(/,/g, ""));
  if (m[2]) num *= /k/i.test(m[2]) ? 1_000 : 1_000_000;
  if (!Number.isFinite(num) || num <= 0) return { value: null, raw: c };
  return { value: String(Math.round(num)), raw: c };
}

const MONTHS: Record<string, number> = {
  jan: 1, feb: 2, mar: 3, apr: 4, may: 5, jun: 6,
  jul: 7, aug: 8, sep: 9, oct: 10, nov: 11, dec: 12,
};

function isoDate(y: number, mo: number, d: number): string | null {
  if (y < 100) y += 2000;
  if (y < 1990 || y > 2100 || mo < 1 || mo > 12 || d < 1 || d > 31) return null;
  return `${y}-${String(mo).padStart(2, "0")}-${String(d).padStart(2, "0")}`;
}

/** US-Formate: 6/25/26, 06-25-2026, "June 25, 2026" → YYYY-MM-DD. */
function normDate(raw: string | null): string | null {
  const c = cleanStr(raw);
  if (!c) return null;
  let m = c.match(/^(\d{1,2})[\/\-.](\d{1,2})[\/\-.](\d{2}|\d{4})$/);
  if (m) return isoDate(parseInt(m[3], 10), parseInt(m[1], 10), parseInt(m[2], 10));
  m = c.match(/^([A-Za-z]{3,9})\.?\s+(\d{1,2})(?:st|nd|rd|th)?,?\s+(\d{2}|\d{4})$/);
  if (m) {
    const mo = MONTHS[m[1].slice(0, 3).toLowerCase()];
    if (mo) return isoDate(parseInt(m[3], 10), mo, parseInt(m[2], 10));
  }
  m = c.match(/^(\d{4})-(\d{2})-(\d{2})$/);
  if (m) return isoDate(parseInt(m[1], 10), parseInt(m[2], 10), parseInt(m[3], 10));
  return null;
}

// ---------------------------------------------------------------------------
// PDF-Index (rekursiv, Basename → voller Pfad) + Rendering
// ---------------------------------------------------------------------------

async function buildPdfIndex(root: string): Promise<Map<string, string>> {
  const index = new Map<string, string>();
  const stack = [root];
  while (stack.length) {
    const dir = stack.pop()!;
    let entries: fs.Dirent[];
    try {
      entries = await fsp.readdir(dir, { withFileTypes: true });
    } catch {
      continue;
    }
    for (const e of entries) {
      const p = path.join(dir, e.name);
      if (e.isDirectory()) stack.push(p);
      else if (e.isFile() && e.name.toLowerCase().endsWith(".pdf")) {
        if (!index.has(e.name)) index.set(e.name, p);
      }
    }
  }
  return index;
}

async function renderPdf(pdfPath: string, dpi: number, maxPages: number, tmpDir: string): Promise<string[]> {
  const prefix = path.join(tmpDir, "page");
  await execFileP("pdftoppm", ["-png", "-r", String(dpi), "-l", String(maxPages), pdfPath, prefix], {
    timeout: 120_000,
  });
  const pageNum = (f: string) => parseInt(f.match(/-(\d+)\.png$/)?.[1] ?? "0", 10);
  const files = (await fsp.readdir(tmpDir))
    .filter((f) => f.endsWith(".png"))
    .sort((a, b) => pageNum(a) - pageNum(b))
    .map((f) => path.join(tmpDir, f));
  if (files.length === 0) throw new Error("pdftoppm hat keine Seiten erzeugt");
  return files;
}

// ---------------------------------------------------------------------------
// VLM-Aufruf (llama-server, OpenAI-kompatibel, JSON-Schema)
// ---------------------------------------------------------------------------

const nullableString = { type: ["string", "null"] };
const nullableInt = { type: ["integer", "null"] };

const RESPONSE_SCHEMA = {
  type: "object",
  additionalProperties: false,
  required: [
    "doc_type", "info_sheet_count", "info_page", "ca_page", "name_company",
    "prospective_buyer", "company", "phone", "cell", "email", "address",
    "state", "how_did_you_hear", "interested_in_updates",
    "types_of_business_raw", "background_experience",
    "total_purchase_price", "down_payment_raw", "date_of_introduction_raw",
  ],
  properties: {
    doc_type: { type: "string", enum: ["buyer_sheet", "ca_only", "other"] },
    info_sheet_count: { type: "integer" },
    info_page: nullableInt,
    ca_page: nullableInt,
    name_company: nullableString,
    prospective_buyer: nullableString,
    company: nullableString,
    phone: nullableString,
    cell: nullableString,
    email: nullableString,
    address: nullableString,
    state: nullableString,
    how_did_you_hear: nullableString,
    interested_in_updates: nullableString,
    types_of_business_raw: nullableString,
    background_experience: nullableString,
    total_purchase_price: nullableString,
    down_payment_raw: nullableString,
    date_of_introduction_raw: nullableString,
  },
} as const;

const SYSTEM_PROMPT =
  "You are a precise document transcription engine for scanned business forms. " +
  "You transcribe handwritten and typed form fields EXACTLY as written (verbatim), " +
  "including typos. You never guess, infer, or invent values. If a field is blank " +
  "or unreadable, you return null.";

const USER_PROMPT = `You see all pages of one PDF, in order (image 1 = page 1).

Task: Classify the document and transcribe form fields from a business brokerage "BUYER INFORMATION SHEET" package.

1. doc_type:
   - "buyer_sheet": at least one page with the heading "BUYER INFORMATION SHEET" exists.
   - "ca_only": no such page, but a confidentiality agreement page exists (text like "PROSPECTIVE BUYER AGREES TO KEEP AND HOLD CONFIDENTIAL").
   - "other": neither (notes, listing, letter, other form) — then return null for every field and 0 for info_sheet_count.
2. info_sheet_count: how many separate BUYER INFORMATION SHEET pages the document contains (some files contain two filled sheets). 0 if none. If more than one, transcribe the fields from the FIRST sheet only.
3. info_page: page number (1-based) of the (first) BUYER INFORMATION SHEET page, else null.
4. ca_page: page number of the (first) confidentiality agreement page, else null.
5. If doc_type is "buyer_sheet", transcribe from the info page VERBATIM (exactly as written, do not normalize, do not expand abbreviations):
   - name_company: value of the "Name/Company" line
   - prospective_buyer: value of the "Prospective Buyer" line
   - company: value of a separate "Company" line if present
   - phone, cell, email, address, state
   - how_did_you_hear: "How did you hear about us"
   - interested_in_updates: answer/checkbox for receiving updates (transcribe what is marked, e.g. "Yes" or "No"), else null
   - types_of_business_raw: "Type(s) of business interested in" (may span multiple lines — join with a space)
   - background_experience: "Background/Experience" (may span multiple lines)
   - total_purchase_price: "Total Purchase Price" as written
   - down_payment_raw: "Down Payment" as written (e.g. "$350,000", "1.5M", "TBD")
6. If doc_type is "ca_only": transcribe what the confidentiality agreement page offers — the prospective buyer's printed or signed name into prospective_buyer, plus address/phone/email if they appear on that page. Everything not present stays null.
7. date_of_introduction_raw: the date written next to the buyer's signature on the confidentiality agreement page, verbatim (e.g. "6/25/26"), else null.

A blank field, "N/A", "n", or an empty line = transcribe it as written; if truly empty, use null. Return only the JSON object.`;

async function fetchModelId(api: string): Promise<string> {
  try {
    const r = await fetch(`${api}/models`);
    const j = (await r.json()) as { data?: Array<{ id?: string }> };
    return j?.data?.[0]?.id ?? "unknown";
  } catch {
    return "unknown";
  }
}

async function callVision(api: string, images: string[], timeoutMs: number): Promise<VisionRaw> {
  const content: Array<Record<string, unknown>> = [{ type: "text", text: USER_PROMPT }];
  for (const img of images) {
    const b64 = await fsp.readFile(img, { encoding: "base64" });
    content.push({ type: "image_url", image_url: { url: `data:image/png;base64,${b64}` } });
  }
  const body = {
    model: "qwen3.6",
    temperature: 0.1,
    top_p: 0.95,
    max_tokens: 1500,
    chat_template_kwargs: { enable_thinking: false },
    response_format: {
      type: "json_schema",
      json_schema: { name: "buyer_sheet", strict: true, schema: RESPONSE_SCHEMA },
    },
    messages: [
      { role: "system", content: SYSTEM_PROMPT },
      { role: "user", content },
    ],
  };
  const ctl = new AbortController();
  const timer = setTimeout(() => ctl.abort(), timeoutMs);
  try {
    const r = await fetch(`${api}/chat/completions`, {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify(body),
      signal: ctl.signal,
    });
    if (!r.ok) throw new Error(`HTTP ${r.status}: ${(await r.text()).slice(0, 300)}`);
    const j = (await r.json()) as { choices?: Array<{ message?: { content?: string } }> };
    const text = j?.choices?.[0]?.message?.content;
    if (!text) throw new Error("Leere Antwort vom Server");
    return JSON.parse(text) as VisionRaw;
  } finally {
    clearTimeout(timer);
  }
}

// ---------------------------------------------------------------------------
// Merge: deterministischer Datensatz + Vision-Rohwerte → Zielstruktur
// ---------------------------------------------------------------------------

function mergeRecord(det: BuyerRecord, vis: VisionRaw, model: string): BuyerRecord {
  const base: BuyerRecord = {
    ...det,
    _parser: "vision",
    _vision_model: model,
    _vision_ts: new Date().toISOString(),
    _vision_error: undefined,
  };
  delete (base as Record<string, unknown>)["_vision_error"];

  if (vis.doc_type === "other") {
    return { ...base, is_buyer_sheet: false, _doc_type: "other", _info_page: null, _ca_page: null };
  }

  // buyer_sheet ODER ca_only: alles übernehmen, was das Dokument hergibt
  const dp = normDownPayment(vis.down_payment_raw);
  return {
    ...base,
    is_buyer_sheet: vis.doc_type === "buyer_sheet",
    _doc_type: vis.doc_type,
    _info_sheet_count: vis.info_sheet_count ?? (vis.doc_type === "buyer_sheet" ? 1 : 0),
    name_company: cleanStr(vis.name_company),
    prospective_buyer: cleanStr(vis.prospective_buyer),
    company: cleanStr(vis.company),
    phone: cleanStr(vis.phone),
    cell: cleanStr(vis.cell),
    email: cleanStr(vis.email),
    address: cleanStr(vis.address),
    state: normState(vis.state),
    how_did_you_hear: cleanStr(vis.how_did_you_hear),
    interested_in_updates: cleanStr(vis.interested_in_updates),
    types_of_business_raw: cleanStr(vis.types_of_business_raw),
    background_experience: cleanStr(vis.background_experience),
    total_purchase_price: cleanStr(vis.total_purchase_price),
    down_payment: dp.value,
    down_payment_raw: dp.raw,
    // Vision-Datum bevorzugt; Fallback: deterministischer Wert (z.B. aus Dateinamen)
    date_of_introduction: normDate(vis.date_of_introduction_raw) ?? (det.date_of_introduction as string | null) ?? null,
    _info_page: vis.info_page,
    _ca_page: vis.ca_page,
  };
}

// ---------------------------------------------------------------------------
// Main
// ---------------------------------------------------------------------------

const RETRIES = 3;

function progress(line: string): void {
  // Im Log-Modus (Pipe/tee): vollständige Zeilen mit Timestamp statt \r-Überschreiben
  if (!process.stderr.isTTY) {
    process.stderr.write(`[${new Date().toISOString()}] ${line}\n`);
    return;
  }
  const cols = process.stderr.columns ?? 120;
  process.stderr.write("\r" + line.slice(0, cols - 1).padEnd(cols - 1));
}

async function main(): Promise<void> {
  const args = parseArgs();

  // pdftoppm vorhanden?
  try {
    await execFileP("pdftoppm", ["-v"]);
  } catch {
    console.error("Fehler: pdftoppm nicht gefunden. Installieren: sudo apt install poppler-utils");
    process.exit(1);
  }

  const all: BuyerRecord[] = JSON.parse(await fsp.readFile(args.input, "utf8"));

  const visionPath = path.join(args.outDir, "buyers_vision.json");
  const mergedPath = path.join(args.outDir, "buyers_merged.json");
  const done = new Map<string, BuyerRecord>();
  try {
    const prev: BuyerRecord[] = JSON.parse(await fsp.readFile(visionPath, "utf8"));
    for (const r of prev) done.set(r.file_name, r);
  } catch {
    /* kein Resume-Stand */
  }

  let candidates = all.filter((r) => r.is_buyer_sheet === false);
  if (args.only) candidates = candidates.filter((r) => r.file_name === args.only);

  // Resume-Skip VOR dem Limit: bereits fehlerfrei Verarbeitete zählen nicht mit
  const pending = candidates.filter((r) => {
    const prev = done.get(r.file_name);
    return !(prev && !prev["_vision_error"] && !args.force);
  });
  const skipped = candidates.length - pending.length;
  const targets = args.limit > 0 ? pending.slice(0, args.limit) : pending;

  console.error(`Indexiere PDFs unter ${args.pdfRoot} ...`);
  const pdfIndex = await buildPdfIndex(args.pdfRoot);
  console.error(
    `${pdfIndex.size} PDFs gefunden. ${candidates.length} Kandidaten, ${skipped} bereits verarbeitet, ${targets.length} in diesem Lauf.`
  );

  const model = await fetchModelId(args.api);
  console.error(`Modell: ${model} @ ${args.api}`);

  let ok = 0, caOnly = 0, notSheet = 0, errors = 0;

  for (let i = 0; i < targets.length; i++) {
    const det = targets[i];
    const tag = `[${i + 1}/${targets.length}] ${det.file_name}`;

    const pdfPath = pdfIndex.get(det.file_name);
    if (!pdfPath) {
      done.set(det.file_name, { ...det, _vision_error: "PDF nicht gefunden", _vision_ts: new Date().toISOString() });
      errors++;
      progress(`${tag} … FEHLER: PDF nicht gefunden`);
      continue;
    }

    const tmpDir = await fsp.mkdtemp(path.join(os.tmpdir(), "bvs-"));
    try {
      progress(`${tag} … rendere`);
      const images = await renderPdf(pdfPath, args.dpi, args.maxPages, tmpDir);

      let vis: VisionRaw | null = null;
      let lastErr = "";
      for (let attempt = 1; attempt <= RETRIES; attempt++) {
        try {
          progress(`${tag} … VLM (${images.length} Seiten, Versuch ${attempt})`);
          vis = await callVision(args.api, images, args.timeoutMs);
          break;
        } catch (e) {
          lastErr = e instanceof Error ? e.message : String(e);
          if (attempt < RETRIES) await new Promise((res) => setTimeout(res, 5000 * attempt));
        }
      }

      if (!vis) {
        done.set(det.file_name, { ...det, _vision_error: lastErr, _vision_ts: new Date().toISOString() });
        errors++;
        progress(`${tag} … FEHLER: ${lastErr}`);
      } else {
        const merged = mergeRecord(det, vis, model);
        done.set(det.file_name, merged);
        if (merged["_doc_type"] === "ca_only") { caOnly++; progress(`${tag} … CA only`); }
        else if (merged.is_buyer_sheet) { ok++; progress(`${tag} … OK`); }
        else { notSheet++; progress(`${tag} … kein Buyer Sheet`); }
      }
    } catch (e) {
      const msg = e instanceof Error ? e.message : String(e);
      done.set(det.file_name, { ...det, _vision_error: msg, _vision_ts: new Date().toISOString() });
      errors++;
      progress(`${tag} … FEHLER: ${msg}`);
    } finally {
      await fsp.rm(tmpDir, { recursive: true, force: true });
    }

    // Inkrementell sichern (Resume-fähig)
    await fsp.mkdir(args.outDir, { recursive: true });
    await fsp.writeFile(visionPath, JSON.stringify([...done.values()], null, 2));
  }

  // Merge: Gleis A + Gleis B
  const merged = all.map((r) => done.get(r.file_name) ?? r);
  await fsp.writeFile(mergedPath, JSON.stringify(merged, null, 2));

  process.stderr.write("\n");
  console.error(
    `Fertig. OK: ${ok}, CA only: ${caOnly}, kein Buyer Sheet: ${notSheet}, Fehler: ${errors}, übersprungen: ${skipped}`
  );
  console.error(`→ ${visionPath}\n→ ${mergedPath}`);
}

main().catch((e) => {
  console.error("\nAbbruch:", e instanceof Error ? e.message : e);
  process.exit(1);
});