#!/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 {
  is_buyer_sheet: boolean;
  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: [
    "is_buyer_sheet", "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: {
    is_buyer_sheet: { type: "boolean" },
    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: Decide whether this document contains a business brokerage "BUYER INFORMATION SHEET" form, and if so, transcribe its fields.

1. is_buyer_sheet: true only if a page with the heading "BUYER INFORMATION SHEET" exists. If the document is something else (notes, listing, letter, other form), return is_buyer_sheet=false and null for every field.
2. info_page: page number (1-based) of the BUYER INFORMATION SHEET page, else null.
3. ca_page: page number of the confidentiality agreement page containing text like "PROSPECTIVE BUYER AGREES TO KEEP AND HOLD CONFIDENTIAL", else null.
4. From the info page, transcribe 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")
5. 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_error: undefined,
  };
  delete (base as Record<string, unknown>)["_vision_error"];

  if (!vis.is_buyer_sheet) {
    return { ...base, is_buyer_sheet: false, _info_page: null, _ca_page: null };
  }

  const dp = normDownPayment(vis.down_payment_raw);
  return {
    ...base,
    is_buyer_sheet: true,
    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 {
  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"));
  let targets = all.filter((r) => r.is_buyer_sheet === false);
  if (args.only) targets = targets.filter((r) => r.file_name === args.only);
  if (args.limit > 0) targets = targets.slice(0, args.limit);

  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 */
  }

  console.error(`Indexiere PDFs unter ${args.pdfRoot} ...`);
  const pdfIndex = await buildPdfIndex(args.pdfRoot);
  console.error(`${pdfIndex.size} PDFs gefunden. ${targets.length} Einträge zu verarbeiten.`);

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

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

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

    const prev = done.get(det.file_name);
    if (prev && !prev["_vision_error"] && !args.force) {
      skipped++;
      progress(`${tag} … übersprungen (bereits verarbeitet)`);
      continue;
    }

    const pdfPath = pdfIndex.get(det.file_name);
    if (!pdfPath) {
      done.set(det.file_name, { ...det, _vision_error: "PDF nicht gefunden" });
      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 });
        errors++;
        progress(`${tag} … FEHLER: ${lastErr}`);
      } else {
        const merged = mergeRecord(det, vis, model);
        done.set(det.file_name, merged);
        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 });
      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}, 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);
});