package file import ( "context" "fusenapi/model/gmodel" "fusenapi/utils/basic" "net/http" "time" "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 } type UploadBaseReq struct { ResourceId string Refresh int64 Source string FileHash string FileData string Metadata string UploadBucket int64 ApiType int64 UserId int64 GuestId int64 FileByte []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 default: bucketName = basic.StorageBucketName } // 设置AWS会话的区域 upload.AwsSession.Config.Region = aws.String("us-west-1") // 创建新的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, must-revalidate"), }, ) // 设置请求体为文件数据 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 default: bucketName = basic.StorageBucketName } // 设置AWS会话的区域 upload.AwsSession.Config.Region = aws.String("us-west-1") // 创建新的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, must-revalidate"), }, ) // 设置请求体为文件数据 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 }