Upload 10 files

internal/service/provider/anthropic.go

@@ -0,0 +1,164 @@
+package provider
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+
+	"github.com/anthropics/anthropic-sdk-go"
+	"github.com/anthropics/anthropic-sdk-go/option"
+	"github.com/anthropics/anthropic-sdk-go/packages/ssestream"
+	"zencoder-2api/internal/model"
+)
+
+const DefaultAnthropicBaseURL = "https://api.anthropic.com"
+
+type AnthropicProvider struct {
+	client *anthropic.Client
+	config Config
+}
+
+func NewAnthropicProvider(cfg Config) *AnthropicProvider {
+	if cfg.BaseURL == "" {
+		cfg.BaseURL = DefaultAnthropicBaseURL
+	}
+
+	opts := []option.RequestOption{
+		option.WithAPIKey(cfg.APIKey),
+		option.WithBaseURL(cfg.BaseURL),
+	}
+
+	for k, v := range cfg.ExtraHeaders {
+		opts = append(opts, option.WithHeader(k, v))
+	}
+
+	if cfg.Proxy != "" {
+		httpClient := NewHTTPClient(cfg.Proxy, 0)
+		opts = append(opts, option.WithHTTPClient(httpClient))
+	}
+
+	return &AnthropicProvider{
+		client: anthropic.NewClient(opts...),
+		config: cfg,
+	}
+}
+
+func (p *AnthropicProvider) Name() string {
+	return "anthropic"
+}
+
+func (p *AnthropicProvider) ValidateToken() error {
+	_, err := p.client.Messages.New(context.Background(), anthropic.MessageNewParams{
+		Model:     anthropic.F(anthropic.ModelClaude3_5SonnetLatest),
+		MaxTokens: anthropic.F(int64(1)),
+		Messages:  anthropic.F([]anthropic.MessageParam{{Role: anthropic.F(anthropic.MessageParamRoleUser), Content: anthropic.F([]anthropic.ContentBlockParamUnion{anthropic.NewTextBlock("hi")})}}),
+	})
+	return err
+}
+
+func (p *AnthropicProvider) Chat(req *model.ChatCompletionRequest) (*model.ChatCompletionResponse, error) {
+	messages := p.convertMessages(req.Messages)
+
+	maxTokens := int64(4096)
+	if req.MaxTokens > 0 {
+		maxTokens = int64(req.MaxTokens)
+	}
+
+	resp, err := p.client.Messages.New(context.Background(), anthropic.MessageNewParams{
+		Model:     anthropic.F(req.Model),
+		MaxTokens: anthropic.F(maxTokens),
+		Messages:  anthropic.F(messages),
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return p.convertResponse(resp), nil
+}
+
+func (p *AnthropicProvider) convertMessages(msgs []model.ChatMessage) []anthropic.MessageParam {
+	var messages []anthropic.MessageParam
+	for _, msg := range msgs {
+		if msg.Role == "system" {
+			continue
+		}
+		role := anthropic.MessageParamRoleUser
+		if msg.Role == "assistant" {
+			role = anthropic.MessageParamRoleAssistant
+		}
+		messages = append(messages, anthropic.MessageParam{
+			Role:    anthropic.F(role),
+			Content: anthropic.F([]anthropic.ContentBlockParamUnion{anthropic.NewTextBlock(msg.Content)}),
+		})
+	}
+	return messages
+}
+
+func (p *AnthropicProvider) convertResponse(resp *anthropic.Message) *model.ChatCompletionResponse {
+	content := ""
+	for _, block := range resp.Content {
+		if block.Type == anthropic.ContentBlockTypeText {
+			content += block.Text
+		}
+	}
+
+	return &model.ChatCompletionResponse{
+		ID:      resp.ID,
+		Object:  "chat.completion",
+		Model:   string(resp.Model),
+		Choices: []model.Choice{{
+			Index:        0,
+			Message:      model.ChatMessage{Role: "assistant", Content: content},
+			FinishReason: string(resp.StopReason),
+		}},
+		Usage: model.Usage{
+			PromptTokens:     int(resp.Usage.InputTokens),
+			CompletionTokens: int(resp.Usage.OutputTokens),
+			TotalTokens:      int(resp.Usage.InputTokens + resp.Usage.OutputTokens),
+		},
+	}
+}
+
+func (p *AnthropicProvider) ChatStream(req *model.ChatCompletionRequest, writer http.ResponseWriter) error {
+	messages := p.convertMessages(req.Messages)
+
+	maxTokens := int64(4096)
+	if req.MaxTokens > 0 {
+		maxTokens = int64(req.MaxTokens)
+	}
+
+	stream := p.client.Messages.NewStreaming(context.Background(), anthropic.MessageNewParams{
+		Model:     anthropic.F(req.Model),
+		MaxTokens: anthropic.F(maxTokens),
+		Messages:  anthropic.F(messages),
+	})
+
+	return p.handleStream(stream, writer)
+}
+
+func (p *AnthropicProvider) handleStream(stream *ssestream.Stream[anthropic.MessageStreamEvent], writer http.ResponseWriter) error {
+	flusher, ok := writer.(http.Flusher)
+	if !ok {
+		return fmt.Errorf("streaming not supported")
+	}
+
+	writer.Header().Set("Content-Type", "text/event-stream")
+	writer.Header().Set("Cache-Control", "no-cache")
+	writer.Header().Set("Connection", "keep-alive")
+
+	for stream.Next() {
+		event := stream.Current()
+		data, _ := json.Marshal(event)
+		fmt.Fprintf(writer, "data: %s\n\n", data)
+		flusher.Flush()
+	}
+
+	if err := stream.Err(); err != nil {
+		return err
+	}
+
+	fmt.Fprintf(writer, "data: [DONE]\n\n")
+	flusher.Flush()
+	return nil
+}

internal/service/provider/client.go

@@ -0,0 +1,69 @@
+package provider
+
+import (
+	"log"
+	"net/http"
+	"net/url"
+	"strings"
+	"time"
+)
+
+// NewHTTPClient 创建HTTP客户端
+// 支持HTTP和SOCKS5代理
+func NewHTTPClient(proxy string, timeout time.Duration) *http.Client {
+	// 如果代理是SOCKS5格式，使用新的代理客户端创建函数
+	if strings.HasPrefix(proxy, "socks5://") {
+		client, err := NewHTTPClientWithProxy(proxy, timeout)
+		if err != nil {
+			log.Printf("创建SOCKS5代理客户端失败: %v, 使用默认客户端", err)
+			client, _ := NewHTTPClientWithProxy("", timeout)
+			return client
+		}
+		return client
+	}
+	
+	// 原有的HTTP代理逻辑
+	transport := &http.Transport{}
+
+	if proxy != "" {
+		if proxyURL, err := url.Parse(proxy); err == nil {
+			transport.Proxy = http.ProxyURL(proxyURL)
+		}
+	}
+
+	if timeout == 0 {
+		timeout = 600 * time.Second // 10分钟超时，支持长时间流式响应
+	}
+
+	return &http.Client{
+		Transport: transport,
+		Timeout:   timeout,
+	}
+}
+
+// NewHTTPClientWithPoolProxy 使用代理池创建HTTP客户端
+func NewHTTPClientWithPoolProxy(useProxy bool, timeout time.Duration) *http.Client {
+	if !useProxy {
+		// 不使用代理
+		client, _ := NewHTTPClientWithProxy("", timeout)
+		return client
+	}
+	
+	pool := GetProxyPool()
+	if !pool.HasProxies() {
+		// 没有可用代理，使用默认客户端
+		client, _ := NewHTTPClientWithProxy("", timeout)
+		return client
+	}
+	
+	// 获取下一个代理
+	proxyURL := pool.GetNextProxy()
+	client, err := NewHTTPClientWithProxy(proxyURL, timeout)
+	if err != nil {
+		log.Printf("使用代理 %s 创建客户端失败: %v, 使用默认客户端", proxyURL, err)
+		client, _ := NewHTTPClientWithProxy("", timeout)
+		return client
+	}
+	
+	return client
+}

internal/service/provider/errors.go

@@ -0,0 +1,10 @@
+package provider
+
+import "errors"
+
+var (
+	ErrStreamNotSupported = errors.New("streaming not supported")
+	ErrInvalidToken       = errors.New("invalid token")
+	ErrRequestFailed      = errors.New("request failed")
+	ErrUnknownProvider    = errors.New("unknown provider")
+)

internal/service/provider/factory.go

@@ -0,0 +1,27 @@
+package provider
+
+import "fmt"
+
+type ProviderType string
+
+const (
+	ProviderOpenAI    ProviderType = "openai"
+	ProviderAnthropic ProviderType = "anthropic"
+	ProviderGemini    ProviderType = "gemini"
+	ProviderGrok      ProviderType = "grok"
+)
+
+func NewProvider(providerType ProviderType, cfg Config) (Provider, error) {
+	switch providerType {
+	case ProviderOpenAI:
+		return NewOpenAIProvider(cfg), nil
+	case ProviderAnthropic:
+		return NewAnthropicProvider(cfg), nil
+	case ProviderGemini:
+		return NewGeminiProvider(cfg)
+	case ProviderGrok:
+		return NewGrokProvider(cfg), nil
+	default:
+		return nil, fmt.Errorf("%w: %s", ErrUnknownProvider, providerType)
+	}
+}

internal/service/provider/gemini.go

@@ -0,0 +1,124 @@
+package provider
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+
+	"github.com/google/generative-ai-go/genai"
+	"google.golang.org/api/option"
+	"zencoder-2api/internal/model"
+)
+
+type GeminiProvider struct {
+	client *genai.Client
+	config Config
+}
+
+func NewGeminiProvider(cfg Config) (*GeminiProvider, error) {
+	ctx := context.Background()
+
+	opts := []option.ClientOption{
+		option.WithAPIKey(cfg.APIKey),
+	}
+
+	if cfg.BaseURL != "" {
+		opts = append(opts, option.WithEndpoint(cfg.BaseURL))
+	}
+
+	client, err := genai.NewClient(ctx, opts...)
+	if err != nil {
+		return nil, err
+	}
+
+	return &GeminiProvider{
+		client: client,
+		config: cfg,
+	}, nil
+}
+
+func (p *GeminiProvider) Name() string {
+	return "gemini"
+}
+
+func (p *GeminiProvider) ValidateToken() error {
+	model := p.client.GenerativeModel("gemini-1.5-flash")
+	_, err := model.GenerateContent(context.Background(), genai.Text("hi"))
+	return err
+}
+
+func (p *GeminiProvider) Chat(req *model.ChatCompletionRequest) (*model.ChatCompletionResponse, error) {
+	geminiModel := p.client.GenerativeModel(req.Model)
+
+	var parts []genai.Part
+	for _, msg := range req.Messages {
+		parts = append(parts, genai.Text(msg.Content))
+	}
+
+	resp, err := geminiModel.GenerateContent(context.Background(), parts...)
+	if err != nil {
+		return nil, err
+	}
+
+	return p.convertResponse(resp, req.Model), nil
+}
+
+func (p *GeminiProvider) convertResponse(resp *genai.GenerateContentResponse, modelName string) *model.ChatCompletionResponse {
+	content := ""
+	if len(resp.Candidates) > 0 && resp.Candidates[0].Content != nil {
+		for _, part := range resp.Candidates[0].Content.Parts {
+			if text, ok := part.(genai.Text); ok {
+				content += string(text)
+			}
+		}
+	}
+
+	return &model.ChatCompletionResponse{
+		ID:      "gemini-" + modelName,
+		Object:  "chat.completion",
+		Model:   modelName,
+		Choices: []model.Choice{{
+			Index:        0,
+			Message:      model.ChatMessage{Role: "assistant", Content: content},
+			FinishReason: "stop",
+		}},
+	}
+}
+
+func (p *GeminiProvider) ChatStream(req *model.ChatCompletionRequest, writer http.ResponseWriter) error {
+	geminiModel := p.client.GenerativeModel(req.Model)
+
+	var parts []genai.Part
+	for _, msg := range req.Messages {
+		parts = append(parts, genai.Text(msg.Content))
+	}
+
+	iter := geminiModel.GenerateContentStream(context.Background(), parts...)
+	return p.handleStream(iter, writer)
+}
+
+func (p *GeminiProvider) handleStream(iter *genai.GenerateContentResponseIterator, writer http.ResponseWriter) error {
+	flusher, ok := writer.(http.Flusher)
+	if !ok {
+		return fmt.Errorf("streaming not supported")
+	}
+
+	writer.Header().Set("Content-Type", "text/event-stream")
+	writer.Header().Set("Cache-Control", "no-cache")
+	writer.Header().Set("Connection", "keep-alive")
+
+	for {
+		resp, err := iter.Next()
+		if err != nil {
+			break
+		}
+		data, _ := json.Marshal(resp)
+		fmt.Fprintf(writer, "data: %s\n\n", data)
+		flusher.Flush()
+	}
+
+	fmt.Fprintf(writer, "data: [DONE]\n\n")
+	flusher.Flush()
+	return nil
+}

internal/service/provider/grok.go

@@ -0,0 +1,149 @@
+package provider
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+
+	"github.com/openai/openai-go"
+	"github.com/openai/openai-go/option"
+	"github.com/openai/openai-go/packages/ssestream"
+	"zencoder-2api/internal/model"
+)
+
+const DefaultGrokBaseURL = "https://api.x.ai/v1"
+
+type GrokProvider struct {
+	client *openai.Client
+	config Config
+}
+
+func NewGrokProvider(cfg Config) *GrokProvider {
+	if cfg.BaseURL == "" {
+		cfg.BaseURL = DefaultGrokBaseURL
+	}
+
+	opts := []option.RequestOption{
+		option.WithAPIKey(cfg.APIKey),
+		option.WithBaseURL(cfg.BaseURL),
+	}
+
+	for k, v := range cfg.ExtraHeaders {
+		opts = append(opts, option.WithHeader(k, v))
+	}
+
+	if cfg.Proxy != "" {
+		httpClient := NewHTTPClient(cfg.Proxy, 0)
+		opts = append(opts, option.WithHTTPClient(httpClient))
+	}
+
+	return &GrokProvider{
+		client: openai.NewClient(opts...),
+		config: cfg,
+	}
+}
+
+func (p *GrokProvider) Name() string {
+	return "grok"
+}
+
+func (p *GrokProvider) ValidateToken() error {
+	_, err := p.client.Models.List(context.Background())
+	return err
+}
+
+func (p *GrokProvider) Chat(req *model.ChatCompletionRequest) (*model.ChatCompletionResponse, error) {
+	messages := make([]openai.ChatCompletionMessageParamUnion, len(req.Messages))
+	for i, msg := range req.Messages {
+		switch msg.Role {
+		case "system":
+			messages[i] = openai.SystemMessage(msg.Content)
+		case "user":
+			messages[i] = openai.UserMessage(msg.Content)
+		case "assistant":
+			messages[i] = openai.AssistantMessage(msg.Content)
+		}
+	}
+
+	resp, err := p.client.Chat.Completions.New(context.Background(), openai.ChatCompletionNewParams{
+		Model:    openai.F(req.Model),
+		Messages: openai.F(messages),
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return p.convertResponse(resp), nil
+}
+
+func (p *GrokProvider) convertResponse(resp *openai.ChatCompletion) *model.ChatCompletionResponse {
+	choices := make([]model.Choice, len(resp.Choices))
+	for i, c := range resp.Choices {
+		choices[i] = model.Choice{
+			Index:        int(c.Index),
+			Message:      model.ChatMessage{Role: string(c.Message.Role), Content: c.Message.Content},
+			FinishReason: string(c.FinishReason),
+		}
+	}
+
+	return &model.ChatCompletionResponse{
+		ID:      resp.ID,
+		Object:  string(resp.Object),
+		Created: resp.Created,
+		Model:   resp.Model,
+		Choices: choices,
+		Usage: model.Usage{
+			PromptTokens:     int(resp.Usage.PromptTokens),
+			CompletionTokens: int(resp.Usage.CompletionTokens),
+			TotalTokens:      int(resp.Usage.TotalTokens),
+		},
+	}
+}
+
+func (p *GrokProvider) ChatStream(req *model.ChatCompletionRequest, writer http.ResponseWriter) error {
+	messages := make([]openai.ChatCompletionMessageParamUnion, len(req.Messages))
+	for i, msg := range req.Messages {
+		switch msg.Role {
+		case "system":
+			messages[i] = openai.SystemMessage(msg.Content)
+		case "user":
+			messages[i] = openai.UserMessage(msg.Content)
+		case "assistant":
+			messages[i] = openai.AssistantMessage(msg.Content)
+		}
+	}
+
+	stream := p.client.Chat.Completions.NewStreaming(context.Background(), openai.ChatCompletionNewParams{
+		Model:    openai.F(req.Model),
+		Messages: openai.F(messages),
+	})
+
+	return p.handleStream(stream, writer)
+}
+
+func (p *GrokProvider) handleStream(stream *ssestream.Stream[openai.ChatCompletionChunk], writer http.ResponseWriter) error {
+	flusher, ok := writer.(http.Flusher)
+	if !ok {
+		return fmt.Errorf("streaming not supported")
+	}
+
+	writer.Header().Set("Content-Type", "text/event-stream")
+	writer.Header().Set("Cache-Control", "no-cache")
+	writer.Header().Set("Connection", "keep-alive")
+
+	for stream.Next() {
+		chunk := stream.Current()
+		data, _ := json.Marshal(chunk)
+		fmt.Fprintf(writer, "data: %s\n\n", data)
+		flusher.Flush()
+	}
+
+	if err := stream.Err(); err != nil {
+		return err
+	}
+
+	fmt.Fprintf(writer, "data: [DONE]\n\n")
+	flusher.Flush()
+	return nil
+}

internal/service/provider/manager.go

@@ -0,0 +1,73 @@
+package provider
+
+import (
+	"fmt"
+	"sync"
+
+	"zencoder-2api/internal/model"
+)
+
+// Manager Provider管理器，缓存已创建的provider实例
+type Manager struct {
+	mu        sync.RWMutex
+	providers map[string]Provider
+}
+
+var defaultManager = &Manager{
+	providers: make(map[string]Provider),
+}
+
+// GetManager 获取默认管理器
+func GetManager() *Manager {
+	return defaultManager
+}
+
+// GetProvider 根据账号和模型获取或创建Provider
+func (m *Manager) GetProvider(accountID uint, zenModel model.ZenModel, cfg Config) (Provider, error) {
+	key := m.buildKey(accountID, zenModel.ProviderID)
+
+	m.mu.RLock()
+	if p, ok := m.providers[key]; ok {
+		m.mu.RUnlock()
+		return p, nil
+	}
+	m.mu.RUnlock()
+
+	return m.createProvider(key, zenModel.ProviderID, cfg)
+}
+
+func (m *Manager) buildKey(accountID uint, providerID string) string {
+	return fmt.Sprintf("%d:%s", accountID, providerID)
+}
+
+func (m *Manager) createProvider(key, providerID string, cfg Config) (Provider, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	// 双重检查
+	if p, ok := m.providers[key]; ok {
+		return p, nil
+	}
+
+	var providerType ProviderType
+	switch providerID {
+	case "openai":
+		providerType = ProviderOpenAI
+	case "anthropic":
+		providerType = ProviderAnthropic
+	case "gemini":
+		providerType = ProviderGemini
+	case "xai":
+		providerType = ProviderGrok
+	default:
+		providerType = ProviderAnthropic
+	}
+
+	p, err := NewProvider(providerType, cfg)
+	if err != nil {
+		return nil, err
+	}
+
+	m.providers[key] = p
+	return p, nil
+}

internal/service/provider/openai.go

@@ -0,0 +1,149 @@
+package provider
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+
+	"github.com/openai/openai-go"
+	"github.com/openai/openai-go/option"
+	"github.com/openai/openai-go/packages/ssestream"
+	"zencoder-2api/internal/model"
+)
+
+const DefaultOpenAIBaseURL = "https://api.openai.com/v1"
+
+type OpenAIProvider struct {
+	client *openai.Client
+	config Config
+}
+
+func NewOpenAIProvider(cfg Config) *OpenAIProvider {
+	if cfg.BaseURL == "" {
+		cfg.BaseURL = DefaultOpenAIBaseURL
+	}
+
+	opts := []option.RequestOption{
+		option.WithAPIKey(cfg.APIKey),
+		option.WithBaseURL(cfg.BaseURL),
+	}
+
+	for k, v := range cfg.ExtraHeaders {
+		opts = append(opts, option.WithHeader(k, v))
+	}
+
+	if cfg.Proxy != "" {
+		httpClient := NewHTTPClient(cfg.Proxy, 0)
+		opts = append(opts, option.WithHTTPClient(httpClient))
+	}
+
+	return &OpenAIProvider{
+		client: openai.NewClient(opts...),
+		config: cfg,
+	}
+}
+
+func (p *OpenAIProvider) Name() string {
+	return "openai"
+}
+
+func (p *OpenAIProvider) ValidateToken() error {
+	_, err := p.client.Models.List(context.Background())
+	return err
+}
+
+func (p *OpenAIProvider) Chat(req *model.ChatCompletionRequest) (*model.ChatCompletionResponse, error) {
+	messages := make([]openai.ChatCompletionMessageParamUnion, len(req.Messages))
+	for i, msg := range req.Messages {
+		switch msg.Role {
+		case "system":
+			messages[i] = openai.SystemMessage(msg.Content)
+		case "user":
+			messages[i] = openai.UserMessage(msg.Content)
+		case "assistant":
+			messages[i] = openai.AssistantMessage(msg.Content)
+		}
+	}
+
+	resp, err := p.client.Chat.Completions.New(context.Background(), openai.ChatCompletionNewParams{
+		Model:    openai.F(req.Model),
+		Messages: openai.F(messages),
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return p.convertResponse(resp), nil
+}
+
+func (p *OpenAIProvider) convertResponse(resp *openai.ChatCompletion) *model.ChatCompletionResponse {
+	choices := make([]model.Choice, len(resp.Choices))
+	for i, c := range resp.Choices {
+		choices[i] = model.Choice{
+			Index:        int(c.Index),
+			Message:      model.ChatMessage{Role: string(c.Message.Role), Content: c.Message.Content},
+			FinishReason: string(c.FinishReason),
+		}
+	}
+
+	return &model.ChatCompletionResponse{
+		ID:      resp.ID,
+		Object:  string(resp.Object),
+		Created: resp.Created,
+		Model:   resp.Model,
+		Choices: choices,
+		Usage: model.Usage{
+			PromptTokens:     int(resp.Usage.PromptTokens),
+			CompletionTokens: int(resp.Usage.CompletionTokens),
+			TotalTokens:      int(resp.Usage.TotalTokens),
+		},
+	}
+}
+
+func (p *OpenAIProvider) ChatStream(req *model.ChatCompletionRequest, writer http.ResponseWriter) error {
+	messages := make([]openai.ChatCompletionMessageParamUnion, len(req.Messages))
+	for i, msg := range req.Messages {
+		switch msg.Role {
+		case "system":
+			messages[i] = openai.SystemMessage(msg.Content)
+		case "user":
+			messages[i] = openai.UserMessage(msg.Content)
+		case "assistant":
+			messages[i] = openai.AssistantMessage(msg.Content)
+		}
+	}
+
+	stream := p.client.Chat.Completions.NewStreaming(context.Background(), openai.ChatCompletionNewParams{
+		Model:    openai.F(req.Model),
+		Messages: openai.F(messages),
+	})
+
+	return p.handleStream(stream, writer)
+}
+
+func (p *OpenAIProvider) handleStream(stream *ssestream.Stream[openai.ChatCompletionChunk], writer http.ResponseWriter) error {
+	flusher, ok := writer.(http.Flusher)
+	if !ok {
+		return fmt.Errorf("streaming not supported")
+	}
+
+	writer.Header().Set("Content-Type", "text/event-stream")
+	writer.Header().Set("Cache-Control", "no-cache")
+	writer.Header().Set("Connection", "keep-alive")
+
+	for stream.Next() {
+		chunk := stream.Current()
+		data, _ := json.Marshal(chunk)
+		fmt.Fprintf(writer, "data: %s\n\n", data)
+		flusher.Flush()
+	}
+
+	if err := stream.Err(); err != nil {
+		return err
+	}
+
+	fmt.Fprintf(writer, "data: [DONE]\n\n")
+	flusher.Flush()
+	return nil
+}

internal/service/provider/proxy.go

@@ -0,0 +1,251 @@
+package provider
+
+import (
+	"context"
+	"fmt"
+	"math/rand"
+	"net"
+	"net/http"
+	"net/url"
+	"os"
+	"strings"
+	"sync"
+	"time"
+
+	"golang.org/x/net/proxy"
+)
+
+// ProxyPool 代理池管理器
+type ProxyPool struct {
+	proxies []string
+	mu      sync.RWMutex
+	index   int
+}
+
+var (
+	globalProxyPool *ProxyPool
+	once           sync.Once
+)
+
+// GetProxyPool 获取全局代理池实例
+func GetProxyPool() *ProxyPool {
+	once.Do(func() {
+		globalProxyPool = NewProxyPool()
+	})
+	return globalProxyPool
+}
+
+// NewProxyPool 创建新的代理池
+func NewProxyPool() *ProxyPool {
+	pool := &ProxyPool{
+		proxies: make([]string, 0),
+	}
+	pool.loadProxiesFromEnv()
+	return pool
+}
+
+// loadProxiesFromEnv 从环境变量加载代理列表
+func (p *ProxyPool) loadProxiesFromEnv() {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	
+	proxyEnv := os.Getenv("SOCKS_PROXY_POOL")
+	if proxyEnv == "" {
+		return
+	}
+	
+	// 解析逗号分隔的代理列表
+	proxiesStr := strings.Split(proxyEnv, ",")
+	for _, proxyStr := range proxiesStr {
+		proxyStr = strings.TrimSpace(proxyStr)
+		if proxyStr != "" {
+			p.proxies = append(p.proxies, proxyStr)
+		}
+	}
+}
+
+// GetNextProxy 获取下一个代理(轮询方式)
+func (p *ProxyPool) GetNextProxy() string {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	
+	if len(p.proxies) == 0 {
+		return ""
+	}
+	
+	proxy := p.proxies[p.index]
+	p.index = (p.index + 1) % len(p.proxies)
+	return proxy
+}
+
+// GetRandomProxy 获取随机代理
+func (p *ProxyPool) GetRandomProxy() string {
+	p.mu.RLock()
+	defer p.mu.RUnlock()
+	
+	if len(p.proxies) == 0 {
+		return ""
+	}
+	
+	index := rand.Intn(len(p.proxies))
+	return p.proxies[index]
+}
+
+// HasProxies 检查是否有可用代理
+func (p *ProxyPool) HasProxies() bool {
+	p.mu.RLock()
+	defer p.mu.RUnlock()
+	return len(p.proxies) > 0
+}
+
+// Count 返回代理数量
+func (p *ProxyPool) Count() int {
+	p.mu.RLock()
+	defer p.mu.RUnlock()
+	return len(p.proxies)
+}
+
+// GetAllProxies 获取所有代理列表(用于测试)
+func (p *ProxyPool) GetAllProxies() []string {
+	p.mu.RLock()
+	defer p.mu.RUnlock()
+	result := make([]string, len(p.proxies))
+	copy(result, p.proxies)
+	return result
+}
+
+// createSOCKS5Transport 创建SOCKS5代理传输层
+func createSOCKS5Transport(proxyURL string, timeout time.Duration) (*http.Transport, error) {
+	// 处理自定义格式：socks5://host:port:username:password
+	// 转换为标准格式：socks5://username:password@host:port
+	if strings.Contains(proxyURL, "socks5://") && strings.Count(proxyURL, ":") == 4 {
+		// 解析自定义格式
+		parts := strings.Split(proxyURL, ":")
+		if len(parts) == 5 {
+			// parts[0] = "socks5", parts[1] = "//host", parts[2] = "port", parts[3] = "username", parts[4] = "password"
+			host := strings.TrimPrefix(parts[1], "//")
+			port := parts[2]
+			username := parts[3]
+			password := parts[4]
+			
+			// 重构为标准URL格式
+			proxyURL = fmt.Sprintf("socks5://%s:%s@%s:%s", username, password, host, port)
+		}
+	}
+	
+	u, err := url.Parse(proxyURL)
+	if err != nil {
+		return nil, fmt.Errorf("解析代理URL失败: %v", err)
+	}
+	
+	if u.Scheme != "socks5" {
+		return nil, fmt.Errorf("仅支持SOCKS5代理")
+	}
+	
+	// 解析用户名和密码
+	var auth *proxy.Auth
+	if u.User != nil {
+		password, _ := u.User.Password()
+		auth = &proxy.Auth{
+			User:     u.User.Username(),
+			Password: password,
+		}
+	}
+	
+	// 创建SOCKS5拨号器
+	dialer, err := proxy.SOCKS5("tcp", u.Host, auth, proxy.Direct)
+	if err != nil {
+		return nil, fmt.Errorf("创建SOCKS5拨号器失败: %v", err)
+	}
+	
+	transport := &http.Transport{
+		DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
+			return dialer.Dial(network, addr)
+		},
+		MaxIdleConns:        100,
+		IdleConnTimeout:     90 * time.Second,
+		TLSHandshakeTimeout: 10 * time.Second,
+	}
+	
+	return transport, nil
+}
+
+// parseCustomProxyURL 解析自定义代理URL格式
+func parseCustomProxyURL(proxyURL string) string {
+	// 处理自定义格式：socks5://host:port:username:password
+	// 转换为标准格式：socks5://username:password@host:port
+	if strings.Contains(proxyURL, "socks5://") && strings.Count(proxyURL, ":") == 4 {
+		// 解析自定义格式
+		parts := strings.Split(proxyURL, ":")
+		if len(parts) == 5 {
+			// parts[0] = "socks5", parts[1] = "//host", parts[2] = "port", parts[3] = "username", parts[4] = "password"
+			host := strings.TrimPrefix(parts[1], "//")
+			port := parts[2]
+			username := parts[3]
+			password := parts[4]
+			
+			// 重构为标准URL格式
+			return fmt.Sprintf("socks5://%s:%s@%s:%s", username, password, host, port)
+		}
+	}
+	return proxyURL
+}
+
+// NewHTTPClientWithProxy 创建带指定代理的HTTP客户端
+func NewHTTPClientWithProxy(proxyURL string, timeout time.Duration) (*http.Client, error) {
+	if timeout == 0 {
+		timeout = 600 * time.Second // 10分钟超时，支持长时间流式响应
+	}
+	
+	if proxyURL == "" {
+		// 没有代理，使用默认客户端
+		return &http.Client{
+			Transport: &http.Transport{
+				DialContext: (&net.Dialer{
+					Timeout:   30 * time.Second,
+					KeepAlive: 30 * time.Second,
+				}).DialContext,
+				MaxIdleConns:        100,
+				IdleConnTimeout:     90 * time.Second,
+				TLSHandshakeTimeout: 10 * time.Second,
+			},
+			Timeout: timeout,
+		}, nil
+	}
+	
+	// 转换自定义格式到标准格式
+	standardURL := parseCustomProxyURL(proxyURL)
+	
+	// 解析代理URL
+	u, err := url.Parse(standardURL)
+	if err != nil {
+		return nil, fmt.Errorf("解析代理URL失败: %v", err)
+	}
+	
+	var transport *http.Transport
+	
+	if u.Scheme == "socks5" {
+		// SOCKS5代理 - 使用转换后的标准URL
+		transport, err = createSOCKS5Transport(standardURL, timeout)
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		// HTTP代理
+		transport = &http.Transport{
+			Proxy: http.ProxyURL(u),
+			DialContext: (&net.Dialer{
+				Timeout:   30 * time.Second,
+				KeepAlive: 30 * time.Second,
+			}).DialContext,
+			MaxIdleConns:        100,
+			IdleConnTimeout:     90 * time.Second,
+			TLSHandshakeTimeout: 10 * time.Second,
+		}
+	}
+	
+	return &http.Client{
+		Transport: transport,
+		Timeout:   timeout,
+	}, nil
+}