368 lines
11 KiB
TypeScript
368 lines
11 KiB
TypeScript
import {
|
|
safeJsonParse,
|
|
stableStringify,
|
|
type JsonValue,
|
|
} from "@add-ideas/toolbox-helpers";
|
|
import { isObject } from "./data";
|
|
export type StageType =
|
|
"select" | "rename" | "filter" | "map" | "sort" | "group" | "join" | "format";
|
|
export interface Stage {
|
|
id: string;
|
|
type: StageType;
|
|
enabled: boolean;
|
|
config: Record<string, string>;
|
|
}
|
|
export interface Snapshot {
|
|
name: string;
|
|
rows: Record<string, JsonValue>[];
|
|
losses: string[];
|
|
stage?: Stage;
|
|
}
|
|
export interface PipelineResult {
|
|
snapshots: Snapshot[];
|
|
rows: Record<string, JsonValue>[];
|
|
recipe: string;
|
|
}
|
|
const MAX_STAGES = 30,
|
|
MAX_ROWS = 10000,
|
|
MAX_JOIN = 20000;
|
|
export function records(values: JsonValue[]): Record<string, JsonValue>[] {
|
|
if (values.length > MAX_ROWS) throw new Error("Input exceeds 10,000 rows.");
|
|
return values.map((value, index) =>
|
|
isObject(value) ? { ...value } : { value, __row: index + 1 },
|
|
);
|
|
}
|
|
export function runPipeline(
|
|
input: Record<string, JsonValue>[],
|
|
stages: Stage[],
|
|
right: Record<string, JsonValue>[] = [],
|
|
): PipelineResult {
|
|
if (stages.length > MAX_STAGES) throw new Error("Recipe exceeds 30 stages.");
|
|
let rows = input.map((row) => ({ ...row }));
|
|
const snapshots: Snapshot[] = [{ name: "Input", rows, losses: [] }];
|
|
for (const stage of stages) {
|
|
if (!stage.enabled) {
|
|
snapshots.push({
|
|
name: `${stage.type} (disabled)`,
|
|
rows,
|
|
losses: [],
|
|
stage,
|
|
});
|
|
continue;
|
|
}
|
|
const output = apply(rows, stage, right);
|
|
rows = output.rows;
|
|
if (rows.length > MAX_JOIN)
|
|
throw new Error(`Stage ${stage.id} expands beyond 20,000 rows.`);
|
|
snapshots.push({ name: stage.type, rows, losses: output.losses, stage });
|
|
}
|
|
return { snapshots, rows, recipe: exportRecipe(stages) };
|
|
}
|
|
function apply(
|
|
rows: Record<string, JsonValue>[],
|
|
stage: Stage,
|
|
right: Record<string, JsonValue>[],
|
|
) {
|
|
const c = stage.config,
|
|
losses: string[] = [];
|
|
if (stage.type === "select") {
|
|
const fields = list(c.fields);
|
|
if (!fields.length) throw new Error("Select needs comma-separated fields.");
|
|
return {
|
|
rows: rows.map((row) =>
|
|
Object.fromEntries(fields.map((field) => [field, row[field] ?? null])),
|
|
),
|
|
losses: [
|
|
`Fields outside ${fields.join(", ")} were intentionally discarded.`,
|
|
],
|
|
};
|
|
}
|
|
if (stage.type === "rename") {
|
|
const pairs = list(c.pairs).map((item) =>
|
|
item.split("=").map((x) => x.trim()),
|
|
);
|
|
return {
|
|
rows: rows.map((row, index) => {
|
|
const out = { ...row };
|
|
for (const [left, target] of pairs) {
|
|
if (!left || !target) throw new Error("Rename pairs use old=new.");
|
|
if (target in out && target !== left)
|
|
losses.push(
|
|
`Row ${index + 1}: ${target} was overwritten by rename.`,
|
|
);
|
|
out[target] = out[left] ?? null;
|
|
delete out[left];
|
|
}
|
|
return out;
|
|
}),
|
|
losses,
|
|
};
|
|
}
|
|
if (stage.type === "filter") {
|
|
const field = required(c.field, "Filter field"),
|
|
op = c.operator ?? "=",
|
|
literal = parseLiteral(c.value ?? "");
|
|
return {
|
|
rows: rows.filter((row) => test(row[field], op, literal)),
|
|
losses: [
|
|
`${rows.length} input rows were tested; non-matches were discarded.`,
|
|
],
|
|
};
|
|
}
|
|
if (stage.type === "map") {
|
|
const target = required(c.target, "Map target"),
|
|
operation = c.operation ?? "copy",
|
|
source = c.source ?? "",
|
|
literal = c.value ?? "";
|
|
return {
|
|
rows: rows.map((row, index) => {
|
|
const out = { ...row };
|
|
let value: JsonValue =
|
|
operation === "literal"
|
|
? parseLiteral(literal)
|
|
: (row[source] ?? null);
|
|
if (
|
|
operation === "upper" ||
|
|
operation === "lower" ||
|
|
operation === "trim"
|
|
)
|
|
value =
|
|
operation === "upper"
|
|
? String(value ?? "").toUpperCase()
|
|
: operation === "lower"
|
|
? String(value ?? "").toLowerCase()
|
|
: String(value ?? "").trim();
|
|
if (operation === "number") {
|
|
const n = Number(value);
|
|
if (!Number.isFinite(n)) {
|
|
losses.push(
|
|
`Row ${index + 1}: ${source} could not become a number; null emitted.`,
|
|
);
|
|
value = null;
|
|
} else value = n;
|
|
}
|
|
out[target] = value;
|
|
return out;
|
|
}),
|
|
losses,
|
|
};
|
|
}
|
|
if (stage.type === "format") {
|
|
const field = required(c.field, "Format field"),
|
|
as = c.as ?? "string";
|
|
return {
|
|
rows: rows.map((row, index) => {
|
|
const out = { ...row },
|
|
value = row[field];
|
|
if (as === "string")
|
|
out[field] =
|
|
value === null || value === undefined
|
|
? ""
|
|
: typeof value === "object"
|
|
? stableStringify(value)
|
|
: String(value);
|
|
else if (as === "number") {
|
|
const n = Number(value);
|
|
out[field] = Number.isFinite(n) ? n : null;
|
|
if (!Number.isFinite(n))
|
|
losses.push(`Row ${index + 1}: ${field} became null.`);
|
|
} else if (as === "boolean")
|
|
out[field] = value === true || String(value).toLowerCase() === "true";
|
|
else throw new Error("Format type must be string, number or boolean.");
|
|
return out;
|
|
}),
|
|
losses,
|
|
};
|
|
}
|
|
if (stage.type === "sort") {
|
|
const field = required(c.field, "Sort field"),
|
|
direction = c.direction === "desc" ? -1 : 1;
|
|
return {
|
|
rows: rows
|
|
.map((row, index) => ({ row, index }))
|
|
.sort(
|
|
(a, b) =>
|
|
compare(a.row[field], b.row[field]) * direction ||
|
|
a.index - b.index,
|
|
)
|
|
.map((item) => item.row),
|
|
losses,
|
|
};
|
|
}
|
|
if (stage.type === "group") {
|
|
const key = required(c.key, "Group key"),
|
|
fn = (c.fn ?? "count").toLowerCase(),
|
|
field = c.field ?? "";
|
|
const groups = new Map<
|
|
string,
|
|
{ value: JsonValue; rows: Record<string, JsonValue>[] }
|
|
>();
|
|
for (const row of rows) {
|
|
const value = row[key] ?? null,
|
|
id = stableStringify(value),
|
|
group = groups.get(id) ?? { value, rows: [] };
|
|
group.rows.push(row);
|
|
groups.set(id, group);
|
|
}
|
|
return {
|
|
rows: [...groups.values()].map((group) => {
|
|
const values = group.rows
|
|
.map((row) => row[field])
|
|
.filter(
|
|
(x): x is number => typeof x === "number" && Number.isFinite(x),
|
|
);
|
|
let aggregate: JsonValue;
|
|
if (fn === "count") aggregate = group.rows.length;
|
|
else if (!values.length) {
|
|
aggregate = null;
|
|
losses.push(
|
|
`Group ${String(group.value)} has no numeric ${field} values.`,
|
|
);
|
|
} else if (fn === "sum") aggregate = values.reduce((a, b) => a + b, 0);
|
|
else if (fn === "avg")
|
|
aggregate = values.reduce((a, b) => a + b, 0) / values.length;
|
|
else if (fn === "min") aggregate = Math.min(...values);
|
|
else if (fn === "max") aggregate = Math.max(...values);
|
|
else
|
|
throw new Error(
|
|
"Group function must be count, sum, avg, min or max.",
|
|
);
|
|
return { [key]: group.value, [c.output || fn]: aggregate };
|
|
}),
|
|
losses,
|
|
};
|
|
}
|
|
if (stage.type === "join") {
|
|
const leftKey = required(c.left, "Left join field"),
|
|
rightKey = required(c.right, "Right join field"),
|
|
mode = c.mode === "left" ? "left" : "inner",
|
|
prefix = c.prefix ?? "right_",
|
|
index = new Map<string, Record<string, JsonValue>[]>();
|
|
for (const row of right) {
|
|
const id = stableStringify(row[rightKey] ?? null),
|
|
items = index.get(id) ?? [];
|
|
items.push(row);
|
|
index.set(id, items);
|
|
}
|
|
const joined: Record<string, JsonValue>[] = [];
|
|
for (const row of rows) {
|
|
const matches = index.get(stableStringify(row[leftKey] ?? null)) ?? [];
|
|
if (!matches.length && mode === "left") joined.push({ ...row });
|
|
for (const match of matches) {
|
|
const out = { ...row };
|
|
for (const [key, value] of Object.entries(match))
|
|
out[`${prefix}${key}`] = value;
|
|
joined.push(out);
|
|
if (joined.length > MAX_JOIN)
|
|
throw new Error("Join output exceeds 20,000 rows.");
|
|
}
|
|
}
|
|
return {
|
|
rows: joined,
|
|
losses: [
|
|
mode === "inner"
|
|
? "Unmatched left rows were discarded."
|
|
: "Unmatched left rows were retained; missing right fields are absent.",
|
|
],
|
|
};
|
|
}
|
|
throw new Error(`Unsupported stage: ${stage.type}`);
|
|
}
|
|
export function exportRecipe(stages: Stage[]): string {
|
|
return stableStringify(
|
|
{ schema: "de.add-ideas.flow-tools.recipe.v1", stages },
|
|
2,
|
|
);
|
|
}
|
|
export function importRecipe(source: string): Stage[] {
|
|
const value = safeJsonParse(source, {
|
|
maxTextChars: 256 * 1024,
|
|
maxDepth: 12,
|
|
maxNodes: 10000,
|
|
});
|
|
if (
|
|
!isObject(value) ||
|
|
value.schema !== "de.add-ideas.flow-tools.recipe.v1" ||
|
|
!Array.isArray(value.stages)
|
|
)
|
|
throw new Error("Not a Flow Tools recipe v1.");
|
|
if (value.stages.length > MAX_STAGES)
|
|
throw new Error("Recipe exceeds 30 stages.");
|
|
return value.stages.map((raw, index) => {
|
|
if (
|
|
!isObject(raw) ||
|
|
typeof raw.type !== "string" ||
|
|
![
|
|
"select",
|
|
"rename",
|
|
"filter",
|
|
"map",
|
|
"sort",
|
|
"group",
|
|
"join",
|
|
"format",
|
|
].includes(raw.type) ||
|
|
!raw.config ||
|
|
!isObject(raw.config)
|
|
)
|
|
throw new Error(`Invalid recipe stage ${index + 1}.`);
|
|
const config: Record<string, string> = {};
|
|
for (const [key, item] of Object.entries(raw.config))
|
|
if (typeof item === "string") config[key] = item;
|
|
else throw new Error(`Stage ${index + 1} config values must be strings.`);
|
|
return {
|
|
id: typeof raw.id === "string" ? raw.id : `stage-${index + 1}`,
|
|
type: raw.type as StageType,
|
|
enabled: raw.enabled !== false,
|
|
config,
|
|
};
|
|
});
|
|
}
|
|
function list(value = "") {
|
|
return value
|
|
.split(",")
|
|
.map((x) => x.trim())
|
|
.filter(Boolean);
|
|
}
|
|
function required(value: string | undefined, label: string) {
|
|
if (!value?.trim()) throw new Error(`${label} is required.`);
|
|
return value.trim();
|
|
}
|
|
function parseLiteral(value: string): JsonValue {
|
|
const v = value.trim();
|
|
if (
|
|
(v.startsWith('"') && v.endsWith('"')) ||
|
|
(v.startsWith("'") && v.endsWith("'"))
|
|
)
|
|
return v.slice(1, -1);
|
|
if (/^-?(?:\d+\.?\d*|\.\d+)$/u.test(v)) return Number(v);
|
|
if (/^(true|false)$/iu.test(v)) return v.toLowerCase() === "true";
|
|
if (v === "null") return null;
|
|
return value;
|
|
}
|
|
function test(left: JsonValue | undefined, op: string, right: JsonValue) {
|
|
const c = compare(left, right);
|
|
return op === "="
|
|
? left === right
|
|
: op === "!="
|
|
? left !== right
|
|
: op === ">"
|
|
? c > 0
|
|
: op === ">="
|
|
? c >= 0
|
|
: op === "<"
|
|
? c < 0
|
|
: op === "<="
|
|
? c <= 0
|
|
: op === "contains"
|
|
? String(left ?? "").includes(String(right))
|
|
: false;
|
|
}
|
|
function compare(a: JsonValue | undefined, b: JsonValue | undefined) {
|
|
if (a === b) return 0;
|
|
if (a == null) return -1;
|
|
if (b == null) return 1;
|
|
if (typeof a === "number" && typeof b === "number") return a - b;
|
|
return String(a).localeCompare(String(b), undefined, { numeric: true });
|
|
}
|