package service import ( "bufio" "bytes" "encoding/json" "fmt" "io" "net/http" "net/url" "strings" "github.com/Wei-Shaw/sub2api/internal/pkg/claude" "github.com/Wei-Shaw/sub2api/internal/pkg/openai" "github.com/gin-gonic/gin" ) const accountTestSuppressCompletionContextKey = "account_test_suppress_completion" // testCNProviderAdaptiveConnection verifies every native endpoint used by an // adaptive CN-provider account. Zhipu uses Chat Completions plus Anthropic; // DeepSeek and Kimi additionally use their native Responses endpoints. func (s *AccountTestService) testCNProviderAdaptiveConnection(c *gin.Context, account *Account, modelID string, prompt string) error { testModelID := strings.TrimSpace(modelID) if testModelID == "" { testModelID = openai.DefaultTestModel } testModelID = account.GetMappedModel(testModelID) authToken := strings.TrimSpace(account.GetOpenAIProtocolAPIKey()) if authToken == "" { return s.sendErrorAndEnd(c, "No API key available") } // The existing Chat probe owns the SSE lifecycle. Suppress intermediate // completion events until every native adaptive endpoint has passed. c.Set(accountTestSuppressCompletionContextKey, true) defer c.Set(accountTestSuppressCompletionContextKey, false) if err := s.testCNProviderChatCompletionsConnection(c, account, modelID, prompt); err != nil { return err } if err := s.testCNProviderAdaptiveAnthropicConnection(c, account, testModelID, authToken); err != nil { return err } if account.SupportsNativeCNResponses() { if err := s.testCNProviderAdaptiveResponsesConnection(c, account, testModelID, authToken); err != nil { return err } } c.Set(accountTestSuppressCompletionContextKey, false) s.sendEvent(c, TestEvent{Type: "test_complete", Success: true}) return nil } func (s *AccountTestService) testCNProviderAdaptiveAnthropicConnection(c *gin.Context, account *Account, testModelID string, authToken string) error { ctx := c.Request.Context() baseURL, err := s.validateUpstreamBaseURL(account.GetCNProtocolBaseURL(APIProtocolAnthropic)) if err != nil { return s.sendErrorAndEnd(c, fmt.Sprintf("Invalid adaptive Anthropic base URL: %s", err.Error())) } apiURL := strings.TrimRight(baseURL, "/") + "/v1/messages" payload, err := createTestPayload(testModelID) if err != nil { return s.sendErrorAndEnd(c, "Failed to create adaptive Anthropic test payload") } applyIntelligentPayloadPrompt(ctx, payload) payloadBytes, _ := json.Marshal(payload) s.sendEvent(c, TestEvent{Type: "status", Text: "正在通过原生 /v1/messages 测试自适应 Anthropic 端点"}) req, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, bytes.NewReader(payloadBytes)) if err != nil { return s.sendErrorAndEnd(c, "Failed to create adaptive Anthropic request") } req.Header.Set("Content-Type", "application/json") req.Header.Set("Accept", "text/event-stream") req.Header.Set("anthropic-version", "2023-06-01") for key, value := range claude.DefaultHeaders { req.Header.Set(key, value) } req.Header.Set("anthropic-beta", claude.APIKeyBetaHeader) // Ollama Cloud Anthropic 兼容端点按 adaptive 实际选用的 Anthropic // base_url 强制 Bearer,其余保持 extra/default 行为。 setAnthropicAPIKeyAuthHeader(req.Header, account, authToken, account.GetCNProtocolBaseURL(APIProtocolAnthropic)) account.ApplyHeaderOverrides(req.Header) resp, err := s.doCNProviderAdaptiveRequest(req, account) if err != nil { return s.sendErrorAndEnd(c, fmt.Sprintf("Adaptive Anthropic endpoint request failed: %s", err.Error())) } defer func() { _ = resp.Body.Close() }() if resp.StatusCode != http.StatusOK { body, _ := io.ReadAll(resp.Body) errMsg := fmt.Sprintf("Adaptive Anthropic endpoint returned %d: %s", resp.StatusCode, string(body)) if resp.StatusCode == http.StatusUnauthorized && s.accountRepo != nil { _ = s.accountRepo.SetError(ctx, account.ID, errMsg) } return s.sendErrorAndEnd(c, errMsg) } if err := s.processCNProviderAdaptiveAnthropicStream(c, resp.Body); err != nil { return err } s.sendEvent(c, TestEvent{Type: "status", Text: "已通过原生 /v1/messages 验证"}) return nil } func (s *AccountTestService) processCNProviderAdaptiveAnthropicStream(c *gin.Context, body io.Reader) error { reader := bufio.NewReader(body) for { line, err := reader.ReadString('\n') if err != nil { if err == io.EOF { return s.sendErrorAndEnd(c, "Adaptive Anthropic stream ended before message_stop") } return s.sendErrorAndEnd(c, fmt.Sprintf("Adaptive Anthropic stream read error: %s", err.Error())) } line = strings.TrimSpace(line) if line == "" || !sseDataPrefix.MatchString(line) { continue } jsonStr := sseDataPrefix.ReplaceAllString(line, "") if jsonStr == "[DONE]" { return nil } var data map[string]any if err := json.Unmarshal([]byte(jsonStr), &data); err != nil { continue } switch eventType, _ := data["type"].(string); eventType { case "content_block_delta": if delta, ok := data["delta"].(map[string]any); ok { if text, ok := delta["text"].(string); ok && text != "" { s.sendEvent(c, TestEvent{Type: "content", Text: text}) } } case "message_stop": return nil case "error": errorMsg := "Unknown error" if errData, ok := data["error"].(map[string]any); ok { if message, ok := errData["message"].(string); ok && message != "" { errorMsg = message } } return s.sendErrorAndEnd(c, fmt.Sprintf("Adaptive Anthropic endpoint error: %s", errorMsg)) } } } func (s *AccountTestService) testCNProviderAdaptiveResponsesConnection(c *gin.Context, account *Account, testModelID string, authToken string) error { ctx := c.Request.Context() baseURL, err := s.validateUpstreamBaseURL(account.GetCNProtocolBaseURL(APIProtocolResponses)) if err != nil { return s.sendErrorAndEnd(c, fmt.Sprintf("Invalid adaptive Responses base URL: %s", err.Error())) } apiURL := buildOpenAIResponsesURLForPlatform(account.Platform, baseURL) payload := createOpenAITestPayload(testModelID, false) // DeepSeek / Kimi native Responses endpoints are stateless and do not need // the OpenAI probe's synthetic instructions. delete(payload, "instructions") applyIntelligentPayloadPrompt(ctx, payload) payloadBytes, _ := json.Marshal(payload) payloadBytes = normalizeDeepSeekResponsesRequestBody(account, payloadBytes) s.sendEvent(c, TestEvent{Type: "status", Text: "正在通过原生 /responses 测试自适应 Responses 端点"}) req, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, bytes.NewReader(payloadBytes)) if err != nil { return s.sendErrorAndEnd(c, "Failed to create adaptive Responses request") } req = req.WithContext(WithHTTPUpstreamProfile(req.Context(), HTTPUpstreamProfileOpenAI)) req.Header.Set("Content-Type", "application/json") req.Header.Set("Accept", "text/event-stream") req.Header.Set("Authorization", "Bearer "+authToken) applyOpenAICodexProbeHeaders(req.Header) account.ApplyHeaderOverrides(req.Header) resp, err := s.doCNProviderAdaptiveRequest(req, account) if err != nil { return s.sendErrorAndEnd(c, fmt.Sprintf("Adaptive Responses endpoint request failed: %s", err.Error())) } defer func() { _ = resp.Body.Close() }() if resp.StatusCode != http.StatusOK { body, _ := io.ReadAll(resp.Body) errMsg := fmt.Sprintf("Adaptive Responses endpoint returned %d: %s", resp.StatusCode, string(body)) if resp.StatusCode == http.StatusUnauthorized && s.accountRepo != nil { _ = s.accountRepo.SetError(ctx, account.ID, errMsg) } return s.sendErrorAndEnd(c, errMsg) } if err := s.processOpenAIStream(c, resp.Body); err != nil { return err } s.sendEvent(c, TestEvent{Type: "status", Text: "已通过原生 /responses 验证"}) return nil } func (s *AccountTestService) doCNProviderAdaptiveRequest(req *http.Request, account *Account) (*http.Response, error) { proxyURL := "" if account.ProxyID != nil && account.Proxy != nil { proxyURL = account.Proxy.URL() } return s.httpUpstream.DoWithTLS(WithAccountTrafficRequest(req, account), proxyURL, account.ID, account.Concurrency, s.tlsFPProfileService.ResolveTLSProfile(account)) } // testCNProviderAnthropicConnection verifies the native Anthropic endpoint of a // CN-provider account explicitly configured with api_protocol=anthropic. Before // this path existed such accounts fell through to the generic Claude tester,// which (a) appended ?beta=true and (b) defaulted a missing base_url to // https://api.anthropic.com — sending the provider's API key to Anthropic // instead of the provider's own Anthropic-compatible endpoint. The probe uses // GetAnthropicProtocolBaseURL (same resolution as real /v1/messages forwarding, // including per-platform defaults) and the shared API-key auth header. func (s *AccountTestService) testCNProviderAnthropicConnection(c *gin.Context, account *Account, modelID string) error { ctx := c.Request.Context() testModelID := strings.TrimSpace(modelID) if testModelID == "" { testModelID = claude.DefaultTestModel } testModelID = account.GetMappedModel(testModelID) authToken := strings.TrimSpace(account.GetOpenAIProtocolAPIKey()) if authToken == "" { return s.sendErrorAndEnd(c, "No API key available") } baseURL, err := s.validateUpstreamBaseURL(account.GetAnthropicProtocolBaseURL()) if err != nil { return s.sendErrorAndEnd(c, fmt.Sprintf("Invalid Anthropic base URL: %s", err.Error())) } if hint := cnAnthropicBaseURLMisconfigHint(baseURL); hint != "" { return s.sendErrorAndEnd(c, hint) } apiURL := strings.TrimRight(baseURL, "/") + "/v1/messages" c.Writer.Header().Set("Content-Type", "text/event-stream") c.Writer.Header().Set("Cache-Control", "no-cache") c.Writer.Header().Set("Connection", "keep-alive") c.Writer.Header().Set("X-Accel-Buffering", "no") c.Writer.Flush() payload, err := createTestPayload(testModelID) if err != nil { return s.sendErrorAndEnd(c, "Failed to create Anthropic test payload") } applyIntelligentPayloadPrompt(ctx, payload) payloadBytes, _ := json.Marshal(payload) s.sendEvent(c, TestEvent{Type: "test_start", Model: testModelID}) req, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, bytes.NewReader(payloadBytes)) if err != nil { return s.sendErrorAndEnd(c, "Failed to create Anthropic test request") } req.Header.Set("Content-Type", "application/json") req.Header.Set("Accept", "text/event-stream") req.Header.Set("anthropic-version", "2023-06-01") for key, value := range claude.DefaultHeaders { req.Header.Set(key, value) } // Ollama Cloud Anthropic 兼容端点按实际 base_url 强制 Bearer,其余保持 // extra/default 行为。 setAnthropicAPIKeyAuthHeader(req.Header, account, authToken, account.GetAnthropicProtocolBaseURL()) account.ApplyHeaderOverrides(req.Header) resp, err := s.doCNProviderAdaptiveRequest(req, account) if err != nil { return s.sendErrorAndEnd(c, fmt.Sprintf("Anthropic endpoint request failed: %s", err.Error())) } defer func() { _ = resp.Body.Close() }() if resp.StatusCode != http.StatusOK { body, _ := io.ReadAll(resp.Body) errMsg := fmt.Sprintf("Anthropic endpoint returned %d: %s", resp.StatusCode, string(body)) if (resp.StatusCode == http.StatusUnauthorized || resp.StatusCode == http.StatusForbidden) && s.accountRepo != nil { _ = s.accountRepo.SetError(ctx, account.ID, errMsg) } return s.sendErrorAndEnd(c, errMsg) } return s.processClaudeStream(c, resp.Body) } // cnAnthropicBaseURLMisconfigHint reports an actionable error when an // anthropic-protocol account's base_url still points at an OpenAI-compatible // endpoint (paas path, version segment, or chat/completions / responses // suffix). The naive {base}/v1/messages join would 404 (e.g. // .../api/paas/v4/v1/messages) with no hint about the actual misconfiguration. func cnAnthropicBaseURLMisconfigHint(baseURL string) string { parsed, err := url.Parse(strings.TrimSpace(baseURL)) if err != nil || parsed.Scheme == "" || parsed.Host == "" { return "" } path := strings.ToLower(strings.TrimRight(parsed.Path, "/")) if path == "" { return "" } openAICompatShaped := strings.Contains(path, "/paas/") || strings.HasSuffix(path, "/chat/completions") || strings.HasSuffix(path, "/responses") || openAIBaseURLHasVersionSuffix(path) if !openAICompatShaped { return "" } return fmt.Sprintf( "API protocol is anthropic but base_url (%s) looks like an OpenAI-compatible endpoint; "+ "requests would hit {base}/v1/messages and 404. Set base_url to the provider's Anthropic endpoint "+ "(e.g. https://open.bigmodel.cn/api/anthropic) or switch api_protocol to chat_completions/adaptive.", baseURL, ) }