// 直播结束后自动合并录播和弹幕层 package main import ( "bufio" "encoding/json" "errors" "fmt" "os" "path/filepath" "strings" "time" ) // 弹幕和视频录制结束时都会调用:标记对应部分已结束,两部分都自然结束后启动自动合并 func finishPart(liveID string, danmu bool) { lInfoMap.Lock() info, ok := lInfoMap.info[liveID] if !ok { lInfoMap.Unlock() lPrintWarnf("找不到 liveID 为 %s 的直播信息,跳过录制结束的处理(记录可能已被清理)", liveID) return } info.doneAt = time.Now().UnixNano() if danmu { info.isDanmu = false info.danmuDone = true } else { info.isRecording = false info.recordDone = true } // 弹幕文件写坏过就别合并了:成片会缺一大截弹幕,界面上看不出是缺还是本来就没弹幕 broke := info.danmuWriteFailed shouldMerge := config.AutoMerge && mainCtx.Err() == nil && !broke && !info.danmuStopped && !info.recordStopped && info.danmuDone && info.recordDone && !info.mergeStarted && info.assFile != "" && info.firstRecordFile != "" if broke && config.AutoMerge && info.danmuDone && info.recordDone && !info.mergeStarted { lPrintWarn("本场弹幕文件在录制时写盘出错,跳过自动合并(录播和弹幕文件都留着)") } var segs []recordSegment if shouldMerge { info.mergeStarted = true // 复制一份分段列表:合并是在另一个 goroutine 里做的,不能让它读到会被后续修改的切片 segs = append([]recordSegment(nil), info.recordSegments...) if len(segs) == 0 { segs = []recordSegment{{file: info.firstRecordFile, startNs: info.recordStartNs}} } } lInfoMap.info[liveID] = info lInfoMap.Unlock() if shouldMerge { _, jsonlFile := nickJsonlFile(info.assFile) uid := info.uid danmuStartNs := info.cfg.StartTime // 合并是最重的任务,必须兜底 recover,否则一次 panic 会终止整个程序并中断其它主播的录制 safeGo("自动合并"+longID(uid), func() { autoMerge(uid, segs, jsonlFile, danmuStartNs) }) } } // 弹幕录制结束 func finishDanmuPart(liveID string) { finishPart(liveID, true) } // 视频录制结束 func finishRecordPart(liveID string) { finishPart(liveID, false) } // 查找文件实际位置:优先原路径,其次 directory(录制结束时文件可能已被移动过去) func findMovedFile(path string) string { if path == "" { return "" } if _, err := os.Stat(path); err == nil { return path } // 录制结束时文件可能已经被移动到保存目录下的子文件夹 p := filepath.Join(recordSaveDir(), filepath.Base(path)) if _, err := os.Stat(p); err == nil { return p } return "" } // 录播因为断流可能分成多段:单段直接使用,多段先拼成完整视频并把弹幕时间映射到拼接后的时间轴。 // 返回合并要用的视频、字幕文件和弹幕相对视频起点的时间偏移、**实际参与合并的源文件**, // 以及清理临时文件的函数。used 是给「合并完删源视频」用的:中途回退成「只合并第一个分段」时, // 后面的分段并没有进成片,按原分段列表全删就等于把后半场永久丢了。 func prepareForMerge(ffmpegFile string, segs []recordSegment, jsonlFile string, danmuStartNs int64) (string, string, float64, []string, func(), error) { noop := func() {} // 录制结束时文件可能已经被移动到保存目录,逐个找实际位置 files := make([]string, 0, len(segs)) starts := make([]int64, 0, len(segs)) for _, s := range segs { if p := findMovedFile(s.file); p != "" { files = append(files, p) starts = append(starts, s.startNs) } } if len(files) == 0 { return "", "", 0, nil, noop, errors.New("找不到录播视频文件") } if len(files) == 1 { offset := float64(danmuStartNs-starts[0]) / 1e9 return files[0], jsonlFile, offset, files, noop, nil } lPrintf("本场录播中途中断过,共 %d 个分段,先拼接成完整视频再合并", len(files)) // 每段的真实时长:拼接后的时间轴是按各段时长累加得到的,弹幕时间要照着映射过去 durs := make([]float64, len(files)) for i, f := range files { durs[i] = probeDuration(ffmpegFile, f) if durs[i] <= 0 { lPrintWarnf("无法获取分段 %s 的时长,本场只合并第一个分段(后半场没有弹幕)", filepath.Base(f)) offset := float64(danmuStartNs-starts[0]) / 1e9 return files[0], jsonlFile, offset, files[:1], noop, nil } } dir := filepath.Dir(jsonlFile) // 拼接出来的中间文件沿用分段的封装格式,concat 用 -c copy 时容器一致最稳 joined := filepath.Join(dir, chunkTempPrefix+"joined"+filepath.Ext(files[0])) jobSetPhase(jobPhaseJoin) joinedSec := 0.0 for _, d := range durs { joinedSec += d } jobSetTotal(joinedSec) if err := concatChunks(ffmpegFile, files, joined, dir, func(outSec, speed, fps float64, size int64) { if joinedSec > 0 { jobSetPhasePercent(outSec / joinedSec) } }); err != nil { // 任务被手动停止时不要再退化成「只合并第一个分段」, // 也别把拼到一半的中间文件留下 _ = os.Remove(joined) if errors.Is(err, errJobStopped) { return "", "", 0, nil, noop, errJobStopped } lPrintWarnf("拼接录播分段失败,本场只合并第一个分段(后半场没有弹幕):%v", err) offset := float64(danmuStartNs-starts[0]) / 1e9 return files[0], jsonlFile, offset, files[:1], noop, nil } newJsonl := filepath.Join(dir, chunkTempPrefix+"remapped.jsonl") if err := remapJsonlTimes(jsonlFile, newJsonl, starts, durs, danmuStartNs); err != nil { _ = os.Remove(joined) lPrintWarnf("重映射弹幕时间失败,本场只合并第一个分段(后半场没有弹幕):%v", err) offset := float64(danmuStartNs-starts[0]) / 1e9 return files[0], jsonlFile, offset, files[:1], noop, nil } cleanup := func() { _ = os.Remove(joined) _ = os.Remove(newJsonl) } return joined, newJsonl, 0, files, cleanup, nil } // 把弹幕时间从「真实时间轴」映射到「分段拼接后的时间轴」 func remapJsonlTimes(srcFile, dstFile string, starts []int64, durs []float64, danmuStartNs int64) error { // 每段的真实起点(相对弹幕开始录制)和它在拼接后时间轴上的起点 realStart := make([]float64, len(starts)) concatStart := make([]float64, len(starts)) sum := 0.0 for i := range starts { realStart[i] = float64(starts[i]-danmuStartNs) / 1e9 concatStart[i] = sum sum += durs[i] } in, err := os.Open(srcFile) if err != nil { return err } defer in.Close() out, err := os.Create(dstFile) if err != nil { return err } defer out.Close() w := bufio.NewWriter(out) scanner := bufio.NewScanner(in) scanner.Buffer(make([]byte, 1024*1024), 1024*1024) first := true dropped, moved := 0, 0 for scanner.Scan() { line := scanner.Bytes() if first { // 首行是元数据,原样保留 w.Write(line) w.WriteByte('\n') first = false continue } var r danmuRecord if json.Unmarshal(line, &r) != nil || r.Content == "" { continue } t := r.Time idx := -1 for i := range realStart { if t >= realStart[i] && t < realStart[i]+durs[i] { idx = i break } } switch { case idx >= 0: r.Time = concatStart[idx] + (t - realStart[idx]) default: // 落在断流空档里的弹幕,挪到下一段的开头 for i := range realStart { if t < realStart[i] { idx = i break } } if idx < 0 { // 晚于最后一段:视频里没有对应画面,丢弃 dropped++ continue } r.Time = concatStart[idx] moved++ } b, err := json.Marshal(r) if err != nil { continue } w.Write(b) w.WriteByte('\n') } if err := scanner.Err(); err != nil { return err } if err := w.Flush(); err != nil { return err } if moved > 0 || dropped > 0 { lPrintf("弹幕时间轴已对齐拼接后的视频:%d 条落在断流空档里(挪到下一段开头),%d 条超出视频(丢弃)", moved, dropped) } return nil } // 自动合并入口:排队跑完一场,把结果写进任务进度;日志措辞仍由内部负责 func autoMerge(uid int, segs []recordSegment, jsonlFile string, danmuStartNs int64) { // 多个主播可能同时下播,而全局只有一份任务状态:先排队,跑到了再登记进度 if jobActive() { lPrintf("已有合并任务在跑(%s),本场先排队等待", jobBusyName()) } err := runMergeExclusive(func() error { return autoMergeJob(uid, segs, jsonlFile, danmuStartNs) }) if err == errJobStopped { // 停止不是故障,内部不打印,这里补一句 lPrintWarn("自动合并已被手动停止" + longID(uid)) } jobFinish(jobErrText(err)) // 失败要推一条,否则只有盯着控制台才知道这场没合成出来 notifyMergeFailure(uid, autoMergeOutFile(jsonlFile), err) } // 把错误翻成给界面看的原因:正常结束和「没有弹幕」都不算失败 func jobErrText(err error) string { switch err { case nil, errNoDanmu: return "" case errJobStopped: return errJobStopped.Error() } return err.Error() } // 自动合并成片的最终路径:成片目录 + 与弹幕文件同名的「_弹幕合并.mp4」 func autoMergeOutFile(jsonlFile string) string { name := filepath.Base(strings.TrimSuffix(jsonlFile, ".jsonl") + "_弹幕合并.mp4") return filepath.Join(mergeSaveDir(), name) } // 执行自动合并及后续的删除/移动。日志都在内部按原措辞打印,返回值只给上层写进度用 func autoMergeJob(uid int, segs []recordSegment, jsonlFile string, danmuStartNs int64) error { lPrintln("开始自动合并" + longID(uid) + "的录播和弹幕层") // 视频和弹幕是一次录制产生的,文件名去掉扩展名后应该一致(扩展名由 recordFormat 决定) if len(segs) > 0 { videoBase := filepath.Base(strings.TrimSuffix(segs[0].file, filepath.Ext(segs[0].file))) jsonlBase := filepath.Base(strings.TrimSuffix(jsonlFile, ".jsonl")) if videoBase != jsonlBase { lPrintWarnf("视频文件(%s)和弹幕文件(%s)不是同一场录制,取消自动合并", segs[0].file, jsonlFile) return errors.New("视频文件和弹幕文件不是同一场录制") } } ffmpegFile := getFFmpeg() if ffmpegFile == "" { lPrintErr("自动合并失败:找不到 ffmpeg,请安装并加入 PATH,或在 web UI 设置里填写 ffmpeg 路径") return errors.New("找不到 ffmpeg,请安装并加入 PATH,或在 web UI 设置里填写 ffmpeg 路径") } // 成片先在本地合成,再落到成片文件夹:成片目录可能是网络盘或外接盘, // 直接往里写、写到一半失败会在那里留下一个看起来正常的半成品 mergeDir := mergeSaveDir() if err := os.MkdirAll(mergeDir, 0755); err != nil { lPrintWarnf("创建成片文件夹 %s 失败:%v", mergeDir, err) return fmt.Errorf("创建成片文件夹 %s 失败:%v", mergeDir, err) } outFile := autoMergeOutFile(jsonlFile) tmpOut := tempMergeOutput(jsonlFile, filepath.Base(outFile)) // 输入大小按各分段实际位置合计(录制结束时文件可能已被移动到保存目录) inputBytes := int64(0) for _, s := range segs { if p := findMovedFile(s.file); p != "" { inputBytes += fileSizeOrZero(p) } } if !jobStart(jobKindAutoMerge, filepath.Base(outFile), inputBytes, 0) { // 走到这里说明没先拿 mergeSlot,与其把别人的进度冲掉不如直接放弃这一场 return errors.New("开始任务失败:还有别的合并任务没结束,本场合并已跳过") } video, jsonl, offset, used, cleanup, err := prepareForMerge(ffmpegFile, segs, jsonlFile, danmuStartNs) defer cleanup() if err != nil { lPrintErrf("自动合并失败:%v", err) return err } // 标记成正在写入,避免在 web UI 的文件管理里把还没写完的成片删掉 markFileActive(tmpOut, true) // 自动合并没有单独的参数,全部沿用全局设置 burnErr := burnDanmu(jsonl, video, tmpOut, offset, mergeOptions{}) markFileActive(tmpOut, false) if burnErr != nil { _ = os.Remove(tmpOut) switch burnErr { case errJobStopped: return errJobStopped case errNoDanmu: lPrintWarn("弹幕文件里没有弹幕,跳过自动合并") return errNoDanmu } lPrintErrf("自动合并失败:%v", burnErr) return burnErr } jobSetPhase(jobPhasePlace) if err := placeOutput(tmpOut, outFile); err != nil { _ = os.Remove(tmpOut) lPrintErrf("自动合并失败:%v", err) return err } lPrintln("自动合并完成:" + outFile) sendNotify(notifyEventMergeDone, uid, nameForLog(uid), "", "", outFile, longID(uid)+" 的录播和弹幕合并完成\n"+outFile) if config.MergeDelVideo { // 只删真正进了成片的分段:拼接中途回退成「只合并第一段」时,后面的分段没被合并, // 按原分段列表全删就等于把后半场永久丢掉 for _, p := range used { if p == "" { continue } if err := os.Remove(p); err == nil { lPrintln("已删除源视频:" + p) } else { lPrintWarnf("删除源视频失败:%v", err) } } } if config.MergeDelDanmu { base := strings.TrimSuffix(jsonlFile, ".jsonl") for _, p := range []string{base + ".ass", base + ".nick.ass", base + ".jsonl"} { if err := os.Remove(p); err == nil { lPrintln("已删除弹幕文件:" + p) } } } return nil }