From 61abe70df2c26fe20dc8fffa6a7213eeebdbf3fd Mon Sep 17 00:00:00 2001 From: Joftal <42807405+Joftal@users.noreply.github.com> Date: Wed, 12 Aug 2026 18:03:54 +0800 Subject: [PATCH] =?UTF-8?q?feat(upload):=20=E6=89=8B=E5=8A=A8=E4=B8=8A?= =?UTF-8?q?=E4=BC=A0=E6=94=B9=E7=94=A8=20Pipeline=20=E4=BB=BB=E5=8A=A1?= =?UTF-8?q?=EF=BC=8C=E4=B8=8E=E8=87=AA=E5=8A=A8=E4=B8=8A=E4=BC=A0=E8=A1=8C?= =?UTF-8?q?=E4=B8=BA=E4=B8=80=E8=87=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 手动上传（文件管理界面）原来直接调用 OpenList 客户端上传， 路径推断逻辑与自动上传（CloudUploadStage）不一致，且无任务追踪。 改为创建单阶段 cloud_upload PipelineTask，复用 CloudUploadStage 的 完整逻辑：路径模板渲染、上传标记持久化、任务追踪、重试能力。 主要改动： - 新增 EnqueueUploadTask 方法，接收绝对路径列表创建 Pipeline 任务 - 新增 upload_handler.go，通过 Pipeline 系统处理手动上传请求 - 提取 RenderUploadPath 为公共函数，消除 extract_cover.go 中的重复代码 - 上传成功后持久化标记到 metadata DB，文件列表显示已上传标签 - 文件删除/重命名时自动清除/迁移上传标记 - Pipeline 任务错误信息支持多行展示 Co-Authored-By: Kscc --- src/pipeline/executor.go | 15 +++ src/pipeline/manager.go | 91 +++++++++++++ src/pipeline/stages/extract_cover.go | 90 +++++-------- src/pipeline/upload_utils.go | 67 ++++++++++ src/pkg/metadata/store.go | 2 + src/servers/handler.go | 98 +++++++++++++- src/servers/server.go | 3 +- src/servers/upload_handler.go | 125 ++++++++++++++++++ src/webapp/src/component/file-list/index.tsx | 96 +++++++++++++- .../component/pipeline-task-list/index.tsx | 12 +- src/webapp/src/utils/api.ts | 16 +++ 11 files changed, 550 insertions(+), 65 deletions(-) create mode 100644 src/pipeline/upload_utils.go create mode 100644 src/servers/upload_handler.go diff --git a/src/pipeline/executor.go b/src/pipeline/executor.go index 0949aec..02323be 100644 --- a/src/pipeline/executor.go +++ b/src/pipeline/executor.go @@ -9,6 +9,8 @@ import ( "sync" "time" + "github.com/bililive-go/bililive-go/src/configs" + "github.com/bililive-go/bililive-go/src/pkg/metadata" bilisentry "github.com/bililive-go/bililive-go/src/pkg/sentry" "github.com/sirupsen/logrus" ) @@ -185,6 +187,9 @@ func (e *Executor) deleteMarkedFiles(files []FileInfo) []FileInfo { deleteAll := false // 标记是否为 deleteAll 模式 deleteAfter := false // 标记是否为 deleteAfter 模式（仅删除已上传文件） + // 获取输出路径用于计算相对路径（清理 DB 上传标记） + cfg := configs.GetCurrentConfig() + for _, f := range files { // 检查删除模式（CloudUploadStage 标记） if f.Metadata != nil { @@ -212,6 +217,16 @@ func (e *Executor) deleteMarkedFiles(files []FileInfo) []FileInfo { } } else { e.logger.Infof("pipeline cleanup: 已删除 %s", f.Path) + // 清除 DB 中的上传标记 + if cfg != nil { + outPutPath, _ := filepath.Abs(cfg.OutPutPath) + absFilePath, _ := filepath.Abs(f.Path) + if relPath, relErr := filepath.Rel(outPutPath, absFilePath); relErr == nil { + if delErr := metadata.GetStore().Delete(context.Background(), metadata.NamespaceUploaded, filepath.ToSlash(relPath)); delErr != nil { + e.logger.Warnf("pipeline cleanup: 清除上传标记失败 %s: %v", f.Path, delErr) + } + } + } } } else { kept = append(kept, f) diff --git a/src/pipeline/manager.go b/src/pipeline/manager.go index 4643c1e..5f72e3d 100644 --- a/src/pipeline/manager.go +++ b/src/pipeline/manager.go @@ -4,14 +4,18 @@ import ( "context" "fmt" "os" + "path/filepath" + "strings" "sync" "time" + "github.com/bililive-go/bililive-go/src/configs" "github.com/bililive-go/bililive-go/src/instance" "github.com/bililive-go/bililive-go/src/live" "github.com/bililive-go/bililive-go/src/pkg/events" "github.com/bililive-go/bililive-go/src/pkg/livelogger" bilisentry "github.com/bililive-go/bililive-go/src/pkg/sentry" + "github.com/bililive-go/bililive-go/src/types" "github.com/sirupsen/logrus" ) @@ -584,6 +588,93 @@ type ManagerStats struct { CancelledCount int `json:"cancelled_count"` } +// EnqueueUploadTask 创建手动上传的 Pipeline 任务并入队 +// 行为与自动上传（CloudUploadStage）完全一致：使用相同的路径模板、配置选项 +// 每个文件创建一个独立的单阶段 cloud_upload 任务 +// absPaths: 已校验的绝对路径列表（调用方应先通过 getSafePath 校验） +func (m *Manager) EnqueueUploadTask(absPaths []string) (enqueued int, skipped []string, taskIDs []int64, err error) { + config := configs.GetCurrentConfig() + if config == nil { + return 0, nil, nil, fmt.Errorf("配置未初始化") + } + + cu := config.OnRecordFinished.CloudUpload + if !cu.Enable { + return 0, nil, nil, fmt.Errorf("云上传功能未启用，请先在设置中开启") + } + if cu.StorageName == "" { + return 0, nil, nil, fmt.Errorf("未配置存储名称") + } + + // 构建单阶段 cloud_upload PipelineConfig，与 ConvertLegacyConfig 一致 + uploadConfig := &PipelineConfig{ + Stages: []StageConfig{ + { + Name: StageNameCloudUpload, + Options: map[string]any{ + OptionStorage: cu.StorageName, + OptionPathTemplate: cu.UploadPathTmpl, + OptionDeleteAfter: cu.DeleteAfterUpload, + OptionDeleteAllAfter: cu.DeleteAllAfterUpload, + OptionUploadTiming: string(config.OnRecordFinished.UploadTiming), + OptionFileTypes: []string{string(FileTypeVideo), string(FileTypeCover)}, + }, + }, + }, + } + + outputPath, _ := filepath.Abs(config.OutPutPath) + skipped = []string{} + + for _, absPath := range absPaths { + // 检查文件存在 + info, statErr := os.Stat(absPath) + if statErr != nil { + skipped = append(skipped, filepath.Base(absPath)+" - 文件不存在或无法访问") + continue + } + if info.IsDir() { + skipped = append(skipped, filepath.Base(absPath)+" - 不支持上传文件夹") + continue + } + + // 从相对路径推断 RecordInfo（用于上传路径模板渲染） + relPath, _ := filepath.Rel(outputPath, absPath) + platform, hostName := inferUploadPathInfo(relPath) + + recordInfo := RecordInfo{ + LiveID: types.LiveID("manual-upload"), + Platform: platform, // 用于上传路径模板：{{ .Platform }} + HostName: hostName, // 用于上传路径模板：{{ .HostName }} + RoomName: filepath.Base(absPath), // 显示文件名，便于在任务列表中识别 + StartTime: info.ModTime(), + } + + files := []FileInfo{NewVideoFileInfo(absPath)} + task := NewPipelineTask(recordInfo, uploadConfig, files) + + if enqueueErr := m.EnqueueTask(task); enqueueErr != nil { + skipped = append(skipped, filepath.Base(absPath)+" - 入队失败: "+enqueueErr.Error()) + continue + } + + enqueued++ + taskIDs = append(taskIDs, task.ID) + } + + return enqueued, skipped, taskIDs, nil +} + +// inferUploadPathInfo 从文件相对路径推断平台和主播名 +// 目录结构约定：{Platform}/{HostName}/... +func inferUploadPathInfo(relPath string) (platform, hostName string) { + parts := strings.Split(filepath.ToSlash(relPath), "/") + if len(parts) >= 2 { + return parts[0], parts[1] + } + return "", "" +} + // GetManager 从实例获取管道管理器 func GetManager(inst *instance.Instance) *Manager { if inst == nil || inst.PipelineManager == nil { diff --git a/src/pipeline/stages/extract_cover.go b/src/pipeline/stages/extract_cover.go index a7157dc..221fe38 100644 --- a/src/pipeline/stages/extract_cover.go +++ b/src/pipeline/stages/extract_cover.go @@ -1,17 +1,16 @@ package stages import ( - "bytes" "fmt" "os" "path" "path/filepath" "strings" - "text/template" "time" "github.com/bililive-go/bililive-go/src/configs" "github.com/bililive-go/bililive-go/src/pipeline" + "github.com/bililive-go/bililive-go/src/pkg/metadata" "github.com/bililive-go/bililive-go/src/pkg/openlist" "github.com/bililive-go/bililive-go/src/tools" ) @@ -173,6 +172,7 @@ func (s *CloudUploadStage) Execute(ctx *pipeline.PipelineContext, input []pipeli var output []pipeline.FileInfo var uploadFailed bool usedPaths := map[string]bool{} // 检测远程路径冲突 + var failureDetails []string // 收集每条失败详情，用于返回给前端 // 上传所有非 Deletable 的视频文件和封面文件 // Deletable 检查已能正确区分中间产物（如 ConvertMp4 标记的原始 FLV） @@ -194,7 +194,9 @@ func (s *CloudUploadStage) Execute(ctx *pipeline.PipelineContext, input []pipeli // 检查文件是否存在 if _, err := os.Stat(file.Path); os.IsNotExist(err) { - s.logs += fmt.Sprintf("文件不存在: %s\n", file.Path) + errMsg := fmt.Sprintf("文件不存在: %s", file.Path) + s.logs += errMsg + "\n" + failureDetails = append(failureDetails, errMsg) output = append(output, file) uploadFailed = true continue @@ -203,7 +205,9 @@ func (s *CloudUploadStage) Execute(ctx *pipeline.PipelineContext, input []pipeli // 渲染目标路径 targetPath := s.renderTargetPath(ctx, file) if targetPath == "" { - s.logs += fmt.Sprintf("无法生成目标路径: %s\n", file.Path) + errMsg := fmt.Sprintf("无法生成目标路径: %s", file.Path) + s.logs += errMsg + "\n" + failureDetails = append(failureDetails, errMsg) output = append(output, file) uploadFailed = true continue @@ -232,6 +236,7 @@ func (s *CloudUploadStage) Execute(ctx *pipeline.PipelineContext, input []pipeli } // 确保存储名和路径之间没有双斜杠 fullRemotePath := "/" + s.storageName + remotePath + fullRemotePath = strings.ReplaceAll(fullRemotePath, "//", "/") // 确保远程目录存在（递归创建） // 使用 path.Dir 而非 filepath.Dir，因为远程路径始终用正斜杠 @@ -250,7 +255,9 @@ func (s *CloudUploadStage) Execute(ctx *pipeline.PipelineContext, input []pipeli }) if err != nil { - s.logs += fmt.Sprintf("上传失败: %s -> %s/%s - %s\n", fileName, s.storageName, targetPath, err.Error()) + errMsg := fmt.Sprintf("上传失败: %s -> %s/%s - %s", fileName, s.storageName, targetPath, err.Error()) + s.logs += errMsg + "\n" + failureDetails = append(failureDetails, errMsg) ctx.Logger.Errorf("上传失败: %s - %v", file.Path, err) // 上传失败，保留文件 output = append(output, file) @@ -262,6 +269,17 @@ func (s *CloudUploadStage) Execute(ctx *pipeline.PipelineContext, input []pipeli ctx.Logger.Infof("文件上传成功: %s -> %s/%s", fileName, s.storageName, targetPath) uploadedIndices[len(output)] = true // 记录上传成功的文件索引 output = append(output, file) + + // 持久化上传标记（统一用正斜杠存储，与 URL 路径一致） + if cfg := configs.GetCurrentConfig(); cfg != nil { + outPutPath, _ := filepath.Abs(cfg.OutPutPath) + absFilePath, _ := filepath.Abs(file.Path) + if relPath, relErr := filepath.Rel(outPutPath, absFilePath); relErr == nil { + if markErr := metadata.GetStore().Set(ctx.Ctx, metadata.NamespaceUploaded, filepath.ToSlash(relPath), "1"); markErr != nil { + ctx.Logger.Warnf("保存上传标记失败: %v", markErr) + } + } + } } // 全部上传成功后，根据配置标记文件为可删除（由 Executor 在管道全部成功后统一删除） @@ -337,8 +355,9 @@ func (s *CloudUploadStage) Execute(ctx *pipeline.PipelineContext, input []pipeli } // 任意上传失败，返回错误（让 Executor 将任务标记为 failed） + // 将每条失败详情拼入 error message，前端可直接展示具体原因 if uploadFailed { - return output, fmt.Errorf("部分文件上传失败，详见日志") + return output, fmt.Errorf("部分文件上传失败:\n%s", strings.Join(failureDetails, "\n")) } return output, nil @@ -356,29 +375,12 @@ func (s *CloudUploadStage) matchFileType(fileType pipeline.FileType) bool { // renderTargetPath 渲染目标路径 func (s *CloudUploadStage) renderTargetPath(ctx *pipeline.PipelineContext, file pipeline.FileInfo) string { - if s.pathTemplate == "" { - // 默认路径：/录播归档/{平台}/{主播名}/{文件名} - return fmt.Sprintf("/录播归档/%s/%s/%s", - ctx.RecordInfo.Platform, - ctx.RecordInfo.HostName, - filepath.Base(file.Path), - ) - } - - // 获取扩展名 ext := filepath.Ext(file.Path) if len(ext) > 0 && ext[0] == '.' { ext = ext[1:] } - // 模板数据 - data := struct { - Platform string - HostName string - RoomName string - FileName string - Ext string - }{ + data := pipeline.UploadPathData{ Platform: ctx.RecordInfo.Platform, HostName: ctx.RecordInfo.HostName, RoomName: ctx.RecordInfo.RoomName, @@ -386,44 +388,14 @@ func (s *CloudUploadStage) renderTargetPath(ctx *pipeline.PipelineContext, file Ext: ext, } - // 使用 Go 模板引擎 - funcMap := template.FuncMap{ - "date": func(format string, t time.Time) string { - return t.Format(format) - }, - "now": func() time.Time { - return ctx.RecordInfo.StartTime - }, - "trimSuffix": func(suffix, s string) string { - return strings.TrimSuffix(s, suffix) - }, - "ext": func(path string) string { - return filepath.Ext(path) - }, - "filenameFilter": func(s string) string { - // 过滤文件名中的非法字符 - replacer := strings.NewReplacer( - "/", "_", "\\", "_", ":", "_", "*", "_", - "?", "_", "\"", "_", "<", "_", ">", "_", "|", "_", - ) - return replacer.Replace(s) - }, - } - - tmpl, err := template.New("path").Funcs(funcMap).Parse(s.pathTemplate) + path, err := pipeline.RenderUploadPath(s.pathTemplate, data, func() time.Time { + return ctx.RecordInfo.StartTime + }) if err != nil { - // 模板语法错误，返回空让调用方跳过上传，避免上传到错误路径 - ctx.Logger.Errorf("上传路径模板解析失败: %v", err) + ctx.Logger.Errorf("%v", err) return "" } - - var buf bytes.Buffer - if err := tmpl.Execute(&buf, data); err != nil { - ctx.Logger.Errorf("上传路径模板执行失败: %v", err) - return "" - } - - return buf.String() + return path } func (s *CloudUploadStage) GetCommands() []string { diff --git a/src/pipeline/upload_utils.go b/src/pipeline/upload_utils.go new file mode 100644 index 0000000..d6c0b88 --- /dev/null +++ b/src/pipeline/upload_utils.go @@ -0,0 +1,67 @@ +package pipeline + +import ( + "bytes" + "fmt" + "path/filepath" + "strings" + "text/template" + "time" +) + +// UploadPathData 上传路径模板数据 +type UploadPathData struct { + Platform string + HostName string + RoomName string + FileName string + Ext string +} + +// DefaultUploadPathFuncs 返回上传路径模板的默认函数集 +// nowFn 返回当前时间（可从 RecordInfo.StartTime 或 time.Now() 传入） +func DefaultUploadPathFuncs(nowFn func() time.Time) template.FuncMap { + return template.FuncMap{ + "date": func(format string, t time.Time) string { + return t.Format(format) + }, + "now": nowFn, + "trimSuffix": func(suffix, s string) string { + return strings.TrimSuffix(s, suffix) + }, + "ext": func(path string) string { + return filepath.Ext(path) + }, + "filenameFilter": func(s string) string { + replacer := strings.NewReplacer( + "/", "_", "\\", "_", ":", "_", "*", "_", + "?", "_", "\"", "_", "<", "_", ">", "_", "|", "_", + ) + return replacer.Replace(s) + }, + } +} + +// RenderUploadPath 渲染上传目标路径 +// pathTemplate 为空时使用默认路径 /录播归档/{Platform}/{HostName}/{FileName} +// 返回空字符串表示模板解析失败 +func RenderUploadPath(pathTemplate string, data UploadPathData, nowFn func() time.Time) (string, error) { + if pathTemplate == "" { + path := fmt.Sprintf("/录播归档/%s/%s/%s", data.Platform, data.HostName, data.FileName) + return path, nil + } + + funcMap := DefaultUploadPathFuncs(nowFn) + + tmpl, err := template.New("path").Funcs(funcMap).Parse(pathTemplate) + if err != nil { + return "", fmt.Errorf("上传路径模板解析失败: %w", err) + } + + var buf bytes.Buffer + if err := tmpl.Execute(&buf, data); err != nil { + return "", fmt.Errorf("上传路径模板执行失败: %w", err) + } + + return buf.String(), nil +} diff --git a/src/pkg/metadata/store.go b/src/pkg/metadata/store.go index d1c377e..7912307 100644 --- a/src/pkg/metadata/store.go +++ b/src/pkg/metadata/store.go @@ -206,6 +206,8 @@ const ( NamespaceUpdate = "update" // NamespaceMigration 数据库迁移状态 NamespaceMigration = "migration" + // NamespaceUploaded 已上传文件标记（key=相对路径, value="1"） + NamespaceUploaded = "uploaded" ) // 预定义的键常量 diff --git a/src/servers/handler.go b/src/servers/handler.go index a620efd..2cfd0b9 100644 --- a/src/servers/handler.go +++ b/src/servers/handler.go @@ -35,6 +35,7 @@ import ( applog "github.com/bililive-go/bililive-go/src/log" "github.com/bililive-go/bililive-go/src/pkg/livelogger" "github.com/bililive-go/bililive-go/src/pkg/memstats" + "github.com/bililive-go/bililive-go/src/pkg/metadata" "github.com/bililive-go/bililive-go/src/pkg/ratelimit" "github.com/bililive-go/bililive-go/src/pkg/utils" "github.com/bililive-go/bililive-go/src/recorders" @@ -2419,6 +2420,7 @@ func getFileInfo(writer http.ResponseWriter, r *http.Request) { LastModified int64 `json:"last_modified"` Size int64 `json:"size"` SubtitleFile string `json:"subtitle_file,omitempty"` + Uploaded bool `json:"uploaded,omitempty"` } // First pass: separate ASS files and build base-name -> ASS file map @@ -2461,6 +2463,14 @@ func getFileInfo(writer http.ResponseWriter, r *http.Request) { if assName, ok := assFiles[baseName]; ok { jf.SubtitleFile = assName } + // Check if this file has been uploaded to cloud + relPath := fe.dir.Name() + if path != "" { + relPath = path + "/" + fe.dir.Name() // key 统一用正斜杠 + } + if val, err := metadata.GetStore().Get(r.Context(), metadata.NamespaceUploaded, relPath); err == nil && val != "" { + jf.Uploaded = true + } } jsonFiles = append(jsonFiles, jf) } @@ -2580,6 +2590,30 @@ func renameFile(writer http.ResponseWriter, r *http.Request) { } } + // 迁移上传标记（key 统一用正斜杠） + oldRel := path + newRelPath, _ := filepath.Rel(base, newAbsPath) + newRel := filepath.ToSlash(newRelPath) + if info.IsDir() { + // 目录重命名：迁移该目录下所有文件的上传标记 + allMarks, _ := metadata.GetStore().GetAll(r.Context(), metadata.NamespaceUploaded) + oldPrefix := oldRel + "/" + for key, val := range allMarks { + if strings.HasPrefix(key, oldPrefix) { + suffix := key[len(oldPrefix):] + newKey := newRel + "/" + suffix + metadata.GetStore().Set(r.Context(), metadata.NamespaceUploaded, newKey, val) + metadata.GetStore().Delete(r.Context(), metadata.NamespaceUploaded, key) + } + } + } else { + // 文件重命名：迁移单个上传标记 + if val, err := metadata.GetStore().Get(r.Context(), metadata.NamespaceUploaded, oldRel); err == nil { + metadata.GetStore().Set(r.Context(), metadata.NamespaceUploaded, newRel, val) + metadata.GetStore().Delete(r.Context(), metadata.NamespaceUploaded, oldRel) + } + } + writeJSON(writer, commonResp{Data: "OK"}) } @@ -2599,8 +2633,14 @@ func deleteFile(writer http.ResponseWriter, r *http.Request) { return } + // 记录是否为目录（删除前检查） + isDir := false + if info, err := os.Stat(absPath); err == nil { + isDir = info.IsDir() + } + // 删除关联的 ASS 弹幕文件 - if info, err := os.Stat(absPath); err == nil && !info.IsDir() { + if !isDir { assPath := strings.TrimSuffix(absPath, filepath.Ext(absPath)) + ".ass" if _, err := os.Stat(assPath); err == nil { os.Remove(assPath) @@ -2612,6 +2652,20 @@ func deleteFile(writer http.ResponseWriter, r *http.Request) { return } + // 清除上传标记 + if isDir { + // 目录：清除该目录下所有文件的上传标记 + allMarks, _ := metadata.GetStore().GetAll(r.Context(), metadata.NamespaceUploaded) + prefix := path + "/" // key 统一用正斜杠 + for key := range allMarks { + if strings.HasPrefix(key, prefix) { + metadata.GetStore().Delete(r.Context(), metadata.NamespaceUploaded, key) + } + } + } else { + metadata.GetStore().Delete(r.Context(), metadata.NamespaceUploaded, path) + } + writeJSON(writer, commonResp{Data: "OK"}) } @@ -2698,6 +2752,28 @@ func batchRenameFiles(writer http.ResponseWriter, r *http.Request) { os.Rename(oldAss, newAss) } } + + // 迁移上传标记（key 统一用正斜杠） + oldRel := path + newRelPath, _ := filepath.Rel(base, newAbsPath) + newRel := filepath.ToSlash(newRelPath) + if info.IsDir() { + allMarks, _ := metadata.GetStore().GetAll(r.Context(), metadata.NamespaceUploaded) + oldPrefix := oldRel + "/" + for key, val := range allMarks { + if strings.HasPrefix(key, oldPrefix) { + suffix := key[len(oldPrefix):] + newKey := newRel + "/" + suffix + metadata.GetStore().Set(r.Context(), metadata.NamespaceUploaded, newKey, val) + metadata.GetStore().Delete(r.Context(), metadata.NamespaceUploaded, key) + } + } + } else { + if val, err := metadata.GetStore().Get(r.Context(), metadata.NamespaceUploaded, oldRel); err == nil { + metadata.GetStore().Set(r.Context(), metadata.NamespaceUploaded, newRel, val) + metadata.GetStore().Delete(r.Context(), metadata.NamespaceUploaded, oldRel) + } + } } } @@ -2734,8 +2810,14 @@ func batchDeleteFiles(writer http.ResponseWriter, r *http.Request) { continue } + // 记录是否为目录（删除前检查） + isDir := false + if info, err := os.Stat(absPath); err == nil { + isDir = info.IsDir() + } + // 删除关联的 ASS 弹幕文件 - if info, err := os.Stat(absPath); err == nil && !info.IsDir() { + if !isDir { assPath := strings.TrimSuffix(absPath, filepath.Ext(absPath)) + ".ass" if _, err := os.Stat(assPath); err == nil { os.Remove(assPath) @@ -2745,6 +2827,18 @@ func batchDeleteFiles(writer http.ResponseWriter, r *http.Request) { if err := os.RemoveAll(absPath); err != nil { results = append(results, Result{Path: path, Success: false, Message: translateOSError(err)}) } else { + // 清除上传标记 + if isDir { + allMarks, _ := metadata.GetStore().GetAll(r.Context(), metadata.NamespaceUploaded) + prefix := path + "/" // key 统一用正斜杠 + for key := range allMarks { + if strings.HasPrefix(key, prefix) { + metadata.GetStore().Delete(r.Context(), metadata.NamespaceUploaded, key) + } + } + } else { + metadata.GetStore().Delete(r.Context(), metadata.NamespaceUploaded, path) + } results = append(results, Result{Path: path, Success: true, Message: "成功"}) } } diff --git a/src/servers/server.go b/src/servers/server.go index e24191b..fcf7c4a 100644 --- a/src/servers/server.go +++ b/src/servers/server.go @@ -153,10 +153,11 @@ func initMux(ctx context.Context) *mux.Router { apiRoute.HandleFunc("/openlist/status", getOpenListStatus).Methods("GET") apiRoute.HandleFunc("/openlist/check-storage", checkOpenListStorageHealth).Methods("GET") - // Pipeline 任务路由 + // Pipeline 任务路由 + 手动上传路由 inst := instance.GetInstance(ctx) if pm := pipeline.GetManager(inst); pm != nil { RegisterPipelineHandlers(apiRoute, pm) + RegisterUploadHandlers(apiRoute, pm) } // OSRP 开放直播录制协议路由 diff --git a/src/servers/upload_handler.go b/src/servers/upload_handler.go new file mode 100644 index 0000000..f1dc1a6 --- /dev/null +++ b/src/servers/upload_handler.go @@ -0,0 +1,125 @@ +package servers + +import ( + "encoding/json" + "net/http" + "path/filepath" + + "github.com/gorilla/mux" + + "github.com/bililive-go/bililive-go/src/configs" + "github.com/bililive-go/bililive-go/src/pipeline" +) + +// uploadResult 手动上传的响应结构 +type uploadResult struct { + Enqueued int `json:"enqueued"` + Skipped []string `json:"skipped"` + TaskIDs []int64 `json:"task_ids"` +} + +// RegisterUploadHandlers 注册手动上传相关的 HTTP 处理器 +// 通过 Pipeline 系统执行上传，与自动上传行为完全一致 +func RegisterUploadHandlers(r *mux.Router, pm *pipeline.Manager) { + if pm == nil { + return + } + + r.HandleFunc("/file/upload", makeUploadFileHandler(pm)).Methods("POST") + r.HandleFunc("/batch/file/upload", makeBatchUploadFilesHandler(pm)).Methods("POST") +} + +// makeUploadFileHandler 单文件上传到云盘（通过 Pipeline） +func makeUploadFileHandler(pm *pipeline.Manager) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + var req struct { + Path string `json:"path"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeJSON(w, commonResp{ErrMsg: "请求参数错误"}) + return + } + if req.Path == "" { + writeJSON(w, commonResp{ErrMsg: "文件路径不能为空"}) + return + } + + // 路径校验（复用 handler.go 中的 getSafePath） + cfg := configs.GetCurrentConfig() + if cfg == nil { + writeJSON(w, commonResp{ErrMsg: "配置未初始化"}) + return + } + absPath, err := getSafePath(cfg.OutPutPath, req.Path) + if err != nil { + writeJSON(w, commonResp{ErrMsg: "无效路径: " + err.Error()}) + return + } + + enqueued, skipped, taskIDs, err := pm.EnqueueUploadTask([]string{absPath}) + if err != nil { + writeJSON(w, commonResp{ErrMsg: err.Error()}) + return + } + + writeJSON(w, commonResp{Data: uploadResult{ + Enqueued: enqueued, + Skipped: skipped, + TaskIDs: taskIDs, + }}) + } +} + +// makeBatchUploadFilesHandler 批量上传文件到云盘（通过 Pipeline） +func makeBatchUploadFilesHandler(pm *pipeline.Manager) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + var req struct { + Paths []string `json:"paths"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeJSON(w, commonResp{ErrMsg: "请求参数错误"}) + return + } + + // 批量路径校验 + cfg := configs.GetCurrentConfig() + if cfg == nil { + writeJSON(w, commonResp{ErrMsg: "配置未初始化"}) + return + } + + absPaths := make([]string, 0, len(req.Paths)) + skipped := []string{} + for _, p := range req.Paths { + absPath, err := getSafePath(cfg.OutPutPath, p) + if err != nil { + skipped = append(skipped, filepath.Base(p)+" - 路径越权") + continue + } + absPaths = append(absPaths, absPath) + } + + // 如果所有路径都校验失败，直接返回 + if len(absPaths) == 0 { + writeJSON(w, commonResp{Data: uploadResult{ + Enqueued: 0, + Skipped: skipped, + TaskIDs: []int64{}, + }}) + return + } + + enqueued, moreSkipped, taskIDs, err := pm.EnqueueUploadTask(absPaths) + if err != nil { + writeJSON(w, commonResp{ErrMsg: err.Error()}) + return + } + skipped = append(skipped, moreSkipped...) + + writeJSON(w, commonResp{Data: uploadResult{ + Enqueued: enqueued, + Skipped: skipped, + TaskIDs: taskIDs, + }}) + } +} diff --git a/src/webapp/src/component/file-list/index.tsx b/src/webapp/src/component/file-list/index.tsx index ef5432d..3d24a05 100644 --- a/src/webapp/src/component/file-list/index.tsx +++ b/src/webapp/src/component/file-list/index.tsx @@ -11,7 +11,9 @@ import { // @ts-ignore EditOutlined, // @ts-ignore - DeleteOutlined + DeleteOutlined, + // @ts-ignore + CloudUploadOutlined } from "@ant-design/icons"; import { Link, useNavigate, useParams } from "react-router-dom"; import Utils from "../../utils/common"; @@ -479,6 +481,7 @@ type CurrentFolderFile = { last_modified: number; size: number; subtitle_file?: string; + uploaded?: boolean; } const FileList: React.FC = () => { @@ -780,6 +783,71 @@ const FileList: React.FC = () => { .catch(err => message.error("删除失败: " + err)); }; + const handleUpload = (record: CurrentFolderFile, e?: React.MouseEvent) => { + if (e) e.stopPropagation(); + let fullPath = record.name; + if (pathParam) { + fullPath = pathParam + "/" + record.name; + } + + message.loading({ content: `正在上传 ${record.name}...`, key: "upload" }); + api.uploadFile(fullPath) + .then((rsp: any) => { + if (rsp.data) { + const data = rsp.data; + if (data.enqueued > 0) { + message.success({ content: `已入队上传任务: ${record.name}`, key: "upload" }); + } else if (data.skipped && data.skipped.length > 0) { + message.error({ content: data.skipped[0] || "上传失败", key: "upload" }); + } else { + message.error({ content: rsp.err_msg || "上传失败", key: "upload" }); + } + } else { + message.error({ content: rsp.err_msg || "上传失败", key: "upload" }); + } + }) + .catch((err: any) => message.error({ content: "上传失败: " + (err.message || err), key: "upload" })); + }; + + const handleBatchUpload = () => { + if (selectedRowKeys.length === 0) return; + const paths = selectedRowKeys.map(key => { + const fileName = key.toString(); + return pathParam ? `${pathParam}/${fileName}` : fileName; + }); + + message.loading({ content: `正在上传 ${paths.length} 个文件...`, key: "batchUpload" }); + api.batchUploadFiles(paths) + .then((rsp: any) => { + if (rsp.data) { + const data = rsp.data; + const enqueued = data.enqueued || 0; + const skipped = data.skipped || []; + const failCount = skipped.length; + if (failCount === 0) { + message.success({ content: `全部 ${enqueued} 个文件已入队上传`, key: "batchUpload" }); + } else if (enqueued > 0) { + message.warning({ content: `已入队 ${enqueued} 个，跳过 ${failCount} 个`, key: "batchUpload" }); + Modal.info({ + title: '部分文件已跳过', + content: ( +
+ {skipped.map((s: string, i: number) => ( +
{s}
+ ))} +
+ ), + }); + } else { + message.error({ content: skipped[0] || "批量上传失败", key: "batchUpload" }); + } + } else { + message.error({ content: rsp.err_msg || "批量上传失败", key: "batchUpload" }); + } + }) + .catch((err: any) => message.error({ content: "批量上传失败: " + (err.message || err), key: "batchUpload" })); + }; + const handleBatchDelete = () => { if (selectedRowKeys.length === 0) return; const paths = selectedRowKeys.map(key => { @@ -1153,6 +1221,11 @@ const FileList: React.FC = () => { 无字幕 )} + {record.uploaded && ( + + 已上传 + + )} ); } @@ -1194,6 +1267,18 @@ const FileList: React.FC = () => { > 重命名 + {!record.is_folder && ( + + )} handleDelete(record)} @@ -1401,6 +1486,15 @@ const FileList: React.FC = () => { ); })()} + { {result.error_message && ( + {result.error_message} + + } type="error" style={{ marginBottom: 8 }} /> @@ -574,7 +578,11 @@ class PipelineTaskList extends Component { {task.error_message && ( + {task.error_message} + + } type="error" /> )} diff --git a/src/webapp/src/utils/api.ts b/src/webapp/src/utils/api.ts index 183fccd..a39ff32 100644 --- a/src/webapp/src/utils/api.ts +++ b/src/webapp/src/utils/api.ts @@ -170,6 +170,22 @@ class API { return utils.requestPost(`${BASE_URL}/pipeline/batch-burn`, { paths }); } + /** + * 上传单个文件到云盘 + * @param path 文件路径 + */ + uploadFile(path: string) { + return utils.requestPost(`${BASE_URL}/file/upload`, { path }); + } + + /** + * 批量上传文件到云盘 + * @param paths 文件路径列表 + */ + batchUploadFiles(paths: string[]) { + return utils.requestPost(`${BASE_URL}/batch/file/upload`, { paths }); + } + /** * 获取Cookie列表 */