From 9f1603ed2f1c3edea57c4cd2650f92ca922b22c2 Mon Sep 17 00:00:00 2001 From: Gaurav Bajpai Date: Thu, 23 Apr 2026 22:44:59 +0530 Subject: [PATCH] feat(bdt): Python parser core + CLI subcommands @W-22196528@ MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds scripts/bdt_analyze.py — a generic, stdlib-only DAG parser and query CLI for Salesforce Data Cloud BDT JSON. 1,379 lines. Parser primitives: - DataTransform.from_path / from_dict — accepts three input shapes: editor export ({version, nodes, ui, ...}), Connect API single-definition ({name, label, type, definition: {...}}), and Connect API multi-definition ({name, definitions: [{name, label, type, definition}, ...]}). - Node dataclass with ui_label / ui_description fallback resolution. - roots() / sinks() / topo_order() — Kahn's algorithm with deterministic tie-break for reproducible output. - upstream() / downstream() traversal resilient to broken refs and cycles (does not infinite-loop on self-edges or cycles). - broken_references(), fields_produced(), fields_consumed(), _scrape_field_refs() heuristics for field-trace discovery. CLI (argparse, 10 subcommands, each supports --json for machine output): - summary — node counts, source/sink counts, stage totals. - sources — list source nodes (no upstream). - outputs — list sink nodes (no downstream). - stages — topologically ordered stages. - nodes — flat node listing with labels. - node — per-node detail (action, inputs, outputs, fields, UI label/description). - lineage — upstream + downstream chain from a node. - field-trace — which nodes produce/consume a given field. - formula — extract formulas/expressions from a node. - definitions — lists definitions in multi-definition payloads; every other subcommand accepts --definition N (default 0) to route into a specific definition within the payload. Error contract: - Exit 0 on success, 2 on unknown node/field, 3 on malformed input. - BdtInputError (exit 3) and BdtNotFoundError (exit 2) classes centralize error handling so the CLI shell stays thin. - Field-trace narrowing refinements prevent false positives from substring matches in formula bodies. - Upstream/downstream walkers harden against broken refs discovered during internal-BDT audit. Design invariants: - Python owns truth (parsing, DAG math, field discovery). LLM owns narrative (explaining what the structure means to a user). - No external dependencies — stdlib only: argparse, json, pathlib, re, sys, collections, dataclasses, typing. - Output size budgets: every subcommand caps its default-mode output so summaries fit in a single LLM context window; --json dumps everything for agents that need raw data. @W-22196528@ --- .../scripts/bdt_analyze.py | 1379 +++++++++++++++++ 1 file changed, 1379 insertions(+) create mode 100644 skills/explaining-batch-data-transform/scripts/bdt_analyze.py diff --git a/skills/explaining-batch-data-transform/scripts/bdt_analyze.py b/skills/explaining-batch-data-transform/scripts/bdt_analyze.py new file mode 100644 index 0000000..919f5ca --- /dev/null +++ b/skills/explaining-batch-data-transform/scripts/bdt_analyze.py @@ -0,0 +1,1379 @@ +#!/usr/bin/env python3 +"""Analyze Salesforce Data Cloud Batch Data Transform (BDT) JSON. + +This module provides a generic typed-DAG parser over BDT JSON definitions. +Grounded against the canonical BDT Connect API InputRepresentation classes +(core-264-public). Design principle: structural facts come from deterministic +graph walks here; narrative explanations are built by the host LLM. + +Usage (CLI lands in a later plan): + python bdt_analyze.py [...] + +Requirements: + - Python 3.9+ (standard library only). +""" +from __future__ import annotations + +import argparse +import json +import pathlib +import re +import sys +from collections import Counter +from dataclasses import dataclass, field +from typing import Any, Dict, List, Mapping, Optional, Set, Tuple + + +# --------- Exit codes --------------------------------------------------------- +EXIT_OK = 0 +EXIT_NOT_FOUND = 2 # unknown node / field +EXIT_BAD_INPUT = 3 # malformed JSON / bad shape / cycle + + +# --------- Exceptions -------------------------------------------------------- +class BdtInputError(Exception): + """Raised on any malformed input. CLI maps this to exit code 3.""" + + +class BdtNotFoundError(Exception): + """Raised when a queried node or field doesn't exist. CLI maps to exit 2.""" + + +@dataclass +class Node: + """One entry in BDT `nodes{}`.""" + name: str + action: str + sources: List[str] = field(default_factory=list) + parameters: Dict[str, Any] = field(default_factory=dict) + schema: Optional[Dict[str, Any]] = None + ui_label: Optional[str] = None + ui_description: Optional[str] = None + + @property + def display_name(self) -> str: + """UI label if present and non-empty; else node name.""" + if self.ui_label: + return f"{self.ui_label} ({self.name})" + return self.name + + +@dataclass +class Definition: + """One parsed STL block from a BDT payload. + + A `DataTransform` always holds at least one of these (length 1 for the + editor-export and single-definition Connect API shapes, length N for the + multi-definition Connect API shape). + """ + index: int + version: Optional[str] + nodes: Dict[str, Node] + ui: Dict[str, Any] + name: Optional[str] = None + label: Optional[str] = None + + +@dataclass +class DataTransform: + """Parsed BDT JSON, with a typed DAG view. + + The parser accepts both the editor-export shape (`nodes` at top level) and + the Connect API create-payload shapes (`definition` or `definitions[]` + wrappers around the editor-export content). When a wrapper is detected the + parser descends into the STL definition(s) transparently; outer metadata + (name, label, type, dataSpaceName) is preserved as optional attributes. + + For multi-definition payloads, every definition is parsed and kept on + `self.definitions`. The currently-selected definition (default index 0) + drives the pass-through properties `nodes`, `version`, and `ui`, so all + existing methods operate on the selected block with no signature changes. + Callers can hop between definitions with `select_definition(index)`. + """ + definitions: List[Definition] + # Outer-wrapper metadata (only populated for Connect API input shapes). + name: Optional[str] = None + label: Optional[str] = None + data_transform_type: Optional[str] = None + data_space_name: Optional[str] = None + # Which definition the accessor properties return; 0 by default. + selected_index: int = 0 + + # --- Backward-compat pass-through properties --------------------------- + # Existing callers (tests + CLI handlers) read `.nodes`, `.version`, `.ui` + # as simple attributes; keep that shape by delegating to the selected + # definition. + + @property + def nodes(self) -> Dict[str, Node]: + return self.definitions[self.selected_index].nodes + + @property + def version(self) -> Optional[str]: + return self.definitions[self.selected_index].version + + @property + def ui(self) -> Dict[str, Any]: + return self.definitions[self.selected_index].ui + + @property + def definition_count(self) -> int: + return len(self.definitions) + + @property + def definition_index(self) -> int: + return self.selected_index + + def select_definition(self, index: int) -> None: + """Pick which definition the accessor properties (and the rest of the + API surface) operate on. Raises `BdtNotFoundError` if `index` is out + of range.""" + if not isinstance(index, int) or index < 0 or index >= len(self.definitions): + raise BdtNotFoundError( + f"Definition index {index} out of range; payload has " + f"{len(self.definitions)} definitions " + f"(indices 0-{len(self.definitions) - 1})." + ) + self.selected_index = index + + @classmethod + def from_path(cls, path: pathlib.Path) -> "DataTransform": + path = pathlib.Path(path) + try: + raw = json.loads(path.read_text()) + except OSError as e: + raise BdtInputError(f"Cannot read {path}: {e}") from e + except json.JSONDecodeError as e: + raise BdtInputError( + f"Invalid JSON in {path} at line {e.lineno}, col {e.colno}: {e.msg}" + ) from e + return cls.from_dict(raw) + + @classmethod + def from_dict(cls, raw: Mapping[str, Any]) -> "DataTransform": + if not isinstance(raw, Mapping): + raise BdtInputError("Top-level BDT JSON must be an object") + + # --- Shape detection ------------------------------------------------- + # 1. Editor-export shape: `nodes` at top level — exactly one definition. + # 2. Connect API single-definition shape: `definition` object with + # `nodes` — one definition, plus outer wrapper metadata. + # 3. Connect API multi-definition shape: non-empty `definitions` array + # whose entries each have `nodes` — N definitions, plus outer + # wrapper metadata. Every definition is parsed. + outer_name: Optional[str] = None + outer_label: Optional[str] = None + outer_type: Optional[str] = None + outer_data_space: Optional[str] = None + raw_definitions: List[Mapping[str, Any]] + + if isinstance(raw.get("nodes"), Mapping): + raw_definitions = [raw] + elif (isinstance(raw.get("definition"), Mapping) + and isinstance(raw["definition"].get("nodes"), Mapping)): + raw_definitions = [raw["definition"]] + outer_name = raw.get("name") if isinstance(raw.get("name"), str) else None + outer_label = raw.get("label") if isinstance(raw.get("label"), str) else None + outer_type = raw.get("type") if isinstance(raw.get("type"), str) else None + outer_data_space = ( + raw.get("dataSpaceName") if isinstance(raw.get("dataSpaceName"), str) else None + ) + elif (isinstance(raw.get("definitions"), list) + and len(raw["definitions"]) > 0 + and all( + isinstance(d, Mapping) and isinstance(d.get("nodes"), Mapping) + for d in raw["definitions"] + )): + raw_definitions = list(raw["definitions"]) + outer_name = raw.get("name") if isinstance(raw.get("name"), str) else None + outer_label = raw.get("label") if isinstance(raw.get("label"), str) else None + outer_type = raw.get("type") if isinstance(raw.get("type"), str) else None + outer_data_space = ( + raw.get("dataSpaceName") if isinstance(raw.get("dataSpaceName"), str) else None + ) + else: + raise BdtInputError( + "Expected 'nodes' at top level, or 'definition.nodes', " + "or 'definitions[].nodes'; none found" + ) + + definitions: List[Definition] = [] + for idx, def_raw in enumerate(raw_definitions): + definitions.append( + cls._parse_definition( + idx, def_raw, + fallback_name=outer_name, + fallback_label=outer_label, + ) + ) + + return cls( + definitions=definitions, + name=outer_name, + label=outer_label, + data_transform_type=outer_type, + data_space_name=outer_data_space, + selected_index=0, + ) + + @classmethod + def _parse_definition( + cls, + index: int, + definition: Mapping[str, Any], + fallback_name: Optional[str] = None, + fallback_label: Optional[str] = None, + ) -> Definition: + """Parse a single STL block into a `Definition`. + + `name`/`label` come from the definition itself when present; otherwise + fall back to the outer wrapper's values (so a single-definition API + payload still carries the transform's name on `definitions[0]`). + """ + nodes_raw = definition.get("nodes") + if not isinstance(nodes_raw, Mapping): + # Defensive: the shape-detection step should keep this out. + raise BdtInputError( + "Expected 'nodes' at top level, or 'definition.nodes', " + "or 'definitions[].nodes'; none found" + ) + ui = definition.get("ui") or {} + ui_nodes = (ui.get("nodes") if isinstance(ui, Mapping) else {}) or {} + + nodes: Dict[str, Node] = {} + for name, body in nodes_raw.items(): + if not isinstance(body, Mapping): + raise BdtInputError(f"Node {name!r} body is not an object") + action = body.get("action") + if not isinstance(action, str) or not action: + raise BdtInputError(f"Node {name!r} missing string 'action'") + srcs = body.get("sources") or [] + if not isinstance(srcs, list): + raise BdtInputError(f"Node {name!r} 'sources' is not a list") + ui_entry = ui_nodes.get(name) or {} + nodes[name] = Node( + name=name, + action=action, + sources=list(srcs), + parameters=dict(body.get("parameters") or {}), + schema=body.get("schema"), + ui_label=(ui_entry.get("label") if isinstance(ui_entry, Mapping) else None) or None, + ui_description=(ui_entry.get("description") if isinstance(ui_entry, Mapping) else None) or None, + ) + + def_name = definition.get("name") if isinstance(definition.get("name"), str) else None + def_label = definition.get("label") if isinstance(definition.get("label"), str) else None + + return Definition( + index=index, + version=definition.get("version"), + nodes=nodes, + ui=dict(ui) if isinstance(ui, Mapping) else {}, + name=def_name or fallback_name, + label=def_label or fallback_label, + ) + + def warnings(self) -> List[str]: + """Non-fatal advisories about the parsed BDT. + + The multi-definition case used to emit an advisory here ("payload + contains N definitions; this skill explains the first only"). With + full multi-definition support that warning is gone; the method is + preserved for future advisories and currently returns an empty list. + """ + return [] + + def roots(self) -> List[str]: + """Node names with no sources (data enters here).""" + return [name for name, n in self.nodes.items() if not n.sources] + + def sinks(self) -> List[str]: + """Node names that are not listed as a source by any other node (data exits here).""" + referenced = set() + for n in self.nodes.values(): + referenced.update(n.sources) + return [name for name in self.nodes if name not in referenced] + + def topo_order(self) -> List[str]: + """Kahn's algorithm. Raises BdtInputError on cycles.""" + # Adjacency: for each node, its forward edges (consumers) and in-degree + in_degree: Dict[str, int] = {name: 0 for name in self.nodes} + forward: Dict[str, List[str]] = {name: [] for name in self.nodes} + for name, n in self.nodes.items(): + for s in n.sources: + if s not in self.nodes: + # Broken reference — counted as in-edge so the dependent node stays blocked + # (we surface broken refs via a separate method; topo should not crash here). + in_degree[name] += 1 + continue + forward[s].append(name) + in_degree[name] += 1 + + # Start with all zero-in-degree nodes in deterministic order (sorted) + ready = sorted([name for name, d in in_degree.items() if d == 0]) + order: List[str] = [] + while ready: + # Pop deterministically so output is stable across runs + ready.sort() + current = ready.pop(0) + order.append(current) + for consumer in sorted(forward[current]): + in_degree[consumer] -= 1 + if in_degree[consumer] == 0: + ready.append(consumer) + + if len(order) != len(self.nodes): + stuck = [n for n in self.nodes if n not in order] + raise BdtInputError( + f"Cycle or broken reference detected; unresolved nodes: {sorted(stuck)}" + ) + return order + + def _topo_order_subset(self, subset: Set[str]) -> List[str]: + """Kahn's algorithm restricted to `subset`. + + Only edges whose endpoints are both in `subset` are considered. This + tolerates broken references that sit outside the queried sub-DAG and + avoids crashing on unrelated cycles. Assumes the sub-DAG itself is + acyclic (caller guarantees this by restricting to a reachable set + via a walk that skips broken refs). + """ + in_degree: Dict[str, int] = {name: 0 for name in subset} + forward: Dict[str, List[str]] = {name: [] for name in subset} + for name in subset: + n = self.nodes[name] + for s in n.sources: + if s in subset: + forward[s].append(name) + in_degree[name] += 1 + ready = sorted([name for name, d in in_degree.items() if d == 0]) + order: List[str] = [] + while ready: + ready.sort() + current = ready.pop(0) + order.append(current) + for consumer in sorted(forward[current]): + in_degree[consumer] -= 1 + if in_degree[consumer] == 0: + ready.append(consumer) + return order + + def upstream(self, node_name: str) -> List[str]: + """All nodes transitively feeding `node_name`, in topological order.""" + if node_name not in self.nodes: + raise BdtNotFoundError( + f"No node named {node_name!r}. Available: " + f"{sorted(list(self.nodes.keys()))[:20]}" + ) + visited: Set[str] = set() + stack = list(self.nodes[node_name].sources) + while stack: + cur = stack.pop() + if cur in visited: + continue + if cur not in self.nodes: + # Broken reference — skip (surfaced via broken_references()) + continue + visited.add(cur) + stack.extend(self.nodes[cur].sources) + # Sort only the reachable sub-DAG (+ the queried node itself), which is + # guaranteed clean. Drop the queried node from the returned list so + # upstream() returns just the ancestors. + sub = visited | {node_name} + topo = self._topo_order_subset(sub) + return [n for n in topo if n in visited] + + def downstream(self, node_name: str) -> List[str]: + """All nodes transitively consuming `node_name`, in topological order.""" + if node_name not in self.nodes: + raise BdtNotFoundError( + f"No node named {node_name!r}. Available: " + f"{sorted(list(self.nodes.keys()))[:20]}" + ) + # Build reverse adjacency on the fly + consumers: Dict[str, List[str]] = {name: [] for name in self.nodes} + for name, n in self.nodes.items(): + for s in n.sources: + if s in consumers: + consumers[s].append(name) + visited: Set[str] = set() + stack = list(consumers[node_name]) + while stack: + cur = stack.pop() + if cur in visited: + continue + visited.add(cur) + stack.extend(consumers[cur]) + # Sort only the reachable sub-DAG; tolerate unrelated broken refs/cycles. + sub = visited | {node_name} + topo = self._topo_order_subset(sub) + return [n for n in topo if n in visited] + + def broken_references(self) -> List[Tuple[str, str]]: + """List of (node_name, missing_source) pairs. Empty if BDT is well-formed.""" + out = [] + for name, n in self.nodes.items(): + for s in n.sources: + if s not in self.nodes: + out.append((name, s)) + return sorted(out) + + +# --------- CLI --------------------------------------------------------------- + + +def _build_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser( + prog="bdt_analyze", + description="Explain and investigate Salesforce Data Cloud BDT JSON.", + ) + sub = p.add_subparsers(dest="subcommand", required=True) + + # Common: --json flag + path positional + --definition selector. + # `--definition N` picks which entry in a multi-definition Connect API + # payload the subcommand operates on. Default 0 (the first definition, + # matching the historical behavior). + def _common(sp, with_definition: bool = True): + sp.add_argument("path", help="Path to BDT JSON file") + sp.add_argument("--json", action="store_true", dest="as_json", + help="Emit machine-readable JSON instead of Markdown") + if with_definition: + sp.add_argument( + "--definition", type=int, default=0, dest="definition", + help=( + "Which definition to operate on in a multi-definition " + "Connect API payload (0-indexed, default 0). Use the " + "'definitions' subcommand to list them." + ), + ) + + for name, help_text in [ + ("summary", "High-level digest: counts, sources, outputs"), + ("sources", "All graph-root nodes with their dataset details"), + ("outputs", "All graph-sink nodes with their target details"), + ("stages", "Topologically ordered list of nodes with action + hint"), + ("nodes", "Every node in topo order with a one-line parameter digest"), + ]: + s = sub.add_parser(name, help=help_text) + _common(s) + if name == "nodes": + s.add_argument("--limit", type=int, default=0, + help="Cap row count (0 = no limit)") + + # `node ` + n = sub.add_parser("node", help="Full detail of one node") + _common(n) + n.add_argument("node_name", help="The node name (key in `nodes` map)") + + # Lineage / tracing subcommands + lg = sub.add_parser("lineage", help="All upstream nodes feeding a target node") + _common(lg) + lg.add_argument("node_name", help="The target node name") + + ft = sub.add_parser("field-trace", help="Trace a field back through the DAG") + _common(ft) + ft.add_argument("field_name", help="Field name (qualified or unqualified)") + + fm = sub.add_parser("formula", help="Focused formula expression + upstream field deps") + _common(fm) + fm.add_argument("node_name", help="A formula or computeRelative node name") + + # `definitions ` — list every definition in a payload. Always + # operates on the whole set, so no --definition flag here. + dfs = sub.add_parser("definitions", + help="List every definition in the payload (one row per definition)") + _common(dfs, with_definition=False) + + return p + + +def main(argv: Optional[List[str]] = None) -> int: + """CLI entry point. Returns exit code.""" + parser = _build_parser() + args = parser.parse_args(argv) + + try: + bdt = DataTransform.from_path(args.path) + except BdtInputError as e: + print(str(e), file=sys.stderr) + return EXIT_BAD_INPUT + + # Select the requested definition (default 0). `definitions` subcommand + # doesn't expose the flag — it always reports on the full list. + definition_idx = getattr(args, "definition", 0) + if definition_idx != 0 or args.subcommand != "definitions": + # Only validate/select when the flag is present (it's missing on + # `definitions`, which always walks the full list). The 0-default + # also goes through here for consistency; `select_definition(0)` is + # always safe given we guarantee at least one definition exists. + try: + bdt.select_definition(definition_idx) + except BdtNotFoundError as e: + # Spec: out-of-range `--definition` exits 2 with a message that + # points the user at the `definitions` subcommand. + msg = ( + f"Definition index {definition_idx} out of range; payload " + f"has {bdt.definition_count} definitions " + f"(indices 0-{bdt.definition_count - 1}). " + f"Use 'definitions ' to list them." + ) + # Preserve the underlying error type for completeness, but surface + # the user-facing message on stderr. + _ = e + print(msg, file=sys.stderr) + return EXIT_NOT_FOUND + + dispatch = { + "summary": cmd_summary, + "sources": cmd_sources, + "outputs": cmd_outputs, + "stages": cmd_stages, + "nodes": cmd_nodes, + "node": cmd_node, + "lineage": cmd_lineage, + "field-trace": cmd_field_trace, + "formula": cmd_formula, + "definitions": cmd_definitions, + } + handler = dispatch.get(args.subcommand) + if handler is None: + print(f"Unknown subcommand: {args.subcommand!r}", file=sys.stderr) + return EXIT_BAD_INPUT + try: + output = handler(bdt, args) + except BdtNotFoundError as e: + print(str(e), file=sys.stderr) + return EXIT_NOT_FOUND + except BdtInputError as e: + print(str(e), file=sys.stderr) + return EXIT_BAD_INPUT + print(output) + return EXIT_OK + + +# --------- Field-discovery helpers ------------------------------------------ + +# Matches an unquoted identifier reasonable as a field name: optional qualifier +# (e.g., "SalesOrder.ssot__Id__c") followed by the identifier itself. +_FIELD_NAME_RE = re.compile(r'\b(?:[A-Za-z_][A-Za-z0-9_]*\.)?[A-Za-z_][A-Za-z0-9_]*\b') + +# Reserved identifiers we exclude when scraping a formula expression for field refs. +_SQL_RESERVED = { + "case", "when", "then", "else", "end", "and", "or", "not", "null", "true", "false", + "if", "is", "in", "like", "between", "cast", "as", + # Common SFSQL functions — callers can add more if needed. + "abs", "ceiling", "exp", "floor", "log", "max", "min", "mod", "power", "round", "sqrt", "trunc", + "concat", "contains", "ends", "length", "lower", "ltrim", "rtrim", "substitute", "substr", + "text", "trim", "upper", "uuid", "value", "begins", + "adddays", "addmonths", "datediff", "datetimevalue", "datevalue", "day", "monthdiff", + "now", "today", "weekday", + "blankvalue", "isblank", "isnull", "nullvalue", + "sequence", "explode", + "row_number", "rank", "dense_rank", "lag", "lead", "first_value", "last_value", + "nth_value", "ntile", "percent_rank", "cume_dist", + "coalesce", + # Type names + "number", "text", "boolean", "date", "datetime", +} + + +def fields_produced(node: "Node") -> List[str]: + """Which field names this node emits into its output stream. + + Heuristic per action type. Unknown actions fall back to [] (caller must cope). + """ + p = node.parameters if isinstance(node.parameters, dict) else {} + if node.action == "load": + return list(p.get("fields") or []) + if node.action in ("formula", "computeRelative"): + return [f.get("name") for f in (p.get("fields") or []) + if isinstance(f, dict) and f.get("name")] + if node.action == "outputD360": + return [m.get("targetField") for m in (p.get("fieldsMappings") or []) + if isinstance(m, dict) and m.get("targetField")] + if node.action == "schema": + # Schema nodes rename/drop; renamed names come from slice.fields[].newProperties.name + fs = (p.get("fields") or []) + out = [] + for f in fs: + if isinstance(f, dict): + np_ = f.get("newProperties") or {} + out.append(np_.get("name") or f.get("name")) + return [x for x in out if x] + if node.action == "aggregate": + # Aggregations add their output names; groupings pass through as-is + out = list(p.get("groupings") or []) + for a in (p.get("aggregations") or []): + if isinstance(a, dict) and a.get("name"): + out.append(a["name"]) + return out + # Default: we can't guess — caller must inspect parameters directly. + return [] + + +def fields_consumed(node: "Node") -> List[str]: + """Which upstream field names this node reads. + + For pass-through nodes we return [] — callers that need "what flows through + this node" should look at `fields_produced` on the source(s). + """ + p = node.parameters if isinstance(node.parameters, dict) else {} + if node.action == "load": + return [] + if node.action == "join": + out = list(p.get("leftKeys") or []) + list(p.get("rightKeys") or []) + return out + if node.action == "filter": + return [e.get("field") for e in (p.get("filterExpressions") or []) + if isinstance(e, dict) and e.get("field")] + if node.action in ("formula", "computeRelative"): + consumed = set() + for f in (p.get("fields") or []): + if isinstance(f, dict): + expr = f.get("formulaExpression") or "" + consumed.update(_scrape_field_refs(expr)) + # computeRelative also consumes partitionBy + orderBy fields + if node.action == "computeRelative": + consumed.update(p.get("partitionBy") or []) + for ob in (p.get("orderBy") or []): + if isinstance(ob, dict) and ob.get("fieldName"): + consumed.add(ob["fieldName"]) + return sorted(consumed) + if node.action == "aggregate": + out = list(p.get("groupings") or []) + for a in (p.get("aggregations") or []): + if isinstance(a, dict) and a.get("source"): + out.append(a["source"]) + # Also hierarchical fields + for f_key in ("selfField", "parentField", "percentageField"): + v = p.get(f_key) + if v: + out.append(v) + return out + if node.action == "outputD360": + return [m.get("sourceField") for m in (p.get("fieldsMappings") or []) + if isinstance(m, dict) and m.get("sourceField")] + if node.action == "schema": + return [f.get("name") for f in (p.get("fields") or []) + if isinstance(f, dict) and f.get("name")] + return [] + + +def _scrape_field_refs(expression: str) -> List[str]: + """Pull likely field identifiers out of a SQL-ish formula expression. + + This is a heuristic: it returns identifiers that aren't reserved keywords or + function names. Callers should still treat results as a *candidate* set, not + a verified one. In particular, we also pick up quoted identifiers. + """ + if not isinstance(expression, str): + return [] + # Also capture "Quoted.Name" style identifiers + quoted = re.findall(r'"([A-Za-z_][\w.]*)"', expression) + bare = _FIELD_NAME_RE.findall(expression) + candidates = set(quoted) | set(bare) + # Drop reserved tokens (case-insensitive) and numeric-looking tokens + out = [] + for c in candidates: + last = c.rsplit(".", 1)[-1] + if last.lower() in _SQL_RESERVED: + continue + if last.isdigit(): + continue + out.append(c) + # Deduplicate while preserving order + seen = set() + dedup = [] + for c in out: + if c not in seen: + seen.add(c) + dedup.append(c) + return dedup + + +def _field_specific_upstream_deps(node: "Node", field: str) -> List[str]: + """Deps relevant to `field` specifically, not all deps the node consumes. + + `fields_consumed(node)` gives the union of every field the node reads + across *all* of its outputs. For field-trace narration we want a much + tighter set: only the upstream fields that feed *this one* field. This + keeps the trace output small and focused, which matters for real BDTs + where a single `outputD360` may have 40+ mappings or a `formula` node + may define a dozen unrelated output fields. + """ + p = node.parameters if isinstance(node.parameters, dict) else {} + action = node.action + if action in ("formula", "computeRelative"): + out: List[str] = [] + for f in (p.get("fields") or []): + if isinstance(f, dict) and f.get("name") == field: + out.extend(_scrape_field_refs(f.get("formulaExpression") or "")) + break + if action == "computeRelative": + out.extend(p.get("partitionBy") or []) + for ob in (p.get("orderBy") or []): + if isinstance(ob, dict) and ob.get("fieldName"): + out.append(ob["fieldName"]) + # Deduplicate, preserve order + seen: set = set() + dedup: List[str] = [] + for x in out: + if x not in seen: + seen.add(x) + dedup.append(x) + return dedup + if action == "outputD360": + for m in (p.get("fieldsMappings") or []): + if isinstance(m, dict) and m.get("targetField") == field: + src = m.get("sourceField") + return [src] if src else [] + return [] + if action == "aggregate": + # If the field is one of the aggregations, return just its source. + for a in (p.get("aggregations") or []): + if isinstance(a, dict) and a.get("name") == field: + src = a.get("source") + return [src] if src else [] + # If it's one of the groupings (pass-through), the dep is itself. + if field in (p.get("groupings") or []): + return [field] + return [] + if action == "schema": + # Renames: if this node renamed some X to `field`, the dep is X. + for f in (p.get("fields") or []): + if isinstance(f, dict): + np_ = f.get("newProperties") or {} + if np_.get("name") == field: + return [f.get("name")] if f.get("name") else [] + return [] + if action == "load": + return [] + # Fallback for unknown / unhandled actions — reuse the generic consumed set + return fields_consumed(node) + + +def cmd_definitions(bdt: "DataTransform", args) -> str: + """List every definition in the payload — one row per definition. + + Safe to call on editor-export or single-definition inputs: they always + have exactly one row. On multi-definition payloads this is the entry + point for the SKILL.md flow that offers the user a choice. + """ + rows = [] + for d in bdt.definitions: + rows.append({ + "index": d.index, + "name": d.name, + "label": d.label, + "version": d.version, + "node_count": len(d.nodes), + }) + + if getattr(args, "as_json", False): + return json.dumps(rows, indent=2) + + lines = [f"# Definitions ({len(rows)})", ""] + lines.append("| Index | Name | Label | Version | Nodes |") + lines.append("|---|---|---|---|---|") + for r in rows: + lines.append( + f"| {r['index']} | " + f"{r['name'] or '(no name)'} | " + f"{r['label'] or '(no label)'} | " + f"{r['version'] or '(unknown)'} | " + f"{r['node_count']} |" + ) + return "\n".join(lines) + + +def cmd_summary(bdt: "DataTransform", args) -> str: + """High-level digest of the BDT.""" + action_counts = Counter(n.action for n in bdt.nodes.values()) + roots = bdt.roots() + sinks = bdt.sinks() + + if getattr(args, "as_json", False): + data = { + "version": bdt.version, + "total_nodes": len(bdt.nodes), + "action_counts": dict(action_counts), + "sources": [_source_digest(bdt, r) for r in roots], + "outputs": [_output_digest(bdt, s) for s in sinks], + } + return json.dumps(data, indent=2) + + # Markdown + lines = [] + lines.append("# BDT Summary") + lines.append(f"Version: {bdt.version or '(unknown)'}") + lines.append(f"Total nodes: {len(bdt.nodes)}") + if action_counts: + counts_str = ", ".join(f"{a}: {c}" for a, c in action_counts.most_common()) + lines.append(f"By action: {counts_str}") + lines.append("") + if roots: + lines.append(f"## Sources ({len(roots)} nodes)") + lines.append("| Node | Label | Dataset | Type | Field count |") + lines.append("|---|---|---|---|---|") + for r in roots: + d = _source_digest(bdt, r) + lines.append( + f"| {r} | {d['label'] or '(no label)'} | " + f"{d['dataset_name'] or '(none)'} | " + f"{d['dataset_type'] or '(none)'} | " + f"{d['field_count']} |" + ) + lines.append("") + if sinks: + lines.append(f"## Outputs ({len(sinks)} nodes)") + lines.append("| Node | Label | Target | Type | Mapping count |") + lines.append("|---|---|---|---|---|") + for s in sinks: + d = _output_digest(bdt, s) + lines.append( + f"| {s} | {d['label'] or '(no label)'} | " + f"{d['target_name'] or '(none)'} | " + f"{d['target_type'] or '(none)'} | " + f"{d['mapping_count']} |" + ) + return "\n".join(lines) + + +def _source_digest(bdt: "DataTransform", name: str) -> dict: + n = bdt.nodes[name] + ds = (n.parameters.get("dataset") or {}) if isinstance(n.parameters, dict) else {} + fields = n.parameters.get("fields") or [] if isinstance(n.parameters, dict) else [] + return { + "name": name, + "label": n.ui_label, + "dataset_name": ds.get("name") if isinstance(ds, dict) else None, + "dataset_type": ds.get("type") if isinstance(ds, dict) else None, + "field_count": len(fields) if isinstance(fields, list) else 0, + } + + +def _output_digest(bdt: "DataTransform", name: str) -> dict: + n = bdt.nodes[name] + p = n.parameters if isinstance(n.parameters, dict) else {} + mappings = p.get("fieldsMappings") or [] + return { + "name": name, + "label": n.ui_label, + "target_name": p.get("name"), + "target_type": p.get("type"), + "write_mode": p.get("writeMode"), + "mapping_count": len(mappings) if isinstance(mappings, list) else 0, + } + + +def cmd_sources(bdt: "DataTransform", args) -> str: + roots = bdt.roots() + enriched = [] + for r in roots: + n = bdt.nodes[r] + p = n.parameters if isinstance(n.parameters, dict) else {} + ds = p.get("dataset") or {} + enriched.append({ + "name": r, + "label": n.ui_label, + "dataset_name": ds.get("name") if isinstance(ds, dict) else None, + "dataset_type": ds.get("type") if isinstance(ds, dict) else None, + "fields": list(p.get("fields") or []), + "action": n.action, + }) + if getattr(args, "as_json", False): + return json.dumps(enriched, indent=2) + lines = [f"# Sources ({len(enriched)} node(s))", ""] + for e in enriched: + lines.append(f"## {e['name']}" + (f" — {e['label']}" if e['label'] else "")) + lines.append(f"- action: `{e['action']}`") + lines.append(f"- dataset: `{e['dataset_name']}` ({e['dataset_type']})") + if e['fields']: + lines.append(f"- fields ({len(e['fields'])}): " + ", ".join(f"`{f}`" for f in e['fields'])) + else: + lines.append("- fields: _(none declared)_") + lines.append("") + return "\n".join(lines).rstrip() + "\n" + + +def cmd_outputs(bdt: "DataTransform", args) -> str: + sinks = bdt.sinks() + enriched = [] + for s in sinks: + n = bdt.nodes[s] + p = n.parameters if isinstance(n.parameters, dict) else {} + mappings = p.get("fieldsMappings") or [] + enriched.append({ + "name": s, + "label": n.ui_label, + "action": n.action, + "target_name": p.get("name"), + "target_type": p.get("type"), + "write_mode": p.get("writeMode"), + "dedup_order": list(p.get("dedupOrder") or []), + "mappings": [ + { + "source": m.get("sourceField") if isinstance(m, dict) else None, + "target": m.get("targetField") if isinstance(m, dict) else None, + } + for m in (mappings if isinstance(mappings, list) else []) + ], + }) + if getattr(args, "as_json", False): + return json.dumps(enriched, indent=2) + lines = [f"# Outputs ({len(enriched)} node(s))", ""] + for e in enriched: + lines.append(f"## {e['name']}" + (f" — {e['label']}" if e['label'] else "")) + lines.append(f"- action: `{e['action']}`") + lines.append(f"- target: `{e['target_name']}` ({e['target_type']})") + lines.append(f"- writeMode: `{e['write_mode']}`") + if e['dedup_order']: + lines.append(f"- dedupOrder: {e['dedup_order']}") + if e['mappings']: + lines.append("") + lines.append(f"### Field mappings ({len(e['mappings'])})") + lines.append("| Source field | Target field |") + lines.append("|---|---|") + for m in e['mappings']: + lines.append(f"| `{m['source']}` | `{m['target']}` |") + lines.append("") + return "\n".join(lines).rstrip() + "\n" + + +# One-line purpose hint per known action. Unknown actions fall back to the raw action name. +_ACTION_HINT = { + "load": "Load data from a DMO or DLO", + "filter": "Keep rows matching filter criteria", + "sqlFilter": "Keep rows matching a raw SQL predicate", + "join": "Join two upstream streams", + "formula": "Add derived columns via per-row formulas", + "computeRelative": "Compute a window function over partitioned/ordered rows", + "aggregate": "Group-by aggregation", + "extractGrains": "Fan out rows across time grains", + "extractTable": "Extract a table from flattened JSON", + "schema": "Rename/drop/reshape columns", + "outputD360": "Write rows to a target DMO/DLO", + "appendV2": "Union rows from multiple upstream streams", + "flatten": "Flatten nested structure", + "flattenJson": "Flatten a JSON field into rows/columns", + "split": "Split rows across multiple outputs", + "typeCast": "Cast field types", + "update": "Update records", + "bucket": "Bucket field values", + "formatDate": "Reformat date values", + "extension": "Run custom extension logic", + "extensionFunction":"Run custom extension function", + "cdpPredict": "Apply a prediction model", + "jsonAggregate": "Aggregate JSON fields", + "save": "Save intermediate result", + "recommendation": "Apply a recommendation model", +} + + +def cmd_stages(bdt: "DataTransform", args) -> str: + order = bdt.topo_order() + rows = [] + for name in order: + n = bdt.nodes[name] + hint = _ACTION_HINT.get(n.action, f"({n.action} — not in known catalog)") + rows.append({ + "name": name, + "label": n.ui_label, + "action": n.action, + "sources": list(n.sources), + "hint": hint, + }) + if getattr(args, "as_json", False): + return json.dumps(rows, indent=2) + lines = [f"# Stages ({len(rows)} node(s), topological order)", ""] + for i, r in enumerate(rows, 1): + srcs = ", ".join(f"`{s}`" for s in r["sources"]) if r["sources"] else "_(root)_" + label_part = f" — {r['label']}" if r['label'] else "" + lines.append(f"{i}. **{r['name']}**{label_part} `{r['action']}`") + lines.append(f" - sources: {srcs}") + lines.append(f" - {r['hint']}") + return "\n".join(lines) + + +def cmd_nodes(bdt: "DataTransform", args) -> str: + order = bdt.topo_order() + limit = getattr(args, "limit", 0) or 0 + limited = limit > 0 and len(order) > limit + shown = order[:limit] if limited else order + + rows = [] + for name in shown: + n = bdt.nodes[name] + rows.append({ + "name": name, + "label": n.ui_label, + "action": n.action, + "sources": list(n.sources), + "param_digest": _param_digest(n), + }) + + if getattr(args, "as_json", False): + return json.dumps({"total": len(order), "shown": len(rows), "nodes": rows}, indent=2) + + lines = [f"# Nodes ({len(order)} total)"] + if limited: + lines.append(f"_Output limited to {limit} of {len(order)} — re-run without --limit for full list._") + lines.append("") + lines.append("| Node | Label | Action | Sources | Digest |") + lines.append("|---|---|---|---|---|") + for r in rows: + srcs = ", ".join(r["sources"]) if r["sources"] else "_(root)_" + label = r["label"] or "" + digest = r["param_digest"].replace("|", "\\|") + lines.append(f"| {r['name']} | {label} | {r['action']} | {srcs} | {digest} |") + return "\n".join(lines) + + +def _param_digest(n: "Node") -> str: + """One-line Markdown-safe parameter summary. Never exceeds ~120 chars.""" + p = n.parameters if isinstance(n.parameters, dict) else {} + if n.action == "load": + ds = p.get("dataset") or {} + ds_name = ds.get("name") if isinstance(ds, dict) else "?" + fc = len(p.get("fields") or []) + return f"dataset=`{ds_name}`, {fc} field(s)" + if n.action == "join": + return (f"type=`{p.get('joinType')}` " + f"leftKeys={p.get('leftKeys')} rightKeys={p.get('rightKeys')} " + f"rightQualifier=`{p.get('rightQualifier')}`") + if n.action == "filter": + exprs = p.get("filterExpressions") or [] + return f"{len(exprs)} expression(s), logic=`{p.get('filterBooleanLogic') or 'all AND'}`" + if n.action in ("formula", "computeRelative"): + fs = p.get("fields") or [] + names = ", ".join(f.get("name", "?") for f in fs if isinstance(f, dict)) + extra = "" + if n.action == "computeRelative": + extra = f" partitionBy={p.get('partitionBy')} orderBy={p.get('orderBy')}" + return f"{len(fs)} output field(s): {names}{extra}" + if n.action == "aggregate": + aggs = p.get("aggregations") or [] + groupings = p.get("groupings") or [] + return f"{len(aggs)} aggregation(s), groupBy={groupings}" + if n.action == "outputD360": + mc = len(p.get("fieldsMappings") or []) + return f"target=`{p.get('name')}` ({p.get('type')}) writeMode=`{p.get('writeMode')}` {mc} mapping(s)" + if n.action == "schema": + slc = p.get("slice") or {} + return f"slice.mode=`{slc.get('mode')}` {len(slc.get('fields') or [])} field(s)" + # Fallback: show param keys only + keys = list(p.keys())[:5] + return f"parameters keys: {keys}" + + +def cmd_node(bdt: "DataTransform", args) -> str: + name = args.node_name + if name not in bdt.nodes: + available = sorted(bdt.nodes.keys())[:20] + raise BdtNotFoundError( + f"No node named {name!r}. Available (first 20): {available}" + ) + n = bdt.nodes[name] + # Compute consumers (nodes that list this one as a source) + consumers = [m for m, other in bdt.nodes.items() if name in other.sources] + + if getattr(args, "as_json", False): + return json.dumps({ + "name": n.name, + "label": n.ui_label, + "description": n.ui_description, + "action": n.action, + "sources": n.sources, + "consumers": consumers, + "parameters": n.parameters, + "schema": n.schema, + }, indent=2) + + lines = [f"# {n.name}" + (f" — {n.ui_label}" if n.ui_label else "")] + if n.ui_description: + lines.append(f"> {n.ui_description}") + lines.append("") + lines.append(f"- **action:** `{n.action}`") + lines.append(f"- **hint:** {_ACTION_HINT.get(n.action, '(not in known catalog)')}") + lines.append(f"- **sources:** {n.sources or '_(root)_'}") + lines.append(f"- **direct consumers:** {consumers or '_(sink — no downstream)_'}") + lines.append("") + lines.append("## Parameters") + lines.append("```json") + lines.append(json.dumps(n.parameters, indent=2, default=str)) + lines.append("```") + if n.schema: + lines.append("") + lines.append("## Schema block") + lines.append("```json") + lines.append(json.dumps(n.schema, indent=2, default=str)) + lines.append("```") + return "\n".join(lines) + + +def cmd_lineage(bdt: "DataTransform", args) -> str: + target = args.node_name + if target not in bdt.nodes: + available = sorted(bdt.nodes.keys())[:20] + raise BdtNotFoundError( + f"No node named {target!r}. Available (first 20): {available}" + ) + upstream_names = bdt.upstream(target) # topo-ordered + + rows = [] + for name in upstream_names: + n = bdt.nodes[name] + rows.append({ + "name": name, + "action": n.action, + "label": n.ui_label, + "sources": list(n.sources), + }) + + if getattr(args, "as_json", False): + return json.dumps({"target": target, "upstream": rows}, indent=2) + + lines = [f"# Lineage of {target}"] + t = bdt.nodes[target] + if t.ui_label: + lines.append(f"_{t.ui_label}_") + lines.append("") + if not upstream_names: + lines.append(f"`{target}` is a graph root (no upstream).") + return "\n".join(lines) + lines.append(f"{len(upstream_names)} upstream node(s), topological order:") + lines.append("") + for r in rows: + arrow = " ← " if r["sources"] else " " + label_part = f" — {r['label']}" if r['label'] else "" + lines.append(f"{arrow}**{r['name']}** `{r['action']}`{label_part}") + if r['sources']: + srcs = ", ".join(f"`{s}`" for s in r['sources']) + lines.append(f" (fed by {srcs})") + lines.append("") + lines.append(f" → **{target}** `{t.action}`" + (f" — {t.ui_label}" if t.ui_label else "")) + return "\n".join(lines) + + +def cmd_field_trace(bdt: "DataTransform", args) -> str: + field = args.field_name + # Cache topo order once — `bdt.topo_order()` walks the whole graph. + topo = bdt.topo_order() + # 1. Find all nodes that "produce" this field (per fields_produced heuristic). + definers = [] + for name in topo: + n = bdt.nodes[name] + produced = fields_produced(n) + if field in produced: + definers.append(n) + + # 2. If nothing matched, also try matching the unqualified part (drop any "Qualifier." prefix) + if not definers: + unq = field.rsplit(".", 1)[-1] + for name in topo: + n = bdt.nodes[name] + produced = fields_produced(n) + if unq in produced: + definers.append(n) + field = unq # report with the unqualified name we actually matched + + if not definers: + available_sample = [] + for name in list(bdt.nodes.keys())[:10]: + available_sample.extend(fields_produced(bdt.nodes[name])) + raise BdtNotFoundError( + f"Field {args.field_name!r} is not defined by any node in this BDT. " + f"Sample of visible field names (from the first 10 nodes): " + f"{sorted(set(available_sample))[:20]}" + ) + + # 3. For each definer, compute: (a) the formula or mapping that defines it, if any; + # (b) its direct upstream field dependencies (`fields_consumed`). + def _definition_of(n: "Node", fname: str) -> dict: + p = n.parameters if isinstance(n.parameters, dict) else {} + detail = {"kind": n.action} + if n.action in ("formula", "computeRelative"): + for f in (p.get("fields") or []): + if isinstance(f, dict) and f.get("name") == fname: + detail["formulaExpression"] = f.get("formulaExpression") + detail["type"] = f.get("type") + detail["label"] = f.get("label") + break + elif n.action == "outputD360": + for m in (p.get("fieldsMappings") or []): + if isinstance(m, dict) and m.get("targetField") == fname: + detail["sourceField"] = m.get("sourceField") + break + elif n.action == "load": + ds = p.get("dataset") or {} + detail["dataset"] = {"name": ds.get("name"), "type": ds.get("type")} + elif n.action == "aggregate": + for a in (p.get("aggregations") or []): + if isinstance(a, dict) and a.get("name") == fname: + detail["aggregation"] = { + "action": a.get("action"), + "source": a.get("source"), + } + break + elif n.action == "schema": + for f in (p.get("fields") or []): + if isinstance(f, dict): + np_ = f.get("newProperties") or {} + if (np_.get("name") or f.get("name")) == fname: + detail["renamed_from"] = f.get("name") + detail["renamed_to"] = np_.get("name") + break + return detail + + result = [] + for n in definers: + entry = { + "node": n.name, + "label": n.ui_label, + "action": n.action, + "definition": _definition_of(n, field), + # Narrowed to deps of *this specific field*, not every field the + # node consumes. For output/formula/aggregate nodes with many + # unrelated output fields this dramatically reduces trace size + # and noise. The broader node-level context is still available + # via `upstream_chain` below. + "upstream_deps": _field_specific_upstream_deps(n, field), + "sources": list(n.sources), + } + result.append(entry) + + # Upstream node chain for the first definer (for narration convenience) + chain = [] + if definers: + chain = bdt.upstream(definers[0].name) + [definers[0].name] + + if getattr(args, "as_json", False): + return json.dumps({ + "field": field, + "definitions": result, + "upstream_chain": chain, + }, indent=2) + + lines = [f"# Field trace: `{field}`"] + if len(definers) > 1: + lines.append(f"> Field is defined in {len(definers)} places (first in topological order listed first).") + lines.append("") + for i, entry in enumerate(result, 1): + header = f"## {i}. Defined by {entry['node']}" + if entry['label']: + header += f" — {entry['label']}" + lines.append(header) + lines.append(f"- action: `{entry['action']}`") + d = entry['definition'] + if d.get("formulaExpression"): + lines.append(f"- formula: `{d['formulaExpression']}`") + if d.get("type"): + lines.append(f"- type: `{d['type']}`") + if d.get("sourceField"): + lines.append(f"- sourced from: `{d['sourceField']}`") + if d.get("dataset"): + lines.append(f"- loaded from: `{d['dataset']['name']}` ({d['dataset']['type']})") + if d.get("aggregation"): + a = d['aggregation'] + lines.append(f"- aggregation: `{a['action']}` over `{a['source']}`") + if d.get("renamed_from"): + lines.append(f"- renamed: `{d['renamed_from']}` → `{d['renamed_to']}`") + if entry['upstream_deps']: + lines.append(f"- upstream field deps: {entry['upstream_deps']}") + lines.append(f"- upstream nodes: {entry['sources'] or '_(root)_'}") + lines.append("") + if chain: + lines.append("## Upstream node chain") + lines.append(" → ".join(f"`{c}`" for c in chain)) + return "\n".join(lines) + + +def cmd_formula(bdt: "DataTransform", args) -> str: + name = args.node_name + if name not in bdt.nodes: + raise BdtNotFoundError( + f"No node named {name!r}. Available (first 20): " + f"{sorted(bdt.nodes.keys())[:20]}" + ) + n = bdt.nodes[name] + if n.action not in ("formula", "computeRelative"): + raise BdtInputError( + f"Node {name!r} has action {n.action!r}; `formula` subcommand " + f"only applies to `formula` and `computeRelative` nodes. " + f"Use `node {name}` instead." + ) + p = n.parameters if isinstance(n.parameters, dict) else {} + fields_out = p.get("fields") or [] + consumed = fields_consumed(n) + + # For each consumed field, find the node that defines it (if any) — give + # the LLM ready context for I2-style reasoning. Hoist `bdt.upstream(name)` + # out of the loop — it's a pure function of `name` and walks the graph. + upstream_names = bdt.upstream(name) + upstream_defs = {} + for f in consumed: + # Prefer exact match; fall back to unqualified + matches = [] + for up_name in upstream_names: + up = bdt.nodes[up_name] + prod = fields_produced(up) + if f in prod or f.rsplit(".", 1)[-1] in prod: + matches.append({ + "node": up_name, + "action": up.action, + "label": up.ui_label, + }) + if matches: + upstream_defs[f] = matches[-1] # closest definer (last in topo order) + + payload = { + "node": name, + "label": n.ui_label, + "action": n.action, + "formula_fields": [ + { + "name": f.get("name"), + "label": f.get("label"), + "type": f.get("type"), + "formulaExpression": f.get("formulaExpression"), + } + for f in fields_out if isinstance(f, dict) + ], + "consumed_fields": consumed, + "upstream_field_definitions": upstream_defs, + } + if n.action == "computeRelative": + payload["partitionBy"] = list(p.get("partitionBy") or []) + payload["orderBy"] = p.get("orderBy") or [] + + if getattr(args, "as_json", False): + return json.dumps(payload, indent=2) + + lines = [f"# Formula extract: {name}" + (f" — {n.ui_label}" if n.ui_label else "")] + lines.append(f"- action: `{n.action}`") + if n.action == "computeRelative": + lines.append(f"- partitionBy: {payload['partitionBy']}") + lines.append(f"- orderBy: {payload['orderBy']}") + lines.append("") + lines.append("## Output fields") + for f in payload["formula_fields"]: + lines.append(f"### `{f['name']}`" + (f" _{f['label']}_" if f['label'] else "")) + lines.append(f"- type: `{f['type']}`") + lines.append(f"- expression: `{f['formulaExpression']}`") + lines.append("") + lines.append("## Upstream field dependencies") + if not consumed: + lines.append("_(none detected — expression may reference only literals)_") + else: + for f in consumed: + defn = upstream_defs.get(f) + if defn: + label = f" _{defn['label']}_" if defn.get('label') else "" + lines.append(f"- `{f}` ← **{defn['node']}** (`{defn['action']}`){label}") + else: + lines.append(f"- `{f}` ← (no upstream definition found — may be a literal or a passthrough)") + return "\n".join(lines) + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:]))