mirror of
https://github.com/johndoe6345789/AutoMetabuilder.git
synced 2026-04-25 06:15:01 +00:00
90 lines
3.0 KiB
Python
90 lines
3.0 KiB
Python
"""Build a node/edge view of n8n workflows for visualization."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from typing import Any, Dict, Iterable, List
|
|
|
|
from autometabuilder.data import get_workflow_content, load_metadata
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _parse_workflow_definition() -> Dict[str, Any]:
|
|
payload = get_workflow_content()
|
|
if not payload:
|
|
return {"name": "Empty", "nodes": [], "connections": {}}
|
|
try:
|
|
parsed = json.loads(payload)
|
|
except json.JSONDecodeError as exc:
|
|
logger.warning("Invalid workflow JSON: %s", exc)
|
|
return {"name": "Invalid", "nodes": [], "connections": {}}
|
|
return parsed if isinstance(parsed, dict) else {"name": "Invalid", "nodes": [], "connections": {}}
|
|
|
|
|
|
def _gather_n8n_nodes(
|
|
nodes: Iterable[Dict[str, Any]],
|
|
plugin_map: Dict[str, Any]
|
|
) -> List[Dict[str, Any]]:
|
|
"""Extract nodes from n8n format."""
|
|
collected = []
|
|
for node in nodes:
|
|
node_id = node.get("id", node.get("name", f"node-{len(collected)}"))
|
|
node_type = node.get("type", "unknown")
|
|
metadata = plugin_map.get(node_type, {})
|
|
|
|
collected.append({
|
|
"id": node_id,
|
|
"name": node.get("name", node_id),
|
|
"type": node_type,
|
|
"label_key": metadata.get("label"),
|
|
"parent": None,
|
|
"position": node.get("position", [0, 0]),
|
|
})
|
|
return collected
|
|
|
|
|
|
def _build_n8n_edges(
|
|
connections: Dict[str, Any],
|
|
nodes: List[Dict[str, Any]]
|
|
) -> List[Dict[str, str]]:
|
|
"""Build edges from n8n connections."""
|
|
# Build name to ID mapping
|
|
name_to_id = {node["name"]: node["id"] for node in nodes}
|
|
|
|
edges = []
|
|
for source_name, outputs in connections.items():
|
|
source_id = name_to_id.get(source_name, source_name)
|
|
|
|
for output_type, indices in outputs.items():
|
|
for index, targets in indices.items():
|
|
for target in targets:
|
|
target_name = target["node"]
|
|
target_id = name_to_id.get(target_name, target_name)
|
|
|
|
edges.append({
|
|
"from": source_id,
|
|
"to": target_id,
|
|
"type": target.get("type", "main"),
|
|
"output_index": index,
|
|
"input_index": target.get("index", 0),
|
|
})
|
|
return edges
|
|
|
|
|
|
def build_workflow_graph() -> Dict[str, Any]:
|
|
"""Build workflow graph from n8n format (breaking change: legacy format removed)."""
|
|
definition = _parse_workflow_definition()
|
|
plugin_map = load_metadata().get("workflow_plugins", {})
|
|
|
|
# Only support n8n format now
|
|
nodes = _gather_n8n_nodes(definition.get("nodes", []), plugin_map)
|
|
edges = _build_n8n_edges(definition.get("connections", {}), nodes)
|
|
|
|
logger.debug("Built workflow graph with %d nodes and %d edges", len(nodes), len(edges))
|
|
return {
|
|
"nodes": nodes,
|
|
"edges": edges,
|
|
"count": {"nodes": len(nodes), "edges": len(edges)},
|
|
}
|