backend/scripts/generate_openapi.py

#!/usr/bin/env python3
"""Generate a compact OpenAPI spec from Drogon route macros and metadata blocks."""

from __future__ import annotations

import argparse
import json
import re
from pathlib import Path


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_WITH_JSON_BODY = {"post", "put", "patch"}
REQUIRED_METADATA = {"tags", "summary", "description", "auth", "body", "response"}


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


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

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

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


def parse_metadata(header: Path, text: str, route_start: int) -> dict[str, str]:
    prefix = text[:route_start].rstrip()
    if not prefix.endswith("*/"):
        raise ValueError(f"{header}: route is missing an openapi metadata block")

    comment_start = prefix.rfind("/*")
    if comment_start == -1:
        raise ValueError(f"{header}: route is missing an openapi metadata block")

    block = prefix[comment_start + 2:-2]
    metadata: dict[str, str] = {}
    in_openapi_block = False

    for raw_line in block.splitlines():
        line = raw_line.strip()
        if line.startswith("*"):
            line = line[1:].strip()
        if not line:
            continue
        if line == "openapi:":
            in_openapi_block = True
            continue
        if not in_openapi_block:
            continue

        match = re.match(r"([A-Za-z_]+):\s*(.+)", line)
        if match:
            metadata[match.group(1)] = match.group(2).strip()

    missing = sorted(REQUIRED_METADATA - metadata.keys())
    if missing:
        names = ", ".join(missing)
        raise ValueError(f"{header}: openapi metadata block is missing: {names}")

    return metadata


def request_body(body_kind: str, method: str) -> dict | None:
    if body_kind == "none":
        return None
    if body_kind == "multipart":
        return {
            "required": True,
            "content": {
                "multipart/form-data": {
                    "schema": {
                        "type": "object",
                        "properties": {
                            "files": {
                                "type": "array",
                                "items": {"type": "string", "format": "binary"},
                            }
                        },
                    }
                }
            },
        }
    if body_kind == "json" and method in METHODS_WITH_JSON_BODY:
        return {
            "required": True,
            "content": {
                "application/json": {"schema": {"type": "object"}}
            },
        }
    if body_kind == "json":
        raise ValueError(f"JSON body is not supported for {method.upper()} routes")
    raise ValueError(f"Unsupported request body kind: {body_kind}")


def parse_tags(value: str) -> list[str]:
    return [tag.strip() for tag in value.split(",") if tag.strip()]


def is_public(metadata: dict[str, str]) -> bool:
    auth = metadata["auth"]
    if auth == "public":
        return True
    if auth == "bearer":
        return False
    raise ValueError(f"Unsupported auth kind: {auth}")


def operation(
    controller: str,
    handler: str,
    raw_path: str,
    method: str,
    metadata: dict[str, str],
) -> dict:
    path, params = openapi_path(raw_path)
    op = {
        "tags": parse_tags(metadata["tags"]) or [controller_tag(controller)],
        "operationId": f"{controller}.{handler}",
        "summary": metadata["summary"],
        "description": metadata["description"],
        "parameters": params,
        "responses": {
            "200": {
                "description": metadata["response"],
                "content": {"application/json": {"schema": {"type": "object"}}},
            }
        },
    }
    body = request_body(metadata["body"], method)
    if body:
        op["requestBody"] = body
    if not is_public(metadata):
        op["security"] = [{"bearerAuth": []}]
    if method == "post":
        op["responses"]["201"] = op["responses"]["200"]
    return path, op


def generate(controllers_dir: Path, title: str, version: str) -> dict:
    spec = {
        "openapi": "3.0.3",
        "info": {
            "title": title,
            "version": version,
            "description": "Generated from Drogon controller route macros.",
        },
        "servers": [{"url": "http://localhost:5555"}],
        "components": {
            "securitySchemes": {
                "bearerAuth": {"type": "http", "scheme": "bearer"}
            }
        },
        "paths": {},
    }
    for header in sorted(controllers_dir.glob("*.h")):
        text = header.read_text(encoding="utf-8")
        for match in ROUTE_RE.finditer(text):
            controller, handler, raw_path, drogon_method = match.groups()
            metadata = parse_metadata(header, text, match.start())
            method = drogon_method.lower()
            path, op = operation(controller, handler, raw_path, method, metadata)
            if "/photos" in path:
                raise ValueError(f"{header}: generated OpenAPI path still uses /photos")
            spec["paths"].setdefault(path, {})[method] = op
    return spec


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

    spec = generate(args.controllers, args.title, args.version)
    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()