beige/backend/app/api/v1/api_keys.py

"""
app/api/v1/api_keys.py — API Key 路由
只录不显余额（CLAUDE.md 红线）。
只返回 provider + key_last4，不返回 encrypted_key。
url 字段不接收，token站固定自家站（架构方案§二）。
"""

import logging
from typing import Annotated

from fastapi import APIRouter, Depends
from pydantic import BaseModel, field_validator
from sqlalchemy.orm import Session

from app.core.database import get_db
from app.core.response import ok, raise_not_found
from app.core.security import encrypt_api_key, mask_api_key
from app.middleware.workspace_guard import CurrentUser, require_write_permission
from app.models.workspace import UserApiKey

logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api-keys", tags=["api-keys"])


# ── DTO ────────────────────────────────────────────────────
class CreateApiKeyRequest(BaseModel):
    provider: str
    api_key: str

    @field_validator("provider")
    @classmethod
    def provider_not_empty(cls, v: str) -> str:
        if not v or not v.strip():
            raise ValueError("provider 不能为空")
        return v.strip().lower()

    @field_validator("api_key")
    @classmethod
    def key_not_empty(cls, v: str) -> str:
        if not v or len(v) < 8:
            raise ValueError("api_key 无效")
        return v
    # 注：不接收 url 字段（token站固定自家站，基石B）


class TestApiKeyRequest(CreateApiKeyRequest):
    pass


def _format_key(k: UserApiKey) -> dict:
    """只显 provider + key_last4，不暴露余额/用量/encrypted_key（红线）。"""
    return {
        "id": k.id,
        "provider": k.provider,
        "key_last4": k.key_last4,
        "created_at": k.created_at.isoformat(),
    }


# ── 路由 ───────────────────────────────────────────────────
@router.get("")
def list_api_keys(
    current_user: Annotated[CurrentUser, Depends(require_write_permission)],
    db: Session = Depends(get_db),
):
    """列出当前用户在此 workspace 的 API Key（只显后4位）。"""
    keys = (
        db.query(UserApiKey)
        .filter(
            UserApiKey.user_id == current_user.user_id,
            UserApiKey.workspace_id == current_user.workspace_id,
        )
        .all()
    )
    return ok({"items": [_format_key(k) for k in keys]})


@router.post("")
def create_api_key(
    body: CreateApiKeyRequest,
    current_user: Annotated[CurrentUser, Depends(require_write_permission)],
    db: Session = Depends(get_db),
):
    """录入/更新 API Key（Fernet 加密存储，只保存后4位明文用于展示）。

    覆盖式（倩倩姐2026-06-23拍板）：同 user+workspace+provider 已存在则直接更新，
    不报错——「切换/修改 key」无需先删再录，前端点「修改」重录即覆盖。
    """
    encrypted = encrypt_api_key(body.api_key)
    last4 = mask_api_key(body.api_key)
    # 检查同 user+workspace+provider 是否已有
    existing = (
        db.query(UserApiKey)
        .filter(
            UserApiKey.user_id == current_user.user_id,
            UserApiKey.workspace_id == current_user.workspace_id,
            UserApiKey.provider == body.provider,
        )
        .first()
    )
    if existing:
        existing.encrypted_key = encrypted
        existing.key_last4 = last4
        db.commit()
        db.refresh(existing)
        logger.info("API key updated: user=%s provider=%s", current_user.user_id, body.provider)
        return ok(_format_key(existing))

    key_obj = UserApiKey(
        user_id=current_user.user_id,
        workspace_id=current_user.workspace_id,
        provider=body.provider,
        encrypted_key=encrypted,
        key_last4=last4,
    )
    db.add(key_obj)
    db.commit()
    db.refresh(key_obj)
    logger.info("API key created: user=%s provider=%s", current_user.user_id, body.provider)
    return ok(_format_key(key_obj))


@router.post("/test")
async def test_api_key(
    body: TestApiKeyRequest,
    current_user: Annotated[CurrentUser, Depends(require_write_permission)],
):
    """用客户输入的明文 key 做一次轻量连通性测试，不落库。"""
    import httpx
    from app.core.config import get_settings
    from app.core.response import raise_business

    settings = get_settings()
    provider = body.provider
    if provider in ("openai", "apiports"):
        base = (settings.APIPORTS_BASE_URL or settings.IMAGE_API_BASE).rstrip("/")
        model = settings.MODEL_TEXT
    elif provider == "codeproxy":
        base = settings.CODEPROXY_BASE_URL.rstrip("/")
        model = "gpt-5.5"
    else:
        raise_business("该 Provider 暂不支持在线测试")
    if not base:
        raise_business("服务端未配置该 Provider 的 Base URL")

    try:
        async with httpx.AsyncClient(timeout=12) as client:
            resp = await client.post(
                f"{base}/chat/completions",
                headers={"Authorization": f"Bearer {body.api_key}"},
                json={
                    "model": model,
                    "messages": [{"role": "user", "content": "ping"}],
                    "max_tokens": 4,
                },
            )
            resp.raise_for_status()
    except Exception as exc:
        raise_business(f"Key 测试失败：{type(exc).__name__}")

    logger.info("API key test ok: user=%s provider=%s", current_user.user_id, provider)
    return ok({"ok": True, "provider": provider})


@router.delete("/{key_id}")
def delete_api_key(
    key_id: int,
    current_user: Annotated[CurrentUser, Depends(require_write_permission)],
    db: Session = Depends(get_db),
):
    """删除 API Key（只能删自己的）。"""
    key_obj = (
        db.query(UserApiKey)
        .filter(
            UserApiKey.id == key_id,
            UserApiKey.user_id == current_user.user_id,
            UserApiKey.workspace_id == current_user.workspace_id,
        )
        .first()
    )
    if not key_obj:
        raise_not_found("API Key 不存在")
    db.delete(key_obj)
    db.commit()
    return ok({"deleted": key_id})