core/api/routes.py

"""
OpenAI 协议路由。

支持：
- /openai/{provider}/v1/chat/completions
- /openai/{provider}/v1/models
- 旧路径 /{provider}/v1/...（等价于 OpenAI 协议）
"""

import json
import time
from collections.abc import AsyncIterator
from typing import Any

from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi.responses import JSONResponse, StreamingResponse

from core.api.auth import require_api_key
from core.api.chat_handler import ChatHandler
from core.config.repository import APP_SETTING_ENABLE_PRO_MODELS
from core.plugin.base import PluginRegistry
from core.protocol.openai import OpenAIProtocolAdapter
from core.protocol.schemas import CanonicalChatRequest
from core.protocol.service import CanonicalChatService


def get_chat_handler(request: Request) -> ChatHandler:
    """从 app state 取出 ChatHandler。"""
    handler = getattr(request.app.state, "chat_handler", None)
    if handler is None:
        raise HTTPException(status_code=503, detail="服务未就绪")
    return handler


def resolve_request_model(
    provider: str,
    canonical_req: CanonicalChatRequest,
) -> CanonicalChatRequest:
    resolved = PluginRegistry.resolve_model(provider, canonical_req.model)
    canonical_req.model = resolved.public_model
    canonical_req.metadata["upstream_model"] = resolved.upstream_model
    return canonical_req


def check_pro_model_access(
    request: Request,
    provider: str,
    model: str,
) -> JSONResponse | None:
    """Return 403 JSONResponse if model requires Pro and Pro is disabled, else None."""
    plugin = PluginRegistry.get(provider)
    if plugin is None:
        return None
    pro_models = getattr(plugin, "PRO_MODELS", frozenset())
    if model not in pro_models:
        return None
    config_repo = getattr(request.app.state, "config_repo", None)
    if config_repo is None:
        return None
    enabled = config_repo.get_app_setting(APP_SETTING_ENABLE_PRO_MODELS)
    if enabled == "true":
        return None
    return JSONResponse(
        status_code=403,
        content={
            "error": {
                "message": (
                    f"Model '{model}' requires a Claude Pro subscription. "
                    "Enable Pro models in the config page at /config."
                ),
                "type": "model_not_available",
                "code": "pro_model_required",
            }
        },
    )


def create_router() -> APIRouter:
    """创建 OpenAI 协议路由。"""
    router = APIRouter(dependencies=[Depends(require_api_key)])
    adapter = OpenAIProtocolAdapter()

    def _list_models(provider: str) -> dict[str, Any]:
        try:
            metadata = PluginRegistry.model_metadata(provider)
        except ValueError as exc:
            raise HTTPException(status_code=404, detail=str(exc)) from exc
        now = int(time.time())
        return {
            "object": "list",
            "data": [
                {
                    "id": mid,
                    "object": "model",
                    "created": now,
                    "owned_by": provider,
                }
                for mid in metadata["public_models"]
            ],
        }

    @router.get("/openai/{provider}/v1/models")
    def list_models(provider: str) -> dict[str, Any]:
        return _list_models(provider)

    @router.get("/{provider}/v1/models")
    def list_models_legacy(provider: str) -> dict[str, Any]:
        return _list_models(provider)

    async def _chat_completions(
        provider: str,
        request: Request,
        handler: ChatHandler,
    ) -> Any:
        raw_body = await request.json()
        try:
            canonical_req = resolve_request_model(
                provider,
                adapter.parse_request(provider, raw_body),
            )
        except Exception as exc:
            status, payload = adapter.render_error(exc)
            return JSONResponse(status_code=status, content=payload)

        pro_err = check_pro_model_access(request, provider, canonical_req.model)
        if pro_err is not None:
            return pro_err

        service = CanonicalChatService(handler)
        if canonical_req.stream:

            async def sse_stream() -> AsyncIterator[str]:
                try:
                    async for event in adapter.render_stream(
                        canonical_req,
                        service.stream_raw(canonical_req),
                    ):
                        yield event
                except Exception as exc:
                    status, payload = adapter.render_error(exc)
                    del status
                    yield f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"

            return StreamingResponse(
                sse_stream(),
                media_type="text/event-stream",
                headers={
                    "Cache-Control": "no-cache",
                    "Connection": "keep-alive",
                    "X-Accel-Buffering": "no",
                },
            )

        try:
            raw_events = await service.collect_raw(canonical_req)
            return adapter.render_non_stream(canonical_req, raw_events)
        except Exception as exc:
            status, payload = adapter.render_error(exc)
            return JSONResponse(status_code=status, content=payload)

    @router.post("/openai/{provider}/v1/chat/completions")
    async def chat_completions(
        provider: str,
        request: Request,
        handler: ChatHandler = Depends(get_chat_handler),
    ) -> Any:
        return await _chat_completions(provider, request, handler)

    @router.post("/{provider}/v1/chat/completions")
    async def chat_completions_legacy(
        provider: str,
        request: Request,
        handler: ChatHandler = Depends(get_chat_handler),
    ) -> Any:
        return await _chat_completions(provider, request, handler)

    return router