okooooooo/bililive-go
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(upload): 手动上传改用 Pipeline 任务，与自动上传行为一致

Browse Source 
手动上传（文件管理界面）原来直接调用 OpenList 客户端上传，
路径推断逻辑与自动上传（CloudUploadStage）不一致，且无任务追踪。

改为创建单阶段 cloud_upload PipelineTask，复用 CloudUploadStage 的
完整逻辑：路径模板渲染、上传标记持久化、任务追踪、重试能力。

主要改动：
- 新增 EnqueueUploadTask 方法，接收绝对路径列表创建 Pipeline 任务
- 新增 upload_handler.go，通过 Pipeline 系统处理手动上传请求
- 提取 RenderUploadPath 为公共函数，消除 extract_cover.go 中的重复代码
- 上传成功后持久化标记到 metadata DB，文件列表显示已上传标签
- 文件删除/重命名时自动清除/迁移上传标记
- Pipeline 任务错误信息支持多行展示

Co-Authored-By: Kscc <noreply@owtffssent.com>
This commit is contained in:
JoftalandKscc committed 2026-08-12 18:03:54 +08:00
1 parent d92070a89c
commit 61abe70df2
11 files changed
+550 -65
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/pipeline/executor.go
+15
View File
@@ -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)
src/pipeline/manager.go
+91
View File
@@ -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 {
src/pipeline/stages/extract_cover.go
+31 -59
View File
@@ -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 {
src/pipeline/upload_utils.go
+67
View File
@@ -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
}
src/pkg/metadata/store.go
+2
View File
@@ -206,6 +206,8 @@ const (
NamespaceUpdate = "update"
// NamespaceMigration 数据库迁移状态
NamespaceMigration = "migration"
// NamespaceUploaded 已上传文件标记（key=相对路径, value="1"）
NamespaceUploaded = "uploaded"
)
// 预定义的键常量
src/servers/handler.go
+96 -2
View File
@@ -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: "成功"})
}
}
src/servers/server.go
+2 -1
View File
@@ -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 开放直播录制协议路由
src/servers/upload_handler.go
+125
View File
@@ -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,
}})
}
}
src/webapp/src/component/file-list/index.tsx
+95 -1
View File
@@ -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: (
<div style={{ maxHeight: 300, overflow: 'auto' }}>
{skipped.map((s: string, i: number) => (
<div key={i} style={{ fontSize: '12px', color: '#8c8c8c' }}>{s}</div>
))}
</div>
),
});
} 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 = () => {
无字幕
</span>
)}
{record.uploaded && (
<span style={{ marginLeft: 6, fontSize: 11, color: '#52c41a', background: '#f6ffed', padding: '1px 6px', borderRadius: 4, cursor: 'default' }}>
已上传
</span>
)}
</div>
);
}
@@ -1194,6 +1267,18 @@ const FileList: React.FC = () => {
>
重命名
</Button>
{!record.is_folder && (
<Button
type="link"
size="small"
// @ts-ignore
icon={<CloudUploadOutlined />}
onClick={(e) => handleUpload(record, e)}
className="action-btn"
>
上传
</Button>
)}
<Popconfirm
title={`确定要删除${record.is_folder ? '文件夹' : '文件'} "${record.name}" 吗？`}
onConfirm={() => handleDelete(record)}
@@ -1401,6 +1486,15 @@ const FileList: React.FC = () => {
</Tooltip>
);
})()}
<Button
size="small"
type="default"
// @ts-ignore
icon={<CloudUploadOutlined />}
onClick={handleBatchUpload}
>
批量上传
</Button>
<Popconfirm
title={`确定要删除选中的 ${selectedRowKeys.length} 个项目吗？`}
onConfirm={handleBatchDelete}
src/webapp/src/component/pipeline-task-list/index.tsx
+10 -2
View File
@@ -495,7 +495,11 @@ class PipelineTaskList extends Component<object, PipelineTaskListState> {
{result.error_message && (
<Alert
message="错误信息"
description={result.error_message}
description={
<div style={{ whiteSpace: 'pre-wrap', maxHeight: 200, overflow: 'auto' }}>
{result.error_message}
</div>
}
type="error"
style={{ marginBottom: 8 }}
/>
@@ -574,7 +578,11 @@ class PipelineTaskList extends Component<object, PipelineTaskListState> {
{task.error_message && (
<Alert
message="任务失败"
description={task.error_message}
description={
<div style={{ whiteSpace: 'pre-wrap', maxHeight: 300, overflow: 'auto' }}>
{task.error_message}
</div>
}
type="error"
/>
)}
src/webapp/src/utils/api.ts
+16
View File
@@ -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列表
*/