src/opencode_api/routes/provider.py

from typing import List, Dict
from fastapi import APIRouter, HTTPException
import os
from dotenv import load_dotenv

# .env 파일에서 환경변수 로드
load_dotenv()

from ..provider import list_providers, get_provider
from ..provider.provider import ProviderInfo, ModelInfo


router = APIRouter(prefix="/provider", tags=["Provider"])


# Provider별 필요 환경변수 매핑
PROVIDER_API_KEYS = {
    "anthropic": "ANTHROPIC_API_KEY",
    "openai": "OPENAI_API_KEY",
    "gemini": ["GOOGLE_API_KEY", "GEMINI_API_KEY"],
    "blablador": "BLABLADOR_API_KEY",
    "litellm": None,  # LiteLLM은 개별 모델별로 체크
}

# LiteLLM 모델별 필요 환경변수
LITELLM_MODEL_KEYS = {
    "claude-": "ANTHROPIC_API_KEY",
    "gpt-": "OPENAI_API_KEY",
    "o1": "OPENAI_API_KEY",
    "gemini/": ["GOOGLE_API_KEY", "GEMINI_API_KEY"],
    "groq/": "GROQ_API_KEY",
    "deepseek/": "DEEPSEEK_API_KEY",
    "openrouter/": "OPENROUTER_API_KEY",
    "zai/": "ZAI_API_KEY",
}


def has_api_key(provider_id: str) -> bool:
    """Check if provider has required API key configured"""
    keys = PROVIDER_API_KEYS.get(provider_id)
    if keys is None:
        return True  # No key required (like litellm container)
    if isinstance(keys, list):
        return any(os.environ.get(k) for k in keys)
    return bool(os.environ.get(keys))


def filter_litellm_models(models: Dict[str, ModelInfo]) -> Dict[str, ModelInfo]:
    """Filter LiteLLM models based on available API keys"""
    filtered = {}
    for model_id, model_info in models.items():
        for prefix, env_key in LITELLM_MODEL_KEYS.items():
            if model_id.startswith(prefix):
                if isinstance(env_key, list):
                    if any(os.environ.get(k) for k in env_key):
                        filtered[model_id] = model_info
                elif os.environ.get(env_key):
                    filtered[model_id] = model_info
                break
    return filtered


@router.get("/", response_model=List[ProviderInfo])
async def get_providers():
    """Get available providers (filtered by API key availability)"""
    all_providers = list_providers()
    available = []

    for provider in all_providers:
        if provider.id == "litellm":
            # LiteLLM: 개별 모델별 필터링
            filtered_models = filter_litellm_models(provider.models)
            if filtered_models:
                provider.models = filtered_models
                available.append(provider)
        elif has_api_key(provider.id):
            available.append(provider)

    return available


@router.get("/{provider_id}", response_model=ProviderInfo)
async def get_provider_info(provider_id: str):
    provider = get_provider(provider_id)
    if not provider:
        raise HTTPException(status_code=404, detail=f"Provider not found: {provider_id}")
    return provider.get_info()


@router.get("/{provider_id}/model", response_model=List[ModelInfo])
async def get_provider_models(provider_id: str):
    provider = get_provider(provider_id)
    if not provider:
        raise HTTPException(status_code=404, detail=f"Provider not found: {provider_id}")
    return list(provider.models.values())


@router.get("/{provider_id}/model/{model_id}", response_model=ModelInfo)
async def get_model_info(provider_id: str, model_id: str):
    provider = get_provider(provider_id)
    if not provider:
        raise HTTPException(status_code=404, detail=f"Provider not found: {provider_id}")

    model = provider.models.get(model_id)
    if not model:
        raise HTTPException(status_code=404, detail=f"Model not found: {model_id}")

    return model