diff --git a/avatar_cache_test.go b/avatar_cache_test.go new file mode 100644 index 0000000..178aae4 --- /dev/null +++ b/avatar_cache_test.go @@ -0,0 +1,377 @@ +package main + +import ( + "errors" + "fmt" + "image" + "image/png" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync/atomic" + "testing" + "time" +) + +// 换掉真的反查动作,返回调用次数 +func stubLookupAvatarURL(t *testing.T, f func(uid int64) string) *int64 { + t.Helper() + var calls int64 + old := lookupAvatarURL + lookupAvatarURL = func(uid int64) string { + atomic.AddInt64(&calls, 1) + return f(uid) + } + t.Cleanup(func() { lookupAvatarURL = old }) + return &calls +} + +// 清空进程内的链接记忆,模拟「重启进程」,用完恢复 +func resetAvatarURLMemo(t *testing.T) { + t.Helper() + avatarURLMemo.Lock() + prevM, prevAt, prevDir, prevLoaded := avatarURLMemo.m, avatarURLMemo.at, avatarURLMemo.dir, avatarURLMemo.loaded + avatarURLMemo.m, avatarURLMemo.at, avatarURLMemo.dir, avatarURLMemo.loaded = nil, nil, "", false + avatarURLMemo.Unlock() + t.Cleanup(func() { + avatarURLMemo.Lock() + avatarURLMemo.m, avatarURLMemo.at, avatarURLMemo.dir, avatarURLMemo.loaded = prevM, prevAt, prevDir, prevLoaded + avatarURLMemo.Unlock() + }) +} + +// 直接写落盘文件(绕过 append,方便造历史记录) +func writeAvatarURLCacheFile(t *testing.T, dir string, lines []string) string { + t.Helper() + path := filepath.Join(dir, avatarURLCacheFile) + if err := os.MkdirAll(dir, 0755); err != nil { + t.Fatal(err) + } + body := strings.Join(lines, "\n") + if body != "" { + body += "\n" + } + if err := os.WriteFile(path, []byte(body), 0644); err != nil { + t.Fatal(err) + } + return path +} + +// 查到的链接要落盘,重启进程后不再重复反查 +func TestAvatarURLCacheSurvivesRestart(t *testing.T) { + dir := t.TempDir() + resetAvatarURLMemo(t) + calls := stubLookupAvatarURL(t, func(uid int64) string { + return fmt.Sprintf("https://example.invalid/%d.png", uid) + }) + + url1, fetched1 := avatarURLByUID(42, dir) + if !fetched1 { + t.Fatal("第一次应当真的去查") + } + if url1 != "https://example.invalid/42.png" { + t.Fatalf("拿到的链接不对:%q", url1) + } + if atomic.LoadInt64(calls) != 1 { + t.Fatalf("第一次应当查 1 次,实际 %d", atomic.LoadInt64(calls)) + } + + // 同进程第二次:内存记忆命中,不再查 + if _, fetched := avatarURLByUID(42, dir); fetched { + t.Fatal("同进程内第二次不该再查") + } + + // 模拟重启:清空内存,只留落盘文件 + resetAvatarURLMemo(t) + url2, fetched2 := avatarURLByUID(42, dir) + if fetched2 { + t.Fatal("重启后应当直接用落盘记录,不再向 AcFun 查") + } + if url2 != url1 { + t.Fatalf("落盘记录里的链接应当和当初查到的一样,实际 %q", url2) + } + if n := atomic.LoadInt64(calls); n != 1 { + t.Fatalf("整个过程只该查 1 次,实际 %d 次", n) + } +} + +// 过期的记录要重新查(给主播换头像留出路),没到期的直接用 +func TestAvatarURLCacheTTL(t *testing.T) { + dir := t.TempDir() + now := time.Now() + writeAvatarURLCacheFile(t, dir, []string{ + fmt.Sprintf(`{"uid":1,"url":"https://old/1.png","at":%d}`, now.Add(-8*24*time.Hour).Unix()), + fmt.Sprintf(`{"uid":2,"url":"https://fresh/2.png","at":%d}`, now.Add(-time.Hour).Unix()), + }) + resetAvatarURLMemo(t) + stubLookupAvatarURL(t, func(uid int64) string { return "https://rechecked/x.png" }) + + if u, fetched := avatarURLByUID(2, dir); fetched || u != "https://fresh/2.png" { + t.Fatalf("没到期的记录应当直接用,实际 (%q, fetched=%v)", u, fetched) + } + if u, fetched := avatarURLByUID(1, dir); !fetched || u != "https://rechecked/x.png" { + t.Fatalf("过了有效期的记录应当重新查,实际 (%q, fetched=%v)", u, fetched) + } +} + +// 文件里有重复行、过期行和坏行时,加载后顺手压一遍,别让它一直长 +func TestAvatarURLCacheCompactsFile(t *testing.T) { + dir := t.TempDir() + now := time.Now() + path := writeAvatarURLCacheFile(t, dir, []string{ + fmt.Sprintf(`{"uid":7,"url":"https://a/7.png","at":%d}`, now.Add(-3*time.Minute).Unix()), + fmt.Sprintf(`{"uid":7,"url":"https://b/7.png","at":%d}`, now.Add(-2*time.Minute).Unix()), + fmt.Sprintf(`{"uid":7,"url":"https://c/7.png","at":%d}`, now.Add(-time.Minute).Unix()), + fmt.Sprintf(`{"uid":8,"url":"https://old/8.png","at":%d}`, now.Add(-30*24*time.Hour).Unix()), + `这不是 json`, + }) + resetAvatarURLMemo(t) + stubLookupAvatarURL(t, func(uid int64) string { return "" }) + + if u, fetched := avatarURLByUID(7, dir); fetched || u != "https://c/7.png" { + t.Fatalf("同一 uid 应当以后写的为准(c),实际 (%q, fetched=%v)", u, fetched) + } + + data, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + lines := strings.Split(strings.TrimSpace(string(data)), "\n") + if len(lines) != 1 { + t.Fatalf("压缩后应当只剩 1 行有效记录,实际 %d 行:%v", len(lines), lines) + } + if !strings.Contains(lines[0], "https://c/7.png") { + t.Fatalf("压缩后保留的应当是最新的那条:%s", lines[0]) + } +} + +// 新查到的链接要追加到落盘文件里 +func TestAvatarURLCacheAppendsNewEntries(t *testing.T) { + dir := t.TempDir() + resetAvatarURLMemo(t) + stubLookupAvatarURL(t, func(uid int64) string { return fmt.Sprintf("https://x/%d.png", uid) }) + + if _, fetched := avatarURLByUID(11, dir); !fetched { + t.Fatal("第一次应当去查") + } + data, err := os.ReadFile(filepath.Join(dir, avatarURLCacheFile)) + if err != nil { + t.Fatalf("落盘文件应当已生成:%v", err) + } + if !strings.Contains(string(data), `"uid":11`) || !strings.Contains(string(data), "https://x/11.png") { + t.Fatalf("落盘内容不对:%s", data) + } +} + +// 头像并发获取:首次全部下载,第二次全部命中缓存,不会重复下载 +func TestFetchAvatarsReusesCache(t *testing.T) { + withJobFlags(t, true, false, false) + dir := t.TempDir() + + var hits int64 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt64(&hits, 1) + _ = png.Encode(w, image.NewRGBA(image.Rect(0, 0, 8, 8))) + })) + defer srv.Close() + + const n = 20 + uids := make([]int64, 0, n) + urlByUID := make(map[int64]string, n) + for i := 0; i < n; i++ { + uid := int64(1000 + i) + uids = append(uids, uid) + urlByUID[uid] = fmt.Sprintf("%s/av%d.png", srv.URL, uid) + } + + paths, stat, err := fetchAvatars(uids, urlByUID, dir, "") + if err != nil { + t.Fatalf("获取头像不该出错:%v", err) + } + if len(paths) != n { + t.Fatalf("应当拿到 %d 个头像,实际 %d", n, len(paths)) + } + if stat.imgDownloaded != n || stat.imgReused != 0 || stat.failed != 0 { + t.Fatalf("首次应当全部下载:下载 %d 复用 %d 失败 %d", stat.imgDownloaded, stat.imgReused, stat.failed) + } + if got := atomic.LoadInt64(&hits); got != n { + t.Fatalf("首次应当请求 %d 次,实际 %d", n, got) + } + + // 第二次:全部命中缓存,一次网络都不该发 + paths2, stat2, err := fetchAvatars(uids, urlByUID, dir, "") + if err != nil { + t.Fatalf("第二次获取头像不该出错:%v", err) + } + if len(paths2) != n { + t.Fatalf("第二次也应当拿到 %d 个头像,实际 %d", n, len(paths2)) + } + if stat2.imgReused != n || stat2.imgDownloaded != 0 { + t.Fatalf("第二次应当全部复用缓存:复用 %d 下载 %d", stat2.imgReused, stat2.imgDownloaded) + } + if got := atomic.LoadInt64(&hits); got != n { + t.Fatalf("第二次不该再下载,累计请求数应当还是 %d,实际 %d", n, got) + } +} + +// 统计要能区分「链接复用」和「真的去查了」 +func TestFetchAvatarsCountsLinkLookups(t *testing.T) { + withJobFlags(t, true, false, false) + dir := t.TempDir() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _ = png.Encode(w, image.NewRGBA(image.Rect(0, 0, 8, 8))) + })) + defer srv.Close() + + resetAvatarURLMemo(t) + lookups := stubLookupAvatarURL(t, func(uid int64) string { return srv.URL + "/looked.png" }) + + uids := []int64{1, 2, 3} + urlByUID := map[int64]string{ + 1: srv.URL + "/own.png", // 自带的链接 + 2: "", + 3: "", + } + paths, stat, err := fetchAvatars(uids, urlByUID, dir, "") + if err != nil { + t.Fatalf("不该出错:%v", err) + } + if len(paths) != 3 { + t.Fatalf("三个用户都该拿到头像,实际 %d", len(paths)) + } + if stat.fetched != 2 { + t.Fatalf("应当有 2 个走了反查,实际 %d", stat.fetched) + } + if stat.reusedLink != 0 { + t.Fatalf("第一次没有可复用的链接,实际 %d", stat.reusedLink) + } + if got := atomic.LoadInt64(lookups); got != 2 { + t.Fatalf("反查动作应当只发生 2 次,实际 %d", got) + } + + // 再跑一次:这两个 uid 的链接已经记下了,不该再查 + _, stat2, err := fetchAvatars(uids, urlByUID, dir, "") + if err != nil { + t.Fatalf("第二次不该出错:%v", err) + } + if stat2.reusedLink != 2 || stat2.fetched != 0 { + t.Fatalf("第二次应当复用链接:复用 %d 反查 %d", stat2.reusedLink, stat2.fetched) + } + if got := atomic.LoadInt64(lookups); got != 2 { + t.Fatalf("第二次不该再查链接,累计应当还是 2 次,实际 %d", got) + } +} + +// 任务被停止时,获取头像要立刻收工并返回 errJobStopped +func TestFetchAvatarsStopsWhenAsked(t *testing.T) { + withJobFlags(t, true, false, true) + dir := t.TempDir() + _, _, err := fetchAvatars([]int64{1}, map[int64]string{1: "http://127.0.0.1:1/x.png"}, dir, "") + if !errors.Is(err, errJobStopped) { + t.Fatalf("应当返回 errJobStopped,实际 %v", err) + } +} + +// 统计里的「新下载」要能识别出「这个文件本来就在」 +func TestFetchAvatarsCountsReuseAfterRestart(t *testing.T) { + withJobFlags(t, true, false, false) + dir := t.TempDir() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _ = png.Encode(w, image.NewRGBA(image.Rect(0, 0, 8, 8))) + })) + defer srv.Close() + + uids := []int64{5001} + urlByUID := map[int64]string{5001: srv.URL + "/a.png"} + + // 先下载一次,再把内存里的链接记忆清掉(模拟重启后再合并同一场) + if _, stat, err := fetchAvatars(uids, urlByUID, dir, ""); err != nil || stat.imgDownloaded != 1 { + t.Fatalf("首次应当下载 1 个:%v %+v", err, stat) + } + resetAvatarURLMemo(t) + if _, stat, err := fetchAvatars(uids, urlByUID, dir, ""); err != nil || stat.imgReused != 1 || stat.imgDownloaded != 0 { + t.Fatalf("重启后应当复用缓存:%v %+v", err, stat) + } +} + +// 落盘文件写坏/读不出时不能影响合并,只是这次多查几遍 +func TestAvatarURLCacheBrokenFileIsIgnored(t *testing.T) { + dir := t.TempDir() + writeAvatarURLCacheFile(t, dir, []string{"{坏的一行", "还是坏的"}) + resetAvatarURLMemo(t) + calls := stubLookupAvatarURL(t, func(uid int64) string { return "https://ok/1.png" }) + + u, fetched := avatarURLByUID(1, dir) + if !fetched || u != "https://ok/1.png" { + t.Fatalf("坏文件应当忽略并照常反查,实际 (%q, fetched=%v)", u, fetched) + } + if atomic.LoadInt64(calls) != 1 { + t.Fatalf("应当查 1 次,实际 %d", atomic.LoadInt64(calls)) + } +} + +// 确认落盘用的是头像缓存目录里的固定文件名(别改到别处去) +func TestAvatarURLCacheFilePath(t *testing.T) { + dir := t.TempDir() + resetAvatarURLMemo(t) + stubLookupAvatarURL(t, func(uid int64) string { return "https://x/1.png" }) + avatarURLByUID(1, dir) + + want := filepath.Join(dir, avatarURLCacheFile) + if _, err := os.Stat(want); err != nil { + t.Fatalf("落盘文件应当在 %s:%v", want, err) + } + // 顺带确认它能被当成普通 jsonl 读出来(一行一条) + f, err := os.Open(want) + if err != nil { + t.Fatal(err) + } + defer f.Close() + data, err := io.ReadAll(f) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(data), "https://x/1.png") { + t.Fatalf("内容不对:%s", data) + } +} + +// 头像确实是并发获取的:串行做几百个用户要好几分钟,这条锁住「同时有多个请求在跑」 +func TestFetchAvatarsRunsConcurrently(t *testing.T) { + withJobFlags(t, true, false, false) + dir := t.TempDir() + + var active, maxActive int64 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + cur := atomic.AddInt64(&active, 1) + for { + old := atomic.LoadInt64(&maxActive) + if cur <= old || atomic.CompareAndSwapInt64(&maxActive, old, cur) { + break + } + } + time.Sleep(40 * time.Millisecond) // 让并发的重叠看得见 + atomic.AddInt64(&active, -1) + _ = png.Encode(w, image.NewRGBA(image.Rect(0, 0, 8, 8))) + })) + defer srv.Close() + + const n = 12 + uids := make([]int64, 0, n) + urlByUID := make(map[int64]string, n) + for i := 0; i < n; i++ { + uid := int64(2000 + i) + uids = append(uids, uid) + urlByUID[uid] = fmt.Sprintf("%s/c%d.png", srv.URL, uid) + } + + if _, _, err := fetchAvatars(uids, urlByUID, dir, ""); err != nil { + t.Fatalf("不该出错:%v", err) + } + if got := atomic.LoadInt64(&maxActive); got < 2 { + t.Fatalf("头像应当是并发获取的,同时最多只有 %d 个请求在跑(串行会把几百人拖成几分钟)", got) + } +} diff --git a/command_len_test.go b/command_len_test.go new file mode 100644 index 0000000..f5dbc2a --- /dev/null +++ b/command_len_test.go @@ -0,0 +1,76 @@ +package main + +import ( + "fmt" + "path/filepath" + "strings" + "testing" +) + +// 按最坏情况(每条弹幕一个不同用户、每个都带头像)渲染一块,返回这一块的命令行长度。 +// 长度从 renderChunk 自己的自检错误里读出来;没触发自检就返回 (0, false)。 +// +// 线上的那次事故就是这里没守住:单块 200 条时实测 48288 字符,远超 Windows 的 32767, +// 而 exec 只回一句 "The filename or extension is too long",看不出跟弹幕数量有关。 +func chunkCommandLenProbe(t *testing.T, n int) (length int, over bool) { + t.Helper() + dir := t.TempDir() + avatars := make(map[int64]string, n) + records := make([]danmuRecord, 0, n) + for i := 0; i < n; i++ { + uid := int64(1000000 + i) + // 头像路径按真实缓存目录的形态造(uid_12位摘要.png),故意做得和线上一样长 + avatars[uid] = filepath.Join(dir, avatarCacheDir, + fmt.Sprintf("%d_%012x.png", uid, i*0x9e3779b1)) + records = append(records, danmuRecord{ + Time: float64(i) * 0.05, UID: uid, Name: "观众甲", + Content: "这条弹幕稍微长一点测试宽度", Lane: i % 20, Width: 480, + }) + } + ctx := &burnContext{ + ffmpegFile: filepath.Join(dir, "没有这个-ffmpeg.exe"), // 自检在 exec 之前,跑不到它 + outFile: filepath.Join(dir, "out.mp4"), + meta: danmuMeta{PlayResX: 1920, PlayResY: 1080, FontSize: 42}, + avatars: avatars, + dir: dir, + } + err := ctx.renderChunk(danmuChunk{start: 0, end: 60, records: records}, "chunk.ass", ctx.outFile, false, nil) + if err == nil { + t.Fatal("用不存在的 ffmpeg 路径不该成功") + } + if !strings.Contains(err.Error(), "命令过长") { + return 0, false + } + if _, scanErr := fmt.Sscanf(err.Error(), "ffmpeg 命令过长(%d 字符", &length); scanErr != nil { + t.Fatalf("没能从错误里读出长度:%v / %s", scanErr, err) + } + return length, true +} + +// 按硬上限切出来的块,在最坏情况下也不能顶破命令行上限 +func TestChunkCommandLengthUnderLimit(t *testing.T) { + if length, over := chunkCommandLenProbe(t, maxDanmuPerChunkHard); over { + t.Fatalf("按硬上限 %d 条渲染,命令长达 %d 字符,已经超过安全值 %d", + maxDanmuPerChunkHard, length, maxCommandLineLen) + } +} + +// 块大到一定程度必须被自检拦住,并报出「多少条弹幕、多少个头像」这种能定位的信息 +func TestRenderChunkReportsTooLongCommand(t *testing.T) { + length, over := chunkCommandLenProbe(t, 300) + if !over { + t.Fatal("300 条的最坏情况应当被自检拦住,否则交给 exec 只会报一句看不出原因的错") + } + if length <= maxCommandLineLen { + t.Fatalf("自检报出的长度 %d 应当大于阈值 %d", length, maxCommandLineLen) + } +} + +// 硬上限本身要留出足够的余量:按实测每条约 241 字符,上限条数不能逼近 32767 +func TestChunkHardLimitLeavesHeadroom(t *testing.T) { + const perRecordWorstCase = 241 // 2026-09-26 真机实测(每条一个不同用户) + if est := maxDanmuPerChunkHard*perRecordWorstCase + 200; est > maxCommandLineLen { + t.Fatalf("硬上限 %d 条按最坏情况估算要 %d 字符,超过了安全值 %d", + maxDanmuPerChunkHard, est, maxCommandLineLen) + } +} diff --git a/danmuburn.go b/danmuburn.go index 960c1c8..2587e62 100644 --- a/danmuburn.go +++ b/danmuburn.go @@ -34,6 +34,21 @@ import ( // 头像缓存目录名 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 { @@ -66,11 +81,30 @@ const maxDanmuPerChunk = 80 // 一条弹幕在 -filter_complex 里约占 110 字节,几百条足以顶破上面那条命令行上限, // 表现同样是「合并失败且看不出原因」,所以并入也要有个头。 // 到上限就把块尾收到下一条弹幕出现之前,让剩下的弹幕落到下一块。 -const maxDanmuPerChunkHard = 200 +// +// 上限值按最坏情况实测定的:一块里每条弹幕都来自不同用户时,每条要占 +// 「-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 @@ -293,8 +327,6 @@ func burnDanmu(jsonlFile, videoFile, outFile string, offset float64) error { // 弹道按最终字号统一重排,避免还原弹幕或改字号后同弹道追尾 relayoutLanes(records, meta) - // 下载并圆形裁切头像 - avatarPaths := make(map[int64]string) // 缓存放在程序所在目录,换保存目录(甚至换盘)都不用重新下载; // 旧版本放在弹幕文件同目录,作为回退来源读一次并搬过去 cacheDir := avatarCachePath() @@ -314,34 +346,9 @@ func burnDanmu(jsonlFile, videoFile, outFile string, offset float64) error { urlByUID[r.UID] = r.Avatar } } - looked := 0 - jobSetPhase(jobPhaseAvatar) - jobAvatars(0, len(uids)) - for i, uid := range uids { - // 这一阶段既要下网络又要解图片,允许在这里暂停 / 停止 - if err := jobGate(); err != nil { - return err - } - url := urlByUID[uid] - if url == "" { - // 从 .ass 还原出来的弹幕没有头像链接,按 uid 反查一次 - url = avatarURLByUID(uid) - looked++ - // 反查比较慢,定期报一下进度,避免看起来像卡住了 - if looked%50 == 0 || i+1 == len(uids) { - lPrintf("正在获取头像:已处理 %d/%d 个用户", i+1, len(uids)) - } - } - jobAvatars(i+1, len(uids)) - if url == "" { - continue - } - if path := ensureAvatar(uid, url, cacheDir, legacyCache); path != "" { - avatarPaths[uid] = path - } - } - if looked > 0 { - lPrintf("有 %d 个用户没有自带头像链接(多半是从 .ass 还原的弹幕),已按 uid 向 AcFun 反查", looked) + avatarPaths, _, err := fetchAvatars(uids, urlByUID, cacheDir, legacyCache) + if err != nil { + return err } if len(avatarPaths) == 0 { lPrintWarn("没有任何头像可用,弹幕将只烧录文字,请检查网络后重试合并") @@ -395,6 +402,111 @@ func burnDanmu(jsonlFile, videoFile, outFile string, offset float64) error { 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 还原的弹幕没把昵称和头像算进宽度(占位提前放行了同弹道的下一条, // 画面上就追尾叠字),改过字号后更是全对不上。这里用和录制时相同的碰撞公式、 @@ -814,6 +926,15 @@ func (c *burnContext) renderChunk(chunk danmuChunk, assBase, out string, cut boo 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) @@ -1054,17 +1175,28 @@ func fileSizeOrZero(path string) int64 { 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 路径,失败返回空字符串。 // legacyDirs 是旧版本放缓存的位置,命中就搬过来用,省得换保存目录后重新下一次。 func ensureAvatar(uid int64, url, cacheDir string, legacyDirs ...string) string { - // 文件名里带上头像链接的摘要,主播换头像后不会一直用着旧缓存 - sum := sha1.Sum([]byte(url)) - name := fmt.Sprintf("%d_%x.png", uid, sum[:6]) + name := avatarCacheName(uid, url) path := filepath.Join(cacheDir, name) if _, err := os.Stat(path); err == nil { return path @@ -1189,36 +1321,187 @@ func decodeAvatar(data []byte, contentType string) (image.Image, error) { return src, nil } -// uid → 头像链接的记忆,包含查不到的结果,避免同一个 uid 反复去问 +// uid → 头像链接的记录,包含查不到的结果,避免同一个 uid 反复去问。 +// 查到的结果还会落盘(见 avatarURLCacheFile),否则进程一重启就要把整场 uid 重查一遍。 var avatarURLMemo struct { sync.Mutex - m map[int64]string + 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 -// 按 uid 反查头像链接。原版 .ass 里没有头像链接,只能这样补回来 -func avatarURLByUID(uid int64) string { +// 读取落盘的反查记录。返回整理后的记录,以及文件里原有的行数(行数多于记录数 +// 说明有重复或过期的行,调用方会顺手把文件压一遍) +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 "" + return "", false } avatarURLMemo.Lock() - if avatarURLMemo.m == nil { - avatarURLMemo.m = make(map[int64]string) - } + loadAvatarURLCacheInto(cacheDir) if url, ok := avatarURLMemo.m[uid]; ok { avatarURLMemo.Unlock() - return url + 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 + return url, true } // 查头像链接用的 client,同样复用连接