Files
acfunlive-remake/danmuburn.go
T
tamakyi 3eb4aa3102 让 ffmpeg 的报错能进日志,并堵住三处会直接毁掉整块渲染的坑
背景:线上第 153/648 块失败,三个编码器依次报同一个 "exit status 0xffffffea"
(EINVAL)。可用的线索只有这一句 —— 因为 ffmpeg 自己的输出走的是 os.Stderr,
服务模式下没人看,等于丢掉了。所以这轮先解决"看不到原因",再顺手补几处防御。

一、把 ffmpeg 的报错留在日志里
   runFFmpegWith 给 stderr 挂一个只保留最后 40 行的 writer(进度行用 \r 刷新,
   一并按行处理),失败时把这段附在错误上再由上层 lPrintErrf 写进日志。
   顺带过滤掉每次启动都打印的版本/编译配置那几行(configuration 一行就一千多字符,
   留着会把真正的报错挤出去)。实测效果:报错里直接能看到
   "moov atom not found" / "Error opening input: Invalid data found"。

二、跟编码器无关的失败不再挨个回退
   滤镜图建不起来、输入文件打不开这类问题,换个编码器结果一模一样。
   原来会把同一件事重跑三遍(hevc_qsv → h264_qsv → libx264),既浪费时间
   又在日志里刷三行看不出区别的错误。现在命中这些关键词就直接报出完整错误。

三、头像 split 的份数必须和真正用到的份数一致
   统计头像用量时按整块弹幕算,生成 overlay 时才做可见性裁剪:
   同一 uid 在本块有多条、其中一条整条落在块外的话,split 会多出一个没人接收的输出 pad,
   ffmpeg 直接 EINVAL 退出(报错完全看不出跟弹幕有关)。抽出 danmuVisibleInChunk
   让统计、生成 overlay、写块字幕三处用同一个判断。

四、缓存命中要校验文件完整,下载改成原子写
   原来只看"文件在不在",被写一半的 PNG 会被当成可用缓存塞给 ffmpeg 当输入。
   现在要求 PNG 以 IEND 收尾(16 字节以下直接判无效),下载先写 .part 再改名,
   旧位置搬过来的文件同样要过这道校验。

五、头像直径至少为 1
   有人把 danmuFontSize 填成 1、2 时 avatarSize 会算出 0,而 scale=0:0
   会让 ffmpeg 直接 EINVAL、整块渲染失败。

验证:
- gofmt / go vet / go build 通过;go test 全绿(16.8s);go test -race 全绿(22.8s);
  staticcheck -checks=all 零告警
- 新增 diagnostics_test.go(10 个)与 split_pad_test.go(1 个)用例;
  反向验证扩到 21 项全过
- 隔离实例:e2e.py 37 项、smoke_extra.py 33 项、smoke_late_danmu.py(越界弹幕场景)全过;
  另用一个损坏的 mp4 实测,任务失败时日志里能看到 ffmpeg 的原话,
  且没有出现编码器回退
2026-09-26 12:15:14 +08:00

1779 lines
59 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
// 弹幕合成相关:读取 jsonl,下载头像并圆形裁切,调用 ffmpeg 合成带头像的弹幕层
package main
import (
"bufio"
"bytes"
"crypto/sha1"
"encoding/json"
"errors"
"fmt"
"image"
"image/color"
"image/draw"
_ "image/gif"
_ "image/jpeg"
"image/png"
"io"
"math"
"net/http"
"os"
"os/exec"
"path/filepath"
"regexp"
"sort"
"strconv"
"strings"
"sync"
"time"
xdraw "golang.org/x/image/draw"
_ "golang.org/x/image/webp"
)
// 头像缓存目录名
const avatarCacheDir = "avatar_cache"
// 把这个阶段的任务分几路并发。查链接和下图都要走网络,一个 uid 的链接反查实测约 1 秒,
// 串行做几百人的场次要几分钟;AcFun 的接口和图片 CDN 都吃得住几路并发,先给 6 路。
// 暂停/停止仍然在每个任务开头过闸门,所以并发不影响暂停和停止的响应
const avatarFetchWorkers = 6
// 头像这一步的统计,只用来打一句看得懂的汇总
type avatarStat struct {
done int // 已处理
fetched int // 真向 AcFun 查过链接的
reusedLink int // 链接直接来自上次的记录(没发请求)
imgReused int // 图片用的已有缓存
imgDownloaded int // 图片这次才下下来
failed int // 链接或图片拿不到,这个用户只烧文字
}
// 头像缓存放程序所在目录:换保存目录(甚至换盘)都还能复用,
// 不用为同一批观众重新下一次。旧版本放在弹幕文件同目录,读取时会搬过来。
func avatarCachePath() string {
if exeDir == "" {
return avatarCacheDir
}
return filepath.Join(exeDir, avatarCacheDir)
}
// 命令行 -vbitrate 指定的合成目标码率(Kbps),配合 config.MergeBitrate 使用
var burnBitrate int
// 缓存头像的边长
const avatarCacheSize = 256
// jsonl 里没有弹幕记录
var errNoDanmu = errors.New("jsonl 里没有弹幕")
// 弹幕在屏幕上的停留时长(秒),与 danmu.go 里的 danmuShowDur 一致
const danmuShowSeconds = 10.0
// 每块最多包含的弹幕条数
// 每条弹幕都要生成一段 overlay 并作为 -filter_complex 的单个参数传给 ffmpeg,
// 而 Windows 的命令行总长上限是 32767 字符(Linux 是单个参数 128KB),
// 弹幕一多整条命令就会超限且报错完全看不出原因,所以超过这个数量就分块渲染再拼接。
const maxDanmuPerChunk = 80
// 单块条数的硬上限。块尾会把「显示时间跨过块尾」的弹幕一并并入,
// 弹幕密集时(10 秒停留窗口里塞进几百条)单块就会膨胀到几百条 ——
// 一条弹幕在 -filter_complex 里约占 110 字节,几百条足以顶破上面那条命令行上限,
// 表现同样是「合并失败且看不出原因」,所以并入也要有个头。
// 到上限就把块尾收到下一条弹幕出现之前,让剩下的弹幕落到下一块。
//
// 上限值按最坏情况实测定的:一块里每条弹幕都来自不同用户时,每条要占
// 「-i 头像路径 + scale 行 + overlay 行」约 241 字符(2026-09-26 在真机量的),
// 120 条 ≈ 28864 字符,给中文路径和其它参数留出 3KB 余量;原来写 200 条时
// 实测 48288 字符,已经远超 Windows 的 32767,线上就是这里崩的。
const maxDanmuPerChunkHard = 120
// 分块渲染产生的临时文件前缀
const chunkTempPrefix = "acfunlive_chunk_"
// Windows 的 CreateProcess 限制整个命令行 32767 字符(Linux 是单个参数 128KB)。
// 留一点余量,超了提早报错。filter 内容已经走 -filter_complex_script 文件,
// 这里管的是剩下的参数 —— 每个头像一个 -i 路径是大头
const maxCommandLineLen = 30000
// 估算交给 exec 的命令行总长:每个参数按「一个空格 + 内容」算
func commandLength(exe string, args []string) int {
n := len(exe)
for _, a := range args {
n += len(a) + 1
}
return n
}
// 预估成片体积时给音频留的码率(Kbps):分块时音频固定重编码成 192k,
// 单块时是直接拷流,这里统一按 192 估,反正渲染过 5% 后会被实测外推值顶掉
const estimateAudioKbps = 192
// 一个待渲染的弹幕块
type danmuChunk struct {
start float64 // 在视频时间轴上的起点(秒)
end float64 // 终点(秒),math.Inf(1) 表示一直到视频结尾
records []danmuRecord // 属于这一块的弹幕
}
// 渲染分块时共用的参数
type burnContext struct {
ffmpegFile string
videoFile string // 为空表示输出透明头像层
outFile string
offset float64
meta danmuMeta
avatars map[int64]string
bitrate int
dir string
encoder string // 已经验证可用的编码器,分块时避免每块都重新回退
}
// 头像直径,与文字的实际墨迹高度一致(字号是 em 框,墨迹约为 0.8 倍)
// 头像直径:跟随弹幕字号。字号特别小时(有人把 danmuFontSize 填成 1、2),
// 算出来是 0,而 scale=0:0 会让 ffmpeg 直接以 EINVAL 退出、整块渲染失败
func avatarSize(fontSize int) int {
n := fontSize * 4 / 5
if n < 1 {
n = 1
}
return n
}
// 头像和弹幕文字之间的间隔
func avatarGap(fontSize int) int {
return fontSize / 6
}
// 画布短边:横屏取高、竖屏取宽。弹幕文字和头像的尺寸都以它为准 ——
// 竖屏画面窄,按高度算出来的字会大到一条弹幕就铺满整行。
func shortSide(w, h int) int {
if w <= 0 || h <= 0 {
return 0
}
if w < h {
return w
}
return h
}
// 自动字号的比例:短边 × 60/1080 ≈ 5.56%,
// 和录制端的档位预设一致(1920x1080 → 60、1280x720 → 40、960x540 → 30)
const autoFontRatioNum, autoFontRatioDen = 60, 1080
// 按画布短边算自动字号,拿不到尺寸时返回 0
func autoFontSize(w, h int) int {
short := shortSide(w, h)
if short <= 0 {
return 0
}
fs := int(math.Round(float64(short) * autoFontRatioNum / autoFontRatioDen))
if fs < 1 {
fs = 1
}
return fs
}
// 定成片字号。设置里填了 >0 就按填的(成片像素);填 0(默认)按画布短边自动算,
// 这样横屏和竖屏的字看起来一样大,竖屏成片不会字大到铺满整行。头像随字号等比缩放。
func applyFontSize(meta *danmuMeta, override int) {
if override > 0 {
if override != meta.FontSize {
lPrintf("弹幕字号按设置使用 %d(弹幕文件里是 %d)", override, meta.FontSize)
}
meta.FontSize = override
return
}
auto := autoFontSize(meta.PlayResX, meta.PlayResY)
if auto <= 0 || auto == meta.FontSize {
return
}
orient := "横屏"
if meta.PlayResY > meta.PlayResX {
orient = "竖屏"
}
lPrintf("%s画布 %dx%d,弹幕字号自动取 %d(短边 × %d/%d,弹幕文件里是 %d)",
orient, meta.PlayResX, meta.PlayResY, auto, autoFontRatioNum, autoFontRatioDen, meta.FontSize)
meta.FontSize = auto
}
// 弹幕合成入口(命令行)
// 弹幕文件可以给 .jsonl,也可以给 .ass(先还原成 jsonl 再合成)
func runDanmuburn(danmuFile, videoFile, outFile string, offset float64) {
src, err := prepareDanmuSource(danmuFile, true)
if err != nil {
if err == errNoDanmu {
lPrintWarn(err.Error())
os.Exit(0)
}
lPrintErr(err)
os.Exit(1)
}
// 还原结果写不进原目录时会落在系统临时目录里,收工前删掉。
// 这里到处都是 os.Exit,defer 不会被执行,所以显式调用
if err := burnDanmu(src.jsonlPath, videoFile, outFile, offset); err != nil {
dropTempDanmu(src)
if err == errNoDanmu {
lPrintWarn(err.Error())
os.Exit(0)
}
lPrintErr(err)
os.Exit(1)
}
dropTempDanmu(src)
}
// 合成一个 jsonl(和可选的视频)为带头像的弹幕层
// videoFile 为空时输出透明头像层(ProRes 4444),否则把弹幕层压进视频
func burnDanmu(jsonlFile, videoFile, outFile string, offset float64) error {
jsonlFile, err := filepath.Abs(jsonlFile)
if err != nil {
return err
}
if _, err := os.Stat(jsonlFile); err != nil {
return fmt.Errorf("jsonl 文件不存在:%s", jsonlFile)
}
// 读取 jsonl,首行是元数据
f, err := os.Open(jsonlFile)
if err != nil {
return err
}
defer f.Close()
scanner := bufio.NewScanner(f)
scanner.Buffer(make([]byte, 1024*1024), 1024*1024)
var meta danmuMeta
var records []danmuRecord
first := true
for scanner.Scan() {
line := scanner.Bytes()
if first {
if err := json.Unmarshal(line, &meta); err != nil {
return err
}
first = false
continue
}
var r danmuRecord
if json.Unmarshal(line, &r) == nil && r.Content != "" {
records = append(records, r)
}
}
if err := scanner.Err(); err != nil {
return err
}
if meta.PlayResX == 0 || meta.FontSize == 0 {
return errors.New("jsonl 文件格式不对,首行必须包含 playResX、playResY、fontSize")
}
if len(records) == 0 {
return errNoDanmu
}
// 确定输出文件和模式:有视频就把弹幕压进视频,没有视频就输出透明头像层
burnMode := videoFile != ""
if burnMode {
videoFile, err = filepath.Abs(videoFile)
if err != nil {
return err
}
if _, err := os.Stat(videoFile); err != nil {
return fmt.Errorf("录播视频不存在:%s", videoFile)
}
if outFile == "" {
// 成片默认放到单独的文件夹里,不再和录播、弹幕混在一起
outFile = filepath.Join(mergeSaveDir(), filepath.Base(strings.TrimSuffix(jsonlFile, ".jsonl")+"_弹幕合并.mp4"))
}
} else {
if outFile == "" {
outFile = strings.TrimSuffix(jsonlFile, ".jsonl") + "_avatar.mov"
}
}
outFile, err = filepath.Abs(outFile)
if err != nil {
return err
}
ffmpegFile := getFFmpeg()
if ffmpegFile == "" {
return errors.New("找不到 ffmpeg,请先安装并加入 PATH,或把 ffmpeg 放在本程序所在文件夹,或在设置里填写 ffmpeg 路径")
}
// 弹幕可能录得比视频久(视频中途中断过),超出视频时长的部分没法渲染,直接丢掉
videoDuration := 0.0
if burnMode {
videoDuration = probeDuration(ffmpegFile, videoFile)
if videoDuration > 0 {
kept := records[:0]
for _, r := range records {
if r.Time+offset < videoDuration {
kept = append(kept, r)
}
}
if dropped := len(records) - len(kept); dropped > 0 {
lPrintWarnf("有 %d 条弹幕超出了录播视频的时长(%.1f 秒),已跳过", dropped, videoDuration)
}
records = kept
if len(records) == 0 {
return errNoDanmu
}
}
}
// 弹幕文件的坐标系和视频对不上时,统一到视频坐标系(见 unifyDanmuSpace);
// 顺便拿到源片码率,用于「合成码率填 0」时的自动码率
sourceKbps := 0
if burnMode {
var vw, vh int
vw, vh, sourceKbps = probeVideoStream(ffmpegFile, videoFile)
unifyDanmuSpace(&meta, vw, vh)
lPrintf("视频编码器优先顺序:%s", strings.Join(availableEncoders(ffmpegFile), " → "))
}
// 字号在坐标系统一之后再定,此时 meta.PlayRes 就是成片画布
applyFontSize(&meta, config.DanmuFontSize)
// 弹道按最终字号统一重排,避免还原弹幕或改字号后同弹道追尾
relayoutLanes(records, meta)
// 缓存放在程序所在目录,换保存目录(甚至换盘)都不用重新下载;
// 旧版本放在弹幕文件同目录,作为回退来源读一次并搬过去
cacheDir := avatarCachePath()
legacyCache := filepath.Join(filepath.Dir(jsonlFile), avatarCacheDir)
// 先去重,顺便记下每个用户自带的头像链接
urlByUID := make(map[int64]string)
uids := make([]int64, 0, 64)
for _, r := range records {
if r.UID == 0 {
continue
}
if old, ok := urlByUID[r.UID]; !ok {
uids = append(uids, r.UID)
urlByUID[r.UID] = r.Avatar
} else if old == "" && r.Avatar != "" {
// 同一个用户的前几条可能没有头像链接,后面带上就补上
urlByUID[r.UID] = r.Avatar
}
}
avatarPaths, _, err := fetchAvatars(uids, urlByUID, cacheDir, legacyCache)
if err != nil {
return err
}
if len(avatarPaths) == 0 {
lPrintWarn("没有任何头像可用,弹幕将只烧录文字,请检查网络后重试合并")
}
// 合成码率:填了值就用它;填 0(默认)按「源片码率 + 弹幕余量」算,探测不到才退回质量模式
bitrate := config.MergeBitrate + burnBitrate
if bitrate == 0 && burnMode {
if auto := autoMergeBitrate(sourceKbps); auto > 0 {
lPrintf("合成码率填的 0,按源片 %d kbps + 弹幕余量 %d kbps 取 %d kbps", sourceKbps, danmuBitrateMargin, auto)
bitrate = auto
} else {
lPrintWarn("探测不到源片码率,改用默认质量模式(ICQ 22 / CRF 18)")
}
}
ctx := &burnContext{
ffmpegFile: ffmpegFile,
videoFile: videoFile,
outFile: outFile,
offset: offset,
meta: meta,
avatars: avatarPaths,
bitrate: bitrate,
dir: filepath.Dir(jsonlFile),
}
chunks := splitDanmuChunks(records, offset, videoDuration)
// 任务总时长:有视频就用视频时长,透明头像层模式没有视频,用各块时长之和
jobTotalSec := videoDuration
if jobTotalSec <= 0 {
for _, ch := range chunks {
jobTotalSec += ctx.chunkDuration(ch)
}
}
jobSetTotal(jobTotalSec)
// 预计成片体积 =(视频码率 + 音频码率)× 时长 ÷ 8;渲染过 5% 之后会被实测外推值顶掉
if bitrate > 0 {
jobPlanSize(int64(float64(bitrate+estimateAudioKbps) * jobTotalSec / 8 * 1000))
}
jobChunks(0, len(chunks))
jobSetPhase(jobPhaseRender)
if len(chunks) <= 1 {
lPrintf("共 %d 条弹幕,%d 个用户的头像可用,开始合成", len(records), len(avatarPaths))
return ctx.burnSingle(records)
}
lPrintf("共 %d 条弹幕,%d 个用户的头像可用;超过单次渲染上限(%d 条),分 %d 块渲染后拼接",
len(records), len(avatarPaths), maxDanmuPerChunk, len(chunks))
return ctx.burnChunked(chunks)
}
// 获取所有用户的头像:没有自带链接的先按 uid 查一次,再决定用已有缓存还是下载。
// 之所以并发:查链接实测约 1 秒/个,几百人的场次串行做要好几分钟。
// 返回 uid → 缓存文件路径、本阶段统计;暂停时在闸门处等待,停止时返回 errJobStopped。
func fetchAvatars(uids []int64, urlByUID map[int64]string, cacheDir, legacyCache string) (map[int64]string, avatarStat, error) {
type avatarTask struct {
uid int64
url string
}
tasks := make([]avatarTask, 0, len(uids))
for _, uid := range uids {
tasks = append(tasks, avatarTask{uid: uid, url: urlByUID[uid]})
}
paths := make(map[int64]string, len(tasks))
var stat avatarStat
jobSetPhase(jobPhaseAvatar)
jobAvatars(0, len(tasks))
if len(tasks) == 0 {
return paths, stat, nil
}
lPrintf("开始获取 %d 个用户的头像(%d 路并发)", len(tasks), avatarFetchWorkers)
workers := avatarFetchWorkers
if workers > len(tasks) {
workers = len(tasks)
}
// 缓冲足够大:worker 因停止提前退出时,投递端也不会卡住
jobs := make(chan avatarTask, len(tasks))
for _, t := range tasks {
jobs <- t
}
close(jobs)
var (
mu sync.Mutex
stopErr error
)
var wg sync.WaitGroup
for w := 0; w < workers; w++ {
wg.Add(1)
go func() {
defer wg.Done()
for t := range jobs {
// 每个任务开头过闸门:暂停时在这里等,停止时直接收工
if err := jobGate(); err != nil {
mu.Lock()
if stopErr == nil {
stopErr = err
}
mu.Unlock()
return
}
url, fetched := t.url, false
if url == "" {
// 从 .ass 还原出来的弹幕没有头像链接,按 uid 查一次(结果会落盘)
url, fetched = avatarURLByUID(t.uid, cacheDir)
}
// 下载前先看这个头像在不在,用来区分「复用缓存」和「真的下载了」
existed := url != "" && fileExists(filepath.Join(cacheDir, avatarCacheName(t.uid, url)))
path := ""
if url != "" {
path = ensureAvatar(t.uid, url, cacheDir, legacyCache)
}
mu.Lock()
stat.done++
if fetched {
stat.fetched++
} else if t.url == "" {
stat.reusedLink++
}
switch {
case existed:
stat.imgReused++
case path != "":
stat.imgDownloaded++
case url != "":
stat.failed++
}
if path != "" {
paths[t.uid] = path
}
done := stat.done
mu.Unlock()
jobAvatars(done, len(tasks))
}
}()
}
wg.Wait()
if stopErr != nil {
return paths, stat, stopErr
}
summary := fmt.Sprintf("头像处理完成:%d 个用户,链接复用 %d / 向 AcFun 查 %d,图片复用缓存 %d / 新下载 %d",
len(tasks), stat.reusedLink, stat.fetched, stat.imgReused, stat.imgDownloaded)
if stat.failed > 0 {
summary += fmt.Sprintf(",%d 个不可用(这些用户只烧文字)", stat.failed)
}
lPrintf("%s", summary)
return paths, stat, nil
}
// 渲染前按最终字号重排弹道:jsonl 里的 lane/width 是录制或 .ass 还原时算的,
// 从原格式 .ass 还原的弹幕没把昵称和头像算进宽度(占位提前放行了同弹道的下一条,
// 画面上就追尾叠字),改过字号后更是全对不上。这里用和录制时相同的碰撞公式、
// 按「昵称(uid):内容 + 头像」的完整宽度重新分配一遍,保证弹道不重叠。
func relayoutLanes(records []danmuRecord, meta danmuMeta) {
sort.SliceStable(records, func(i, j int) bool { return records[i].Time < records[j].Time })
fs := meta.FontSize
show := int64(danmuShowSeconds * 1e9)
resX := int64(meta.PlayResX)
avW, avGap := avatarSize(fs), avatarGap(fs)
// 空弹道用「远古」状态占位:placeLane 的条件是本条出现时间晚于上一条的让位时刻,
// 零值下 0 > 0 不成立,第一条弹幕(还原出来的弹幕时间可能正好是 0)会占不上弹道,
// 占用状态丢掉后后面的弹幕又会跟着落回空弹道,等于白排。录制时弹幕时间总大于 0,
// 所以原逻辑没暴露这个问题。
lanes := make([]laneState, laneNum)
for i := range lanes {
lanes[i] = laneState{appear: math.MinInt64, emerge: math.MinInt64, disappear: math.MinInt64}
}
for i := range records {
r := &records[i]
r.Width = textWidth(speakerText(r.Name, r.UID, r.Content), fs)
total := int64(avW+avGap) + int64(r.Width)
rel := int64(r.Time * 1e9)
dt := laneState{
appear: rel,
emerge: rel + total*show/(resX+total),
disappear: rel + show,
}
left := rel + resX*show/(resX+total)
j := placeLane(lanes, dt, left)
if j < 0 {
// 100 条弹道都还占着:退到最早让出来的那条。录制时是直接不写这条,
// 合并时丢一条等于凭空少一句弹幕,压一点也比丢了强。
j = earliestLane(lanes)
lanes[j] = dt
}
r.Lane = j
}
}
// 最早会空出来的弹道行
func earliestLane(lanes []laneState) int {
best := 0
for i := range lanes {
if lanes[i].disappear < lanes[best].disappear {
best = i
}
}
return best
}
// 把渲染坐标系统一到视频分辨率。
// 弹幕文件里的 PlayResX/PlayResY 是录制当时的画面尺寸,老文件常和现在的视频对不上
// (例如 960x540 的弹幕文件配 720x1280 的竖屏视频)。libass 会按「视频高 / PlayResY」
// 缩放文字和它的坐标,而头像叠加用的是视频像素坐标,两者不在同一个坐标系里 ——
// 字号越大越明显:头像显得小,而且每往下一行错得越多(看起来和文字分开了)。
// 这里只负责改 PlayRes,字号交给 applyFontSize 按最终画布重算。返回是否做过调整。
func unifyDanmuSpace(meta *danmuMeta, vw, vh int) bool {
if vw <= 0 || vh <= 0 || (vw == meta.PlayResX && vh == meta.PlayResY) {
return false
}
oldX, oldY := meta.PlayResX, meta.PlayResY
meta.PlayResX, meta.PlayResY = vw, vh
lPrintf("弹幕文件的坐标系是 %dx%d,视频是 %dx%d,改按视频坐标系渲染", oldX, oldY, vw, vh)
return true
}
// 探测视频流的分辨率和码率(Kbps),拿不到的项返回 0
func probeVideoStream(ffmpegFile, path string) (int, int, int) {
cmd := exec.Command(ffmpegFile, "-hide_banner", "-i", path)
var buf bytes.Buffer
cmd.Stderr = &buf
// 没指定输出文件时 ffmpeg 会返回非 0,但流信息已经打印出来了
_ = cmd.Run()
width, height, kbps := 0, 0, 0
for _, line := range strings.Split(buf.String(), "\n") {
// 容器的总码率(含音频)先记下来,视频流自己没写码率时用它
if strings.Contains(line, "Duration:") {
if m := containerBitrateRe.FindStringSubmatch(line); len(m) == 2 {
kbps, _ = strconv.Atoi(m[1])
}
continue
}
// 只看视频流,跳过封面图(mjpeg 之类的 attached pic)那几行
if !strings.Contains(line, "Video:") || strings.Contains(line, "attached pic") {
continue
}
if m := videoSizeRe.FindStringSubmatch(line); len(m) == 3 {
width, _ = strconv.Atoi(m[1])
height, _ = strconv.Atoi(m[2])
}
if m := videoBitrateRe.FindStringSubmatch(line); len(m) == 2 {
kbps, _ = strconv.Atoi(m[1])
}
}
return width, height, kbps
}
var (
// 逗号后紧跟的「数字x数字」就是分辨率,前面的编码器名里也有 0x… 这种十六进制,靠逗号区分
videoSizeRe = regexp.MustCompile(`, (\d{2,5})x(\d{2,5})[, \[]`)
videoBitrateRe = regexp.MustCompile(`, (\d+) kb/s`)
// 容器总码率,形如「Duration: 00:02:55.07, start: 0.000000, bitrate: 1767 kb/s」
containerBitrateRe = regexp.MustCompile(`bitrate: (\d+) kb/s`)
)
// 合成码率填 0(默认)时的算法:源片码率 + 弹幕余量。
// 烧进去的文字和头像都是高频细节,跟源片同样码率会糊,所以留一点余量;
// 探测不到源片码率时返回 0,让调用方退回质量模式。
func autoMergeBitrate(sourceKbps int) int {
if sourceKbps <= 0 {
return 0
}
br := sourceKbps + danmuBitrateMargin
if br < minMergeBitrate {
br = minMergeBitrate
}
return br
}
const (
// 弹幕层占用的额外码率(Kbps)
danmuBitrateMargin = 500
// 自动码率的下限,源片码率特别低时也别低于这个值
minMergeBitrate = 800
)
// 用 ffmpeg 探测媒体文件的时长(秒),探测不到返回 0
func probeDuration(ffmpegFile, path string) float64 {
cmd := exec.Command(ffmpegFile, "-hide_banner", "-i", path)
var buf bytes.Buffer
cmd.Stderr = &buf
// 没指定输出文件时 ffmpeg 会返回非 0,但时长信息已经打印出来了
_ = cmd.Run()
m := durationRe.FindStringSubmatch(buf.String())
if len(m) != 4 {
return 0
}
h, _ := strconv.Atoi(m[1])
mi, _ := strconv.Atoi(m[2])
s, _ := strconv.ParseFloat(m[3], 64)
return float64(h)*3600 + float64(mi)*60 + s
}
var durationRe = regexp.MustCompile(`Duration: (\d+):(\d+):(\d+\.\d+)`)
// 弹幕不多时直接一次渲染完,同样从 jsonl 生成字幕,不依赖录制时留下的 .nick.ass
func (c *burnContext) burnSingle(records []danmuRecord) error {
chunk := danmuChunk{start: 0, end: math.Inf(1), records: records}
// 临时字幕一律用纯 ASCII 文件名:libass 在部分环境下打不开非 ASCII 的文件名
assName := chunkTempPrefix + "single.ass"
assPath := filepath.Join(c.dir, assName)
if err := writeChunkAss(assPath, c.meta, chunk, c.offset,
avatarSize(c.meta.FontSize), avatarGap(c.meta.FontSize)); err != nil {
return fmt.Errorf("生成字幕文件失败:%v", err)
}
defer os.Remove(assPath)
// 只有一块,ffmpeg 报的时长就是整片进度
return c.renderChunk(chunk, assName, c.outFile, false, func(outSec, speed, fps float64, size int64) {
jobRender(outSec, speed, fps, size)
})
}
// 分块渲染:每块单独渲染成一个小片,最后用 concat 拼成完整成片
func (c *burnContext) burnChunked(chunks []danmuChunk) error {
ext := ".mov"
if c.videoFile != "" {
ext = ".mp4"
}
var parts []string
defer func() {
for _, p := range parts {
_ = os.Remove(p)
}
}()
ad, gap := avatarSize(c.meta.FontSize), avatarGap(c.meta.FontSize)
// 已完成块写出的字节数:加在当前块上,界面上的「已写出」和体积外推才是累计值
var bytesBase int64
var joinedSec float64
for _, ch := range chunks {
joinedSec += c.chunkDuration(ch)
}
for i, ch := range chunks {
// 每块开始时过一下闸门:暂停在这里等,停止在这里直接退出
if err := jobGate(); err != nil {
return err
}
jobChunks(i, len(chunks))
// 每块的字幕单独生成,只包含这一块时间范围内的弹幕
assName := fmt.Sprintf("%schunk_%04d.ass", chunkTempPrefix, i)
assPath := filepath.Join(c.dir, assName)
if err := writeChunkAss(assPath, c.meta, ch, c.offset, ad, gap); err != nil {
return fmt.Errorf("生成第 %d 块的字幕失败:%v", i+1, err)
}
defer os.Remove(assPath)
part := filepath.Join(c.dir, fmt.Sprintf("%spart_%04d%s", chunkTempPrefix, i, ext))
// 先登记再渲染:中途失败或被打断时,defer 里的清理也能把这个半成品删掉
parts = append(parts, part)
lPrintf("正在渲染第 %d/%d 块(%s ~ %s,%d 条弹幕)",
i+1, len(chunks), formatSeconds(ch.start), formatSeconds(ch.end), len(ch.records))
base := bytesBase
if err := c.renderChunk(ch, assName, part, true, func(outSec, speed, fps float64, size int64) {
// 块内时间加上块起点,折算成整片时间轴上的进度
jobRender(ch.start+outSec, speed, fps, base+size)
}); err != nil {
if jobStopRequested() {
return errJobStopped
}
return fmt.Errorf("渲染第 %d/%d 块失败:%v", i+1, len(chunks), err)
}
if n := fileSizeOrZero(part); n > 0 {
bytesBase += n
}
}
jobSetPhase(jobPhasePlace)
lPrintf("正在把 %d 个分块拼接成完整成片", len(parts))
if err := concatChunks(c.ffmpegFile, parts, c.outFile, c.dir, func(outSec, speed, fps float64, size int64) {
if joinedSec > 0 {
jobSetPhasePercent(outSec / joinedSec)
}
}); err != nil {
return err
}
return nil
}
// 把弹幕按时间切成若干块,块的时间范围首尾相接,保证视频不会缺帧
// maxEnd 大于 0 时表示视频总时长,块尾不会超过它
func splitDanmuChunks(records []danmuRecord, offset, maxEnd float64) []danmuChunk {
n := len(records)
if n == 0 {
return nil
}
if n <= maxDanmuPerChunk {
return []danmuChunk{{start: 0, end: math.Inf(1), records: records}}
}
chunks := make([]danmuChunk, 0, n/maxDanmuPerChunk+1)
i := 0
start := 0.0
for i < n {
limit := i + maxDanmuPerChunk
if limit >= n {
chunks = append(chunks, danmuChunk{start: start, end: math.Inf(1), records: records[i:]})
break
}
// 块尾取第 limit 条弹幕显示结束的时刻
end := records[limit-1].Time + offset + danmuShowSeconds
if maxEnd > 0 && end > maxEnd {
end = maxEnd
}
if end <= start {
end = start + danmuShowSeconds
}
// 起点早于块尾的弹幕(显示会跨过块尾)也并入这一块,但不许超过硬上限
j := limit
for j < n && j < i+maxDanmuPerChunkHard && records[j].Time+offset < end {
j++
}
// 被硬上限挡住时,把块尾收紧到下一条弹幕出现之前:它因此落在下一块里,
// 而且不会因为「整条都早于下一块块头」在下一块里被当成无效区间丢掉
if j < n && records[j].Time+offset < end {
if t := records[j].Time + offset; t > start {
end = t
}
}
chunks = append(chunks, danmuChunk{start: start, end: end, records: records[i:j]})
i = j
start = end
}
return chunks
}
// 为分块渲染生成只包含该块弹幕的字幕文件(相对块起点计时)
func writeChunkAss(path string, meta danmuMeta, chunk danmuChunk, offset float64, avW, avGap int) error {
f, err := os.Create(path)
if err != nil {
return err
}
defer f.Close()
// ASS 必须带 [Script Info] 开头,否则 libass 探测格式会失败
fmt.Fprintf(f, scriptInfoTmpl, "", "", filepath.Base(path), meta.PlayResX, meta.PlayResY)
fmt.Fprintf(f, stylesTmpl, meta.FontSize)
fmt.Fprint(f, eventsTmpl)
for _, r := range chunk.records {
visStart, visEnd, visible := danmuVisibleInChunk(r, offset, chunk)
if !visible {
continue
}
text := r.Name + "(" + strconv.FormatInt(r.UID, 10) + "):" + r.Content
y := meta.FontSize * r.Lane
fmt.Fprintf(f, dialogueNickTmpl,
assTime(int64((visStart-chunk.start)*1e9)), assTime(int64((visEnd-chunk.start)*1e9)),
strings.ReplaceAll(r.Name, ",", " "), r.UID,
meta.PlayResX+avW+avGap, y, -r.Width, y, text)
}
return nil
}
// 一条弹幕在这一块里到底会不会被画出来:超出块的部分会被裁掉,裁完没有可见区间就整条跳过。
// 统计头像用量和生成 overlay 必须共用这一个判断 —— 只统计、不裁剪的话,同一个人在这块里
// 会被 split 出多余的份数,那些输出 pad 没人接收,ffmpeg 直接以 EINVAL 退出
// (Windows 上就是 exit status 0xffffffea,报错完全看不出跟弹幕有关,
// 还会连累三个编码器挨个"运行失败")
func danmuVisibleInChunk(r danmuRecord, offset float64, chunk danmuChunk) (visStart, visEnd float64, ok bool) {
t0 := r.Time + offset
visStart = math.Max(t0, chunk.start)
visEnd = math.Min(t0+danmuShowSeconds, chunk.end)
return visStart, visEnd, visEnd > visStart
}
// 渲染一个块:把这一块的弹幕(文字层 + 头像层)压进对应的视频片段
// cut 为 true 时表示这是分块渲染,需要按时长切出视频片段并重编码音频
// cb 用来回传 ffmpeg 的实时进度(已输出时长 / 倍速 / 帧率 / 已写出字节),可为 nil
func (c *burnContext) renderChunk(chunk danmuChunk, assBase, out string, cut bool, cb ffmpegProgress) error {
burnMode := c.videoFile != ""
// 统计这一块用到的头像,按首次出现顺序分配输入序号(输入 0 是视频或背景)。
// 只算真正会画出来的弹幕,见 danmuVisibleInChunk 的说明
uidOrder := []int64{}
uidCount := make(map[int64]int)
uidInput := make(map[int64]int)
for _, r := range chunk.records {
if _, ok := c.avatars[r.UID]; !ok {
continue
}
if _, _, visible := danmuVisibleInChunk(r, c.offset, chunk); !visible {
continue
}
if _, seen := uidCount[r.UID]; !seen {
uidOrder = append(uidOrder, r.UID)
uidInput[r.UID] = len(uidOrder)
}
uidCount[r.UID]++
}
args := []string{"-y"}
if burnMode {
if cut {
// -ss 放在 -i 之前是快速定位,配合重编码能精确切到指定时间点
args = append(args, "-ss", formatSeconds(chunk.start))
}
args = append(args, "-i", c.videoFile)
} else {
args = append(args, "-f", "lavfi", "-i",
fmt.Sprintf("color=c=black@0:s=%dx%d:d=%.2f:r=30,format=rgba",
c.meta.PlayResX, c.meta.PlayResY, c.chunkDuration(chunk)+0.5))
}
for _, uid := range uidOrder {
args = append(args, "-i", c.avatars[uid])
}
fc := &strings.Builder{}
prev := "0:v"
if burnMode {
fmt.Fprintf(fc, "[%s]subtitles=filename='%s'[b0];\n", prev, assBase)
prev = "b0"
}
// 把每个头像缩放到弹幕字号对应的直径,同一个头像被多条弹幕使用时用 split 复制
ad := avatarSize(c.meta.FontSize)
gap := avatarGap(c.meta.FontSize)
for _, uid := range uidOrder {
idx := uidInput[uid]
if uidCount[uid] == 1 {
fmt.Fprintf(fc, "[%d:v]scale=%d:%d[av%d];\n", idx, ad, ad, idx)
} else {
outs := &strings.Builder{}
for j := 0; j < uidCount[uid]; j++ {
fmt.Fprintf(outs, "[av%d_%d]", idx, j)
}
fmt.Fprintf(fc, "[%d:v]scale=%d:%d,split=%d%s;\n", idx, ad, ad, uidCount[uid], outs.String())
}
}
uidCopy := make(map[int64]int) // 每个头像当前用掉的份数
for n, r := range chunk.records {
idx, ok := uidInput[r.UID]
if !ok {
continue
}
// 先判可见性再取 split 的份:被裁掉的弹幕不能占掉一份,
// 否则份数对不上,split 会留下没人接收的输出 pad(见 danmuVisibleInChunk)
visStart, visEnd, visible := danmuVisibleInChunk(r, c.offset, chunk)
if !visible {
continue
}
avLabel := strconv.Itoa(idx)
if uidCount[r.UID] > 1 {
avLabel = strconv.Itoa(idx) + "_" + strconv.Itoa(uidCopy[r.UID])
uidCopy[r.UID]++
}
t0 := r.Time + c.offset
// 头像在文字左侧并留出间隔:ASS 里的文字左边缘是这里加 ad+gap,头像跟着文字一起滚动。
// 位置公式 x = 画面宽 - (画面宽+总宽)*(t-t0)/10,这里 t 是块内时间,把 t0 也换算到块内坐标
total := ad + gap + r.Width
shift := chunk.start - t0
x := fmt.Sprintf("%d-(%d+%d)*(t%+.3f)/10", c.meta.PlayResX, c.meta.PlayResX, total, shift)
y := c.meta.FontSize*r.Lane + (c.meta.FontSize-ad)/2 // 头像与文字墨迹垂直居中
outLabel := fmt.Sprintf("s%d", n)
fmt.Fprintf(fc, "[%s][av%s]overlay=x='%s':y=%d:enable='between(t,%.3f,%.3f)'[%s];\n",
prev, avLabel, x, y, visStart-chunk.start, visEnd-chunk.start, outLabel)
prev = outLabel
}
if burnMode {
fmt.Fprintf(fc, "[%s]format=yuv420p[v]", prev)
} else {
fmt.Fprintf(fc, "[%s]format=yuva444p10le[v]", prev)
}
args = append(args, "-filter_complex", fc.String(), "-map", "[v]")
if burnMode {
args = append(args, "-map", "0:a?")
if cut {
// 分块切割点的音频不在帧边界上,必须重编码,否则拼接后会累积音画错位
args = append(args, "-c:a", "aac", "-b:a", "192k")
} else {
args = append(args, "-c:a", "copy")
}
args = append(args, "-movflags", "+faststart")
} else {
args = append(args, "-c:v", "prores_ks", "-profile:v", "4444", "-pix_fmt", "yuva444p10le")
}
if cut && !math.IsInf(chunk.end, 1) {
args = append(args, "-t", formatSeconds(chunk.end-chunk.start))
}
// 自检:每条弹幕都要生成一段 overlay,条数一多整个命令行就会顶破 Windows 的
// 32767 字符上限,而 exec 只会回一句 "The filename or extension is too long",
// 完全看不出跟弹幕数量有关(这个构建的 ffmpeg 也没有 -filter_complex_script 可以绕)。
// 这里按实际参数估长,超了就给一句能定位的错,而不是让它烂在 exec 里
if n := commandLength(c.ffmpegFile, args); n > maxCommandLineLen {
return fmt.Errorf("ffmpeg 命令过长(%d 字符,Windows 上限 %d 左右):这一块有 %d 条弹幕、%d 个头像,请调小单块弹幕数",
n, maxCommandLineLen, len(chunk.records), len(uidOrder))
}
if !burnMode {
// 透明头像层模式固定用 ProRes 4444,不能再追加编码器覆盖
full := append(append([]string{}, args...), out)
lPrintln("开始运行 ffmpeg 合成透明头像层,输出到 " + out)
if err := runFFmpegWith(c.ffmpegFile, full, c.dir, cb); err != nil {
return err
}
return nil
}
// 烧录模式:优先 Intel QSV 硬件编码,运行失败自动回退 libx264
encoders := buildEncoders(c.bitrate, c.ffmpegFile)
// 已经在这条流水线里验证过可用的编码器直接复用,分块时不用每块都回退一次
begin := 0
if c.encoder != "" {
for i, enc := range encoders {
if enc.name == c.encoder {
begin = i
break
}
}
}
var lastErr error
for i := begin; i < len(encoders); i++ {
enc := encoders[i]
full := append(append([]string{}, args...), "-c:v", enc.name)
full = append(full, enc.opts...)
full = append(full, out)
if i == begin {
lPrintf("开始运行 ffmpeg 合成(编码器 %s),输出到 %s", enc.name, out)
}
if err := runFFmpegWith(c.ffmpegFile, full, c.dir, cb); err != nil {
// 停止任务时杀掉 ffmpeg 也会让它非 0 退出,这时候不该再换编码器重试
if jobStopRequested() {
return errJobStopped
}
lastErr = err
if i+1 < len(encoders) && encoderMightHelp(err) {
lPrintWarnf("编码器 %s 运行失败,回退到 %s:%v", enc.name, encoders[i+1].name, err)
continue
}
// 跟编码器无关的失败(滤镜图建不起来、输入打不开等)换个编码器也是一样的结果,
// 直接把 ffmpeg 的报错写进日志,别让它烂在 exec 的一句 exit status 里
lPrintErrf("渲染失败:%v", err)
return fmt.Errorf("ffmpeg 运行失败:%v", err)
}
c.encoder = enc.name
return nil
}
return fmt.Errorf("ffmpeg 运行失败:%v", lastErr)
}
// 这一块需要渲染的时长(用于透明头像层模式生成等长的黑底)
func (c *burnContext) chunkDuration(chunk danmuChunk) float64 {
if !math.IsInf(chunk.end, 1) {
return chunk.end - chunk.start
}
maxT := chunk.start
for _, r := range chunk.records {
if t := r.Time + c.offset + danmuShowSeconds; t > maxT {
maxT = t
}
}
return maxT - chunk.start
}
// 合成用的视频编码器配置
type videoEncoder struct {
name string
opts []string
}
// 编码器候选顺序:硬件 HEVC(同画质体积最省)→ 硬件 H.264 → CPU 软编
func encoderCandidates(hasHEVCQSV, hasH264QSV bool) []string {
names := make([]string, 0, 3)
if hasHEVCQSV {
names = append(names, "hevc_qsv")
}
if hasH264QSV {
names = append(names, "h264_qsv")
}
return append(names, "libx264")
}
// 本机可用的编码器候选,顺序即优先级
func availableEncoders(ffmpegFile string) []string {
return encoderCandidates(qsvEncoderAvailable(ffmpegFile, "hevc_qsv"),
qsvEncoderAvailable(ffmpegFile, "h264_qsv"))
}
// 某个编码器的输出参数。bitrate 为 0 时用质量模式(ICQ / CRF)
func encoderOpts(encoder string, bitrate int) []string {
var opts []string
isQSV := strings.HasSuffix(encoder, "_qsv")
if bitrate > 0 {
opts = []string{"-b:v", strconv.Itoa(bitrate) + "k"}
if !isQSV {
opts = append(opts, "-preset", "medium")
}
} else if isQSV {
opts = []string{"-global_quality", "22", "-preset", "medium"}
} else {
opts = []string{"-crf", "18", "-preset", "medium"}
}
// HEVC 在 mp4 里默认写成 hev1,Apple 系(QuickTime / Safari / iPhone)不认,统一用 hvc1
if isHEVCCodec(encoder) {
opts = append(opts, "-tag:v", "hvc1")
}
return opts
}
// 这个编码器是不是 HEVC(hevc_qsv / libx265 / hevc_nvenc 之类)
func isHEVCCodec(encoder string) bool {
return strings.Contains(encoder, "hevc") || strings.Contains(encoder, "265")
}
// 按目标码率构造编码器候选列表(顺序即回退顺序)
func buildEncoders(bitrate int, ffmpegFile string) []videoEncoder {
names := availableEncoders(ffmpegFile)
encoders := make([]videoEncoder, 0, len(names))
for _, name := range names {
encoders = append(encoders, videoEncoder{name: name, opts: encoderOpts(name, bitrate)})
}
return encoders
}
// ffmpeg 的进度回调:已输出时长(秒)、实时倍速、帧率、已写出字节
type ffmpegProgress func(outSec, speed, fps float64, totalSize int64)
// 跑一次 ffmpeg。cb 不为空时加 -progress pipe:1 取结构化进度;
// ffmpeg 自己的统计行仍在 stderr 里,照旧进日志,单独用命令行时不会少东西。
func runFFmpegWith(ffmpegFile string, args []string, dir string, cb ffmpegProgress) error {
// 暂停时在这里等,任务被停止时直接返回,不再启动新进程
if err := jobGate(); err != nil {
return err
}
full := args
if cb != nil {
full = append([]string{"-progress", "pipe:1"}, args...)
}
cmd := exec.Command(ffmpegFile, full...)
cmd.Dir = dir
// 一边原样透传到控制台,一边留一份尾部:服务模式下 os.Stderr 没人看,
// 失败时日志里只剩一句 "exit status 0xffffffea",谁也查不出 ffmpeg 在抱怨什么
stderr := newTailWriter(ffmpegStderrKeep)
cmd.Stderr = io.MultiWriter(os.Stderr, stderr)
if cb != nil {
// StdoutPipe 要求 Stdout 还没被设置,所以这里不能先指到 os.Stdout
stdout, err := cmd.StdoutPipe()
if err != nil {
return err
}
if err := cmd.Start(); err != nil {
return err
}
jobAttach(cmd)
defer jobDetach(cmd)
// 读到 EOF(进程退出会关闭管道)再 Wait,避免和 Wait 抢着关管道
scanFFmpegProgress(stdout, cb)
return ffmpegResult(cmd.Wait(), stderr.String())
}
cmd.Stdout = os.Stdout
if err := cmd.Start(); err != nil {
return err
}
jobAttach(cmd)
defer jobDetach(cmd)
return ffmpegResult(cmd.Wait(), stderr.String())
}
// ffmpeg 的 stderr 尾部保留多少行:失败时够说明问题,又不至于把日志刷爆
const ffmpegStderrKeep = 40
// 只留最近若干行的 writer,给 ffmpeg 的 stderr 用(它会不停地刷进度行)
type tailWriter struct {
mu sync.Mutex
lines []string
buf []byte
keep int
}
func newTailWriter(keep int) *tailWriter {
return &tailWriter{keep: keep}
}
func (w *tailWriter) Write(p []byte) (int, error) {
w.mu.Lock()
defer w.mu.Unlock()
for _, b := range p {
// ffmpeg 的进度行用 \r 覆盖,这里把 \r 也当行尾处理
if b == '\n' || b == '\r' {
w.flushLine()
continue
}
w.buf = append(w.buf, b)
}
return len(p), nil
}
func (w *tailWriter) flushLine() {
if len(bytes.TrimSpace(w.buf)) > 0 && !ffmpegNoise(string(w.buf)) {
w.lines = append(w.lines, string(w.buf))
if len(w.lines) > w.keep {
w.lines = w.lines[1:]
}
}
w.buf = w.buf[:0]
}
// ffmpeg 每次启动都要打一段版本和编译配置(configuration 那行有一千多字符),
// 留着它就把真正有用的报错挤出日志了
var ffmpegNoisePrefix = []string{
"ffmpeg version ", " built with ", " configuration:", " libav", " libsw",
}
func ffmpegNoise(line string) bool {
for _, p := range ffmpegNoisePrefix {
if strings.HasPrefix(line, p) {
return true
}
}
return false
}
// 尾部内容(含还没有换行的最后一段)
func (w *tailWriter) String() string {
w.mu.Lock()
defer w.mu.Unlock()
lines := append([]string{}, w.lines...)
if len(bytes.TrimSpace(w.buf)) > 0 {
lines = append(lines, string(w.buf))
}
return strings.Join(lines, "\n")
}
// 把 ffmpeg 自己的报错附在错误上:不然只剩一句 exit status,查不出原因
func ffmpegResult(err error, stderr string) error {
if err == nil {
return nil
}
if msg := strings.TrimSpace(stderr); msg != "" {
return fmt.Errorf("%v\n--- ffmpeg 最后的输出 ---\n%s", err, msg)
}
return err
}
// 这些报错说明问题在滤镜图或输入上,跟用哪个编码器无关:
// 挨个回退只会把同一件事重跑三遍,日志里刷三行、浪费时间还看不出区别
var notEncoderFault = []string{
"Error binding filtergraph",
"Error initializing filtergraph",
"Error reinitializing filters",
"Error parsing filterchain",
"Error opening input",
"Invalid data found when processing input",
}
// 换编码器值不值得:只有跟编码器本身有关的失败才值得回退
func encoderMightHelp(err error) bool {
msg := err.Error()
for _, s := range notEncoderFault {
if strings.Contains(msg, s) {
return false
}
}
return true
}
// 解析 -progress pipe:1 的输出:每行一个 key=value,
// 每个块以 progress=continue / progress=end 收尾,就在这时候回调一次。
func scanFFmpegProgress(r io.Reader, cb ffmpegProgress) {
sc := bufio.NewScanner(r)
sc.Buffer(make([]byte, 4096), 1<<20)
var doneSec, speed, fps float64
var size int64
for sc.Scan() {
key, val, ok := strings.Cut(strings.TrimSpace(sc.Text()), "=")
if !ok {
continue
}
switch key {
case "out_time_us", "out_time_ms":
// 这两个键都是微秒(ffmpeg 的历史遗留),都认一下,谁在后面谁说了算
if us, err := strconv.ParseInt(val, 10, 64); err == nil && us >= 0 {
doneSec = float64(us) / 1e6
}
case "speed":
speed = parseFFmpegSpeed(val)
case "fps":
if v, err := strconv.ParseFloat(val, 64); err == nil && v >= 0 {
fps = v
}
case "total_size":
if v, err := strconv.ParseInt(val, 10, 64); err == nil && v >= 0 {
size = v
}
case "progress":
cb(doneSec, speed, fps, size)
}
}
}
// 解析 ffmpeg 的 speed 字段,形如 "1.37x";还没开始或暂停时是 "N/A"
func parseFFmpegSpeed(v string) float64 {
v = strings.TrimSuffix(strings.TrimSpace(v), "x")
f, err := strconv.ParseFloat(strings.TrimSpace(v), 64)
if err != nil || f < 0 {
return 0
}
return f
}
// 用 concat demuxer 把各分块的成片拼接起来。cb 用来回传拼接进度,可为 nil
func concatChunks(ffmpegFile string, files []string, out, dir string, cb ffmpegProgress) error {
listFile := filepath.Join(dir, chunkTempPrefix+"list.txt")
var b strings.Builder
for _, p := range files {
// concat demuxer 里单引号要写成 '\''
p = filepath.ToSlash(p)
b.WriteString("file '" + strings.ReplaceAll(p, "'", `'\''`) + "'\n")
}
if err := os.WriteFile(listFile, []byte(b.String()), 0644); err != nil {
return fmt.Errorf("写入拼接清单失败:%v", err)
}
defer os.Remove(listFile)
if err := runFFmpegWith(ffmpegFile, []string{"-y", "-f", "concat", "-safe", "0", "-i", listFile, "-c", "copy", out}, dir, cb); err != nil {
// 停止任务时杀掉 ffmpeg 也会让它非 0 退出,这是「已停止」不是「拼接失败」
if jobStopRequested() {
return errJobStopped
}
return fmt.Errorf("拼接各分块失败:%v", err)
}
return nil
}
// 取文件大小,出错时返回 0
func fileSizeOrZero(path string) int64 {
if info, err := os.Stat(path); err == nil {
return info.Size()
}
return 0
}
// 文件存不存在(只关心存在性,不关心是不是目录)
func fileExists(path string) bool {
_, err := os.Stat(path)
return err == nil
}
// 把秒格式化成 ffmpeg 参数,保留毫秒
func formatSeconds(sec float64) string {
return strconv.FormatFloat(sec, 'f', 3, 64)
}
// 头像缓存的文件名:uid + 头像链接的摘要。链接变了文件名跟着变,
// 主播换头像后不会一直用着旧缓存
func avatarCacheName(uid int64, url string) string {
sum := sha1.Sum([]byte(url))
return fmt.Sprintf("%d_%x.png", uid, sum[:6])
}
// 缓存的头像是不是一个完整的 PNG:PNG 必须以 IEND 块收尾。
// 只看「文件在不在」是不够的 —— 下载中断、写盘失败都会留下半截文件,
// 而它下次会被当成可用缓存塞给 ffmpeg 当输入,那一块直接渲染失败,
// 报的还是 exit status 0xffffffea 这类完全看不出跟头像有关的信息
func validCachedAvatar(path string) bool {
f, err := os.Open(path)
if err != nil {
return false
}
defer f.Close()
info, err := f.Stat()
if err != nil || info.Size() < 16 {
return false
}
tail := make([]byte, 12) // IEND 块:4 字节长度 + "IEND" + 4 字节 CRC
if _, err := f.ReadAt(tail, info.Size()-12); err != nil {
return false
}
return bytes.Equal(tail[4:8], []byte("IEND"))
}
// 下载头像并圆形裁切,失败自动重试,返回缓存的 PNG 路径,失败返回空字符串。
// legacyDirs 是旧版本放缓存的位置,命中就搬过来用,省得换保存目录后重新下一次。
func ensureAvatar(uid int64, url, cacheDir string, legacyDirs ...string) string {
name := avatarCacheName(uid, url)
path := filepath.Join(cacheDir, name)
if validCachedAvatar(path) {
return path
}
if err := os.MkdirAll(cacheDir, 0755); err != nil {
lPrintWarnf("创建头像缓存文件夹 %s 失败,弹幕只烧录文字不带头像:%v", cacheDir, err)
return ""
}
// 旧位置(以前存在录播目录里)有同一个头像就直接搬过来,不重复下载
for _, dir := range legacyDirs {
if dir == "" || dir == cacheDir {
continue
}
old := filepath.Join(dir, name)
if _, err := os.Stat(old); err != nil {
continue
}
if err := os.Rename(old, path); err == nil {
if validCachedAvatar(path) {
return path
}
_ = os.Remove(path) // 旧的也是半截文件,删掉重新下
break
}
// 跨盘改名会失败,退回复制
if copyFile(old, path) == nil {
if validCachedAvatar(path) {
_ = os.Remove(old)
return path
}
_ = os.Remove(path)
break
}
}
// 清掉这个用户已经失效的旧头像
for _, dir := range append([]string{cacheDir}, legacyDirs...) {
if dir == "" {
continue
}
olds, err := filepath.Glob(filepath.Join(dir, strconv.FormatInt(uid, 10)+"_*.png"))
if err != nil {
continue
}
for _, p := range olds {
if p != path {
_ = os.Remove(p)
}
}
}
err := runThrice(func() error {
// 每次重试前过闸门:暂停就在这里等着(不会把剩下的重试继续跑完),
// 停止就直接放弃 —— 否则点了暂停/停止还得等这个头像把三次重试耗完
if err := jobGate(); err != nil {
return noRetry(err)
}
return downloadAvatar(uid, url, path)
})
if err != nil {
// 停止不算故障,不用报「头像不可用」
if !errors.Is(err, errJobStopped) {
lPrintWarnf("uid %d 的头像不可用,该用户的弹幕只烧录文字不带头像:%v", uid, err)
}
return ""
}
return path
}
const (
// 头像下载的体积上限。真实头像最大也就几百 KB,这里给个大方的余量。
// 没有上限的话,CDN 一旦返回异常大的响应就能把内存吃光(实测单个头像吃掉 512MB)
avatarMaxBytes = 2 << 20
// 解码后允许的像素数上限(长 × 宽)。2048×2048 对头像来说是夸张地宽裕
avatarMaxPixels = 2048 * 2048
)
// 复用连接:一场直播要下几百个头像,每个都新建 client 就是几百次握手。
// 超时给 10 秒:头像只有几百 KB,超过就说明这个 CDN 节点不通,
// 早点超时能早点重试/过闸门,用户点暂停时也不用干等半分钟
var avatarClient = &http.Client{Timeout: 10 * time.Second}
func downloadAvatar(uid int64, url, path string) error {
req, err := http.NewRequest("GET", url, nil)
if err != nil {
return err
}
req.Header.Set("User-Agent", "Mozilla/5.0")
resp, err := avatarClient.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("HTTP %d", resp.StatusCode)
}
// 多读一个字节才能分辨「刚好到上限」和「超过了上限」
data, err := io.ReadAll(io.LimitReader(resp.Body, avatarMaxBytes+1))
if err != nil {
return err
}
if len(data) > avatarMaxBytes {
return noRetry(fmt.Errorf("头像体积超过 %d MB,不下载", avatarMaxBytes>>20))
}
src, err := decodeAvatar(data, resp.Header.Get("Content-Type"))
if err != nil {
return err
}
// 先写临时文件再改名:写一半失败留下的半截文件不能被当成可用缓存,
// 否则下次命中它就会把残图塞给 ffmpeg 当输入,整块渲染跟着失败
tmp := path + ".part"
out, err := os.Create(tmp)
if err != nil {
return err
}
if err := png.Encode(out, circleAvatar(src, avatarCacheSize)); err != nil {
out.Close()
_ = os.Remove(tmp)
return err
}
if err := out.Close(); err != nil {
_ = os.Remove(tmp)
return err
}
return os.Rename(tmp, path)
}
// 解码头像图片。AcFun 的头像里混着 gif 和 webp:
// gif 由标准库解码,多帧的取第一帧;webp 只支持静态的,动画 webp 会报错,
// 这种情况下这个用户就不烧头像,只烧文字。
func decodeAvatar(data []byte, contentType string) (image.Image, error) {
src, _, err := image.Decode(bytes.NewReader(data))
if err != nil {
return nil, fmt.Errorf("图片解析不了(Content-Type: %s):%v", contentType, err)
}
// 防压缩炸弹:几十 KB 的 PNG/GIF 能解压出上亿像素,几 MB 的文件就能吃掉几 GB 内存。
// 头像最后只会被缩到几十像素(跟随字号),根本用不到这么大的原图
if b := src.Bounds(); b.Dx()*b.Dy() > avatarMaxPixels {
return nil, noRetry(fmt.Errorf("头像尺寸 %dx%d 超出上限(最多 %d 像素)", b.Dx(), b.Dy(), avatarMaxPixels))
}
return src, nil
}
// uid → 头像链接的记录,包含查不到的结果,避免同一个 uid 反复去问。
// 查到的结果还会落盘(见 avatarURLCacheFile),否则进程一重启就要把整场 uid 重查一遍。
var avatarURLMemo struct {
sync.Mutex
m map[int64]string
at map[int64]int64 // 记录写入时刻(Unix 秒),0 表示本进程刚查到的
dir string // 落盘文件所在目录,换目录会重新加载
loaded bool
}
// 反查结果的落盘文件(放在头像缓存目录里)与有效期。
// 有效期是为了让主播换头像后能自己更新过来:过期就重新查一次,查到了用新的
const (
avatarURLCacheFile = "uid_urls.jsonl"
avatarURLCacheTTL = 7 * 24 * time.Hour
)
type avatarURLCacheEntry struct {
UID int64 `json:"uid"`
URL string `json:"url"`
At int64 `json:"at"` // 写入时刻(Unix 秒)
}
// 实际去查头像的动作,单独拎出来方便测试替换掉
var lookupAvatarURL = fetchAvatarURL
// 读取落盘的反查记录。返回整理后的记录,以及文件里原有的行数(行数多于记录数
// 说明有重复或过期的行,调用方会顺手把文件压一遍)
func loadAvatarURLCache(dir string) (map[int64]avatarURLCacheEntry, int, error) {
path := filepath.Join(dir, avatarURLCacheFile)
f, err := os.Open(path)
if err != nil {
return nil, 0, err
}
defer f.Close()
entries := make(map[int64]avatarURLCacheEntry, 256)
total := 0
now := time.Now().Unix()
ttl := int64(avatarURLCacheTTL / time.Second)
scanner := bufio.NewScanner(f)
scanner.Buffer(make([]byte, 64*1024), 64*1024)
for scanner.Scan() {
line := strings.TrimSpace(scanner.Text())
if line == "" {
continue
}
var e avatarURLCacheEntry
if json.Unmarshal([]byte(line), &e) != nil || e.UID <= 0 || e.URL == "" {
continue
}
total++
if now-e.At > ttl {
continue // 过期了:主播可能换过头像,重新查
}
entries[e.UID] = e // 后写的覆盖先写的
}
if err := scanner.Err(); err != nil {
return entries, total, err
}
return entries, total, nil
}
// 重写整个落盘文件(先写临时文件再改名,避免写一半断电留个坏文件)
func saveAvatarURLCache(dir string, entries map[int64]avatarURLCacheEntry) error {
if err := os.MkdirAll(dir, 0755); err != nil {
return err
}
path := filepath.Join(dir, avatarURLCacheFile)
tmp := path + ".tmp"
f, err := os.Create(tmp)
if err != nil {
return err
}
w := bufio.NewWriter(f)
for _, e := range entries {
b, err := json.Marshal(e)
if err != nil {
continue
}
w.Write(b)
w.WriteByte('\n')
}
if err := w.Flush(); err != nil {
f.Close()
return err
}
if err := f.Close(); err != nil {
return err
}
return os.Rename(tmp, path)
}
// 追加一条反查结果(调用方持锁)。写不进去最多下次多查一遍,不影响合并
func appendAvatarURLCache(dir string, uid int64, url string) {
if url == "" {
return
}
if err := os.MkdirAll(dir, 0755); err != nil {
return
}
f, err := os.OpenFile(filepath.Join(dir, avatarURLCacheFile),
os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
if err != nil {
return
}
defer f.Close()
b, err := json.Marshal(avatarURLCacheEntry{UID: uid, URL: url, At: time.Now().Unix()})
if err != nil {
return
}
b = append(b, '\n')
_, _ = f.Write(b)
}
// 把落盘记录读进内存(调用方持锁)。本进程刚查到的结果优先,不被文件里的旧值覆盖
func loadAvatarURLCacheInto(dir string) {
if avatarURLMemo.loaded && avatarURLMemo.dir == dir {
return
}
entries, total, err := loadAvatarURLCache(dir)
if err != nil && !os.IsNotExist(err) {
lPrintWarnf("读取头像链接记录失败,本次仍会按 uid 反查:%v", err)
}
if avatarURLMemo.m == nil {
avatarURLMemo.m = make(map[int64]string, len(entries))
}
if avatarURLMemo.at == nil {
avatarURLMemo.at = make(map[int64]int64, len(entries))
}
added := 0
for uid, e := range entries {
if _, ok := avatarURLMemo.m[uid]; ok {
continue
}
avatarURLMemo.m[uid] = e.URL
avatarURLMemo.at[uid] = e.At
added++
}
avatarURLMemo.dir = dir
avatarURLMemo.loaded = true
if total > len(entries) {
// 有重复或过期的行,顺手压一遍,免得文件一直长
_ = saveAvatarURLCache(dir, entries)
}
if added > 0 {
lPrintf("头像链接记录:从 %s 读到 %d 个用户的链接,这些不用再查", avatarURLCacheFile, added)
}
}
// 按 uid 反查头像链接。原版 .ass 里没有头像链接,只能这样补回来。
// 查到的结果会落盘,重启进程后不用把整场 uid 重新查一遍。
// 第二个返回值表示这次是不是真的发了请求(false = 用了内存或落盘里的记录)
func avatarURLByUID(uid int64, cacheDir string) (string, bool) {
if uid <= 0 {
return "", false
}
avatarURLMemo.Lock()
loadAvatarURLCacheInto(cacheDir)
if url, ok := avatarURLMemo.m[uid]; ok {
avatarURLMemo.Unlock()
return url, false
}
avatarURLMemo.Unlock()
url := lookupAvatarURL(uid)
avatarURLMemo.Lock()
if avatarURLMemo.m == nil {
avatarURLMemo.m = make(map[int64]string)
}
if avatarURLMemo.at == nil {
avatarURLMemo.at = make(map[int64]int64)
}
avatarURLMemo.m[uid] = url
avatarURLMemo.at[uid] = time.Now().Unix()
if url != "" {
appendAvatarURLCache(cacheDir, uid, url)
}
avatarURLMemo.Unlock()
return url, true
}
// 查头像链接用的 client,同样复用连接
var avatarAPIClient = &http.Client{Timeout: 15 * time.Second}
// 查 AcFun 的用户信息接口取头像地址
func fetchAvatarURL(uid int64) string {
addr := fmt.Sprintf("https://www.acfun.cn/rest/pc-direct/user/userInfo?userId=%d", uid)
req, err := http.NewRequest("GET", addr, nil)
if err != nil {
return ""
}
req.Header.Set("User-Agent", "Mozilla/5.0")
resp, err := avatarAPIClient.Do(req)
if err != nil {
return ""
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return ""
}
data, err := io.ReadAll(io.LimitReader(resp.Body, 256*1024))
if err != nil {
return ""
}
var out struct {
Result int `json:"result"`
Profile struct {
HeadURL string `json:"headUrl"`
} `json:"profile"`
}
if json.Unmarshal(data, &out) != nil || out.Result != 0 {
return ""
}
return strings.TrimSpace(out.Profile.HeadURL)
}
// QSV 编码器的可用性探测结果,按 ffmpeg 路径缓存,避免分块时每块都试一次
var qsvCache struct {
sync.Mutex
path string
avail map[string]bool
}
// 探测某个 Intel QSV 编码器(hevc_qsv / h264_qsv)能不能跑,做一次真实的极小编码
func qsvEncoderAvailable(ffmpegFile, encoder string) bool {
qsvCache.Lock()
if qsvCache.avail != nil && qsvCache.path == ffmpegFile {
if ok, tested := qsvCache.avail[encoder]; tested {
qsvCache.Unlock()
return ok
}
}
qsvCache.Unlock()
cmd := exec.Command(ffmpegFile, "-hide_banner", "-loglevel", "error",
"-f", "lavfi", "-i", "color=c=black:s=320x240:d=0.5:r=30",
"-c:v", encoder, "-frames:v", "5", "-f", "null", "-")
ok := cmd.Run() == nil
qsvCache.Lock()
if qsvCache.path != ffmpegFile || qsvCache.avail == nil {
qsvCache.path, qsvCache.avail = ffmpegFile, map[string]bool{}
}
qsvCache.avail[encoder] = ok
qsvCache.Unlock()
return ok
}
// 把头像居中裁方、缩放并圆形裁切,输出带透明通道的 PNG
func circleAvatar(src image.Image, size int) image.Image {
sb := src.Bounds()
side := sb.Dx()
if sb.Dy() < side {
side = sb.Dy()
}
cx, cy := sb.Min.X+sb.Dx()/2, sb.Min.Y+sb.Dy()/2
half := side / 2
sr := image.Rect(cx-half, cy-half, cx+half, cy+half)
scaled := image.NewRGBA(image.Rect(0, 0, size, size))
xdraw.CatmullRom.Scale(scaled, scaled.Bounds(), src, sr, xdraw.Over, nil)
// 圆形遮罩,边缘 2px 抗锯齿过渡
mask := image.NewAlpha(image.Rect(0, 0, size, size))
radius := float64(size) / 2
edge := 2.0
start := 1 - edge/radius
for y := 0; y < size; y++ {
for x := 0; x < size; x++ {
dx := float64(x-size/2) + 0.5
dy := float64(y-size/2) + 0.5
d := math.Sqrt(dx*dx+dy*dy) / radius
switch {
case d >= 1:
mask.SetAlpha(x, y, color.Alpha{A: 0})
case d > start:
a := uint8(255 * (1 - (d-start)/(edge/radius)))
mask.SetAlpha(x, y, color.Alpha{A: a})
default:
mask.SetAlpha(x, y, color.Alpha{A: 255})
}
}
}
dst := image.NewRGBA(image.Rect(0, 0, size, size))
draw.DrawMask(dst, dst.Bounds(), scaled, image.Point{}, mask, image.Point{}, draw.Over)
return dst
}