Từ API spec đến structured metadata: nguồn sự thật cho code generation

# Download spec
curl https://api.example.com/openapi.json -o openapi.json
 
# Hoặc từ GitHub
curl https://raw.githubusercontent.com/example/api/main/openapi.yaml -o openapi.yaml

{
  "methods": [
    {
      "id": "user.list",
      "resource": "user",
      "action": "list",
      "cli_command": "list-users",
      "http_method": "GET",
      "path": "/v1/users",
      "summary": "List all users",
      "description": "Returns a paginated list of users. Supports filtering by role.",
      "parameters": [
        {
          "name": "role",
          "in": "query",
          "type": "string",
          "required": false,
          "enum": ["admin", "member", "viewer"],
          "description": "Filter by user role"
        },
        {
          "name": "limit",
          "in": "query",
          "type": "integer",
          "required": false,
          "default": 100,
          "description": "Maximum number of results"
        }
      ],
      "returns": {
        "type": "array",
        "items": {
          "$ref": "#/definitions/User"
        }
      },
      "paginated": true,
      "idempotent": true,
      "requires_auth": true,
      "tags": ["users", "read"]
    }
  ],
  "definitions": {
    "User": {
      "type": "object",
      "properties": {
        "id": {"type": "integer"},
        "name": {"type": "string"},
        "email": {"type": "string", "format": "email"},
        "role": {"type": "string", "enum": ["admin", "member", "viewer"]},
        "created_at": {"type": "string", "format": "date-time"}
      },
      "required": ["id", "name", "email"]
    }
  },
  "meta": {
    "api_version": "v1",
    "base_url": "https://api.example.com",
    "auth_type": "bearer",
    "extracted_at": "2026-03-17T10:00:00Z",
    "total_methods": 439
  }
}

# tools/extract_openapi.py
import json
import yaml
import re
from pathlib import Path
from datetime import datetime, timezone
 
 
def load_spec(path: str) -> dict:
    """Load OpenAPI spec từ YAML hoặc JSON."""
    content = Path(path).read_text()
    if path.endswith(".yaml") or path.endswith(".yml"):
        return yaml.safe_load(content)
    return json.loads(content)
 
 
def path_to_resource_action(path: str, method: str) -> tuple[str, str]:
    """
    Convert OpenAPI path + HTTP method → (resource, action).
 
    /users         GET  → (user, list)
    /users         POST → (user, create)
    /users/{id}    GET  → (user, get)
    /users/{id}    PUT  → (user, update)
    /users/{id}    DELETE → (user, delete)
    """
    # Remove path params for analysis
    clean = re.sub(r"/\{[^}]+\}", "", path).strip("/")
    parts = clean.split("/")
    resource = parts[-1].rstrip("s")  # Singularize: users → user
 
    method_map = {
        ("get", False): "list",    # GET /users → list
        ("get", True): "get",      # GET /users/{id} → get
        ("post", False): "create",
        ("put", True): "update",
        ("patch", True): "update",
        ("delete", True): "delete",
    }
    has_id = bool(re.search(r"/\{[^}]+\}$", path))
    action = method_map.get((method.lower(), has_id), method.lower())
 
    return resource, action
 
 
def extract_parameters(operation: dict, path_params: list) -> list[dict]:
    """Extract parameters từ operation."""
    params = []
 
    # Path parameters
    for p in path_params:
        params.append({
            "name": p.get("name"),
            "in": "path",
            "type": p.get("schema", {}).get("type", "string"),
            "required": True,
            "description": p.get("description", ""),
        })
 
    # Query/header parameters
    for p in operation.get("parameters", []):
        schema = p.get("schema", {})
        param = {
            "name": p.get("name"),
            "in": p.get("in", "query"),
            "type": schema.get("type", "string"),
            "required": p.get("required", False),
            "description": p.get("description", ""),
        }
        if "enum" in schema:
            param["enum"] = schema["enum"]
        if "default" in schema:
            param["default"] = schema["default"]
        params.append(param)
 
    # Request body parameters (POST/PUT)
    body = operation.get("requestBody", {})
    if body:
        content = body.get("content", {}).get("application/json", {})
        body_schema = content.get("schema", {})
        for prop_name, prop_schema in body_schema.get("properties", {}).items():
            required_props = body_schema.get("required", [])
            param = {
                "name": prop_name,
                "in": "body",
                "type": prop_schema.get("type", "string"),
                "required": prop_name in required_props,
                "description": prop_schema.get("description", ""),
            }
            if "enum" in prop_schema:
                param["enum"] = prop_schema["enum"]
            params.append(param)
 
    return params
 
 
def action_to_cli_command(resource: str, action: str) -> str:
    """(user, list) → list-users"""
    if action == "list":
        return f"list-{resource}s"
    if action == "create":
        return f"create-{resource}"
    if action == "get":
        return f"get-{resource}"
    if action == "update":
        return f"update-{resource}"
    if action == "delete":
        return f"delete-{resource}"
    return f"{action}-{resource}"
 
 
def extract_from_openapi(spec_path: str, output_path: str):
    """Main extractor: OpenAPI spec → metadata JSON."""
    spec = load_spec(spec_path)
    methods = []
    definitions = {}
 
    # Extract component schemas
    components = spec.get("components", {})
    for name, schema in components.get("schemas", {}).items():
        definitions[name] = schema
 
    # Extract paths
    for path, path_item in spec.get("paths", {}).items():
        path_params = path_item.get("parameters", [])
 
        for http_method, operation in path_item.items():
            if http_method in ("parameters", "summary", "description"):
                continue
            if not isinstance(operation, dict):
                continue
 
            resource, action = path_to_resource_action(path, http_method)
            method_id = f"{resource}.{action}"
            cli_command = action_to_cli_command(resource, action)
 
            # Detect pagination
            response_200 = operation.get("responses", {}).get("200", {})
            response_content = response_200.get("content", {}).get("application/json", {})
            response_schema = response_content.get("schema", {})
            paginated = "has_next_page" in str(response_schema) or "next_cursor" in str(response_schema)
 
            # Detect idempotency
            idempotent = http_method.lower() in ("get", "put") or action in ("list", "get", "update")
 
            method = {
                "id": method_id,
                "resource": resource,
                "action": action,
                "cli_command": cli_command,
                "http_method": http_method.upper(),
                "path": path,
                "summary": operation.get("summary", ""),
                "description": operation.get("description", ""),
                "parameters": extract_parameters(operation, path_params),
                "paginated": paginated,
                "idempotent": idempotent,
                "requires_auth": bool(operation.get("security") or spec.get("security")),
                "tags": operation.get("tags", [resource]),
            }
            methods.append(method)
 
    metadata = {
        "methods": methods,
        "definitions": definitions,
        "meta": {
            "api_version": spec.get("info", {}).get("version", "v1"),
            "base_url": next(
                (s.get("url", "") for s in spec.get("servers", [])), ""
            ),
            "auth_type": "bearer",
            "extracted_at": datetime.now(timezone.utc).isoformat(),
            "total_methods": len(methods),
        }
    }
 
    Path(output_path).write_text(json.dumps(metadata, indent=2))
    print(f"✓ Extracted {len(methods)} methods → {output_path}")
 
 
if __name__ == "__main__":
    import sys
    extract_from_openapi(sys.argv[1], sys.argv[2])

python tools/extract_openapi.py openapi.yaml metadata/api-metadata.json
# ✓ Extracted 439 methods → metadata/api-metadata.json

# tools/extract_sdk.py
import ast
import json
import inspect
import importlib
from pathlib import Path
from datetime import datetime, timezone
 
 
def extract_type_annotation(annotation) -> str:
    """Convert Python type annotation → JSON type string."""
    if annotation is None:
        return "string"
    ann_str = str(annotation)
    type_map = {
        "int": "integer", "float": "number", "bool": "boolean",
        "str": "string", "list": "array", "dict": "object",
        "None": "null",
    }
    for py_type, json_type in type_map.items():
        if py_type in ann_str:
            return json_type
    return "string"
 
 
def extract_methods_from_class(cls) -> list[dict]:
    """Extract public methods từ một class dùng inspect."""
    methods = []
 
    for name, method in inspect.getmembers(cls, predicate=inspect.isfunction):
        if name.startswith("_"):
            continue
 
        sig = inspect.signature(method)
        docstring = inspect.getdoc(method) or ""
 
        # Parse parameters từ signature
        params = []
        for param_name, param in sig.parameters.items():
            if param_name in ("self", "cls"):
                continue
 
            param_info = {
                "name": param_name,
                "in": "body",
                "type": extract_type_annotation(param.annotation),
                "required": param.default is inspect.Parameter.empty,
                "description": "",
            }
            params.append(param_info)
 
        # Infer resource và action từ method name
        # list_users → resource=user, action=list
        parts = name.split("_", 1)
        if len(parts) == 2:
            action, resource = parts[0], parts[1].rstrip("s")
        else:
            action, resource = name, "resource"
 
        # Infer CLI command
        cli_command = f"{action}-{resource}s" if action == "list" else f"{action}-{resource}"
 
        method_info = {
            "id": f"{resource}.{action}",
            "resource": resource,
            "action": action,
            "cli_command": cli_command.replace("_", "-"),
            "http_method": "GET" if action in ("list", "get") else "POST",
            "path": f"/{resource}s",
            "summary": docstring.split("\n")[0] if docstring else name,
            "description": docstring,
            "parameters": params,
            "paginated": action == "list",
            "idempotent": action in ("list", "get"),
            "requires_auth": True,
            "tags": [resource],
        }
        methods.append(method_info)
 
    return methods
 
 
def extract_from_sdk(module_name: str, class_name: str, output_path: str):
    """Main extractor: Python SDK class → metadata JSON."""
    module = importlib.import_module(module_name)
    cls = getattr(module, class_name)
 
    methods = extract_methods_from_class(cls)
 
    metadata = {
        "methods": methods,
        "definitions": {},
        "meta": {
            "api_version": "v1",
            "base_url": "",
            "auth_type": "bearer",
            "source": f"{module_name}.{class_name}",
            "extracted_at": datetime.now(timezone.utc).isoformat(),
            "total_methods": len(methods),
        }
    }
 
    Path(output_path).write_text(json.dumps(metadata, indent=2))
    print(f"✓ Extracted {len(methods)} methods from {module_name}.{class_name} → {output_path}")

# tools/normalize_metadata.py
import json
import re
from pathlib import Path
 
 
def normalize_cli_command(command: str) -> str:
    """Đảm bảo kebab-case và không có ký tự đặc biệt."""
    command = command.lower().replace("_", "-")
    command = re.sub(r"[^a-z0-9-]", "", command)
    return command.strip("-")
 
 
def infer_missing_fields(method: dict) -> dict:
    """Fill in fields còn thiếu dựa trên convention."""
    if not method.get("cli_command"):
        action = method.get("action", "")
        resource = method.get("resource", "")
        if action == "list":
            method["cli_command"] = f"list-{resource}s"
        else:
            method["cli_command"] = f"{action}-{resource}"
 
    if "idempotent" not in method:
        method["idempotent"] = method.get("action") in ("list", "get", "update")
 
    if "paginated" not in method:
        method["paginated"] = method.get("action") == "list"
 
    if "requires_auth" not in method:
        method["requires_auth"] = True
 
    return method
 
 
def validate_method(method: dict) -> list[str]:
    """Validate method và trả về list lỗi."""
    errors = []
    required_fields = ["id", "resource", "action", "cli_command", "http_method", "path"]
    for field in required_fields:
        if not method.get(field):
            errors.append(f"Missing required field: {field}")
 
    if "." not in method.get("id", ""):
        errors.append("id must be in format 'resource.action'")
 
    return errors
 
 
def normalize(input_path: str, output_path: str):
    """Normalize metadata JSON."""
    metadata = json.loads(Path(input_path).read_text())
 
    normalized_methods = []
    errors_found = 0
 
    for method in metadata.get("methods", []):
        # Normalize fields
        method["cli_command"] = normalize_cli_command(method.get("cli_command", ""))
        method = infer_missing_fields(method)
 
        # Validate
        errors = validate_method(method)
        if errors:
            print(f"⚠️  {method.get('id', '?')}: {', '.join(errors)}")
            errors_found += 1
 
        normalized_methods.append(method)
 
    metadata["methods"] = normalized_methods
    metadata["meta"]["normalized"] = True
    metadata["meta"]["total_methods"] = len(normalized_methods)
 
    Path(output_path).write_text(json.dumps(metadata, indent=2, ensure_ascii=False))
    print(f"✓ Normalized {len(normalized_methods)} methods ({errors_found} warnings) → {output_path}")

# Đếm tổng số methods
jq '.meta.total_methods' metadata/api-metadata.json
 
# List tất cả CLI commands
jq -r '.methods[].cli_command' metadata/api-metadata.json | sort
 
# Tìm methods có pagination
jq '[.methods[] | select(.paginated == true) | .id]' metadata/api-metadata.json
 
# Methods theo resource
jq '[.methods[] | select(.resource == "user")]' metadata/api-metadata.json
 
# Methods không idempotent (cần --force flag khi generate)
jq '[.methods[] | select(.idempotent == false) | .id]' metadata/api-metadata.json

import json
 
metadata = json.loads(open("metadata/api-metadata.json").read())
 
# Filter methods cho code generation
paginated_methods = [m for m in metadata["methods"] if m["paginated"]]
write_methods = [m for m in metadata["methods"] if m["http_method"] in ("POST", "PUT", "DELETE")]
 
print(f"Total: {metadata['meta']['total_methods']} methods")
print(f"Paginated: {len(paginated_methods)}")
print(f"Write operations: {len(write_methods)}")