diff --git a/src/notify/send.go b/src/notify/send.go index 14db3c4..4175205 100644 --- a/src/notify/send.go +++ b/src/notify/send.go @@ -250,6 +250,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 +285,46 @@ func SendRecordingSummary(logger *livelogger.LiveLogger, hostName, platform stri } title, body := buildRecordingSummaryMessage(hostName, platform, files, outputPath) + sendToAllChannels(cfg, logger, title, body) +} +// SendPipelineRecordingSummary Pipeline 完成后发送录制摘要通知 +// originalFiles: 原始录制文件列表(allUploadedAndDeleted 时用于显示) +// finalFiles: Pipeline 处理后的最终文件列表 +// allUploadedAndDeleted: 所有文件已上传并删除(显示上传状态而非磁盘文件) +func SendPipelineRecordingSummary( + logger *livelogger.LiveLogger, + hostName, platform string, + originalFiles, finalFiles []RecordingFileDetail, + outputPath string, + allUploadedAndDeleted bool, +) { + cfg := configs.GetCurrentConfig() + if cfg == nil || !cfg.Notify.SendRecordingSummary { + return + } + + var title, body string + if allUploadedAndDeleted { + // 所有文件已上传并删除:显示原始文件列表 + 上传状态 + if len(originalFiles) == 0 { + return + } + title = fmt.Sprintf("%s 录制完成", hostName) + body = buildUploadedSummaryBody(platform, originalFiles) + } else { + // 有文件保留或 Pipeline 失败:显示最终文件列表 + if len(finalFiles) == 0 { + return + } + title, body = buildRecordingSummaryMessage(hostName, platform, finalFiles, outputPath) + } + + 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/manager.go b/src/pipeline/manager.go index 1f7fab1..53f4969 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" @@ -277,6 +278,9 @@ func (m *Manager) executeTask(ctx context.Context, task *PipelineTask) { // 广播任务状态变化 m.broadcastTaskUpdate(task) + + // 发送录制摘要通知(仅对录制任务,非手动上传) + m.sendRecordingSummaryIfNeeded(task) } // broadcastTaskUpdate 广播任务更新事件 @@ -286,6 +290,92 @@ func (m *Manager) broadcastTaskUpdate(task *PipelineTask) { } } +// sendRecordingSummaryIfNeeded 在 Pipeline 任务完成后发送录制摘要通知 +// 仅对录制任务(LiveID 非 "manual-upload")发送,手动上传不发通知 +func (m *Manager) sendRecordingSummaryIfNeeded(task *PipelineTask) { + // 跳过手动上传任务 + if task.RecordInfo.LiveID == "manual-upload" { + return + } + + // 仅在完成或失败时发送 + if task.Status != PipelineStatusCompleted && task.Status != PipelineStatusFailed { + return + } + + // 获取输出路径(用于查询剩余磁盘空间) + outputPath := "" + if len(task.InitialFiles) > 0 { + outputPath = filepath.Dir(task.InitialFiles[0].Path) + } + + logger := livelogger.New(livelogger.DefaultBufferSize, logrus.Fields{ + "platform": task.RecordInfo.Platform, + "host": task.RecordInfo.HostName, + }) + + switch task.Status { + case PipelineStatusCompleted: + finalFiles := task.CurrentFiles // deleteMarkedFiles 后的保留文件 + + if len(finalFiles) == 0 { + // 所有文件已上传并删除:从 InitialFiles 构建原始文件列表(Size 已预存) + originalDetails := fileInfoToDetails(task.InitialFiles) + notify.SendPipelineRecordingSummary( + logger, + task.RecordInfo.HostName, + task.RecordInfo.Platform, + originalDetails, // originalFiles(显示用) + nil, // finalFiles + outputPath, + true, // allUploadedAndDeleted + ) + } else { + // 有文件保留:显示最终文件列表 + fileDetails := fileInfoToDetails(finalFiles) + notify.SendPipelineRecordingSummary( + logger, + task.RecordInfo.HostName, + task.RecordInfo.Platform, + nil, // originalFiles(不需要) + fileDetails, // finalFiles + outputPath, + false, + ) + } + + case PipelineStatusFailed: + // Pipeline 失败:文件仍在磁盘,用原始文件列表 + fileDetails := fileInfoToDetails(task.InitialFiles) + if len(fileDetails) > 0 { + notify.SendPipelineRecordingSummary( + logger, + task.RecordInfo.HostName, + task.RecordInfo.Platform, + nil, // originalFiles(不需要) + fileDetails, // finalFiles(失败时=原始文件) + outputPath, + false, + ) + } + } +} + +// 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 { + 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 { diff --git a/src/pipeline/types.go b/src/pipeline/types.go index b2a3e3e..2d76438 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" @@ -30,18 +31,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 创建封面文件信息 diff --git a/src/recorders/recorder.go b/src/recorders/recorder.go index dfc9753..d1a5b1f 100644 --- a/src/recorders/recorder.go +++ b/src/recorders/recorder.go @@ -281,6 +281,10 @@ type recorder struct { currentFileLock sync.RWMutex currentFilePath string + // pipelineEnqueued 标记 Pipeline 任务已入队,由 Pipeline Manager 负责发送录制摘要 + // 避免 recorder 在 Pipeline 处理完成前发送不准确的原始文件列表 + pipelineEnqueued bool + // 当前录制的流信息(来自平台 API) currentStreamInfo *live.AvailableStreamInfo @@ -787,6 +791,7 @@ func (r *recorder) tryRecord(ctx context.Context) { if err := pipelineManager.EnqueueRecordingTask(info, pipelineConfig, outputFiles); err != nil { r.getLogger().WithError(err).Error("failed to enqueue pipeline task") } else { + r.pipelineEnqueued = true // 记录 Pipeline 阶段顺序 var stageNames []string for _, stage := range pipelineConfig.Stages { @@ -960,6 +965,12 @@ func (r *recorder) accumulateRecordedFiles(files ...string) { func (r *recorder) sendAccumulatedSummary() { r.recordedFilesMu.Lock() defer r.recordedFilesMu.Unlock() + // Pipeline 已入队,由 Pipeline Manager 在任务完成后发送摘要, + // 确保通知反映 Pipeline 处理后的最终文件状态(而非原始录制文件) + if r.pipelineEnqueued { + r.getLogger().Infof("Pipeline 任务已入队,跳过录制摘要推送(由 Pipeline Manager 负责)") + return + } if r.suppressSummary { r.getLogger().Infof("录制摘要推送已抑制(分段重启),累积 %d 个文件将传递给新 recorder", len(r.recordedFiles)) return