DataOps-platform/scripts/generate_openapi.py

#!/usr/bin/env python3
"""Generate the current Flask route inventory as an OpenAPI 3.1 contract."""

from __future__ import annotations

import ast
import argparse
import json
import re
from collections import defaultdict
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
OUTPUT = ROOT / "docs" / "architecture" / "OPENAPI.yaml"
PREFIXES = {
    "meta_data": "/api/meta",
    "data_interface": "/api/interface",
    "graph": "/api/graph",
    "system": "/api/system",
    "data_source": "/api/datasource",
    "data_flow": "/api/dataflow",
    "business_domain": "/api/bd",
    "data_factory": "/api/datafactory",
    "data_service": "/api/dataservice",
}


def quoted(value: str) -> str:
    return json.dumps(value, ensure_ascii=False)


def route_path(raw_path: str) -> tuple[str, list[dict[str, str]]]:
    parameters: list[dict[str, str]] = []

    def replace(match: re.Match[str]) -> str:
        converter = match.group(1) or "string"
        name = match.group(2)
        schema_type = "integer" if converter in {"int", "float"} else "string"
        parameters.append({"name": name, "type": schema_type})
        return "{" + name + "}"

    return re.sub(r"<(?:([a-zA-Z_]+):)?([a-zA-Z_][a-zA-Z0-9_]*)>", replace, raw_path), parameters


def extract_routes() -> list[dict[str, object]]:
    routes: list[dict[str, object]] = []
    for module, prefix in PREFIXES.items():
      for route_file in sorted((ROOT / "app" / "api" / module).glob("*.py")):
        if route_file.name == "__init__.py":
            continue
        tree = ast.parse(route_file.read_text(encoding="utf-8"))
        for node in tree.body:
            if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                continue
            for decorator in node.decorator_list:
                if not isinstance(decorator, ast.Call):
                    continue
                function = decorator.func
                if not (
                    isinstance(function, ast.Attribute)
                    and function.attr == "route"
                    and isinstance(function.value, ast.Name)
                    and function.value.id == "bp"
                    and decorator.args
                ):
                    continue
                raw_path = ast.literal_eval(decorator.args[0])
                methods = ["GET"]
                for keyword in decorator.keywords:
                    if keyword.arg == "methods":
                        methods = ast.literal_eval(keyword.value)
                path, parameters = route_path(prefix + raw_path)
                summary = (ast.get_docstring(node) or node.name.replace("_", " ")).splitlines()[0]
                for method in methods:
                    routes.append(
                        {
                            "path": path,
                            "method": method.lower(),
                            "tag": module,
                            "operation_id": f"{module}_{node.name}_{method.lower()}",
                            "summary": summary,
                            "parameters": parameters,
                        }
                    )
    return sorted(routes, key=lambda item: (str(item["path"]), str(item["method"])))


def render(routes: list[dict[str, object]]) -> str:
    by_path: dict[str, list[dict[str, object]]] = defaultdict(list)
    for route in routes:
        by_path[str(route["path"])].append(route)

    lines = [
        "openapi: 3.1.0",
        "info:",
        '  title: "DataOps Platform API（当前代码基线）"',
        '  version: "2026-07-16"',
        '  description: "由 scripts/generate_openapi.py 从 app/api/*/routes.py 生成。请求体与响应细节仍以现有专项 API 文档和代码为准。"',
        f"x-route-count: {len(routes)}",
        "servers:",
        '  - url: "http://localhost:15500"',
        '    description: "全本地隔离测试后端"',
        "tags:",
    ]
    for tag in PREFIXES:
        lines.extend([f"  - name: {tag}", f"    description: {quoted(PREFIXES[tag])}"])
    lines.append("paths:")

    for path, operations in sorted(by_path.items()):
        lines.append(f"  {quoted(path)}:")
        for operation in operations:
            lines.extend(
                [
                    f"    {operation['method']}:",
                    f"      tags: [{operation['tag']}]",
                    f"      operationId: {operation['operation_id']}",
                    f"      summary: {quoted(str(operation['summary']))}",
                    f"      x-source: {quoted('app/api/' + str(operation['tag']) + '/routes.py')}",
                ]
            )
            parameters = operation["parameters"]
            if parameters:
                lines.append("      parameters:")
                for parameter in parameters:
                    lines.extend(
                        [
                            f"        - name: {parameter['name']}",
                            "          in: path",
                            "          required: true",
                            "          schema:",
                            f"            type: {parameter['type']}",
                        ]
                    )
            if operation["method"] in {"post", "put", "patch"}:
                lines.extend(
                    [
                        "      requestBody:",
                        "        required: false",
                        "        content:",
                        "          application/json:",
                        "            schema:",
                        "              type: object",
                        "              additionalProperties: true",
                    ]
                )
            lines.extend(
                [
                    "      responses:",
                    '        "200":',
                    '          description: "请求已由当前实现处理"',
                    "          content:",
                    "            application/json:",
                    "              schema:",
                    "                $ref: '#/components/schemas/ApiEnvelope'",
                    "        default:",
                    '          description: "错误响应"',
                    "          content:",
                    "            application/json:",
                    "              schema:",
                    "                $ref: '#/components/schemas/ApiEnvelope'",
                ]
            )

    lines.extend(
        [
            "components:",
            "  schemas:",
            "    ApiEnvelope:",
            "      type: object",
            "      additionalProperties: true",
            "      properties:",
            "        success:",
            "          type: boolean",
            "        code:",
            "          type: integer",
            "        message:",
            "          type: string",
            "        data: {}",
            "        timestamp:",
            "          type: integer",
            "  securitySchemes:",
            "    bearerAuth:",
            "      type: http",
            "      scheme: bearer",
            "      bearerFormat: JWT",
            "      description: \"下一阶段统一认证方案；当前路由尚未全部接入。\"",
        ]
    )
    return "\n".join(lines) + "\n"


def main(output: Path = OUTPUT) -> None:
    routes = extract_routes()
    if not routes:
        raise SystemExit("No Flask routes found")
    output.parent.mkdir(parents=True, exist_ok=True)
    output.write_text(render(routes), encoding="utf-8")
    try:
        label = output.relative_to(ROOT)
    except ValueError:
        label = output
    print(f"Generated {label} with {len(routes)} operations")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--output", type=Path, default=OUTPUT)
    args = parser.parse_args()
    main(args.output)