zhifu-platform/scripts/export_openapi.py

import argparse
import json
from pathlib import Path

from zhifu.app import create_app
from zhifu.modules.knowledge.inference import app as inference_app
from zhifu.monitoring.api import app as operations_app
from zhifu.oms.api import create_app as create_oms

ROOT = Path(__file__).resolve().parents[1]
OUTPUT = ROOT / "docs" / "api"
METHODS = {"get", "post", "put", "patch", "delete", "head", "options"}


def specifications():
    # 只创建应用与 OpenAPI 定义，不启动 lifespan，也不连接数据库或加载模型权重。
    return {
        "openapi.json": create_app().openapi(),
        "oms-openapi.json": create_oms().openapi(),
        "inference-openapi.json": inference_app.openapi(),
        "operations-openapi.json": operations_app.openapi(),
    }


def catalog(documents):
    lines = [
        "# 接口目录",
        "",
        "由接口定义生成，更新命令：`uv run python scripts/export_openapi.py`。",
        "",
    ]
    for filename, document in documents.items():
        lines.extend(
            [
                f"## {document['info']['title']}",
                "",
                f"字段、参数、响应及错误说明见 [{filename}]({filename})。",
                "",
                "| 方法 | 路径 | 用途 | 认证 | 成功状态 |",
                "|---|---|---|---|---|",
            ]
        )
        for path, methods in document["paths"].items():
            for method, operation in methods.items():
                if method not in METHODS:
                    continue
                success = ", ".join(code for code in operation["responses"] if code.startswith("2"))
                schemes = {
                    scheme
                    for requirement in operation.get("security", [])
                    for scheme in requirement
                }
                # 渠道使用各自签名协议，目录与 OpenAPI security 保持一致。
                security = (
                    "Standard Webhooks" if "StandardWebhooks" in schemes
                    else "HMAC" if "ChannelHMAC" in schemes
                    else "Bearer" if schemes else "无"
                )
                lines.append(
                    f"| {method.upper()} | `{path}` | {operation['summary']} "
                    f"| {security} | {success} |"
                )
        lines.append("")
    return "\n".join(lines)


def main(check=False):
    documents = specifications()
    artifacts = {
        name: json.dumps(document, ensure_ascii=False, indent=2) + "\n"
        for name, document in documents.items()
    }
    artifacts["routes.md"] = catalog(documents)
    for name, content in artifacts.items():
        destination = OUTPUT / name
        if check:
            if destination.read_text(encoding="utf-8") != content:
                raise RuntimeError(f"接口文档需要更新：{name}")
        else:
            destination.write_text(content, encoding="utf-8")
        print(f"{'checked' if check else 'exported'}: {name}")
    counts = {
        name: sum(method in METHODS for methods in document["paths"].values() for method in methods)
        for name, document in documents.items()
    }
    print(json.dumps(counts))


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="导出或核对四类服务的接口文档")
    parser.add_argument("--check", action="store_true")
    main(parser.parse_args().check)