fix(notify): 修复录制摘要通知在 Pipeline 处理前发送导致文件列表不准确
录制摘要通知原来在 recorder.run() 退出时立即发送，此时 Pipeline（转码、修复、上传等） 尚未执行。开启「上传后删除」时，通知中的文件实际已被删除，消息严重失真。 修复方案： - recorder 中添加 pipelineEnqueued 标志，Pipeline 任务入队后跳过摘要推送 - Pipeline Manager 在任务完成/失败后发送摘要，反映最终文件状态 - FileInfo 添加 Size 字段，创建时预存文件大小，确保文件删除后仍可构建摘要 - 提取 sendToAllChannels 消除通道推送重复代码 - 新增 buildUploadedSummaryBody 构造「已上传到云端」场景的消息体 Co-Authored-By: Kscc <noreply@owtffssent.com>
This commit is contained in:
JoftalandKscc committed
1 parent
2b75cd31f9
commit
3699d798a3
4 files changed
+171
-2
No files matched your search
@@ -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)
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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 创建封面文件信息
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in new issue
Block a user