""" app/api/v1/api_keys.py — API Key 路由 只录不显余额（CLAUDE.md 红线）。 只返回 provider + key_last4，不返回 encrypted_key。 url 字段不接收，token站固定自家站（架构方案§二）。 """ import logging from typing import Annotated import httpx 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 不能为空") provider = v.strip().lower() return "apiports" if provider == "openai" else provider @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（红线）。""" provider = "apiports" if k.provider == "openai" else k.provider return { "id": k.id, "provider": provider, "key_last4": k.key_last4, "created_at": k.created_at.isoformat(), } def _provider_aliases(provider: str) -> list[str]: if provider == "apiports": return ["apiports", "openai"] return [provider] def _upstream_error_message(provider: str, exc: Exception) -> str: if isinstance(exc, httpx.HTTPStatusError): status = exc.response.status_code try: detail = exc.response.json() except Exception: detail = (exc.response.text or "")[:160] if status in (401, 403): hint = "Key 无效、权限不足或模型未开通" elif status == 402: hint = "账户余额不足或未开通计费" elif status == 429: hint = "请求过频或额度受限" elif status >= 500: hint = "中转站服务暂时不可用" else: hint = "中转站返回异常" logger.warning("API key test failed provider=%s status=%s detail=%s", provider, status, detail) return f"Key 测试失败：{provider} 返回 {status}，{hint}" if isinstance(exc, httpx.ReadTimeout): return f"Key 测试失败：{provider} 在测试超时时间内无响应，请稍后重试或先保存后跑任务验证" if isinstance(exc, httpx.ConnectError): return f"Key 测试失败：无法连接 {provider} 中转站，请检查服务器网络或 Base URL" return f"Key 测试失败：{type(exc).__name__}" # ── 路由 ─────────────────────────────────────────────────── @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.in_(_provider_aliases(body.provider)), ) .first() ) if existing: existing.provider = body.provider 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 做一次轻量连通性测试，不落库。""" from app.core.config import get_settings from app.core.response import raise_business settings = get_settings() provider = body.provider if provider == "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=45) 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(_upstream_error_message(provider, exc)) 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})