fix(notify): 修复录制摘要通知在 Pipeline 处理前发送导致文件列表不准确

录制摘要通知原来在 recorder.run() 退出时立即发送，此时 Pipeline（转码、修复、上传等）
尚未执行。开启「上传后删除」时，通知中的文件实际已被删除，消息严重失真。

修复方案：
- recorder 中添加 pipelineEnqueued 标志，Pipeline 任务入队后跳过摘要推送
- Pipeline Manager 在任务完成/失败后发送摘要，反映最终文件状态
- FileInfo 添加 Size 字段，创建时预存文件大小，确保文件删除后仍可构建摘要
- 提取 sendToAllChannels 消除通道推送重复代码
- 新增 buildUploadedSummaryBody 构造「已上传到云端」场景的消息体

Co-Authored-By: Kscc <noreply@owtffssent.com>

src/notify/send.go

@@ -16,8 +16,9 @@ import (
 
 // RecordingFileDetail 录制文件详情
 type RecordingFileDetail struct {
-	Name string // 文件名（不含路径）
-	Size int64  // 文件大小（字节）
+	Name     string // 文件名（不含路径）
+	Size     int64  // 文件大小（字节）
+	Uploaded bool   // 是否已上传到云端（用于摘要消息标注）
 }
 
 // SendNotification 发送统一通知函数
@@ -250,6 +251,29 @@ func buildRecordingSummaryMessage(hostName, platform string, files []RecordingFi
 	return
 }
 
+// buildUploadedSummaryBody 构造"已上传到云端"场景的录制摘要消息体
+// 与 buildRecordingSummaryMessage 类似，但标注文件已上传、不显示磁盘空间
+func buildUploadedSummaryBody(platform string, files []RecordingFileDetail) string {
+	const maxDisplayFiles = 30
+
+	var sb strings.Builder
+	fmt.Fprintf(&sb, "平台：%s\n", platform)
+	fmt.Fprintf(&sb, "录制文件：%d 个（已上传到云端）\n", len(files))
+	var totalSize int64
+	for i, f := range files {
+		totalSize += f.Size
+		if i < maxDisplayFiles {
+			fmt.Fprintf(&sb, "  %d. %s (%s)\n", i+1, f.Name, formatFileSize(f.Size))
+		}
+	}
+	if len(files) > maxDisplayFiles {
+		fmt.Fprintf(&sb, "  ... 还有 %d 个文件未显示\n", len(files)-maxDisplayFiles)
+	}
+	fmt.Fprintf(&sb, "总大小：%s\n", formatFileSize(totalSize))
+	fmt.Fprintf(&sb, "本地文件已清理")
+	return sb.String()
+}
+
 // SendRecordingSummary 录制结束后发送录制文件摘要通知
 // outputPath 为录制输出路径，用于获取剩余磁盘空间
 func SendRecordingSummary(logger *livelogger.LiveLogger, hostName, platform string, files []RecordingFileDetail, outputPath string) {
@@ -262,7 +286,55 @@ func SendRecordingSummary(logger *livelogger.LiveLogger, hostName, platform stri
 	}
 
 	title, body := buildRecordingSummaryMessage(hostName, platform, files, outputPath)
+	sendToAllChannels(cfg, logger, title, body)
+}
 
+// SendPipelineRecordingSummary Pipeline 完成后发送录制摘要通知
+// originalFiles: 原始录制文件列表（allUploaded 时用于显示）
+// finalFiles: Pipeline 处理后的最终文件列表（含 Uploaded 标记）
+func SendPipelineRecordingSummary(
+	logger *livelogger.LiveLogger,
+	hostName, platform string,
+	originalFiles, finalFiles []RecordingFileDetail,
+	outputPath string,
+) {
+	cfg := configs.GetCurrentConfig()
+	if cfg == nil || !cfg.Notify.SendRecordingSummary {
+		return
+	}
+
+	// 分离已上传文件和保留文件
+	var kept []RecordingFileDetail
+	allUploaded := len(finalFiles) > 0
+	for _, f := range finalFiles {
+		if f.Uploaded {
+			// 已上传文件不显示在最终列表中（已从磁盘删除）
+		} else {
+			allUploaded = false
+			kept = append(kept, f)
+		}
+	}
+
+	var title, body string
+	if allUploaded {
+		// 所有文件已上传并删除：显示原始文件列表 + 上传状态
+		if len(originalFiles) == 0 {
+			return
+		}
+		title = fmt.Sprintf("%s 录制完成", hostName)
+		body = buildUploadedSummaryBody(platform, originalFiles)
+	} else if len(kept) > 0 {
+		// 有文件保留在磁盘：显示保留的文件列表
+		title, body = buildRecordingSummaryMessage(hostName, platform, kept, outputPath)
+	} else {
+		return
+	}
+
+	sendToAllChannels(cfg, logger, title, body)
+}
+
+// sendToAllChannels 向所有已启用的通知通道推送摘要消息
+func sendToAllChannels(cfg *configs.Config, logger *livelogger.LiveLogger, title, body string) {
 	// Telegram
 	if cfg.Notify.Telegram.Enable {
 		msg := fmt.Sprintf("%s\n%s", title, body)

src/pipeline/executor.go

@@ -114,6 +114,9 @@ func (e *Executor) Execute(
 			output, commands, logs, err = e.executeStage(ctx, stageCfg, files)
 		}
 
+		// 回填新产出文件的 Size（阶段内构造 FileInfo 时可能未 stat）
+		backfillFileSizes(output)
+
 		// 记录结果
 		result := StageResult{
 			StageName:  stageCfg.Name,
@@ -173,6 +176,8 @@ func (e *Executor) Execute(
 		}
 		return results, ctx.Ctx.Err()
 	}
+	// 保存最后阶段输出的快照（用于回调提取上传文件详情），再执行清理
+	ctx.LastStageFiles = files
 	keptFiles := e.deleteMarkedFiles(files)
 	if len(results) > 0 {
 		results[len(results)-1].OutputFiles = keptFiles
@@ -472,6 +477,18 @@ func deduplicateFiles(files []FileInfo) []FileInfo {
 	return result
 }
 
+// backfillFileSizes 对 Size 为 0 的文件执行 os.Stat 回填大小
+// 阶段内构造 FileInfo 时可能未 stat，导致摘要通知漏掉这些文件
+func backfillFileSizes(files []FileInfo) {
+	for i := range files {
+		if files[i].Size == 0 {
+			if fi, err := os.Stat(files[i].Path); err == nil {
+				files[i].Size = fi.Size()
+			}
+		}
+	}
+}
+
 // getTimeNow 获取当前时间（可用于测试时 mock）
 var getTimeNow = func() time.Time {
 	return time.Now()

src/pipeline/manager.go

@@ -13,6 +13,7 @@ import (
 	"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/notify"
 	"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"
@@ -270,6 +271,9 @@ func (m *Manager) executeTask(ctx context.Context, task *PipelineTask) {
 		logrus.WithField("task_id", task.ID).Info("pipeline task completed successfully")
 	}
 
+	// 保存最后阶段输出的快照（清理前），用于回调提取上传文件详情
+	task.LastStageFiles = pipelineCtx.LastStageFiles
+
 	// 更新任务状态
 	if err := m.store.UpdateTask(m.ctx, task); err != nil {
 		logrus.WithError(err).Error("failed to update pipeline task status after execution")
@@ -277,6 +281,11 @@ func (m *Manager) executeTask(ctx context.Context, task *PipelineTask) {
 
 	// 广播任务状态变化
 	m.broadcastTaskUpdate(task)
+
+	// 调用任务完成回调（recorder 用于追踪 Pipeline 状态并统一发送摘要）
+	if task.OnTaskComplete != nil {
+		task.OnTaskComplete(task)
+	}
 }
 
 // broadcastTaskUpdate 广播任务更新事件
@@ -286,6 +295,22 @@ func (m *Manager) broadcastTaskUpdate(task *PipelineTask) {
 	}
 }
 
+// FileInfoToDetails 将 FileInfo 切片转换为 RecordingFileDetail 切片
+// 使用 FileInfo 中预存的 Size，无需再次 stat 文件
+// 仅包含视频和其他类型文件，排除封面（封面不在录制摘要中展示）
+func FileInfoToDetails(files []FileInfo) []notify.RecordingFileDetail {
+	details := make([]notify.RecordingFileDetail, 0, len(files))
+	for _, f := range files {
+		if f.Size > 0 && f.Type != FileTypeCover {
+			details = append(details, notify.RecordingFileDetail{
+				Name: filepath.Base(f.Path),
+				Size: f.Size,
+			})
+		}
+	}
+	return details
+}
+
 // EnqueueTask 添加任务到队列
 func (m *Manager) EnqueueTask(task *PipelineTask) error {
 	if err := m.store.CreateTask(m.ctx, task); err != nil {
@@ -309,10 +334,12 @@ func (m *Manager) EnqueueTask(task *PipelineTask) error {
 
 // EnqueueRecordingTask 创建并入队录制完成后的处理任务
 // 这是 recorder.go 调用的主要入口
+// onTaskComplete: 任务完成回调（可选），用于通知 recorder 任务结束
 func (m *Manager) EnqueueRecordingTask(
 	info *live.Info,
 	pipelineConfig *PipelineConfig,
 	outputFiles []string,
+	onTaskComplete func(task *PipelineTask),
 ) error {
 	// 构建文件信息列表
 	files := make([]FileInfo, len(outputFiles))
@@ -322,6 +349,7 @@ func (m *Manager) EnqueueRecordingTask(
 
 	// 创建任务
 	task := NewPipelineTask(NewRecordInfo(info), pipelineConfig, files)
+	task.OnTaskComplete = onTaskComplete
 
 	return m.EnqueueTask(task)
 }
@@ -644,7 +672,7 @@ func (m *Manager) EnqueueUploadTask(absPaths []string) (enqueued int, skipped []
 		platform, hostName := inferUploadPathInfo(relPath)
 
 		recordInfo := RecordInfo{
-			LiveID:      types.LiveID("manual-upload"),
+			LiveID:      types.LiveID(ManualUploadLiveID),
 			Platform:    platform,              // 用于上传路径模板：{{ .Platform }}
 			HostName:    hostName,              // 用于上传路径模板：{{ .HostName }}
 			RoomName:    "",                    // 手动上传无房间名，不影响路径模板

src/pipeline/types.go

@@ -4,6 +4,7 @@ package pipeline
 
 import (
 	"context"
+	"os"
 	"time"
 
 	"github.com/bililive-go/bililive-go/src/live"
@@ -15,6 +16,9 @@ import (
 // PipelineTaskUpdateEvent 管道任务更新事件
 const PipelineTaskUpdateEvent events.EventType = "PipelineTaskUpdate"
 
+// ManualUploadLiveID 手动上传任务的 LiveID 标识
+const ManualUploadLiveID = "manual-upload"
+
 // FileType 文件类型
 type FileType string
 
@@ -30,18 +34,23 @@ const (
 // FileInfo 表示管道中流转的文件信息
 type FileInfo struct {
 	Path       string         `json:"path"`                  // 文件绝对路径
+	Size       int64          `json:"size,omitempty"`        // 文件大小（字节），在任务创建时记录
 	Type       FileType       `json:"type"`                  // 文件类型
 	SourcePath string         `json:"source_path,omitempty"` // 来源文件路径（用于追踪转换链）
 	Metadata   map[string]any `json:"metadata,omitempty"`    // 额外元数据
 	Deletable  bool           `json:"deletable,omitempty"`   // 标记此文件在管道全部成功后可删除
 }
 
-// NewVideoFileInfo 创建视频文件信息
+// NewVideoFileInfo 创建视频文件信息，自动记录文件大小
 func NewVideoFileInfo(path string) FileInfo {
-	return FileInfo{
+	fi := FileInfo{
 		Path: path,
 		Type: FileTypeVideo,
 	}
+	if stat, err := os.Stat(path); err == nil {
+		fi.Size = stat.Size()
+	}
+	return fi
 }
 
 // NewCoverFileInfo 创建封面文件信息
@@ -84,6 +93,10 @@ type PipelineContext struct {
 
 	// FFmpegPath 是 ffmpeg 可执行文件的路径
 	FFmpegPath string
+
+	// LastStageFiles 最后阶段输出的文件快照（清理前保存）
+	// 用于回调提取上传文件详情，因为 deleteMarkedFiles 会删除文件
+	LastStageFiles []FileInfo
 }
 
 // Stage 管道阶段接口
@@ -243,6 +256,8 @@ type PipelineTask struct {
 	CompletedAt    *time.Time      `json:"completed_at,omitempty"`
 	ErrorMessage   string          `json:"error_message,omitempty"`
 	CanRetry       bool            `json:"can_retry"` // 是否可以重试
+	OnTaskComplete func(task *PipelineTask) `json:"-"` // 任务完成回调（不持久化）
+	LastStageFiles []FileInfo      `json:"-"` // 最后阶段输出文件（清理前快照，不持久化）
 }
 
 // NewPipelineTask 创建新的管道任务

src/recorders/recorder.go

@@ -281,6 +281,18 @@ type recorder struct {
 	currentFileLock sync.RWMutex
 	currentFilePath string
 
+	// pipelineEnqueued 标记 Pipeline 任务已入队，由 Pipeline Manager 负责发送录制摘要
+	// 避免 recorder 在 Pipeline 处理完成前发送不准确的原始文件列表
+	pipelineEnqueued bool
+	// pipelinePendingCount 追踪尚未完成的 Pipeline 任务数量
+	// 仅当所有任务完成且录制结束后，才发送统一的录制摘要
+	pipelinePendingCount int
+	// pipelineMu 保护 pipelinePendingCount 和 pipelineDetails 的并发访问
+	pipelineMu sync.Mutex
+	// pipelineDetails 收集各 Pipeline 任务的最终文件详情
+	// 任务完成时通过回调写入，录制结束时读取用于构造摘要
+	pipelineDetails []notify.RecordingFileDetail
+
 	// 当前录制的流信息（来自平台 API）
 	currentStreamInfo *live.AvailableStreamInfo
 
@@ -784,9 +796,11 @@ func (r *recorder) tryRecord(ctx context.Context) {
 		}
 
 		// 入队 Pipeline 任务
-		if err := pipelineManager.EnqueueRecordingTask(info, pipelineConfig, outputFiles); err != nil {
+		if err := pipelineManager.EnqueueRecordingTask(info, pipelineConfig, outputFiles, r.onPipelineTaskComplete); err != nil {
 			r.getLogger().WithError(err).Error("failed to enqueue pipeline task")
 		} else {
+			r.pipelineEnqueued = true
+			r.pipelinePendingCount++
 			// 记录 Pipeline 阶段顺序
 			var stageNames []string
 			for _, stage := range pipelineConfig.Stages {
@@ -958,22 +972,26 @@ func (r *recorder) accumulateRecordedFiles(files ...string) {
 // sendAccumulatedSummary 录制结束后统一推送录制文件摘要通知
 // 在 run() 退出时通过 defer 调用，确保所有分段文件汇总为一条通知
 func (r *recorder) sendAccumulatedSummary() {
+	r.pipelineMu.Lock()
+	pendingCount := r.pipelinePendingCount
+	r.pipelineMu.Unlock()
+
+	// Pipeline 任务尚未全部完成，等待回调触发最终摘要
+	if r.pipelineEnqueued && pendingCount > 0 {
+		r.getLogger().Infof("Pipeline 任务尚未全部完成（剩余 %d 个），延迟摘要推送", pendingCount)
+		return
+	}
+
 	r.recordedFilesMu.Lock()
 	defer r.recordedFilesMu.Unlock()
 	if r.suppressSummary {
 		r.getLogger().Infof("录制摘要推送已抑制（分段重启），累积 %d 个文件将传递给新 recorder", len(r.recordedFiles))
 		return
 	}
-	if len(r.recordedFiles) == 0 {
+	if len(r.recordedFiles) == 0 && len(r.pipelineDetails) == 0 {
 		r.getLogger().Info("无录制文件，跳过摘要推送")
 		return
 	}
-	obj, err := r.cache.Get(r.Live)
-	if err != nil {
-		r.getLogger().WithError(err).Error("获取直播信息失败，无法推送录制摘要")
-		return
-	}
-	info := obj.(*live.Info)
 
 	// 获取录制输出路径，用于查询剩余磁盘空间
 	cfg := configs.GetCurrentConfig()
@@ -984,8 +1002,114 @@ func (r *recorder) sendAccumulatedSummary() {
 		outputPath = resolved.OutPutPath
 	}
 
-	r.getLogger().Infof("推送录制摘要：%d 个文件", len(r.recordedFiles))
-	notify.SendRecordingSummary(r.getLogger(), info.HostName, r.Live.GetPlatformCNName(), r.recordedFiles, outputPath)
+	// 使用 Pipeline 收集的最终文件详情（如有），否则用录制累积的原始文件
+	if len(r.pipelineDetails) > 0 {
+		r.getLogger().Infof("推送 Pipeline 录制摘要：%d 个文件", len(r.pipelineDetails))
+		hostName := ""
+		platform := r.Live.GetPlatformCNName()
+		if obj, cacheErr := r.cache.Get(r.Live); cacheErr == nil {
+			if info, ok := obj.(*live.Info); ok {
+				hostName = info.HostName
+			}
+		}
+		notify.SendPipelineRecordingSummary(
+			r.getLogger(),
+			hostName,
+			platform,
+			r.recordedFiles,     // originalFiles（allUploaded 时显示用）
+			r.pipelineDetails,   // finalFiles
+			outputPath,
+		)
+	} else {
+		// Pipeline 未产出有效文件详情，回退到录制累积
+		if len(r.recordedFiles) == 0 {
+			return
+		}
+		obj, err := r.cache.Get(r.Live)
+		if err != nil {
+			r.getLogger().WithError(err).Error("获取直播信息失败，无法推送录制摘要")
+			return
+		}
+		info := obj.(*live.Info)
+		r.getLogger().Infof("推送录制摘要：%d 个文件", len(r.recordedFiles))
+		notify.SendRecordingSummary(r.getLogger(), info.HostName, r.Live.GetPlatformCNName(), r.recordedFiles, outputPath)
+	}
+}
+
+// onPipelineTaskComplete Pipeline 任务完成回调
+// 由 Pipeline Manager 在 executeTask 结束时调用
+func (r *recorder) onPipelineTaskComplete(task *pipeline.PipelineTask) {
+	r.pipelineMu.Lock()
+	r.pipelinePendingCount--
+
+	// 收集任务的最终文件详情
+	switch task.Status {
+	case pipeline.PipelineStatusCompleted:
+		finalFiles := task.CurrentFiles
+		if len(finalFiles) == 0 {
+			// 所有文件已上传并删除：从 LastStageFiles 提取真正上传的文件
+			uploadedDetails := extractUploadedDetails(task.LastStageFiles)
+			if len(uploadedDetails) > 0 {
+				r.pipelineDetails = append(r.pipelineDetails, uploadedDetails...)
+			} else {
+				// 回退：无上传标记时用原始文件
+				for _, d := range pipeline.FileInfoToDetails(task.InitialFiles) {
+					d.Uploaded = true
+					r.pipelineDetails = append(r.pipelineDetails, d)
+				}
+			}
+		} else {
+			r.pipelineDetails = append(r.pipelineDetails, pipeline.FileInfoToDetails(finalFiles)...)
+		}
+	case pipeline.PipelineStatusFailed, pipeline.PipelineStatusCancelled:
+		// 失败/取消时优先用 CurrentFiles（最后成功阶段的输出），回退到 InitialFiles
+		if len(task.CurrentFiles) > 0 {
+			r.pipelineDetails = append(r.pipelineDetails, pipeline.FileInfoToDetails(task.CurrentFiles)...)
+		} else {
+			r.pipelineDetails = append(r.pipelineDetails, pipeline.FileInfoToDetails(task.InitialFiles)...)
+		}
+	}
+
+	remaining := r.pipelinePendingCount
+
+	// 防止回调被重复执行（如内存中重试场景）
+	task.OnTaskComplete = nil
+
+	r.pipelineMu.Unlock()
+
+	r.getLogger().Infof("Pipeline 任务完成（状态: %s），剩余 %d 个任务", task.Status, remaining)
+
+	// 所有任务完成且录制已结束，发送统一摘要
+	// 使用 goroutine 避免通知发送（Telegram/Email 等）阻塞 Pipeline 任务槽位
+	if remaining == 0 {
+		select {
+		case <-r.done:
+			// run() 已退出，异步发送摘要（不阻塞 executeTask）
+			go r.sendAccumulatedSummary()
+		default:
+			// run() 仍在运行，等待 run() 退出时由 sendAccumulatedSummary 发送
+		}
+	}
+}
+
+// extractUploadedDetails 从最后阶段输出文件中提取有上传标记的文件详情
+// CloudUploadStage 会为成功上传的文件设置 Metadata["uploaded"]=true
+func extractUploadedDetails(files []pipeline.FileInfo) []notify.RecordingFileDetail {
+	var details []notify.RecordingFileDetail
+	for _, f := range files {
+		if f.Metadata != nil {
+			if uploaded, ok := f.Metadata["uploaded"].(bool); ok && uploaded {
+				if f.Size > 0 {
+					details = append(details, notify.RecordingFileDetail{
+						Name:     filepath.Base(f.Path),
+						Size:     f.Size,
+						Uploaded: true,
+					})
+				}
+			}
+		}
+	}
+	return details
 }
 
 func (r *recorder) logStreamURLRetry(err error) {