package file import ( "bytes" "context" "fusenapi/model/gmodel" "fusenapi/utils/basic" "io" "net/http" "time" "github.com/aliyun/aliyun-oss-go-sdk/oss" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/request" "github.com/aws/aws-sdk-go/aws/session" "github.com/aws/aws-sdk-go/service/s3" "github.com/zeromicro/go-zero/core/logx" "gorm.io/gorm" ) type Upload struct { Ctx context.Context MysqlConn *gorm.DB AwsSession *session.Session UploadType string AliyunOSS AliyunOSS } type AliyunOSS struct { Endpoint string `json:"endpoint"` AccessKeyId string `json:"access_key_id"` AccessKeySecret string `json:"access_key_secret"` BucketName string `json:"bucket_name"` BucketUrl string `json:"bucket_url"` BasePath string `json:"base_path"` } type UploadBaseReq struct { ResourceId string `json:"resource_id"` Refresh int64 `json:"refresh"` Source string `json:"source"` FileHash string `json:"file_hash"` FileData string `json:"file_data"` Metadata string `json:"metadata"` UploadBucket int64 `json:"upload_bucket"` ApiType int64 `json:"api_type"` UserId int64 `json:"user_id"` GuestId int64 `json:"guest_id"` FileByte []byte `json:"file_byte"` } type UploadBaseRes struct { Key string `json:"key"` Status int64 `json:"status"` ApiType int64 `json:"api_type"` ResourceId string `json:"resource_id"` ResourceType string `json:"resource_type"` ResourceUrl string `json:"resource_url"` } func (upload *Upload) UploadFileByBase64(req *UploadBaseReq) (*UploadBaseRes, error) { var metadataB []byte if req.Metadata != "" { metadataB = []byte(req.Metadata) } // 定义存储桶名称 var bucketName *string var apiType int64 = req.ApiType // 根据类别选择存储桶 switch req.UploadBucket { case 2: bucketName = basic.TempfileBucketName case 3: bucketName = basic.StorageBucketName default: bucketName = basic.StorageBucketName } // 设置AWS会话的区域 upload.AwsSession.Config.Region = aws.String("us-east-2") // 创建新的S3服务实例 svc := s3.New(upload.AwsSession) // 定义S3请求和当前时间 var s3req *request.Request var resourceId string = req.FileHash var uploadBaseRes = UploadBaseRes{} // 是否强制刷新 var refresh bool if req.ResourceId != "" { resourceId = req.ResourceId } if req.Refresh == 1 { refresh = true } err := upload.MysqlConn.Transaction(func(tx *gorm.DB) (err error) { var resourceInfo *gmodel.FsResource if !refresh { err = tx.Where("resource_id =?", resourceId).Take(&resourceInfo).Error if err == nil && resourceInfo.ResourceId != "" { uploadBaseRes.Status = 1 uploadBaseRes.ResourceId = resourceId uploadBaseRes.ResourceUrl = *resourceInfo.ResourceUrl } else { refresh = true } } if refresh { dist, contentType, err := FileBase64ToByte(req.FileData) if err != nil { logx.Errorf("err:%+v,desc:%+v", err, "fail.upload.resourceInfoGet.mysql") return err } // 创建S3对象存储请求 s3req, _ = svc.PutObjectRequest( &s3.PutObjectInput{ Bucket: bucketName, Key: &resourceId, CacheControl: aws.String("no-cache"), }, ) // 设置请求体为文件数据 s3req.SetBufferBody(dist) // 发送请求 err = s3req.Send() // 检查是否有错误 if err != nil { logx.Errorf("err:%+v,desc:%+v", err, "fail.upload.s3req") return err } else { var url = s3req.HTTPRequest.URL.String() // 打印请求URL logx.Info(url) uploadBaseRes.Status = 1 uploadBaseRes.ResourceId = resourceId uploadBaseRes.ResourceUrl = url var version string = "0.0.1" var nowTime = time.Now().UTC() if refresh { err = tx.Save(&gmodel.FsResource{ ResourceId: resourceId, UserId: &req.UserId, GuestId: &req.GuestId, ResourceType: &contentType, ResourceUrl: &url, Version: &version, UploadedAt: &nowTime, Metadata: &metadataB, ApiType: &apiType, BucketName: bucketName, Source: &req.Source, }).Error } else { err = tx.Create(&gmodel.FsResource{ ResourceId: resourceId, UserId: &req.UserId, GuestId: &req.GuestId, ResourceType: &contentType, ResourceUrl: &url, Version: &version, UploadedAt: &nowTime, Metadata: &metadataB, ApiType: &apiType, BucketName: bucketName, Source: &req.Source, }).Error } if err != nil { logx.Errorf("err:%+v,desc:%+v", err, "fail.upload.resourceInfoAdd.mysql") return err } } } // 返回 nil 提交事务 return nil }) if err != nil { return nil, err } return &uploadBaseRes, err } func (upload *Upload) UploadFileByByte(req *UploadBaseReq) (*UploadBaseRes, error) { var metadataB []byte if req.Metadata != "" { metadataB = []byte(req.Metadata) } // 定义存储桶名称 var bucketName *string var apiType int64 = req.ApiType // 根据类别选择存储桶 switch req.UploadBucket { case 2: bucketName = basic.TempfileBucketName case 3: bucketName = basic.StorageBucketName default: bucketName = basic.StorageBucketName } // 设置AWS会话的区域 upload.AwsSession.Config.Region = aws.String("us-east-2") // 创建新的S3服务实例 svc := s3.New(upload.AwsSession) // 定义S3请求和当前时间 var s3req *request.Request var resourceId string = req.FileHash var uploadBaseRes = UploadBaseRes{} // 是否强制刷新 var refresh bool if req.ResourceId != "" { resourceId = req.ResourceId } if req.Refresh == 1 { refresh = true } err := upload.MysqlConn.Transaction(func(tx *gorm.DB) (err error) { var resourceInfo *gmodel.FsResource if !refresh { err = tx.Where("resource_id =?", resourceId).Take(&resourceInfo).Error if err == nil && resourceInfo.ResourceId != "" { uploadBaseRes.Status = 1 uploadBaseRes.ResourceId = resourceId uploadBaseRes.ResourceUrl = *resourceInfo.ResourceUrl } else { refresh = true } } if refresh { contentType := http.DetectContentType(req.FileByte) // 创建S3对象存储请求 s3req, _ = svc.PutObjectRequest( &s3.PutObjectInput{ Bucket: bucketName, Key: &resourceId, CacheControl: aws.String("no-cache"), }, ) // 设置请求体为文件数据 s3req.SetBufferBody(req.FileByte) // 发送请求 err = s3req.Send() // 检查是否有错误 if err != nil { logx.Errorf("err:%+v,desc:%+v", err, "fail.upload.s3req") return err } else { var url = s3req.HTTPRequest.URL.String() // 打印请求URL logx.Info(url) uploadBaseRes.Status = 1 uploadBaseRes.ResourceId = resourceId uploadBaseRes.ResourceUrl = url var version string = "0.0.1" var nowTime = time.Now().UTC() if refresh { err = tx.Save(&gmodel.FsResource{ ResourceId: resourceId, UserId: &req.UserId, GuestId: &req.GuestId, ResourceType: &contentType, ResourceUrl: &url, Version: &version, UploadedAt: &nowTime, Metadata: &metadataB, ApiType: &apiType, BucketName: bucketName, Source: &req.Source, }).Error } else { err = tx.Create(&gmodel.FsResource{ ResourceId: resourceId, UserId: &req.UserId, GuestId: &req.GuestId, ResourceType: &contentType, ResourceUrl: &url, Version: &version, UploadedAt: &nowTime, Metadata: &metadataB, ApiType: &apiType, BucketName: bucketName, Source: &req.Source, }).Error } if err != nil { logx.Errorf("err:%+v,desc:%+v", err, "fail.upload.resourceInfoAdd.mysql") return err } } } return nil }) if err != nil { return nil, err } return &uploadBaseRes, err } // 根据URL上传文件 func (upload *Upload) UploadFileByUrl(fileHash string, url string, uploadBucket int64) error { // 定义存储桶名称 var bucketName string // 根据类别选择存储桶 switch uploadBucket { case 2: bucketName = *basic.TempfileBucketName case 3: bucketName = *basic.StorageBucketName default: bucketName = *basic.StorageBucketName } switch upload.UploadType { case "oss": yunFileTmpPath := fileHash // 创建OSSClient实例。 clientOss, err := oss.New(upload.AliyunOSS.Endpoint, upload.AliyunOSS.AccessKeyId, upload.AliyunOSS.AccessKeySecret) if err != nil { return err } bucket, err := clientOss.Bucket(bucketName) if err != nil { return err } // 创建一个 HTTP 客户端 client := &http.Client{} // 发送一个 GET 请求 resp, err := client.Get(url) if err != nil { return err } // 关闭响应 defer resp.Body.Close() // 读取响应内容 buffer, err := io.ReadAll(resp.Body) if err != nil { return err } err = bucket.PutObject(yunFileTmpPath, bytes.NewReader(buffer)) if err != nil { return err } return nil default: return nil } }