import { sha256, simhash, type ConnectorMetadata, type NormalizedContent, type Observation, type SensorEndpoint } from "@websensor/core";
import { httpFetchWithRetry } from "./fetcher";
import { NormalizeError, type WebSensorConnector } from "./types";

/**
 * CSV / TSV connector for open-data files (statistics releases, registers, schedules). Rows become
 * keyed list items so a new row, a removed row or a changed value in `compareColumns` is a list event.
 *
 * Config: { keyColumn?: "id" | ["country","year"] (default: first column; "@row" = row hash),
 *           compareColumns?: ["value","status"] (default: all non-key columns),
 *           titleColumn?: "name", dateColumn?: "date", delimiter?: "," | "\t" | ";" (auto),
 *           skipRows?: 0, commentPrefix?: "#", maxRows?: 2000, tail?: true (keep the last rows), encoding?: "utf8" | "latin1" }
 */
export function parseDelimited(text: string, delimiter?: string): string[][] {
  const d = delimiter ?? detectDelimiter(text);
  const rows: string[][] = [];
  let row: string[] = [];
  let field = "";
  let quoted = false;
  for (let i = 0; i < text.length; i++) {
    const c = text[i]!;
    if (quoted) {
      if (c === '"') {
        if (text[i + 1] === '"') {
          field += '"';
          i++;
        } else quoted = false;
      } else field += c;
      continue;
    }
    if (c === '"') quoted = true;
    else if (c === d) {
      row.push(field);
      field = "";
    } else if (c === "\n" || c === "\r") {
      if (c === "\r" && text[i + 1] === "\n") i++;
      row.push(field);
      field = "";
      if (row.length > 1 || row[0] !== "") rows.push(row);
      row = [];
    } else field += c;
  }
  if (field !== "" || row.length) {
    row.push(field);
    rows.push(row);
  }
  return rows;
}

export function detectDelimiter(text: string): string {
  const head = text.split(/\r?\n/).slice(0, 5).join("\n");
  const counts = [",", "\t", ";", "|"].map((d) => ({ d, n: (head.match(new RegExp(d === "|" ? "\\|" : d, "g")) ?? []).length }));
  counts.sort((a, b) => b.n - a.n);
  return counts[0]!.n > 0 ? counts[0]!.d : ",";
}

export class CsvConnector implements WebSensorConnector {
  mode = "list" as const;
  metadata(): ConnectorMetadata {
    return { key: "csv", name: "CSV / TSV file", sensorTypes: ["FILE"], description: "Delimited open-data files → keyed rows (new/removed rows, changed values)", version: "1.0.0" };
  }
  async fetch(endpoint: SensorEndpoint): Promise<Observation> {
    const cfg = endpoint.config as { headers?: Record<string, string>; timeoutMs?: number };
    return httpFetchWithRetry(endpoint.id, endpoint.url, { etag: endpoint.etag, lastModified: endpoint.lastModified, accept: "text/csv, text/tab-separated-values, text/plain;q=0.9, */*;q=0.5", headers: cfg.headers, timeoutMs: cfg.timeoutMs ?? 60_000, maxBytes: 64 * 1024 * 1024 });
  }
  async normalize(endpoint: SensorEndpoint, obs: Observation): Promise<NormalizedContent> {
    if (!obs.body) throw new NormalizeError("no_body", "Observation has no body");
    const cfg = endpoint.config as { keyColumn?: string | string[]; compareColumns?: string[]; titleColumn?: string; dateColumn?: string; delimiter?: string; skipRows?: number; commentPrefix?: string; maxRows?: number; tail?: boolean; encoding?: BufferEncoding };
    let text = obs.body.toString(cfg.encoding ?? "utf8").replace(/^/, "");
    if (cfg.commentPrefix) text = text.split(/\r?\n/).filter((l) => !l.startsWith(cfg.commentPrefix!)).join("\n");
    if (/^\s*<(!doctype|html)/i.test(text.slice(0, 200))) throw new NormalizeError("html_not_csv", "Endpoint returned HTML instead of a delimited file");
    const rows = parseDelimited(text, cfg.delimiter === "\\t" ? "\t" : cfg.delimiter).slice(cfg.skipRows ?? 0);
    if (rows.length < 2) throw new NormalizeError("bad_shape", "Fewer than 2 rows (header + data) in the file");
    const header = rows[0]!.map((h) => h.trim());
    const idx = (name: string): number => header.findIndex((h) => h.toLowerCase() === name.toLowerCase());
    const keyCols = cfg.keyColumn === "@row" ? [] : (Array.isArray(cfg.keyColumn) ? cfg.keyColumn : [cfg.keyColumn ?? header[0]!]).map((k) => {
      const i = idx(k);
      if (i < 0) throw new NormalizeError("bad_config", `keyColumn "${k}" not found in header [${header.slice(0, 12).join(", ")}]`);
      return i;
    });
    const compareCols = (cfg.compareColumns ?? header.filter((_, i) => !keyCols.includes(i))).map((c) => {
      const i = idx(c);
      if (i < 0) throw new NormalizeError("bad_config", `compareColumns "${c}" not found in header`);
      return i;
    });
    const titleIdx = cfg.titleColumn ? idx(cfg.titleColumn) : -1;
    const dateIdx = cfg.dateColumn ? idx(cfg.dateColumn) : -1;
    const max = cfg.maxRows ?? 2000;
    // `tail: true` keeps the most recent rows of a time series that grows at the end (FRED, ECB, NOAA).
    const data = cfg.tail ? rows.slice(1).slice(-max) : rows.slice(1, 1 + max);
    const items = data.map((r, n) => {
      const key = keyCols.length ? keyCols.map((i) => r[i] ?? "").join("|") : sha256(r.join("
")).slice(0, 16);
      const rec: Record<string, unknown> & { key: string } = { key, row: n + 1 };
      for (const i of compareCols) rec[header[i]!] = r[i] ?? "";
      rec.title = titleIdx >= 0 ? r[titleIdx] : `${header.slice(0, 3).map((h, i) => `${h}=${r[idx(h)] ?? r[i] ?? ""}`).join(" · ")}`;
      rec.summary = compareCols.map((i) => `${header[i]}: ${r[i] ?? ""}`).join(" · ").slice(0, 600);
      rec.url = endpoint.url;
      const dRaw = dateIdx >= 0 ? r[dateIdx] : "";
      const d = dRaw ? new Date(dRaw) : null;
      rec.publishedAt = d && !Number.isNaN(d.getTime()) ? d.toISOString() : null;
      return rec;
    });
    const compareFields = compareCols.map((i) => header[i]!);
    const canonical = items.map((i) => `${i.key}\t${compareFields.map((f) => String(i[f] ?? "")).join("\t")}`).join("\n");
    const newest = items.map((i) => i.publishedAt as string | null).filter((x): x is string => Boolean(x)).sort().at(-1);
    return {
      mode: "list",
      items,
      compareFields,
      title: endpoint.name,
      rawHash: sha256(text),
      canonicalHash: sha256(canonical),
      semanticHash: simhash(items.slice(0, 500).map((i) => String(i.title)).join("\n")),
      publishedAt: newest ? new Date(newest) : null,
      extractionConfidence: 1,
      extra: { rows: data.length, columns: header.length, header: header.slice(0, 40), truncated: rows.length - 1 > data.length },
    };
  }
}