From ce35cfd26ce22b0fd0ffb01582589f06ecc6c212 Mon Sep 17 00:00:00 2001 From: Joftal <42807405+Joftal@users.noreply.github.com> Date: Thu, 13 Aug 2026 16:13:09 +0800 Subject: [PATCH] =?UTF-8?q?fix(notify):=20=E4=BF=AE=E5=A4=8D=E5=BD=95?= =?UTF-8?q?=E5=88=B6=E6=91=98=E8=A6=81=E9=80=9A=E7=9F=A5=E5=9C=A8=20Pipeli?= =?UTF-8?q?ne=20=E5=A4=84=E7=90=86=E5=89=8D=E5=8F=91=E9=80=81=E5=AF=BC?= =?UTF-8?q?=E8=87=B4=E6=96=87=E4=BB=B6=E5=88=97=E8=A1=A8=E4=B8=8D=E5=87=86?= =?UTF-8?q?=E7=A1=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 录制摘要通知原来在 recorder.run() 退出时立即发送，此时 Pipeline（转码、修复、上传等） 尚未执行。开启「上传后删除」时，通知中的文件实际已被删除，消息严重失真。 修复方案： - recorder 中添加 pipelineEnqueued 标志，Pipeline 任务入队后跳过摘要推送 - Pipeline Manager 在任务完成/失败后发送摘要，反映最终文件状态 - FileInfo 添加 Size 字段，创建时预存文件大小，确保文件删除后仍可构建摘要 - 提取 sendToAllChannels 消除通道推送重复代码 - 新增 buildUploadedSummaryBody 构造「已上传到云端」场景的消息体 Co-Authored-By: Kscc --- src/notify/send.go | 76 +++++++++++++++++++- src/pipeline/executor.go | 17 +++++ src/pipeline/manager.go | 30 +++++++- src/pipeline/types.go | 19 ++++- src/recorders/recorder.go | 144 +++++++++++++++++++++++++++++++++++--- 5 files changed, 271 insertions(+), 15 deletions(-) diff --git a/src/notify/send.go b/src/notify/send.go index 14db3c4..da8a647 100644 --- a/src/notify/send.go +++ b/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) diff --git a/src/pipeline/executor.go b/src/pipeline/executor.go index 02323be..793c04d 100644 --- a/src/pipeline/executor.go +++ b/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() diff --git a/src/pipeline/manager.go b/src/pipeline/manager.go index 1f7fab1..13683f5 100644 --- a/src/pipeline/manager.go +++ b/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: "", // 手动上传无房间名，不影响路径模板 diff --git a/src/pipeline/types.go b/src/pipeline/types.go index b2a3e3e..97d9c59 100644 --- a/src/pipeline/types.go +++ b/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 创建新的管道任务 diff --git a/src/recorders/recorder.go b/src/recorders/recorder.go index dfc9753..e1f7fb3 100644 --- a/src/recorders/recorder.go +++ b/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) {