// Package oss 提供统一的 OSS 上传封装。 // // 收敛全仓 4+ 处逐字复制的 multipart 上传(model-gateway/gateway/ai-agent/black-deacon) // 与 media 的裸 client 手动解析,纯 HTTP 调用 oss 服务的 file/uploadFile 接口,不引入 minio SDK。 // // 兼容红线:默认 client 惰性构建并挂载全局注册中心(保留 "oss" 服务名解析), // X-User-Info 注入统一三套旧写法(透传请求头 / ctx 注入 user / 解析 token), // 响应解析对齐 ghttp.DefaultHandlerResponse 的 {code,message,data}。 package oss import ( "bytes" "context" "errors" "fmt" "mime/multipart" "net/http" "sync" "time" "gitea.redpowerfuture.com/red-future/common/utils" "github.com/gogf/gf/v2/net/gclient" "github.com/gogf/gf/v2/net/ghttp" "github.com/gogf/gf/v2/net/gsvc" "github.com/gogf/gf/v2/util/gconv" ) // UploadFileRes 对齐 oss/model/dto/file_dto.go:39 的 UploadFileRes。 type UploadFileRes struct { FileURL string `json:"fileURL" dc:"上传地址"` FileSize int `json:"fileSize" dc:"文件大小"` FileName string `json:"fileName" dc:"文件名称"` FileFormat string `json:"fileFormat" dc:"文件格式"` FileAddressPrefix string `json:"fileAddressPrefix"` FileMinioAddressPrefix string `json:"fileMinioAddressPrefix"` Items []*UploadFileRes `json:"files"` } // defaultEndpoint oss 服务上传端点("oss" 由注册中心解析为实际地址,多副本负载均衡)。 const defaultEndpoint = "oss/file/uploadFile" // uploadConfig 单次上传配置(Option 可变参数收集)。 type uploadConfig struct { endpoint string client *gclient.Client header map[string]string headerTimeout time.Duration // ResponseHeaderTimeout(大文件上传首字节等待) timeout time.Duration // 整体超时 } // defaultClientOnce 惰性构建默认上传客户端(sync.Once 保证只建一次)。 var ( defaultClientOnce sync.Once defaultClient *gclient.Client ) // defaultHTTPClient 返回默认上传客户端: // 生产环境各服务 main 已空导入 common/consul(其 init 内 gsvc.SetRegistry),此处挂上全局 registry, // 使 "oss" 服务名可被解析(等效 commonHttp.Httpclient 的 SetDiscovery 行为); // 测试/无注册中心时 GetRegistry() 为 nil,退化为直连 client。 // 不直接复用 commonHttp.Httpclient:其空导入 common/consul,而 consul init 的 g.Cfg().MustGet // 在无配置文件环境(单元测试)会 panic,本实现避开该副作用。 func defaultHTTPClient() *gclient.Client { defaultClientOnce.Do(func() { client := gclient.New() if reg := gsvc.GetRegistry(); reg != nil { client.SetDiscovery(reg) } defaultClient = client }) return defaultClient } func defaultConfig() *uploadConfig { return &uploadConfig{ endpoint: defaultEndpoint, client: defaultHTTPClient(), } } func buildConfig(opts []Option) *uploadConfig { cfg := defaultConfig() for _, opt := range opts { opt(cfg) } return cfg } // Option 上传可选项。 type Option func(*uploadConfig) // WithEndpoint 覆盖上传端点(默认 oss/file/uploadFile;测试可指 httptest server)。 func WithEndpoint(url string) Option { return func(c *uploadConfig) { c.endpoint = url } } // WithHeader 追加/覆盖请求头(如 X-User-Info),优先级高于透传与 ctx 注入。 func WithHeader(k, v string) Option { return func(c *uploadConfig) { if c.header == nil { c.header = make(map[string]string) } c.header[k] = v } } // WithResponseHeaderTimeout 调大响应头等待超时(大文件上传,media 用 5m)。 func WithResponseHeaderTimeout(d time.Duration) Option { return func(c *uploadConfig) { c.headerTimeout = d } } // WithTimeout 调大整体超时(大文件上传,media 用 10m)。 func WithTimeout(d time.Duration) Option { return func(c *uploadConfig) { c.timeout = d } } // UploadFileBytes 上传文件字节到 OSS,返回完整响应(FileURL/FileFormat 等)。 // multipart field 名固定 "file"(oss 端 dto.UploadFileReq.File *ghttp.UploadFile 按此解析)。 func UploadFileBytes(ctx context.Context, fileName string, data []byte, opts ...Option) (*UploadFileRes, error) { return uploadFileBytesWithCfg(ctx, buildConfig(opts), fileName, data) } // uploadFileBytesWithCfg 内部助手:用已构建的配置上传(TempURLToOSS 复用同一 cfg,保留 header/超时)。 func uploadFileBytesWithCfg(ctx context.Context, cfg *uploadConfig, fileName string, data []byte) (*UploadFileRes, error) { body := &bytes.Buffer{} writer := multipart.NewWriter(body) part, err := writer.CreateFormFile("file", fileName) if err != nil { return nil, err } if _, err = part.Write(data); err != nil { return nil, err } if err = writer.Close(); err != nil { return nil, err } return doPost(ctx, cfg, body.Bytes(), writer.FormDataContentType()) } // doPost 执行 multipart 上传并解析 {code,message,data} 响应。 // 复刻 commonHttp.doRequestRaw 的「先 ContentJson、后 SetHeaderMap 覆盖 Content-Type」顺序(生产已验证); // 超时走 transport clone,避免改动全局共享 transport。 func doPost(ctx context.Context, cfg *uploadConfig, body []byte, contentType string) (*UploadFileRes, error) { client := cfg.clientWithTimeouts().ContentJson() // 透传请求头 + 覆盖 headers := utils.HeadersFromCtx(ctx, utils.HeadersOptions{ResolveToken: true}) if len(cfg.header) > 0 { for k, v := range cfg.header { headers[k] = v } } if contentType != "" { headers["Content-Type"] = contentType } if len(headers) > 0 { client.SetHeaderMap(headers) } response, err := client.DoRequest(ctx, http.MethodPost, cfg.endpoint, body) if err != nil { return nil, fmt.Errorf("[OSS上传] 请求失败: %w", err) } defer response.Close() respBody := response.ReadAll() if response.StatusCode < 200 || response.StatusCode >= 300 { return nil, fmt.Errorf("[OSS上传] 状态码异常: %d, body=%s", response.StatusCode, string(respBody)) } resultStrut := &ghttp.DefaultHandlerResponse{} if err = gconv.Struct(respBody, &resultStrut); err != nil { return nil, errors.New("响应解析失败: " + err.Error()) } if resultStrut.Code != 200 && resultStrut.Code != 0 { return nil, errors.New(gconv.String(resultStrut.Message)) } res := &UploadFileRes{} if err = gconv.Struct(resultStrut.Data, res); err != nil { return nil, errors.New("数据解析失败: " + err.Error()) } return res, nil } // clientWithTimeouts 克隆客户端并按 Option 调大超时;transport 独立拷贝避免污染全局共享实例。 func (c *uploadConfig) clientWithTimeouts() *gclient.Client { client := c.client.Clone() if tr, ok := client.Transport.(*http.Transport); ok && c.headerTimeout > 0 { tr = tr.Clone() tr.ResponseHeaderTimeout = c.headerTimeout client.Transport = tr } if c.timeout > 0 { client = client.Timeout(c.timeout) } return client }