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