mirror of
https://github.com/Such-Software/adaptive-multi-protocol-gateway.git
synced 2026-10-17 15:41:41 +02:00
102 lines
3.4 KiB
Python
102 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Generate an AMPG route manifest from a small app route catalog."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
from typing import Any
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from ampg.route_manifest import ROUTE_MANIFEST_SCHEMA, validate_route_manifest # noqa: E402
|
|
|
|
|
|
PASSTHROUGH_FIELDS = ("tier", "identity", "payments", "realtime", "public_allowed")
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("catalog", type=Path, help="Input app route catalog JSON.")
|
|
parser.add_argument("output", type=Path, help="Output ampg.route-manifest.v1 JSON.")
|
|
parser.add_argument(
|
|
"--check",
|
|
action="store_true",
|
|
help="Fail if output differs instead of writing it.",
|
|
)
|
|
args = parser.parse_args(argv)
|
|
|
|
try:
|
|
catalog = _load_json(args.catalog)
|
|
manifest = route_manifest_from_catalog(catalog)
|
|
content = json.dumps(manifest, indent=2, sort_keys=True) + "\n"
|
|
if args.check:
|
|
existing = args.output.read_text(encoding="utf-8")
|
|
if existing != content:
|
|
print(
|
|
f"ROUTE_MANIFEST_GENERATE status=stale output={args.output}",
|
|
file=sys.stderr,
|
|
)
|
|
return 1
|
|
else:
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.write_text(content, encoding="utf-8")
|
|
print(
|
|
"ROUTE_MANIFEST_GENERATE "
|
|
f"status=ok catalog={args.catalog} output={args.output} routes={len(manifest['routes'])}"
|
|
)
|
|
return 0
|
|
except Exception as exc: # noqa: BLE001 - CLI should print concise failures.
|
|
print(f"ROUTE_MANIFEST_GENERATE status=error message=\"{exc}\"", file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
def route_manifest_from_catalog(catalog: dict[str, Any]) -> dict[str, Any]:
|
|
if not isinstance(catalog, dict):
|
|
raise ValueError("catalog must be a JSON object")
|
|
routes = catalog.get("routes")
|
|
if not isinstance(routes, list):
|
|
raise ValueError("catalog.routes must be an array")
|
|
|
|
manifest: dict[str, Any] = {
|
|
"schema": ROUTE_MANIFEST_SCHEMA,
|
|
"routes": [_route_policy(route, index) for index, route in enumerate(routes)],
|
|
}
|
|
if "default_tier" in catalog:
|
|
manifest["default_tier"] = catalog["default_tier"]
|
|
if "deny_routes" in catalog:
|
|
manifest["deny_routes"] = catalog["deny_routes"]
|
|
|
|
issues = validate_route_manifest(manifest)
|
|
if issues:
|
|
joined = "; ".join(f"{issue.path}: {issue.message}" for issue in issues)
|
|
raise ValueError(f"generated route manifest is invalid: {joined}")
|
|
return manifest
|
|
|
|
|
|
def _route_policy(route: Any, index: int) -> dict[str, Any]:
|
|
if not isinstance(route, dict):
|
|
raise ValueError(f"catalog.routes[{index}] must be an object")
|
|
path = route.get("path", route.get("match"))
|
|
if not isinstance(path, str) or not path:
|
|
raise ValueError(f"catalog.routes[{index}].path must be a non-empty string")
|
|
|
|
policy: dict[str, Any] = {"match": path}
|
|
for field in PASSTHROUGH_FIELDS:
|
|
if field in route:
|
|
policy[field] = route[field]
|
|
return policy
|
|
|
|
|
|
def _load_json(path: Path) -> Any:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|