oadmin/sub2api
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

refactor(upstream): replace upstream account type with apikey, auto-append /antigravity

Browse Source 
Upstream accounts now use the standard APIKey type instead of a dedicated
upstream type. GetBaseURL() and new GetGeminiBaseURL() automatically append
/antigravity for Antigravity platform APIKey accounts, eliminating the need
for separate upstream forwarding methods.

- Remove ForwardUpstream, ForwardUpstreamGemini, testUpstreamConnection
- Remove upstream branch guards in Forward/ForwardGemini/TestConnection
- Add migration 052 to convert existing upstream accounts to apikey
- Update frontend CreateAccountModal to create apikey type
- Add unit tests for GetBaseURL and GetGeminiBaseURL
This commit is contained in:
erio
2026-02-08 13:06:25 +08:00
parent 6ab77f5eb5
commit fb58560d15
9 changed files with 197 additions and 696 deletions
Download Patch File Download Diff File Expand all files Collapse all files

386
backend/internal/service/antigravity_gateway_service.go
View File

@@ -665,9 +665,6 @@ type TestConnectionResult struct {
// TestConnection 测试 Antigravity 账号连接（非流式，无重试、无计费）
// 支持 Claude 和 Gemini 两种协议，根据 modelID 前缀自动选择
func (s *AntigravityGatewayService) TestConnection(ctx context.Context, account *Account, modelID string) (*TestConnectionResult, error) {
if account.Type == AccountTypeUpstream {
return s.testUpstreamConnection(ctx, account, modelID)
}
// 获取 token
if s.tokenProvider == nil {
@@ -986,10 +983,6 @@ func isModelNotFoundError(statusCode int, body []byte) bool {
func (s *AntigravityGatewayService) Forward(ctx context.Context, c *gin.Context, account *Account, body []byte, isStickySession bool) (*ForwardResult, error) {
startTime := time.Now()
if account.Type == AccountTypeUpstream {
return s.ForwardUpstream(ctx, c, account, body, isStickySession)
}
sessionID := getSessionID(c)
prefix := logPrefix(sessionID, account.Name)
@@ -1610,10 +1603,6 @@ func stripSignatureSensitiveBlocksFromClaudeRequest(req *antigravity.ClaudeReque
func (s *AntigravityGatewayService) ForwardGemini(ctx context.Context, c *gin.Context, account *Account, originalModel string, action string, stream bool, body []byte, isStickySession bool) (*ForwardResult, error) {
startTime := time.Now()
if account.Type == AccountTypeUpstream {
return s.ForwardUpstreamGemini(ctx, c, account, originalModel, action, stream, body, isStickySession)
}
sessionID := getSessionID(c)
prefix := logPrefix(sessionID, account.Name)
@@ -3361,378 +3350,3 @@ func filterEmptyPartsFromGeminiRequest(body []byte) ([]byte, error) {
payload["contents"] = filtered
return json.Marshal(payload)
}
// ---------------------------------------------------------------------------
// Upstream 专用转发方法
// upstream 账号直接连接上游 Anthropic/Gemini 兼容端点，不走 Antigravity OAuth 协议转换。
// ---------------------------------------------------------------------------
// testUpstreamConnection 测试 upstream 账号连接
func (s *AntigravityGatewayService) testUpstreamConnection(ctx context.Context, account *Account, modelID string) (*TestConnectionResult, error) {
baseURL := strings.TrimRight(strings.TrimSpace(account.GetCredential("base_url")), "/")
if baseURL == "" {
return nil, errors.New("upstream account missing base_url in credentials")
}
apiKey := strings.TrimSpace(account.GetCredential("api_key"))
if apiKey == "" {
return nil, errors.New("upstream account missing api_key in credentials")
}
mappedModel := s.getMappedModel(account, modelID)
if mappedModel == "" {
return nil, fmt.Errorf("model %s not in whitelist", modelID)
}
// 构建最小 Claude 格式请求
requestBody, _ := json.Marshal(map[string]any{
"model": mappedModel,
"max_tokens": 1,
"messages": []map[string]any{
{"role": "user", "content": "."},
},
"stream": false,
})
apiURL := baseURL + "/antigravity/v1/messages"
req, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, bytes.NewReader(requestBody))
if err != nil {
return nil, fmt.Errorf("构建请求失败: %w", err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", "Bearer "+apiKey)
req.Header.Set("x-api-key", apiKey)
req.Header.Set("anthropic-version", "2023-06-01")
proxyURL := ""
if account.ProxyID != nil && account.Proxy != nil {
proxyURL = account.Proxy.URL()
}
log.Printf("[antigravity-Test-Upstream] account=%s url=%s", account.Name, apiURL)
resp, err := s.httpUpstream.Do(req, proxyURL, account.ID, account.Concurrency)
if err != nil {
return nil, fmt.Errorf("请求失败: %w", err)
}
defer func() { _ = resp.Body.Close() }()
respBody, err := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
if err != nil {
return nil, fmt.Errorf("读取响应失败: %w", err)
}
if resp.StatusCode >= 400 {
return nil, fmt.Errorf("API 返回 %d: %s", resp.StatusCode, string(respBody))
}
// 从 Claude 格式非流式响应中提取文本
var claudeResp struct {
Content []struct {
Text string `json:"text"`
} `json:"content"`
}
text := ""
if json.Unmarshal(respBody, &claudeResp) == nil && len(claudeResp.Content) > 0 {
text = claudeResp.Content[0].Text
}
return &TestConnectionResult{
Text: text,
MappedModel: mappedModel,
}, nil
}
// ForwardUpstream 转发 Claude 协议请求到 upstream（不做协议转换）
func (s *AntigravityGatewayService) ForwardUpstream(ctx context.Context, c *gin.Context, account *Account, body []byte, isStickySession bool) (*ForwardResult, error) {
startTime := time.Now()
sessionID := getSessionID(c)
prefix := logPrefix(sessionID, account.Name)
baseURL := strings.TrimRight(strings.TrimSpace(account.GetCredential("base_url")), "/")
if baseURL == "" {
return nil, s.writeClaudeError(c, http.StatusBadGateway, "api_error", "Upstream account missing base_url")
}
apiKey := strings.TrimSpace(account.GetCredential("api_key"))
if apiKey == "" {
return nil, s.writeClaudeError(c, http.StatusBadGateway, "authentication_error", "Upstream account missing api_key")
}
// 解析请求以获取模型和流式标志
var claudeReq antigravity.ClaudeRequest
if err := json.Unmarshal(body, &claudeReq); err != nil {
return nil, s.writeClaudeError(c, http.StatusBadRequest, "invalid_request_error", "Invalid request body")
}
if strings.TrimSpace(claudeReq.Model) == "" {
return nil, s.writeClaudeError(c, http.StatusBadRequest, "invalid_request_error", "Missing model")
}
originalModel := claudeReq.Model
mappedModel := s.getMappedModel(account, claudeReq.Model)
if mappedModel == "" {
return nil, s.writeClaudeError(c, http.StatusForbidden, "permission_error", fmt.Sprintf("model %s not in whitelist", claudeReq.Model))
}
// 代理 URL
proxyURL := ""
if account.ProxyID != nil && account.Proxy != nil {
proxyURL = account.Proxy.URL()
}
// 统计模型调用次数
if s.cache != nil {
_, _ = s.cache.IncrModelCallCount(ctx, account.ID, mappedModel)
}
apiURL := baseURL + "/antigravity/v1/messages"
log.Printf("%s upstream_forward url=%s model=%s", prefix, apiURL, mappedModel)
// 构建请求：body 原样透传
req, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, bytes.NewReader(body))
if err != nil {
return nil, s.writeClaudeError(c, http.StatusInternalServerError, "api_error", "Failed to build request")
}
// 透传客户端所有请求头（排除 hop-by-hop 和认证头）
if c != nil && c.Request != nil {
for key, values := range c.Request.Header {
if upstreamHopByHopHeaders[strings.ToLower(key)] {
continue
}
for _, v := range values {
req.Header.Add(key, v)
}
}
}
// 覆盖认证头
req.Header.Set("Authorization", "Bearer "+apiKey)
req.Header.Set("x-api-key", apiKey)
if c != nil && len(body) > 0 {
c.Set(OpsUpstreamRequestBodyKey, string(body))
}
// 单次发送，不重试
resp, err := s.httpUpstream.Do(req, proxyURL, account.ID, account.Concurrency)
if err != nil {
return nil, s.writeClaudeError(c, http.StatusBadGateway, "upstream_error", fmt.Sprintf("Upstream request failed: %v", err))
}
defer func() { _ = resp.Body.Close() }()
// 错误响应处理
if resp.StatusCode >= 400 {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
s.handleUpstreamError(ctx, prefix, account, resp.StatusCode, resp.Header, respBody, "", 0, "", isStickySession)
if s.shouldFailoverUpstreamError(resp.StatusCode) {
return nil, &UpstreamFailoverError{StatusCode: resp.StatusCode, ResponseBody: respBody}
}
return nil, s.writeMappedClaudeError(c, account, resp.StatusCode, resp.Header.Get("x-request-id"), respBody)
}
// 成功响应：透传 response header + body
requestID := resp.Header.Get("x-request-id")
// 透传上游响应头（排除 hop-by-hop）
for key, values := range resp.Header {
if upstreamHopByHopHeaders[strings.ToLower(key)] {
continue
}
for _, v := range values {
c.Header(key, v)
}
}
c.Status(resp.StatusCode)
_, copyErr := io.Copy(c.Writer, resp.Body)
if copyErr != nil {
log.Printf("%s status=copy_error error=%v", prefix, copyErr)
}
return &ForwardResult{
RequestID: requestID,
Model: originalModel,
Stream: claudeReq.Stream,
Duration: time.Since(startTime),
}, nil
}
// ForwardUpstreamGemini 转发 Gemini 协议请求到 upstream（不做协议转换）
func (s *AntigravityGatewayService) ForwardUpstreamGemini(ctx context.Context, c *gin.Context, account *Account, originalModel string, action string, stream bool, body []byte, isStickySession bool) (*ForwardResult, error) {
startTime := time.Now()
sessionID := getSessionID(c)
prefix := logPrefix(sessionID, account.Name)
baseURL := strings.TrimRight(strings.TrimSpace(account.GetCredential("base_url")), "/")
if baseURL == "" {
return nil, s.writeGoogleError(c, http.StatusBadGateway, "Upstream account missing base_url")
}
apiKey := strings.TrimSpace(account.GetCredential("api_key"))
if apiKey == "" {
return nil, s.writeGoogleError(c, http.StatusBadGateway, "Upstream account missing api_key")
}
if strings.TrimSpace(originalModel) == "" {
return nil, s.writeGoogleError(c, http.StatusBadRequest, "Missing model in URL")
}
if strings.TrimSpace(action) == "" {
return nil, s.writeGoogleError(c, http.StatusBadRequest, "Missing action in URL")
}
if len(body) == 0 {
return nil, s.writeGoogleError(c, http.StatusBadRequest, "Request body is empty")
}
imageSize := s.extractImageSize(body)
switch action {
case "generateContent", "streamGenerateContent":
// ok
case "countTokens":
c.JSON(http.StatusOK, map[string]any{"totalTokens": 0})
return &ForwardResult{
RequestID: "",
Usage: ClaudeUsage{},
Model: originalModel,
Stream: false,
Duration: time.Since(time.Now()),
FirstTokenMs: nil,
}, nil
default:
return nil, s.writeGoogleError(c, http.StatusNotFound, "Unsupported action: "+action)
}
mappedModel := s.getMappedModel(account, originalModel)
if mappedModel == "" {
return nil, s.writeGoogleError(c, http.StatusForbidden, fmt.Sprintf("model %s not in whitelist", originalModel))
}
// 代理 URL
proxyURL := ""
if account.ProxyID != nil && account.Proxy != nil {
proxyURL = account.Proxy.URL()
}
// 统计模型调用次数
if s.cache != nil {
_, _ = s.cache.IncrModelCallCount(ctx, account.ID, mappedModel)
}
// 构建 upstream URL: base_url + /antigravity/v1beta/models/MODEL:ACTION
apiURL := fmt.Sprintf("%s/antigravity/v1beta/models/%s:%s", baseURL, mappedModel, action)
if stream || action == "streamGenerateContent" {
apiURL += "?alt=sse"
}
log.Printf("%s upstream_forward_gemini url=%s model=%s action=%s", prefix, apiURL, mappedModel, action)
// 构建请求：body 原样透传
req, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, bytes.NewReader(body))
if err != nil {
return nil, s.writeGoogleError(c, http.StatusInternalServerError, "Failed to build request")
}
// 透传客户端所有请求头（排除 hop-by-hop 和认证头）
if c != nil && c.Request != nil {
for key, values := range c.Request.Header {
if upstreamHopByHopHeaders[strings.ToLower(key)] {
continue
}
for _, v := range values {
req.Header.Add(key, v)
}
}
}
// 覆盖认证头
req.Header.Set("Authorization", "Bearer "+apiKey)
if c != nil && len(body) > 0 {
c.Set(OpsUpstreamRequestBodyKey, string(body))
}
// 单次发送，不重试
resp, err := s.httpUpstream.Do(req, proxyURL, account.ID, account.Concurrency)
if err != nil {
return nil, s.writeGoogleError(c, http.StatusBadGateway, fmt.Sprintf("Upstream request failed: %v", err))
}
defer func() { _ = resp.Body.Close() }()
// 错误响应处理
if resp.StatusCode >= 400 {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
contentType := resp.Header.Get("Content-Type")
requestID := resp.Header.Get("x-request-id")
if requestID != "" {
c.Header("x-request-id", requestID)
}
s.handleUpstreamError(ctx, prefix, account, resp.StatusCode, resp.Header, respBody, "", 0, "", isStickySession)
upstreamMsg := strings.TrimSpace(extractAntigravityErrorMessage(respBody))
upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg)
upstreamDetail := s.getUpstreamErrorDetail(respBody)
setOpsUpstreamError(c, resp.StatusCode, upstreamMsg, upstreamDetail)
if s.shouldFailoverUpstreamError(resp.StatusCode) {
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
UpstreamStatusCode: resp.StatusCode,
UpstreamRequestID: requestID,
Kind: "failover",
Message: upstreamMsg,
Detail: upstreamDetail,
})
return nil, &UpstreamFailoverError{StatusCode: resp.StatusCode, ResponseBody: respBody}
}
if contentType == "" {
contentType = "application/json"
}
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
Platform: account.Platform,
AccountID: account.ID,
AccountName: account.Name,
UpstreamStatusCode: resp.StatusCode,
UpstreamRequestID: requestID,
Kind: "http_error",
Message: upstreamMsg,
Detail: upstreamDetail,
})
log.Printf("[antigravity-Forward-Upstream] upstream error status=%d body=%s", resp.StatusCode, truncateForLog(respBody, 500))
c.Data(resp.StatusCode, contentType, respBody)
return nil, fmt.Errorf("antigravity upstream error: %d", resp.StatusCode)
}
// 成功响应：透传 response header + body
requestID := resp.Header.Get("x-request-id")
// 透传上游响应头（排除 hop-by-hop）
for key, values := range resp.Header {
if upstreamHopByHopHeaders[strings.ToLower(key)] {
continue
}
for _, v := range values {
c.Header(key, v)
}
}
c.Status(resp.StatusCode)
_, copyErr := io.Copy(c.Writer, resp.Body)
if copyErr != nil {
log.Printf("%s status=copy_error error=%v", prefix, copyErr)
}
imageCount := 0
if isImageGenerationModel(mappedModel) {
imageCount = 1
}
return &ForwardResult{
RequestID: requestID,
Model: originalModel,
Stream: stream,
Duration: time.Since(startTime),
ImageCount: imageCount,
ImageSize: imageSize,
}, nil
}