TypeScript 55.4%
Python 43.2%
SQL 1.2%
1import { sha256, simhash, type ConnectorMetadata, type NormalizedContent, type Observation, type SensorEndpoint } from "@websensor/core";2import { httpFetchWithRetry } from "./fetcher";3import { NormalizeError, type WebSensorConnector } from "./types";45/**6 * CSV / TSV connector for open-data files (statistics releases, registers, schedules). Rows become7 * keyed list items so a new row, a removed row or a changed value in `compareColumns` is a list event.8 *9 * Config: { keyColumn?: "id" | ["country","year"] (default: first column; "@row" = row hash),10 * compareColumns?: ["value","status"] (default: all non-key columns),11 * titleColumn?: "name", dateColumn?: "date", delimiter?: "," | "\t" | ";" (auto),12 * skipRows?: 0, commentPrefix?: "#", maxRows?: 2000, tail?: true (keep the last rows), encoding?: "utf8" | "latin1" }13 */14export function parseDelimited(text: string, delimiter?: string): string[][] {15 const d = delimiter ?? detectDelimiter(text);16 const rows: string[][] = [];17 let row: string[] = [];18 let field = "";19 let quoted = false;20 for (let i = 0; i < text.length; i++) {21 const c = text[i]!;22 if (quoted) {23 if (c === '"') {24 if (text[i + 1] === '"') {25 field += '"';26 i++;27 } else quoted = false;28 } else field += c;29 continue;30 }31 if (c === '"') quoted = true;32 else if (c === d) {33 row.push(field);34 field = "";35 } else if (c === "\n" || c === "\r") {36 if (c === "\r" && text[i + 1] === "\n") i++;37 row.push(field);38 field = "";39 if (row.length > 1 || row[0] !== "") rows.push(row);40 row = [];41 } else field += c;42 }43 if (field !== "" || row.length) {44 row.push(field);45 rows.push(row);46 }47 return rows;48}4950export function detectDelimiter(text: string): string {51 const head = text.split(/\r?\n/).slice(0, 5).join("\n");52 const counts = [",", "\t", ";", "|"].map((d) => ({ d, n: (head.match(new RegExp(d === "|" ? "\\|" : d, "g")) ?? []).length }));53 counts.sort((a, b) => b.n - a.n);54 return counts[0]!.n > 0 ? counts[0]!.d : ",";55}5657export class CsvConnector implements WebSensorConnector {58 mode = "list" as const;59 metadata(): ConnectorMetadata {60 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" };61 }62 async fetch(endpoint: SensorEndpoint): Promise<Observation> {63 const cfg = endpoint.config as { headers?: Record<string, string>; timeoutMs?: number };64 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 });65 }66 async normalize(endpoint: SensorEndpoint, obs: Observation): Promise<NormalizedContent> {67 if (!obs.body) throw new NormalizeError("no_body", "Observation has no body");68 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 };69 let text = obs.body.toString(cfg.encoding ?? "utf8").replace(/^/, "");70 if (cfg.commentPrefix) text = text.split(/\r?\n/).filter((l) => !l.startsWith(cfg.commentPrefix!)).join("\n");71 if (/^\s*<(!doctype|html)/i.test(text.slice(0, 200))) throw new NormalizeError("html_not_csv", "Endpoint returned HTML instead of a delimited file");72 const rows = parseDelimited(text, cfg.delimiter === "\\t" ? "\t" : cfg.delimiter).slice(cfg.skipRows ?? 0);73 if (rows.length < 2) throw new NormalizeError("bad_shape", "Fewer than 2 rows (header + data) in the file");74 const header = rows[0]!.map((h) => h.trim());75 const idx = (name: string): number => header.findIndex((h) => h.toLowerCase() === name.toLowerCase());76 const keyCols = cfg.keyColumn === "@row" ? [] : (Array.isArray(cfg.keyColumn) ? cfg.keyColumn : [cfg.keyColumn ?? header[0]!]).map((k) => {77 const i = idx(k);78 if (i < 0) throw new NormalizeError("bad_config", `keyColumn "${k}" not found in header [${header.slice(0, 12).join(", ")}]`);79 return i;80 });81 const compareCols = (cfg.compareColumns ?? header.filter((_, i) => !keyCols.includes(i))).map((c) => {82 const i = idx(c);83 if (i < 0) throw new NormalizeError("bad_config", `compareColumns "${c}" not found in header`);84 return i;85 });86 const titleIdx = cfg.titleColumn ? idx(cfg.titleColumn) : -1;87 const dateIdx = cfg.dateColumn ? idx(cfg.dateColumn) : -1;88 const max = cfg.maxRows ?? 2000;89 // `tail: true` keeps the most recent rows of a time series that grows at the end (FRED, ECB, NOAA).90 const data = cfg.tail ? rows.slice(1).slice(-max) : rows.slice(1, 1 + max);91 const items = data.map((r, n) => {92 const key = keyCols.length ? keyCols.map((i) => r[i] ?? "").join("|") : sha256(r.join("")).slice(0, 16);93 const rec: Record<string, unknown> & { key: string } = { key, row: n + 1 };94 for (const i of compareCols) rec[header[i]!] = r[i] ?? "";95 rec.title = titleIdx >= 0 ? r[titleIdx] : `${header.slice(0, 3).map((h, i) => `${h}=${r[idx(h)] ?? r[i] ?? ""}`).join(" · ")}`;96 rec.summary = compareCols.map((i) => `${header[i]}: ${r[i] ?? ""}`).join(" · ").slice(0, 600);97 rec.url = endpoint.url;98 const dRaw = dateIdx >= 0 ? r[dateIdx] : "";99 const d = dRaw ? new Date(dRaw) : null;100 rec.publishedAt = d && !Number.isNaN(d.getTime()) ? d.toISOString() : null;101 return rec;102 });103 const compareFields = compareCols.map((i) => header[i]!);104 const canonical = items.map((i) => `${i.key}\t${compareFields.map((f) => String(i[f] ?? "")).join("\t")}`).join("\n");105 const newest = items.map((i) => i.publishedAt as string | null).filter((x): x is string => Boolean(x)).sort().at(-1);106 return {107 mode: "list",108 items,109 compareFields,110 title: endpoint.name,111 rawHash: sha256(text),112 canonicalHash: sha256(canonical),113 semanticHash: simhash(items.slice(0, 500).map((i) => String(i.title)).join("\n")),114 publishedAt: newest ? new Date(newest) : null,115 extractionConfidence: 1,116 extra: { rows: data.length, columns: header.length, header: header.slice(0, 40), truncated: rows.length - 1 > data.length },117 };118 }119}120