You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
291 lines
9.4 KiB
291 lines
9.4 KiB
package explorer
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
model "github.com/cloudreve/Cloudreve/v3/models"
|
|
"github.com/cloudreve/Cloudreve/v3/pkg/auth"
|
|
"github.com/cloudreve/Cloudreve/v3/pkg/cache"
|
|
"github.com/cloudreve/Cloudreve/v3/pkg/filesystem"
|
|
"github.com/cloudreve/Cloudreve/v3/pkg/filesystem/driver/local"
|
|
"github.com/cloudreve/Cloudreve/v3/pkg/filesystem/fsctx"
|
|
"github.com/cloudreve/Cloudreve/v3/pkg/hashid"
|
|
"github.com/cloudreve/Cloudreve/v3/pkg/serializer"
|
|
"github.com/cloudreve/Cloudreve/v3/pkg/util"
|
|
"github.com/gin-gonic/gin"
|
|
"io/ioutil"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
// CreateUploadSessionService 获取上传凭证服务
|
|
type CreateUploadSessionService struct {
|
|
Path string `json:"path" binding:"required"`
|
|
Size uint64 `json:"size" binding:"min=0"`
|
|
Name string `json:"name" binding:"required"`
|
|
PolicyID string `json:"policy_id" binding:"required"`
|
|
LastModified int64 `json:"last_modified"`
|
|
}
|
|
|
|
// Create 创建新的上传会话
|
|
func (service *CreateUploadSessionService) Create(ctx context.Context, c *gin.Context) serializer.Response {
|
|
// 创建文件系统
|
|
fs, err := filesystem.NewFileSystemFromContext(c)
|
|
if err != nil {
|
|
return serializer.Err(serializer.CodeCreateFSError, "", err)
|
|
}
|
|
|
|
// 取得存储策略的ID
|
|
rawID, err := hashid.DecodeHashID(service.PolicyID, hashid.PolicyID)
|
|
if err != nil {
|
|
return serializer.Err(serializer.CodePolicyNotExist, "", err)
|
|
}
|
|
|
|
if fs.Policy.ID != rawID {
|
|
return serializer.Err(serializer.CodePolicyNotAllowed, "存储策略发生变化，请刷新文件列表并重新添加此任务", nil)
|
|
}
|
|
|
|
file := &fsctx.FileStream{
|
|
Size: service.Size,
|
|
Name: service.Name,
|
|
VirtualPath: service.Path,
|
|
File: ioutil.NopCloser(strings.NewReader("")),
|
|
}
|
|
if service.LastModified > 0 {
|
|
lastModified := time.UnixMilli(service.LastModified)
|
|
file.LastModified = &lastModified
|
|
}
|
|
credential, err := fs.CreateUploadSession(ctx, file)
|
|
if err != nil {
|
|
return serializer.Err(serializer.CodeNotSet, err.Error(), err)
|
|
}
|
|
|
|
return serializer.Response{
|
|
Code: 0,
|
|
Data: credential,
|
|
}
|
|
}
|
|
|
|
// UploadService 本机及从机策略上传服务
|
|
type UploadService struct {
|
|
ID string `uri:"sessionId" binding:"required"`
|
|
Index int `uri:"index" form:"index" binding:"min=0"`
|
|
}
|
|
|
|
// LocalUpload 处理本机文件分片上传
|
|
func (service *UploadService) LocalUpload(ctx context.Context, c *gin.Context) serializer.Response {
|
|
uploadSessionRaw, ok := cache.Get(filesystem.UploadSessionCachePrefix + service.ID)
|
|
if !ok {
|
|
return serializer.Err(serializer.CodeUploadSessionExpired, "", nil)
|
|
}
|
|
|
|
uploadSession := uploadSessionRaw.(serializer.UploadSession)
|
|
|
|
fs, err := filesystem.NewFileSystemFromContext(c)
|
|
if err != nil {
|
|
return serializer.Err(serializer.CodePolicyNotAllowed, err.Error(), err)
|
|
}
|
|
|
|
if uploadSession.UID != fs.User.ID {
|
|
return serializer.Err(serializer.CodeUploadSessionExpired, "", nil)
|
|
}
|
|
|
|
// 查找上传会话创建的占位文件
|
|
file, err := model.GetFilesByUploadSession(service.ID, fs.User.ID)
|
|
if err != nil {
|
|
return serializer.Err(serializer.CodeUploadSessionExpired, "", err)
|
|
}
|
|
|
|
// 重设 fs 存储策略
|
|
if !uploadSession.Policy.IsTransitUpload(uploadSession.Size) {
|
|
return serializer.Err(serializer.CodePolicyNotAllowed, "", err)
|
|
}
|
|
|
|
fs.Policy = &uploadSession.Policy
|
|
if err := fs.DispatchHandler(); err != nil {
|
|
return serializer.Err(serializer.CodePolicyNotExist, "", err)
|
|
}
|
|
|
|
expectedSizeStart := file.Size
|
|
actualSizeStart := uint64(service.Index) * uploadSession.Policy.OptionsSerialized.ChunkSize
|
|
if uploadSession.Policy.OptionsSerialized.ChunkSize == 0 && service.Index > 0 {
|
|
return serializer.Err(serializer.CodeInvalidChunkIndex, "Chunk index cannot be greater than 0", nil)
|
|
}
|
|
|
|
if expectedSizeStart < actualSizeStart {
|
|
return serializer.Err(serializer.CodeInvalidChunkIndex, "Chunk must be uploaded in order", nil)
|
|
}
|
|
|
|
if expectedSizeStart > actualSizeStart {
|
|
util.Log().Info("Trying to overwrite chunk[%d] Start=%d", service.Index, actualSizeStart)
|
|
}
|
|
|
|
return processChunkUpload(ctx, c, fs, &uploadSession, service.Index, file, fsctx.Append)
|
|
}
|
|
|
|
// SlaveUpload 处理从机文件分片上传
|
|
func (service *UploadService) SlaveUpload(ctx context.Context, c *gin.Context) serializer.Response {
|
|
uploadSessionRaw, ok := cache.Get(filesystem.UploadSessionCachePrefix + service.ID)
|
|
if !ok {
|
|
return serializer.Err(serializer.CodeUploadSessionExpired, "", nil)
|
|
}
|
|
|
|
uploadSession := uploadSessionRaw.(serializer.UploadSession)
|
|
|
|
fs, err := filesystem.NewAnonymousFileSystem()
|
|
if err != nil {
|
|
return serializer.Err(serializer.CodeCreateFSError, "", err)
|
|
}
|
|
|
|
fs.Handler = local.Driver{}
|
|
|
|
// 解析需要的参数
|
|
service.Index, _ = strconv.Atoi(c.Query("chunk"))
|
|
mode := fsctx.Append
|
|
if c.GetHeader(auth.CrHeaderPrefix+"Overwrite") == "true" {
|
|
mode |= fsctx.Overwrite
|
|
}
|
|
|
|
return processChunkUpload(ctx, c, fs, &uploadSession, service.Index, nil, mode)
|
|
}
|
|
|
|
func processChunkUpload(ctx context.Context, c *gin.Context, fs *filesystem.FileSystem, session *serializer.UploadSession, index int, file *model.File, mode fsctx.WriteMode) serializer.Response {
|
|
// 取得并校验文件大小是否符合分片要求
|
|
chunkSize := session.Policy.OptionsSerialized.ChunkSize
|
|
isLastChunk := session.Policy.OptionsSerialized.ChunkSize == 0 || uint64(index+1)*chunkSize >= session.Size
|
|
expectedLength := chunkSize
|
|
if isLastChunk {
|
|
expectedLength = session.Size - uint64(index)*chunkSize
|
|
}
|
|
|
|
fileSize, err := strconv.ParseUint(c.Request.Header.Get("Content-Length"), 10, 64)
|
|
if err != nil || (expectedLength != fileSize) {
|
|
return serializer.Err(
|
|
serializer.CodeInvalidContentLength,
|
|
fmt.Sprintf("Invalid Content-Length (expected: %d)", expectedLength),
|
|
err,
|
|
)
|
|
}
|
|
|
|
// 非首个分片时需要允许覆盖
|
|
if index > 0 {
|
|
mode |= fsctx.Overwrite
|
|
}
|
|
|
|
fileData := fsctx.FileStream{
|
|
MIMEType: c.Request.Header.Get("Content-Type"),
|
|
File: c.Request.Body,
|
|
Size: fileSize,
|
|
Name: session.Name,
|
|
VirtualPath: session.VirtualPath,
|
|
SavePath: session.SavePath,
|
|
Mode: mode,
|
|
AppendStart: chunkSize * uint64(index),
|
|
Model: file,
|
|
LastModified: session.LastModified,
|
|
}
|
|
|
|
// 给文件系统分配钩子
|
|
fs.Use("AfterUploadCanceled", filesystem.HookTruncateFileTo(fileData.AppendStart))
|
|
fs.Use("AfterValidateFailed", filesystem.HookTruncateFileTo(fileData.AppendStart))
|
|
|
|
if file != nil {
|
|
fs.Use("BeforeUpload", filesystem.HookValidateCapacity)
|
|
fs.Use("AfterUpload", filesystem.HookChunkUploaded)
|
|
fs.Use("AfterValidateFailed", filesystem.HookChunkUploadFailed)
|
|
if isLastChunk {
|
|
fs.Use("AfterUpload", filesystem.HookPopPlaceholderToFile(""))
|
|
fs.Use("AfterUpload", filesystem.HookDeleteUploadSession(session.Key))
|
|
}
|
|
} else {
|
|
if isLastChunk {
|
|
fs.Use("AfterUpload", filesystem.SlaveAfterUpload(session))
|
|
fs.Use("AfterUpload", filesystem.HookDeleteUploadSession(session.Key))
|
|
}
|
|
}
|
|
|
|
// 执行上传
|
|
uploadCtx := context.WithValue(ctx, fsctx.GinCtx, c)
|
|
err = fs.Upload(uploadCtx, &fileData)
|
|
if err != nil {
|
|
return serializer.Err(serializer.CodeUploadFailed, err.Error(), err)
|
|
}
|
|
|
|
return serializer.Response{}
|
|
}
|
|
|
|
// UploadSessionService 上传会话服务
|
|
type UploadSessionService struct {
|
|
ID string `uri:"sessionId" binding:"required"`
|
|
}
|
|
|
|
// Delete 删除指定上传会话
|
|
func (service *UploadSessionService) Delete(ctx context.Context, c *gin.Context) serializer.Response {
|
|
// 创建文件系统
|
|
fs, err := filesystem.NewFileSystemFromContext(c)
|
|
if err != nil {
|
|
return serializer.Err(serializer.CodeCreateFSError, "", err)
|
|
}
|
|
defer fs.Recycle()
|
|
|
|
// 查找需要删除的上传会话的占位文件
|
|
file, err := model.GetFilesByUploadSession(service.ID, fs.User.ID)
|
|
if err != nil {
|
|
return serializer.Err(serializer.CodeUploadSessionExpired, "", err)
|
|
}
|
|
|
|
// 删除文件
|
|
if err := fs.Delete(ctx, []uint{}, []uint{file.ID}, false, false); err != nil {
|
|
return serializer.Err(serializer.CodeInternalSetting, "Failed to delete upload session", err)
|
|
}
|
|
|
|
return serializer.Response{}
|
|
}
|
|
|
|
// SlaveDelete 从机删除指定上传会话
|
|
func (service *UploadSessionService) SlaveDelete(ctx context.Context, c *gin.Context) serializer.Response {
|
|
// 创建文件系统
|
|
fs, err := filesystem.NewAnonymousFileSystem()
|
|
if err != nil {
|
|
return serializer.Err(serializer.CodeCreateFSError, "", err)
|
|
}
|
|
defer fs.Recycle()
|
|
|
|
session, ok := cache.Get(filesystem.UploadSessionCachePrefix + service.ID)
|
|
if !ok {
|
|
return serializer.Err(serializer.CodeUploadSessionExpired, "", nil)
|
|
}
|
|
|
|
if _, err := fs.Handler.Delete(ctx, []string{session.(serializer.UploadSession).SavePath}); err != nil {
|
|
return serializer.Err(serializer.CodeInternalSetting, "Failed to delete temp file", err)
|
|
}
|
|
|
|
cache.Deletes([]string{service.ID}, filesystem.UploadSessionCachePrefix)
|
|
return serializer.Response{}
|
|
}
|
|
|
|
// DeleteAllUploadSession 删除当前用户的全部上传绘会话
|
|
func DeleteAllUploadSession(ctx context.Context, c *gin.Context) serializer.Response {
|
|
// 创建文件系统
|
|
fs, err := filesystem.NewFileSystemFromContext(c)
|
|
if err != nil {
|
|
return serializer.Err(serializer.CodeCreateFSError, "", err)
|
|
}
|
|
defer fs.Recycle()
|
|
|
|
// 查找需要删除的上传会话的占位文件
|
|
files := model.GetUploadPlaceholderFiles(fs.User.ID)
|
|
fileIDs := make([]uint, len(files))
|
|
for i, file := range files {
|
|
fileIDs[i] = file.ID
|
|
}
|
|
|
|
// 删除文件
|
|
if err := fs.Delete(ctx, []uint{}, fileIDs, false, false); err != nil {
|
|
return serializer.Err(serializer.CodeInternalSetting, "Failed to cleanup upload session", err)
|
|
}
|
|
|
|
return serializer.Response{}
|
|
}
|