acfunlive-remake/automerge.go

// 直播结束后自动合并录播和弹幕层
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
	}
	shouldMerge := config.AutoMerge && mainCtx.Err() == nil &&
		!info.danmuStopped && !info.recordStopped &&
		info.danmuDone && info.recordDone && !info.mergeStarted &&
		info.assFile != "" && info.firstRecordFile != ""
	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 ""
}

// 录播因为断流可能分成多段：单段直接使用，多段先拼成完整视频并把弹幕时间映射到拼接后的时间轴。
// 返回合并要用的视频、字幕文件和弹幕相对视频起点的时间偏移，以及清理临时文件的函数。
func prepareForMerge(ffmpegFile string, segs []recordSegment, jsonlFile string, danmuStartNs int64) (string, string, float64, 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, noop, errors.New("找不到录播视频文件")
	}
	if len(files) == 1 {
		offset := float64(danmuStartNs-starts[0]) / 1e9
		return files[0], jsonlFile, offset, 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, 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 {
		// 任务被手动停止时不要再退化成「只合并第一个分段」，
		// 也别把拼到一半的中间文件留下
		if jobStopRequested() {
			_ = os.Remove(joined)
			return "", "", 0, noop, errJobStopped
		}
		lPrintWarnf("拼接录播分段失败，本场只合并第一个分段（后半场没有弹幕）：%v", err)
		offset := float64(danmuStartNs-starts[0]) / 1e9
		return files[0], jsonlFile, offset, 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, noop, nil
	}

	cleanup := func() {
		_ = os.Remove(joined)
		_ = os.Remove(newJsonl)
	}
	return joined, newJsonl, 0, 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))
}

// 把错误翻成给界面看的原因：正常结束和「没有弹幕」都不算失败
func jobErrText(err error) string {
	switch err {
	case nil, errNoDanmu:
		return ""
	case errJobStopped:
		return errJobStopped.Error()
	}
	return err.Error()
}

// 执行自动合并及后续的删除/移动。日志都在内部按原措辞打印，返回值只给上层写进度用
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 := filepath.Join(mergeDir, filepath.Base(strings.TrimSuffix(jsonlFile, ".jsonl")+"_弹幕合并.mp4"))
	tmpOut := tempMergeOutput(jsonlFile, filepath.Base(outFile))

	// 输入大小按各分段实际位置合计（录制结束时文件可能已被移动到保存目录）
	inputBytes := int64(0)
	for _, s := range segs {
		if p := findMovedFile(s.file); p != "" {
			inputBytes += fileSizeOrZero(p)
		}
	}
	if !jobStart("自动合并", filepath.Base(outFile), inputBytes, 0) {
		// 走到这里说明没先拿 mergeSlot，与其把别人的进度冲掉不如直接放弃这一场
		return errors.New("开始任务失败：还有别的合并任务没结束，本场合并已跳过")
	}

	video, jsonl, offset, 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)
	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 _, s := range segs {
			p := findMovedFile(s.file)
			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
}