awoooi/scripts/generate-schemas.py

#!/usr/bin/env python3
"""
Pydantic → JSON Schema 生成器
=============================
Phase 14.3: 共用型別系統

功能:
- 從 Pydantic 模型生成 JSON Schema
- 正確處理 Pydantic 2.x 的 $defs 格式
- 輸出到 packages/shared-types/schemas/

使用方式:
    cd apps/api
    python ../../scripts/generate-schemas.py

建立: 2026-03-31 (台北時區)
建立者: Claude Code (Phase 14.3)
"""

import json
import sys
from pathlib import Path
from typing import Any

# 加入 apps/api/src 到 Python path
api_src = Path(__file__).parent.parent / "apps" / "api" / "src"
sys.path.insert(0, str(api_src))

# 輸出目錄
OUTPUT_DIR = Path(__file__).parent.parent / "packages" / "shared-types" / "schemas"


def extract_and_merge_defs(schema: dict[str, Any], all_defs: dict[str, Any]) -> dict[str, Any]:
    """
    從單一模型 schema 中提取 $defs 並合併到全局 $defs
    同時修正 $ref 路徑
    """
    # 提取並合併 $defs
    if "$defs" in schema:
        for def_name, def_schema in schema["$defs"].items():
            if def_name not in all_defs:
                # 遞迴處理巢狀 $defs
                cleaned_def = extract_and_merge_defs(def_schema.copy(), all_defs)
                all_defs[def_name] = cleaned_def

    # 移除 schema 中的 $defs (已合併到全局)
    cleaned_schema = {k: v for k, v in schema.items() if k != "$defs"}

    return cleaned_schema


def fix_refs(obj: Any) -> Any:
    """遞迴修正所有 $ref 路徑從 #/$defs/X 到 #/definitions/X"""
    if isinstance(obj, dict):
        result = {}
        for key, value in obj.items():
            if key == "$ref" and isinstance(value, str):
                # 修正 ref 路徑
                result[key] = value.replace("#/$defs/", "#/definitions/")
            else:
                result[key] = fix_refs(value)
        return result
    elif isinstance(obj, list):
        return [fix_refs(item) for item in obj]
    else:
        return obj


def generate_schemas():
    """生成所有模型的 JSON Schema"""

    # 確保輸出目錄存在
    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)

    # 收集所有 definitions
    all_definitions: dict[str, Any] = {}
    model_count = 0

    print("📦 Approval Models:")
    # === Approval Models ===
    try:
        from models.approval import (
            ApprovalRequest,
            ApprovalRequestCreate,
            ApprovalRequestResponse,
            BlastRadius,
            DryRunCheck,
            PendingApprovalsResponse,
            RejectRequest,
            Signature,
            SignRequest,
            SignResponse,
        )

        approval_models = [
            ApprovalRequest,
            ApprovalRequestCreate,
            ApprovalRequestResponse,
            BlastRadius,
            DryRunCheck,
            PendingApprovalsResponse,
            RejectRequest,
            Signature,
            SignRequest,
            SignResponse,
        ]

        for model in approval_models:
            schema = model.model_json_schema()
            cleaned = extract_and_merge_defs(schema, all_definitions)
            all_definitions[model.__name__] = cleaned
            model_count += 1
            print(f"  ✓ {model.__name__}")

    except ImportError as e:
        print(f"  ⚠ Import failed: {e}")

    print("\n📦 Incident Models:")
    # === Incident Models ===
    try:
        from models.incident import (
            AIDecisionChain,
            Incident,
            IncidentCreate,
            IncidentOutcome,
            IncidentResponse,
            IncidentUpdate,
            Signal,
        )

        incident_models = [
            AIDecisionChain,
            Incident,
            IncidentCreate,
            IncidentOutcome,
            IncidentResponse,
            IncidentUpdate,
            Signal,
        ]

        for model in incident_models:
            schema = model.model_json_schema()
            cleaned = extract_and_merge_defs(schema, all_definitions)
            all_definitions[model.__name__] = cleaned
            model_count += 1
            print(f"  ✓ {model.__name__}")

    except ImportError as e:
        print(f"  ⚠ Import failed: {e}")

    print("\n📦 Terminal Models:")
    # === Terminal Models ===
    try:
        from models.terminal import (
            SpatialContext,
            TerminalAbortRequest,
            TerminalAbortResponse,
            TerminalIntentRequest,
            TerminalIntentResponse,
            TerminalStatusResponse,
            TerminalThoughtEvent,
        )

        terminal_models = [
            TerminalIntentRequest,
            TerminalIntentResponse,
            TerminalStatusResponse,
            TerminalAbortRequest,
            TerminalAbortResponse,
            TerminalThoughtEvent,
            SpatialContext,
        ]

        for model in terminal_models:
            schema = model.model_json_schema()
            cleaned = extract_and_merge_defs(schema, all_definitions)
            all_definitions[model.__name__] = cleaned
            model_count += 1
            print(f"  ✓ {model.__name__}")

    except ImportError as e:
        print(f"  ⚠ Import failed: {e}")

    print("\n📦 Playbook Models:")
    # === Playbook Models ===
    try:
        from models.playbook import (
            Playbook,
            PlaybookCreateRequest,
            PlaybookListResponse,
            PlaybookRecommendation,
            PlaybookResponse,
            PlaybookUpdateRequest,
            RepairStep,
            SymptomPattern,
        )

        playbook_models = [
            Playbook,
            PlaybookResponse,
            PlaybookListResponse,
            PlaybookCreateRequest,
            PlaybookUpdateRequest,
            PlaybookRecommendation,
            SymptomPattern,
            RepairStep,
        ]

        for model in playbook_models:
            schema = model.model_json_schema()
            cleaned = extract_and_merge_defs(schema, all_definitions)
            all_definitions[model.__name__] = cleaned
            model_count += 1
            print(f"  ✓ {model.__name__}")

    except ImportError as e:
        print(f"  ⚠ Import failed: {e}")

    print("\n📦 Other Models:")
    # === CSRF Models ===
    try:
        from models.csrf import CSRFTokenResponse

        schema = CSRFTokenResponse.model_json_schema()
        cleaned = extract_and_merge_defs(schema, all_definitions)
        all_definitions["CSRFTokenResponse"] = cleaned
        model_count += 1
        print("  ✓ CSRFTokenResponse")

    except ImportError as e:
        print(f"  ⚠ CSRF import failed: {e}")

    # 修正所有 $ref 路徑
    all_definitions = fix_refs(all_definitions)

    # 組裝最終 schema
    final_schema = {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "title": "AWOOOI API Types",
        "description": "Auto-generated from Pydantic models - DO NOT EDIT",
        "definitions": all_definitions,
    }

    # 寫入檔案
    output_file = OUTPUT_DIR / "api-types.json"
    with open(output_file, "w", encoding="utf-8") as f:
        json.dump(final_schema, f, indent=2, ensure_ascii=False)

    print(f"\n✅ Schema 生成完成: {output_file}")
    print(f"   共 {len(all_definitions)} 個型別定義")

    return final_schema


if __name__ == "__main__":
    print("🔄 開始生成 JSON Schema...\n")
    generate_schemas()