File size: 2,751 Bytes
35bdde1 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 |
import argparse
import sys
from typing import Any
import yaml
def fail(msg: str) -> None:
print(f"ERROR: {msg}", file=sys.stderr)
raise SystemExit(1)
def expect_type(value: Any, t: type, label: str) -> None:
if not isinstance(value, t):
fail(f"{label} must be {t.__name__}, got {type(value).__name__}")
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--path", required=True)
args = ap.parse_args()
with open(args.path, "r", encoding="utf-8") as f:
doc = yaml.safe_load(f)
expect_type(doc, dict, "root")
if doc.get("kind") != "app":
fail("root.kind must be 'app'")
version = doc.get("version")
if not isinstance(version, (str, int, float)):
fail("root.version must be a scalar")
app = doc.get("app")
expect_type(app, dict, "root.app")
if app.get("mode") != "workflow":
fail("app.mode must be 'workflow'")
if not app.get("name"):
fail("app.name is required")
workflow = doc.get("workflow")
expect_type(workflow, dict, "root.workflow")
graph = workflow.get("graph")
expect_type(graph, dict, "workflow.graph")
nodes = graph.get("nodes")
edges = graph.get("edges")
expect_type(nodes, list, "workflow.graph.nodes")
expect_type(edges, list, "workflow.graph.edges")
if not nodes:
fail("workflow.graph.nodes must be non-empty")
node_ids: set[str] = set()
node_types: set[str] = set()
for i, n in enumerate(nodes):
expect_type(n, dict, f"node[{i}]")
nid = n.get("id")
if not isinstance(nid, str) or not nid:
fail(f"node[{i}].id must be non-empty string")
if nid in node_ids:
fail(f"duplicate node id: {nid}")
node_ids.add(nid)
data = n.get("data") or {}
if isinstance(data, dict):
ntype = data.get("type")
if isinstance(ntype, str):
node_types.add(ntype)
# Minimal sanity: ensure a workflow has start/end nodes.
if "start" not in node_types:
fail("no start node found (node.data.type == 'start')")
if "end" not in node_types:
fail("no end node found (node.data.type == 'end')")
for i, e in enumerate(edges):
expect_type(e, dict, f"edge[{i}]")
src = e.get("source")
tgt = e.get("target")
if src not in node_ids:
fail(f"edge[{i}].source references unknown node: {src}")
if tgt not in node_ids:
fail(f"edge[{i}].target references unknown node: {tgt}")
print(f"OK: {args.path} looks like a Dify app DSL (kind=app, mode=workflow, nodes={len(nodes)}, edges={len(edges)}).")
return 0
if __name__ == "__main__":
raise SystemExit(main())
|