File size: 1,845 Bytes
9d33f09 | 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 | #!/usr/bin/env python3
import argparse
import json
from pathlib import Path
from app.models.workflow import WorkflowPackageRequest
from app.services.adapters import adapters
from app.services.intelligence import documentation, package_workflow, quality
from app.services.validation import validator
def load_workflow(path: Path):
return adapters.importers["json"].load(path.read_text(encoding="utf-8"))
def main() -> int:
parser = argparse.ArgumentParser(prog="flowforge", description="Validate and package n8n workflows")
subparsers = parser.add_subparsers(dest="command", required=True)
for command in ("validate", "quality", "docs", "package"):
subparser = subparsers.add_parser(command)
subparser.add_argument("workflow", type=Path)
if command in {"docs", "package"}:
subparser.add_argument("--output", type=Path)
args = parser.parse_args()
workflow = load_workflow(args.workflow)
if args.command == "validate":
result = validator.validate(workflow)
print(result.model_dump_json(indent=2))
return 0 if result.valid else 1
if args.command == "quality":
result = quality(workflow)
print(result.model_dump_json(indent=2))
return 0 if result.overall >= 70 else 1
if args.command == "docs":
content = documentation(workflow).markdown
if args.output:
args.output.write_text(content, encoding="utf-8")
else:
print(content)
return 0
content = package_workflow(WorkflowPackageRequest(workflow=workflow)).model_dump(mode="json")
serialized = json.dumps(content, indent=2)
if args.output:
args.output.write_text(serialized, encoding="utf-8")
else:
print(serialized)
return 0
if __name__ == "__main__":
raise SystemExit(main())
|