| 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) |
|
|
| |
| 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()) |
|
|
|
|
