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