feat: 实现多部分上传功能，支持初始化、上传部分、完成和中止上传，添加媒体资产删除功能
This commit is contained in:
@@ -4,10 +4,13 @@ import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"mime/multipart"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"quyun/v2/app/errorx"
|
||||
@@ -85,12 +88,220 @@ func (s *common) CheckHash(ctx context.Context, userID int64, hash string) (*com
|
||||
return s.composeUploadResult(asset), nil
|
||||
}
|
||||
|
||||
type UploadMeta struct {
|
||||
Filename string
|
||||
Type string
|
||||
MimeType string
|
||||
}
|
||||
|
||||
func (s *common) InitUpload(ctx context.Context, userID int64, form *common_dto.UploadInitForm) (*common_dto.UploadInitResponse, error) {
|
||||
uploadID := uuid.NewString()
|
||||
localPath := s.storage.Config.LocalPath
|
||||
if localPath == "" {
|
||||
localPath = "./storage"
|
||||
}
|
||||
tempDir := filepath.Join(localPath, "temp", uploadID)
|
||||
if err := os.MkdirAll(tempDir, 0755); err != nil {
|
||||
return nil, errorx.ErrInternalError.WithCause(err)
|
||||
}
|
||||
|
||||
// Save metadata
|
||||
meta := UploadMeta{
|
||||
Filename: form.Filename,
|
||||
Type: form.Type, // Ensure form has Type
|
||||
MimeType: form.MimeType,
|
||||
}
|
||||
metaFile, _ := os.Create(filepath.Join(tempDir, "meta.json"))
|
||||
json.NewEncoder(metaFile).Encode(meta)
|
||||
metaFile.Close()
|
||||
|
||||
return &common_dto.UploadInitResponse{
|
||||
UploadID: uploadID,
|
||||
ChunkSize: 5 * 1024 * 1024,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *common) UploadPart(ctx context.Context, userID int64, file *multipart.FileHeader, form *common_dto.UploadPartForm) error {
|
||||
localPath := s.storage.Config.LocalPath
|
||||
if localPath == "" {
|
||||
localPath = "./storage"
|
||||
}
|
||||
partPath := filepath.Join(localPath, "temp", form.UploadID, strconv.Itoa(form.PartNumber))
|
||||
|
||||
src, err := file.Open()
|
||||
if err != nil {
|
||||
return errorx.ErrInternalError.WithCause(err)
|
||||
}
|
||||
defer src.Close()
|
||||
|
||||
dst, err := os.Create(partPath)
|
||||
if err != nil {
|
||||
return errorx.ErrInternalError.WithCause(err)
|
||||
}
|
||||
defer dst.Close()
|
||||
|
||||
if _, err = io.Copy(dst, src); err != nil {
|
||||
return errorx.ErrInternalError.WithCause(err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *common) CompleteUpload(ctx context.Context, userID int64, form *common_dto.UploadCompleteForm) (*common_dto.UploadResult, error) {
|
||||
localPath := s.storage.Config.LocalPath
|
||||
if localPath == "" {
|
||||
localPath = "./storage"
|
||||
}
|
||||
tempDir := filepath.Join(localPath, "temp", form.UploadID)
|
||||
|
||||
// Read Meta
|
||||
var meta UploadMeta
|
||||
metaFile, err := os.Open(filepath.Join(tempDir, "meta.json"))
|
||||
if err != nil {
|
||||
return nil, errorx.ErrRecordNotFound.WithMsg("Upload session expired or invalid")
|
||||
}
|
||||
json.NewDecoder(metaFile).Decode(&meta)
|
||||
metaFile.Close()
|
||||
|
||||
// List parts
|
||||
entries, err := os.ReadDir(tempDir)
|
||||
if err != nil {
|
||||
return nil, errorx.ErrInternalError.WithCause(err)
|
||||
}
|
||||
|
||||
var parts []int
|
||||
for _, e := range entries {
|
||||
if !e.IsDir() && e.Name() != "meta.json" {
|
||||
if i, err := strconv.Atoi(e.Name()); err == nil {
|
||||
parts = append(parts, i)
|
||||
}
|
||||
}
|
||||
}
|
||||
sort.Ints(parts)
|
||||
|
||||
objectKey := uuid.NewString() + "_" + meta.Filename
|
||||
dstPath := filepath.Join(localPath, objectKey)
|
||||
dst, err := os.Create(dstPath)
|
||||
if err != nil {
|
||||
return nil, errorx.ErrInternalError.WithCause(err)
|
||||
}
|
||||
defer dst.Close()
|
||||
|
||||
hasher := sha256.New()
|
||||
var totalSize int64
|
||||
|
||||
for _, partNum := range parts {
|
||||
partPath := filepath.Join(tempDir, strconv.Itoa(partNum))
|
||||
src, err := os.Open(partPath)
|
||||
if err != nil {
|
||||
return nil, errorx.ErrInternalError.WithCause(err)
|
||||
}
|
||||
|
||||
n, err := io.Copy(io.MultiWriter(dst, hasher), src)
|
||||
src.Close()
|
||||
if err != nil {
|
||||
return nil, errorx.ErrInternalError.WithCause(err)
|
||||
}
|
||||
totalSize += n
|
||||
}
|
||||
|
||||
hash := hex.EncodeToString(hasher.Sum(nil))
|
||||
dst.Close(); // Ensure flush before potential removal
|
||||
os.RemoveAll(tempDir)
|
||||
|
||||
// Deduplication Logic (Similar to Upload)
|
||||
t, err := models.TenantQuery.WithContext(ctx).Where(models.TenantQuery.UserID.Eq(userID)).First()
|
||||
var tid int64 = 0
|
||||
if err == nil {
|
||||
tid = t.ID
|
||||
}
|
||||
|
||||
existing, err := models.MediaAssetQuery.WithContext(ctx).Where(models.MediaAssetQuery.Hash.Eq(hash)).First()
|
||||
var asset *models.MediaAsset
|
||||
|
||||
if err == nil {
|
||||
os.Remove(dstPath) // Delete duplicate
|
||||
myExisting, err := models.MediaAssetQuery.WithContext(ctx).
|
||||
Where(models.MediaAssetQuery.Hash.Eq(hash), models.MediaAssetQuery.UserID.Eq(userID)).
|
||||
First()
|
||||
if err == nil {
|
||||
return s.composeUploadResult(myExisting), nil
|
||||
}
|
||||
asset = &models.MediaAsset{
|
||||
TenantID: tid,
|
||||
UserID: userID,
|
||||
Type: consts.MediaAssetType(meta.Type),
|
||||
Status: consts.MediaAssetStatusUploaded,
|
||||
Provider: existing.Provider,
|
||||
Bucket: existing.Bucket,
|
||||
ObjectKey: existing.ObjectKey,
|
||||
Hash: hash,
|
||||
Meta: existing.Meta,
|
||||
}
|
||||
} else {
|
||||
asset = &models.MediaAsset{
|
||||
TenantID: tid,
|
||||
UserID: userID,
|
||||
Type: consts.MediaAssetType(meta.Type),
|
||||
Status: consts.MediaAssetStatusUploaded,
|
||||
Provider: "local",
|
||||
Bucket: "default",
|
||||
ObjectKey: objectKey,
|
||||
Hash: hash,
|
||||
Meta: types.NewJSONType(fields.MediaAssetMeta{
|
||||
Size: totalSize,
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
if err := models.MediaAssetQuery.WithContext(ctx).Create(asset); err != nil {
|
||||
return nil, errorx.ErrDatabaseError.WithCause(err)
|
||||
}
|
||||
|
||||
return s.composeUploadResult(asset), nil
|
||||
}
|
||||
|
||||
func (s *common) DeleteMediaAsset(ctx context.Context, userID int64, id string) error {
|
||||
aid := cast.ToInt64(id)
|
||||
asset, err := models.MediaAssetQuery.WithContext(ctx).
|
||||
Where(models.MediaAssetQuery.ID.Eq(aid), models.MediaAssetQuery.UserID.Eq(userID)).
|
||||
First()
|
||||
if err != nil {
|
||||
return errorx.ErrRecordNotFound
|
||||
}
|
||||
|
||||
// Delete DB record
|
||||
if _, err := models.MediaAssetQuery.WithContext(ctx).Where(models.MediaAssetQuery.ID.Eq(aid)).Delete(); err != nil {
|
||||
return errorx.ErrDatabaseError.WithCause(err)
|
||||
}
|
||||
|
||||
// Check ref count
|
||||
count, _ := models.MediaAssetQuery.WithContext(ctx).
|
||||
Where(models.MediaAssetQuery.ObjectKey.Eq(asset.ObjectKey)).
|
||||
Count()
|
||||
|
||||
if count == 0 {
|
||||
// Physical delete
|
||||
_ = s.storage.Delete(asset.ObjectKey)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *common) AbortUpload(ctx context.Context, userID int64, uploadId string) error {
|
||||
localPath := s.storage.Config.LocalPath
|
||||
if localPath == "" {
|
||||
localPath = "./storage"
|
||||
}
|
||||
tempDir := filepath.Join(localPath, "temp", uploadId)
|
||||
return os.RemoveAll(tempDir)
|
||||
}
|
||||
|
||||
func (s *common) Upload(
|
||||
ctx context.Context,
|
||||
userID int64,
|
||||
file *multipart.FileHeader,
|
||||
typeArg string,
|
||||
) (*common_dto.UploadResult, error) { // Mock Upload to S3/MinIO (Here we just generate key, actual upload handling via direct upload or stream is better)
|
||||
) (*common_dto.UploadResult, error) {
|
||||
// But this Upload endpoint accepts file. So we save it.
|
||||
|
||||
objectKey := uuid.NewString() + "_" + file.Filename
|
||||
|
||||
Reference in New Issue
Block a user