backend/scripts/generate_openapi.py

#!/usr/bin/env python3
"""Generate OpenAPI JSON from Drogon routes plus a manual OpenAPI YAML file."""

from __future__ import annotations

import argparse
import copy
import json
import re
from pathlib import Path
from typing import Any


ROUTE_RE = re.compile(
    r"ADD_METHOD_TO\(\s*([A-Za-z_]\w*)::([A-Za-z_]\w*)\s*,\s*"
    r'"([^"]+)"\s*,\s*drogon::([A-Za-z]+)\s*\)',
    re.MULTILINE | re.DOTALL,
)

METHODS = {"get", "post", "put", "patch", "delete"}


def controller_tag(controller: str) -> str:
    return controller.removesuffix("Controller")


def operation_id(controller: str, handler: str) -> str:
    return f"{controller}.{handler}"


def openapi_path(path: str) -> tuple[str, list[dict[str, Any]]]:
    parameters: list[dict[str, Any]] = []

    def replace(match: re.Match[str]) -> str:
        index = int(match.group(1))
        name = "id" if index == 1 else f"param{index}"
        parameters.append(
            {
                "name": name,
                "in": "path",
                "required": True,
                "schema": {"type": "integer", "format": "int64"},
            }
        )
        return "{" + name + "}"

    return re.sub(r"\{(\d+)\}", replace, path), parameters


def generated_operation(
    controller: str,
    handler: str,
    parameters: list[dict[str, Any]],
) -> dict[str, Any]:
    return {
        "tags": [controller_tag(controller)],
        "operationId": operation_id(controller, handler),
        "summary": handler_to_summary(handler),
        "parameters": parameters,
        "responses": {"200": {"description": "Response"}},
    }


def handler_to_summary(handler: str) -> str:
    words = re.sub(r"([a-z0-9])([A-Z])", r"\1 \2", handler).replace("_", " ")
    return words[:1].upper() + words[1:]


def parse_routes(controllers_dir: Path) -> dict[tuple[str, str], dict[str, Any]]:
    operations: dict[tuple[str, str], dict[str, Any]] = {}
    for header in sorted(controllers_dir.glob("*.h")):
        text = header.read_text(encoding="utf-8")
        for controller, handler, raw_path, drogon_method in ROUTE_RE.findall(text):
            method = drogon_method.lower()
            path, parameters = openapi_path(raw_path)
            key = (path, method)
            if key in operations:
                raise ValueError(f"duplicate route for {method.upper()} {path}")
            operations[key] = generated_operation(controller, handler, parameters)
    return operations


def load_manual(path: Path | None) -> dict[str, Any]:
    if path is None:
        return {}
    text = path.read_text(encoding="utf-8")
    try:
        import yaml  # type: ignore[import-untyped]
    except ImportError as error:
        try:
            data = json.loads(text)
        except json.JSONDecodeError:
            raise SystemExit(
                "PyYAML is required unless the manual OpenAPI YAML uses "
                "JSON-compatible syntax"
            ) from error
    else:
        data = yaml.safe_load(text)

    if data is None:
        return {}
    if not isinstance(data, dict):
        raise ValueError(f"{path}: OpenAPI document must be an object")
    return data


def merge_openapi(
    generated: dict[tuple[str, str], dict[str, Any]],
    manual: dict[str, Any],
    title: str,
    version: str,
    server: str,
) -> dict[str, Any]:
    spec: dict[str, Any] = {
        "openapi": manual.get("openapi", "3.0.3"),
        "info": copy.deepcopy(
            manual.get("info", {"title": title, "version": version})
        ),
        "servers": copy.deepcopy(manual.get("servers", [{"url": server}])),
        "paths": {},
    }

    for key in ("components", "tags", "externalDocs"):
        if key in manual:
            spec[key] = copy.deepcopy(manual[key])

    manual_paths = manual.get("paths", {})
    if not isinstance(manual_paths, dict):
        raise ValueError("manual OpenAPI paths must be an object")

    seen_manual: set[tuple[str, str]] = set()
    for path, path_item in manual_paths.items():
        if not isinstance(path_item, dict):
            raise ValueError(f"{path}: path item must be an object")
        for method, manual_operation in path_item.items():
            method_lower = method.lower()
            if method_lower not in METHODS:
                continue
            key = (path, method_lower)
            seen_manual.add(key)
            if key not in generated and not manual_operation.get("x-manual-only", False):
                raise ValueError(
                    f"{method_upper(method_lower)} {path}: manual operation has no matching Drogon route"
                )

    for key, operation in sorted(generated.items()):
        path, method = key
        manual_operation = manual_paths.get(path, {}).get(method, {})
        merged = copy.deepcopy(operation)
        merged.update(copy.deepcopy(manual_operation))
        merged.pop("x-manual-only", None)
        spec["paths"].setdefault(path, {})[method] = merged

    for key in sorted(seen_manual - generated.keys()):
        path, method = key
        manual_operation = copy.deepcopy(manual_paths[path][method])
        if not manual_operation.get("x-manual-only", False):
            continue
        manual_operation.pop("x-manual-only", None)
        spec["paths"].setdefault(path, {})[method] = manual_operation

    return spec


def method_upper(method: str) -> str:
    return method.upper()


def generate(
    controllers_dir: Path,
    manual_path: Path | None,
    title: str,
    version: str,
    server: str,
) -> dict[str, Any]:
    generated = parse_routes(controllers_dir)
    manual = load_manual(manual_path)
    return merge_openapi(generated, manual, title, version, server)


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--controllers", type=Path, required=True)
    parser.add_argument("--manual", type=Path)
    parser.add_argument("--out", type=Path, required=True)
    parser.add_argument("--title", default="Visione API")
    parser.add_argument("--version", default="0.1.0")
    parser.add_argument("--server", default="http://localhost:5555")
    args = parser.parse_args()

    spec = generate(
        controllers_dir=args.controllers,
        manual_path=args.manual,
        title=args.title,
        version=args.version,
        server=args.server,
    )
    args.out.parent.mkdir(parents=True, exist_ok=True)
    args.out.write_text(
        json.dumps(spec, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )


if __name__ == "__main__":
    main()