diff --git a/README.md b/README.md index fe44f83..2c88e0d 100644 --- a/README.md +++ b/README.md @@ -166,13 +166,26 @@ web UI 默认通过`http://localhost:51880`访问（端口对应`webPort`）， | 分区 | 内容 | | --- | --- | -| 监控 | 添加主播、主播列表（默认打开的分区），以及运行日志 | +| 监控 | 当前转换任务的进度卡片、添加主播、主播列表（默认打开的分区），以及运行日志 | | 录制 | 保存目录和子文件夹、录播格式、默认码率上限、ffmpeg 路径 | | 合并 | 自动合并开关、成片输出目录和合成码率、手动合并 | | 通知 | 推送哪些事件，以及 QQ 机器人 / MeoW 两个渠道的配置和测试连通 | | 文件 | 磁盘剩余容量、录播和成片的文件列表 | | 设置 | 是否允许外网访问、管理口令的查看与更换 | +### 转换任务进度 + +自动合并和手动合并跑起来之后，`监控`分区顶部会出现一张`转换任务`卡片，页面最顶上还有一条很细的全局进度条，切到哪个分区都能看到。卡片里给的是： + +- **阶段步骤条**：准备 → 分段拼接 → 获取头像 → 画面渲染 → 拼接与落盘，走过的打勾、当前的高亮（暂停时变红）； +- **总进度条和百分比**：渲染阶段按「已渲染时长 ÷ 视频总时长」算，其余阶段按各自的完成比例折算后加权； +- **实时指标**：渲染倍速、帧率、已处理时长 / 总时长、已用时间、已写出多少； +- **预计剩余时间**：渲染阶段用「剩余时长 ÷ 当前倍速」，别的阶段按已用时间线性外推； +- **体积对比条**：源片大小和预计成片大小并排显示两段，预计值先按目标码率估，渲染过 5% 之后换成按已写出字节外推（这时候标注会从`按目标码率估算`变成`按已渲染部分推算`）； +- **渲染速度折线**：最近两分钟、每秒一个采样点，能看出硬件编码和 CPU 渲染有没有被别的任务拖慢。 + +任务跑着的时候可以直接**暂停**和**停止**：暂停会挂起 ffmpeg 进程（Windows 用`NtSuspendProcess`、Linux 用`SIGSTOP`），CPU 占用归零、进度冻结，点`继续`从原地接着跑；不在 ffmpeg 环节（比如正在下载头像）时，暂停和停止会在下一个检查点生效。停止会中断这次合并并**丢弃已经渲染的部分**——成片目录不会留下半成品，录播目录里的分块临时文件也会一并清掉，所以需要重新跑一次。 + 开关和下拉框改完立即保存；文本框（目录、地址、口令这些）填完按回车或点右上角的`保存设置`才生效，这一条在每个有文本框的分区里也写了提示。 目录和文件两条路都行：直接在框里填，或者点框右边的`浏览`逐级点选。弹窗里可以一路点进子文件夹、点`C:`/`D:`这类按钮（Linux 上是`/`、`/home`、`/mnt` 这些常用位置）换根位置，地址栏本身也能粘贴路径回车跳转，随时在点选和手填之间切换；选好点`选用这个文件夹`填回设置框，仍然是按回车或点保存才生效。填的是文件时（比如 ffmpeg 路径）列出的是文件，双击即选中并填入。默认不显示以点开头的隐藏项，需要时勾上`显示隐藏项`。 diff --git a/automerge.go b/automerge.go index e03822e..f6496db 100644 --- a/automerge.go +++ b/automerge.go @@ -5,6 +5,7 @@ import ( "bufio" "encoding/json" "errors" + "fmt" "os" "path/filepath" "strings" @@ -117,7 +118,23 @@ func prepareForMerge(ffmpegFile string, segs []recordSegment, jsonlFile string, dir := filepath.Dir(jsonlFile) // 拼接出来的中间文件沿用分段的封装格式，concat 用 -c copy 时容器一致最稳 joined := filepath.Join(dir, chunkTempPrefix+"joined"+filepath.Ext(files[0])) - if err := concatChunks(ffmpegFile, files, joined, dir); err != nil { + 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 @@ -225,8 +242,29 @@ func remapJsonlTimes(srcFile, dstFile string, starts []int64, durs []float64, da return nil } -// 执行自动合并及后续的删除/移动 +// 自动合并入口：把结果写进任务进度；日志措辞仍由内部负责 func autoMerge(uid int, segs []recordSegment, jsonlFile string, danmuStartNs int64) { + err := 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 决定） @@ -235,21 +273,14 @@ func autoMerge(uid int, segs []recordSegment, jsonlFile string, danmuStartNs int jsonlBase := filepath.Base(strings.TrimSuffix(jsonlFile, ".jsonl")) if videoBase != jsonlBase { lPrintWarnf("视频文件（%s）和弹幕文件（%s）不是同一场录制，取消自动合并", segs[0].file, jsonlFile) - return + return errors.New("视频文件和弹幕文件不是同一场录制") } } ffmpegFile := getFFmpeg() if ffmpegFile == "" { lPrintErr("自动合并失败：找不到 ffmpeg，请安装并加入 PATH，或在 web UI 设置里填写 ffmpeg 路径") - return - } - - video, jsonl, offset, cleanup, err := prepareForMerge(ffmpegFile, segs, jsonlFile, danmuStartNs) - defer cleanup() - if err != nil { - lPrintErrf("自动合并失败：%v", err) - return + return errors.New("找不到 ffmpeg，请安装并加入 PATH，或在 web UI 设置里填写 ffmpeg 路径") } // 成片先在本地合成，再落到成片文件夹：成片目录可能是网络盘或外接盘， @@ -257,28 +288,48 @@ func autoMerge(uid int, segs []recordSegment, jsonlFile string, danmuStartNs int mergeDir := mergeSaveDir() if err := os.MkdirAll(mergeDir, 0755); err != nil { lPrintWarnf("创建成片文件夹 %s 失败：%v", mergeDir, err) - return + 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) + } + } + jobStart("自动合并", filepath.Base(outFile), inputBytes, 0) + + 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) - if burnErr == errNoDanmu { + switch burnErr { + case errJobStopped: + return errJobStopped + case errNoDanmu: lPrintWarn("弹幕文件里没有弹幕，跳过自动合并") - return + return errNoDanmu } lPrintErrf("自动合并失败：%v", burnErr) - return + return burnErr } + jobSetPhase(jobPhasePlace) if err := placeOutput(tmpOut, outFile); err != nil { _ = os.Remove(tmpOut) lPrintErrf("自动合并失败：%v", err) - return + return err } lPrintln("自动合并完成：" + outFile) @@ -307,4 +358,5 @@ func autoMerge(uid int, segs []recordSegment, jsonlFile string, danmuStartNs int } } } + return nil } diff --git a/danmuburn.go b/danmuburn.go index f23de40..b4a605c 100644 --- a/danmuburn.go +++ b/danmuburn.go @@ -64,6 +64,10 @@ const maxDanmuPerChunk = 80 // 分块渲染产生的临时文件前缀 const chunkTempPrefix = "acfunlive_chunk_" +// 预估成片体积时给音频留的码率（Kbps）：分块时音频固定重编码成 192k， +// 单块时是直接拷流，这里统一按 192 估，反正渲染过 5% 后会被实测外推值顶掉 +const estimateAudioKbps = 192 + // 一个待渲染的弹幕块 type danmuChunk struct { start float64 // 在视频时间轴上的起点（秒） @@ -300,7 +304,13 @@ func burnDanmu(jsonlFile, videoFile, outFile string, offset float64) error { } } 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 反查一次 @@ -311,6 +321,7 @@ func burnDanmu(jsonlFile, videoFile, outFile string, offset float64) error { lPrintf("正在获取头像：已处理 %d/%d 个用户", i+1, len(uids)) } } + jobAvatars(i+1, len(uids)) if url == "" { continue } @@ -348,6 +359,21 @@ func burnDanmu(jsonlFile, videoFile, outFile string, offset float64) error { } 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) @@ -518,7 +544,10 @@ func (c *burnContext) burnSingle(records []danmuRecord) error { } defer os.Remove(assPath) - return c.renderChunk(chunk, assName, c.outFile, false) + // 只有一块，ffmpeg 报的时长就是整片进度 + return c.renderChunk(chunk, assName, c.outFile, false, func(outSec, speed, fps float64, size int64) { + jobRender(outSec, speed, fps, size) + }) } // 分块渲染：每块单独渲染成一个小片，最后用 concat 拼成完整成片 @@ -536,7 +565,18 @@ func (c *burnContext) burnChunked(chunks []danmuChunk) error { }() 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) @@ -546,16 +586,32 @@ func (c *burnContext) burnChunked(chunks []danmuChunk) error { 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)) - if err := c.renderChunk(ch, assName, part, true); err != nil { + 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) } - parts = append(parts, part) + if n := fileSizeOrZero(part); n > 0 { + bytesBase += n + } } + jobSetPhase(jobPhasePlace) lPrintf("正在把 %d 个分块拼接成完整成片", len(parts)) - if err := concatChunks(c.ffmpegFile, parts, c.outFile, c.dir); err != nil { + 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 @@ -633,7 +689,8 @@ func writeChunkAss(path string, meta danmuMeta, chunk danmuChunk, offset float64 // 渲染一个块：把这一块的弹幕（文字层 + 头像层）压进对应的视频片段 // cut 为 true 时表示这是分块渲染，需要按时长切出视频片段并重编码音频 -func (c *burnContext) renderChunk(chunk danmuChunk, assBase, out string, cut bool) error { +// cb 用来回传 ffmpeg 的实时进度（已输出时长 / 倍速 / 帧率 / 已写出字节），可为 nil +func (c *burnContext) renderChunk(chunk danmuChunk, assBase, out string, cut bool, cb ffmpegProgress) error { burnMode := c.videoFile != "" // 统计这一块用到的头像，按首次出现顺序分配输入序号（输入 0 是视频或背景） @@ -743,7 +800,7 @@ func (c *burnContext) renderChunk(chunk danmuChunk, assBase, out string, cut boo // 透明头像层模式固定用 ProRes 4444，不能再追加编码器覆盖 full := append(append([]string{}, args...), out) lPrintln("开始运行 ffmpeg 合成透明头像层，输出到 " + out) - if err := runFFmpeg(c.ffmpegFile, full, c.dir); err != nil { + if err := runFFmpegWith(c.ffmpegFile, full, c.dir, cb); err != nil { return err } return nil @@ -770,7 +827,11 @@ func (c *burnContext) renderChunk(chunk danmuChunk, assBase, out string, cut boo if i == begin { lPrintf("开始运行 ffmpeg 合成（编码器 %s），输出到 %s", enc.name, out) } - if err := runFFmpeg(c.ffmpegFile, full, c.dir); err != nil { + if err := runFFmpegWith(c.ffmpegFile, full, c.dir, cb); err != nil { + // 停止任务时杀掉 ffmpeg 也会让它非 0 退出，这时候不该再换编码器重试 + if jobStopRequested() { + return errJobStopped + } lastErr = err if i+1 < len(encoders) { lPrintWarnf("编码器 %s 运行失败，回退到 %s：%v", enc.name, encoders[i+1].name, err) @@ -859,16 +920,97 @@ func buildEncoders(bitrate int, ffmpegFile string) []videoEncoder { } // 运行 ffmpeg，输出直接透传到控制台 +// ffmpeg 的进度回调：已输出时长（秒）、实时倍速、帧率、已写出字节 +type ffmpegProgress func(outSec, speed, fps float64, totalSize int64) + func runFFmpeg(ffmpegFile string, args []string, dir string) error { - cmd := exec.Command(ffmpegFile, args...) - cmd.Dir = dir - cmd.Stdout = os.Stdout - cmd.Stderr = os.Stderr - return cmd.Run() + return runFFmpegWith(ffmpegFile, args, dir, nil) } -// 用 concat demuxer 把各分块的成片拼接起来 -func concatChunks(ffmpegFile string, files []string, out, dir string) error { +// 跑一次 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 + cmd.Stderr = os.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 cmd.Wait() + } + cmd.Stdout = os.Stdout + if err := cmd.Start(); err != nil { + return err + } + jobAttach(cmd) + defer jobDetach(cmd) + return cmd.Wait() +} + +// 解析 -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 { @@ -881,12 +1023,20 @@ func concatChunks(ffmpegFile string, files []string, out, dir string) error { } defer os.Remove(listFile) - if err := runFFmpeg(ffmpegFile, []string{"-y", "-f", "concat", "-safe", "0", "-i", listFile, "-c", "copy", out}, dir); err != nil { + if err := runFFmpegWith(ffmpegFile, []string{"-y", "-f", "concat", "-safe", "0", "-i", listFile, "-c", "copy", out}, dir, cb); err != nil { 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 +} + // 把秒格式化成 ffmpeg 参数，保留毫秒 func formatSeconds(sec float64) string { return strconv.FormatFloat(sec, 'f', 3, 64) diff --git a/doc/webapi.md b/doc/webapi.md index b5cf6d0..26326d5 100644 --- a/doc/webapi.md +++ b/doc/webapi.md @@ -103,6 +103,24 @@ web UI（`-webui`，端口看`webPort`）用的是另一套`/api/`接口，全 合并状态通过`GET /api/status`查看：`mergeBusy`（是否在跑）、`mergeOutName`（当前成片名）、`mergeErr`（失败原因）、`mergeDoneAt`（上次完成时间）。合并期间成片会出现在`GET /api/files`里且`inUse`为 true，此时`POST /api/file/remove`会返回 409 拒绝删除。 +`GET /api/job` 返回转换任务（自动合并和手动合并共用一份）的实时进度，页面每秒轮一次它来画监控分区顶部的进度卡片和全局细进度条；空闲时返回`active: false`，但会保留上一次任务的结束信息（`doneAt`、`message`、`err`）方便界面把结果多显示一会儿。字段： + +| 字段 | 含义 | +| --- | --- | +| `kind` / `outName` | 任务类型（`手动合并` / `自动合并`）和成片文件名 | +| `phases` / `phaseIdx` / `phasePercent` | 阶段名列表、当前阶段序号、当前阶段内的完成比例 | +| `percent` | 总进度 0~100，按各阶段权重折算（渲染占 0.75） | +| `paused` / `stopping` / `pausable` | 是否已暂停 / 是否正在停止 / 现在能不能暂停 | +| `speed` / `fps` / `doneSec` / `totalSec` | 渲染倍速、帧率、已渲染时长、视频总时长 | +| `chunkDone` / `chunkTotal` | 分块渲染的进度（`第 3/12 块`） | +| `avatarDone` / `avatarTotal` | 获取头像阶段的用户数进度 | +| `inputBytes` / `estOutBytes` / `estSource` / `outBytes` | 源片总体积、预计成片体积及其来源（`bitrate` 按码率估 / `measured` 按已渲染部分外推）、已写出字节 | +| `etaSec` | 预计剩余秒数，算不出来时为 0 | +| `samples` | 渲染速度折线的采样点（约每秒一个，最多 120 个） | +| `message` / `err` / `doneAt` | 结束提示（`已完成` / `已停止` / `失败`）、失败原因、结束时刻 | + +`POST /api/job/control` 控制当前任务，请求体`{"action":"pause"}`、`{"action":"resume"}`或`{"action":"stop"}`，成功返回`{"ok":true}`。没有任务在跑时返回 409，未知的 action 返回 400。暂停会挂起 ffmpeg 进程（Windows 用`NtSuspendProcess`、Linux 用`SIGSTOP`），CPU 占用归零；不在 ffmpeg 环节时暂停和停止会在下一个检查点生效。**停止会丢弃已经渲染的部分**：不落成片、清掉分块临时文件，任务以`message: "已停止"`结束（`err`为`任务已被手动停止`，界面据此和真失败区分开）。 + `POST /api/config`里的`danmuFontSize`控制合并时弹幕文字的字号（像素），头像按同一比例缩放，填 0 表示按画面分辨率自动：字号取画面**短边**的 5.56%（横屏取高、竖屏取宽），所以 1920x1080 是 60、1280x720 是 40、720x1280 也是 40 —— 竖屏画面窄，若按高度算会得到 71，一条弹幕就铺满整行。显式填写则按填的像素值直接用。渲染前会按最终字号重排弹道：从上游原版`.ass`还原出来的弹幕只按内容宽度占位，而渲染用的是「昵称(uid)：内容 + 头像」的完整宽度，占位时间更长，照搬原弹道会追尾叠字；重排按完整宽度重新分配，同一条弹道上不会再有两条弹幕叠在一起。 `mergeBitrate`填 0（默认）时按「源片码率 + 500 Kbps」自动计算：先探测视频流的码率，加 500 留给烧进去的弹幕；探测不到就退回质量模式。填大于 0 的值则按该码率固定输出。编码器优先硬件 HEVC（`hevc_qsv`，同画质比 H.264 省约三成体积，mp4 里打`hvc1`标记），不支持时回退`h264_qsv`、`libx264`，日志里会打印实际用的那一个。 diff --git a/jobctl_unix.go b/jobctl_unix.go new file mode 100644 index 0000000..ac79d02 --- /dev/null +++ b/jobctl_unix.go @@ -0,0 +1,23 @@ +//go:build !windows + +package main + +import ( + "os" + "syscall" +) + +// 挂起进程：SIGSTOP，进程立刻停止占用 CPU，SIGCONT 后从原地继续 +func suspendProcess(p *os.Process) error { + if p == nil { + return nil + } + return syscall.Kill(p.Pid, syscall.SIGSTOP) +} + +func resumeProcess(p *os.Process) error { + if p == nil { + return nil + } + return syscall.Kill(p.Pid, syscall.SIGCONT) +} diff --git a/jobctl_windows.go b/jobctl_windows.go new file mode 100644 index 0000000..d6f291d --- /dev/null +++ b/jobctl_windows.go @@ -0,0 +1,44 @@ +//go:build windows + +package main + +import ( + "fmt" + "os" + "syscall" +) + +// PROCESS_SUSPEND_RESUME：挂起 / 恢复进程所需的访问权限 +const processSuspendResume = 0x0800 + +var ( + ntdllDLL = syscall.NewLazyDLL("ntdll.dll") + procSuspend = ntdllDLL.NewProc("NtSuspendProcess") + procResume = ntdllDLL.NewProc("NtResumeProcess") +) + +// 挂起进程。Windows 没有公开的 Win32 API，用 ntdll 的 NtSuspendProcess， +// 效果等同任务管理器里的「挂起」：进程不再占用 CPU，恢复后从原地接着跑。 +func suspendProcess(p *os.Process) error { + return ntProcessCall(procSuspend, "NtSuspendProcess", p) +} + +func resumeProcess(p *os.Process) error { + return ntProcessCall(procResume, "NtResumeProcess", p) +} + +func ntProcessCall(proc *syscall.LazyProc, name string, p *os.Process) error { + if p == nil { + return nil + } + h, err := syscall.OpenProcess(processSuspendResume, false, uint32(p.Pid)) + if err != nil { + return fmt.Errorf("打开进程 %d 失败：%v", p.Pid, err) + } + defer syscall.CloseHandle(h) + // NTSTATUS 为 0 表示成功 + if ret, _, _ := proc.Call(uintptr(h)); ret != 0 { + return fmt.Errorf("%s 失败，NTSTATUS=0x%x", name, ret) + } + return nil +} diff --git a/manualmerge.go b/manualmerge.go index e7b380f..64f07a6 100644 --- a/manualmerge.go +++ b/manualmerge.go @@ -459,13 +459,20 @@ func buildMergePlan(req mergeRequest) (mergePlan, error) { return plan, nil } -// 真正执行合并，进度直接打到运行日志里 +// 真正执行合并，进度既打到运行日志里，也同步给 web UI 的任务卡片 func runManualMerge(plan mergePlan) error { ffmpegFile := getFFmpeg() if ffmpegFile == "" { return errors.New("找不到 ffmpeg，请先在设置里填写 ffmpeg 路径") } + // 先把任务的输入大小和总时长登记下来，界面在 ffmpeg 跑起来之前就有东西可显示 + inputBytes := int64(0) + for _, v := range plan.info.Videos { + inputBytes += v.Size + } + jobStart("手动合并", plan.info.OutName, inputBytes, plan.info.Duration) + src, err := prepareDanmuSource(plan.danmu, true) if err != nil { return err @@ -501,7 +508,6 @@ func runManualMerge(plan mergePlan) error { if err != nil { return err } - offset, note := autoOffset, "按文件名里的录制时间自动推算" if plan.offset != nil { offset, note = *plan.offset, "按你填写的偏移" @@ -532,6 +538,7 @@ func runManualMerge(plan mergePlan) error { _ = os.Remove(tmpOut) return burnErr } + jobSetPhase(jobPhasePlace) if err := placeOutput(tmpOut, plan.out); err != nil { _ = os.Remove(tmpOut) return err @@ -601,18 +608,30 @@ func mergeHandler(w http.ResponseWriter, r *http.Request) { mergeRun.doneAt = time.Now().Unix() mergeRun.Unlock() }() - if err := runManualMerge(plan); err != nil { - msg := err.Error() - if err == errNoDanmu { - msg = "弹幕文件里没有弹幕" - lPrintWarn("弹幕文件里没有弹幕，取消手动合并") - } else { - lPrintErrf("手动合并失败：%v", err) - } - mergeRun.Lock() - mergeRun.lastErr = msg - mergeRun.Unlock() + err := runManualMerge(plan) + // 任务卡片挂在这条链路上，无论成败都要收尾，否则进度会一直留在界面上 + jobFinish(jobErrText(err)) + if err == nil { + return } + msg := err.Error() + switch err { + case errNoDanmu: + msg = "弹幕文件里没有弹幕" + lPrintWarn("弹幕文件里没有弹幕，取消手动合并") + case errJobStopped: + // 手动停止不是失败，别让界面弹「合并失败」 + lPrintWarn("手动合并已被手动停止") + mergeRun.Lock() + mergeRun.lastErr = "" + mergeRun.Unlock() + return + default: + lPrintErrf("手动合并失败：%v", err) + } + mergeRun.Lock() + mergeRun.lastErr = msg + mergeRun.Unlock() }) w.Header().Set("Content-Type", "application/json") diff --git a/progress.go b/progress.go new file mode 100644 index 0000000..bf6f6c3 --- /dev/null +++ b/progress.go @@ -0,0 +1,517 @@ +// 转换任务（合并 / 烧录弹幕）的实时进度与暂停、停止控制。 +// 进度只保留一份全局状态：同一时间只允许一个合并任务在跑（mergeRun 已经拦住了）。 +package main + +import ( + "encoding/json" + "errors" + "math" + "net/http" + "os/exec" + "sync" + "time" +) + +// 任务阶段，前端按这个顺序画步骤条 +var jobPhaseNames = []string{"准备", "分段拼接", "获取头像", "画面渲染", "拼接与落盘"} + +// 阶段序号，和 jobPhaseNames 一一对应 +const ( + jobPhasePrepare = iota // 准备（读弹幕、探测视频、生成字幕） + jobPhaseJoin // 分段拼接（断流分成多段时才有事做） + jobPhaseAvatar // 获取头像 + jobPhaseRender // 画面渲染（最耗时） + jobPhasePlace // 拼接分块与落盘 +) + +// 各阶段在总进度里占的权重（加起来是 1）。渲染最耗时，权重给得最多； +// 其余阶段按经验估：拼接分段与头像下载是分钟级，准备和落盘是秒级。 +var jobPhaseWeights = []float64{0.02, 0.08, 0.10, 0.75, 0.05} + +// 渲染进度的采样间隔与折线最多保留的点数（1 秒 1 个，够画两分钟） +const ( + jobSampleInterval = time.Second + jobMaxSamples = 120 +) + +// 任务被手动停止时各层返回这个错误：调用方据此清理临时文件，不再回退其它编码器 +var errJobStopped = errors.New("任务已被手动停止") + +// 速度折线上的一个点 +type jobSample struct { + T float64 `json:"t"` // 任务开始后的秒数 + Speed float64 `json:"speed"` // 当时的渲染倍速 +} + +// 给前端的任务快照 +type jobInfo struct { + Active bool `json:"active"` + Kind string `json:"kind"` // 手动合并 / 自动合并 + OutName string `json:"outName"` // 成片文件名 + PhaseIdx int `json:"phaseIdx"` // 当前阶段序号，-1 表示已全部完成 + Phases []string `json:"phases"` // 阶段名列表 + PhasePercent float64 `json:"phasePercent"` + Percent float64 `json:"percent"` // 总进度 0~100 + Paused bool `json:"paused"` + Stopping bool `json:"stopping"` + Speed float64 `json:"speed"` // 渲染倍速，例如 1.37 + FPS float64 `json:"fps"` + DoneSec float64 `json:"doneSec"` // 已渲染时长 + TotalSec float64 `json:"totalSec"` // 视频总时长 + ChunkDone int `json:"chunkDone"` + ChunkTotal int `json:"chunkTotal"` + AvatarDone int `json:"avatarDone"` + AvatarTotal int `json:"avatarTotal"` + InputBytes int64 `json:"inputBytes"` // 源视频总大小 + EstOutBytes int64 `json:"estOutBytes"` // 预计成片大小 + EstSource string `json:"estSource"` // 预计值的来源：bitrate / measured + OutBytes int64 `json:"outBytes"` // 已经写出多少字节 + ETASec float64 `json:"etaSec"` // 预计剩余秒数，0 表示算不出来 + ElapsedSec float64 `json:"elapsedSec"` // 已用时间 + Samples []jobSample `json:"samples"` // 速度折线 + Message string `json:"message"` // 结束后的提示（完成 / 失败原因） + Err string `json:"err"` // 失败原因，成功为空 + DoneAt int64 `json:"doneAt"` // 结束时刻（Unix 秒） + Pausable bool `json:"pausable"` // 现在是否能暂停 +} + +// 全局唯一的任务状态 +type jobState struct { + sync.Mutex + active bool + kind string + outName string + phaseIdx int + phasePct float64 + started time.Time + paused bool + stopped bool + proc *exec.Cmd // 当前在跑的 ffmpeg，暂停/停止都作用在它身上 + + totalSec float64 + doneSec float64 + speed float64 + fps float64 + + chunkDone int + chunkTotal int + avatarDone int + avatarTotal int + + inputBytes int64 + // 按目标码率算出来的预计成片大小 + planOutBytes int64 + // 按「已写出字节 ÷ 已渲染比例」外推出来的预计成片大小，渲染到 5% 后才有值 + liveOutBytes int64 + outBytes int64 + + samples []jobSample + lastSample time.Time + + errText string + message string + doneAt int64 +} + +var job jobState + +// 开始一个新任务。kind 是任务类型，totalSec 是视频总时长（拿不到填 0） +func jobStart(kind, outName string, inputBytes int64, totalSec float64) { + job.Lock() + defer job.Unlock() + job.active = true + job.kind = kind + job.outName = outName + job.phaseIdx = 0 + job.phasePct = 0 + job.started = time.Now() + job.paused = false + job.stopped = false + job.proc = nil + job.totalSec = totalSec + job.doneSec = 0 + job.speed = 0 + job.fps = 0 + job.chunkDone, job.chunkTotal = 0, 0 + job.avatarDone, job.avatarTotal = 0, 0 + job.inputBytes = inputBytes + job.planOutBytes = 0 + job.liveOutBytes = 0 + job.outBytes = 0 + job.samples = nil + job.lastSample = time.Time{} + job.errText = "" + job.message = "" + job.doneAt = 0 +} + +// 任务结束。errText 为空表示成功，errJobStopped.Error() 表示被手动停止 +func jobFinish(errText string) { + job.Lock() + defer job.Unlock() + job.active = false + job.paused = false + job.stopped = false + job.proc = nil + job.speed = 0 + job.errText = errText + job.doneAt = time.Now().Unix() + switch errText { + case "": + job.phaseIdx = len(jobPhaseNames) + job.phasePct = 1 + job.message = "已完成" + case errJobStopped.Error(): + job.message = "已停止" + default: + job.message = "失败" + } +} + +// 任务是否正在跑 +func jobActive() bool { + job.Lock() + defer job.Unlock() + return job.active +} + +// 是否已经被要求停止 +func jobStopRequested() bool { + job.Lock() + defer job.Unlock() + return job.stopped +} + +// 各阶段在进入下一步前调用这个「闸门」：暂停时在这里等，被停止时立刻返回错误。 +// 放在循环体里，这样下载头像、逐块渲染这些阶段都能被暂停和停止。 +func jobGate() error { + for { + job.Lock() + stopped, paused := job.stopped, job.paused + job.Unlock() + if stopped { + return errJobStopped + } + if !paused { + return nil + } + time.Sleep(120 * time.Millisecond) + } +} + +// 切换阶段。序号只往前走，重复设置同一阶段不会把进度打回 0 +func jobSetPhase(idx int) { + job.Lock() + defer job.Unlock() + if !job.active { + return + } + if idx > job.phaseIdx { + job.phaseIdx = idx + job.phasePct = 0 + } +} + +// 直接设置当前阶段的完成比例（0~1），只取更大值，避免折返 +func jobSetPhasePercent(p float64) { + job.Lock() + defer job.Unlock() + if !job.active { + return + } + if p > job.phasePct { + job.phasePct = clamp01(p) + } +} + +// 渲染进度：doneSec 是累计已渲染的秒数（分块时已折算成整片时间轴）， +// cumBytes 是累计已写出的字节数（分块时是「已完成块 + 当前块」） +func jobRender(doneSec, speed, fps float64, cumBytes int64) { + job.Lock() + defer job.Unlock() + if !job.active { + return + } + if doneSec > job.doneSec { + job.doneSec = doneSec + } + if speed > 0 { + job.speed = speed + } + if fps > 0 { + job.fps = fps + } + if cumBytes > job.outBytes { + job.outBytes = cumBytes + } + ratio := 0.0 + if job.totalSec > 0 { + ratio = clamp01(job.doneSec / job.totalSec) + } + if ratio > job.phasePct { + job.phasePct = ratio + } + // 渲染到 5% 之后，用「已写出 ÷ 已完成比例」外推整片体积，比按码率估的准 + if ratio >= 0.05 && job.outBytes > 0 { + if est := int64(float64(job.outBytes) / ratio); est > 0 { + job.liveOutBytes = est + } + } + // 速度折线：最多 1 秒一个点 + now := time.Now() + if job.lastSample.IsZero() || now.Sub(job.lastSample) >= jobSampleInterval { + job.lastSample = now + job.samples = append(job.samples, jobSample{T: now.Sub(job.started).Seconds(), Speed: speed}) + if len(job.samples) > jobMaxSamples { + job.samples = append([]jobSample(nil), job.samples[len(job.samples)-jobMaxSamples:]...) + } + } +} + +// 已分块渲染的块数，给界面显示「第 3/12 块」 +func jobChunks(done, total int) { + job.Lock() + defer job.Unlock() + if !job.active { + return + } + job.chunkDone, job.chunkTotal = done, total +} + +// 头像下载进度（按用户数） +func jobAvatars(done, total int) { + job.Lock() + defer job.Unlock() + if !job.active { + return + } + job.avatarDone, job.avatarTotal = done, total + if total > 0 { + p := float64(done) / float64(total) + if p > job.phasePct { + job.phasePct = clamp01(p) + } + } +} + +// 设置任务总时长（视频秒数），ETA 和渲染百分比都靠它 +func jobSetTotal(sec float64) { + job.Lock() + defer job.Unlock() + if !job.active { + return + } + job.totalSec = sec +} + +// 按目标码率算出来的预计成片大小（byte） +func jobPlanSize(bytes int64) { + job.Lock() + defer job.Unlock() + if !job.active { + return + } + job.planOutBytes = bytes +} + +// 把当前 ffmpeg 进程注册进来，暂停/停止才作用得到。 +// 如果注册时任务已经处于暂停状态，直接把它挂起（在启动前点了暂停的情况） +func jobAttach(cmd *exec.Cmd) { + job.Lock() + job.proc = cmd + paused := job.paused + job.Unlock() + if paused && cmd != nil && cmd.Process != nil { + _ = suspendProcess(cmd.Process) + } +} + +func jobDetach(cmd *exec.Cmd) { + job.Lock() + if job.proc == cmd { + job.proc = nil + } + job.Unlock() +} + +// 暂停：挂起当前 ffmpeg 进程；不在 ffmpeg 阶段时靠 jobGate 在下一步生效 +func jobPause() error { + job.Lock() + if !job.active { + job.Unlock() + return errors.New("当前没有正在运行的合并任务") + } + if job.paused { + job.Unlock() + return nil + } + job.paused = true + cmd := job.proc + job.Unlock() + if cmd != nil && cmd.Process != nil { + return suspendProcess(cmd.Process) + } + return nil +} + +// 恢复 +func jobResume() error { + job.Lock() + if !job.active { + job.Unlock() + return errors.New("当前没有正在运行的合并任务") + } + if !job.paused { + job.Unlock() + return nil + } + job.paused = false + cmd := job.proc + job.Unlock() + if cmd != nil && cmd.Process != nil { + return resumeProcess(cmd.Process) + } + return nil +} + +// 停止：杀掉当前 ffmpeg，并置位让各层在下一个闸门返回 errJobStopped +func jobStop() error { + job.Lock() + if !job.active { + job.Unlock() + return errors.New("当前没有正在运行的合并任务") + } + job.stopped = true + job.paused = false + cmd := job.proc + job.Unlock() + if cmd != nil && cmd.Process != nil { + _ = cmd.Process.Kill() + } + return nil +} + +func clamp01(v float64) float64 { + if v < 0 { + return 0 + } + if v > 1 { + return 1 + } + return v +} + +// 总进度：已走完的阶段按权重全算，当前阶段按阶段内比例折算 +func (j *jobState) percentLocked() float64 { + p := 0.0 + for i, w := range jobPhaseWeights { + switch { + case i < j.phaseIdx: + p += w + case i == j.phaseIdx: + p += w * clamp01(j.phasePct) + } + } + return math.Min(p*100, 100) +} + +// 预计剩余时间：渲染阶段用「剩余时长 ÷ 倍速」，其它阶段按已用时间线性外推 +func (j *jobState) etaLocked(now time.Time) float64 { + if !j.active { + return 0 + } + if j.phaseIdx == jobPhaseRender && j.speed > 0 && j.totalSec > j.doneSec { + return (j.totalSec - j.doneSec) / j.speed + } + pct := j.percentLocked() + if pct <= 0.5 { + return 0 + } + return now.Sub(j.started).Seconds() * (100 - pct) / pct +} + +// 给页面的快照 +func jobSnapshot() jobInfo { + job.Lock() + defer job.Unlock() + + info := jobInfo{ + Active: job.active, + Kind: job.kind, + OutName: job.outName, + PhaseIdx: job.phaseIdx, + Phases: jobPhaseNames, + Paused: job.paused, + Stopping: job.stopped, + Speed: job.speed, + FPS: job.fps, + DoneSec: job.doneSec, + TotalSec: job.totalSec, + ChunkDone: job.chunkDone, + ChunkTotal: job.chunkTotal, + AvatarDone: job.avatarDone, + AvatarTotal: job.avatarTotal, + InputBytes: job.inputBytes, + OutBytes: job.outBytes, + Message: job.message, + Err: job.errText, + DoneAt: job.doneAt, + } + info.PhasePercent = clamp01(job.phasePct) * 100 + info.Percent = job.percentLocked() + // 预计成片大小：有实测外推就用实测，否则用按码率估的 + info.EstOutBytes, info.EstSource = job.planOutBytes, "bitrate" + if job.liveOutBytes > 0 { + info.EstOutBytes, info.EstSource = job.liveOutBytes, "measured" + } + now := time.Now() + info.ETASec = job.etaLocked(now) + if job.active { + info.ElapsedSec = now.Sub(job.started).Seconds() + } else if job.doneAt > 0 && !job.started.IsZero() { + info.ElapsedSec = time.Unix(job.doneAt, 0).Sub(job.started).Seconds() + } + if len(job.samples) > 0 { + info.Samples = append([]jobSample(nil), job.samples...) + } + // 暂停靠 jobGate 生效：ffmpeg 在跑时直接挂起进程，不在跑时在下一个闸门处停住 + info.Pausable = job.active + return info +} + +// GET /api/job：当前转换任务的快照，页面轮询它来画进度 +func jobHandler(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(jobSnapshot()); err != nil { + lPrintErrf("输出任务进度失败：%v", err) + } +} + +// POST /api/job/control：暂停 / 恢复 / 停止当前任务 +func jobControlHandler(w http.ResponseWriter, r *http.Request) { + var req struct { + Action string `json:"action"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, "请求不是合法的 json："+err.Error(), http.StatusBadRequest) + return + } + var err error + switch req.Action { + case "pause": + err = jobPause() + case "resume": + err = jobResume() + case "stop": + err = jobStop() + default: + http.Error(w, "不认识的操作："+req.Action, http.StatusBadRequest) + return + } + if err != nil { + http.Error(w, err.Error(), http.StatusConflict) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"ok":true}`)) +} diff --git a/progress_test.go b/progress_test.go new file mode 100644 index 0000000..fe85c52 --- /dev/null +++ b/progress_test.go @@ -0,0 +1,308 @@ +package main + +import ( + "errors" + "strings" + "testing" + "time" +) + +// 阶段权重加起来必须正好是 1，否则总进度永远到不了 100% +func TestJobPhaseWeightsSumToOne(t *testing.T) { + if len(jobPhaseNames) != len(jobPhaseWeights) { + t.Fatalf("阶段名有 %d 个，权重有 %d 个，对不上", len(jobPhaseNames), len(jobPhaseWeights)) + } + sum := 0.0 + for _, w := range jobPhaseWeights { + sum += w + } + if sum < 0.999999 || sum > 1.000001 { + t.Errorf("阶段权重之和应为 1，实际 %v", sum) + } +} + +// 总进度：走过的阶段按权重满算，当前阶段按比例折算 +func TestJobPercentByPhase(t *testing.T) { + jobStart("测试", "x.mp4", 0, 100) + t.Cleanup(func() { jobFinish("") }) + + if got := jobSnapshot().Percent; got != 0 { + t.Errorf("刚开始时总进度应为 0，实际 %v", got) + } + // 走到渲染阶段但一帧没渲染：前面三个阶段（准备 / 分段拼接 / 头像）已满算 + jobSetPhase(jobPhaseRender) + want := (jobPhaseWeights[0] + jobPhaseWeights[1] + jobPhaseWeights[2]) * 100 + if got := jobSnapshot().Percent; got < want-0.001 || got > want+0.001 { + t.Errorf("进入渲染阶段时总进度应为 %v，实际 %v", want, got) + } + // 渲染一半 + jobRender(50, 1.5, 25, 0) + want += jobPhaseWeights[jobPhaseRender] * 50 + if got := jobSnapshot().Percent; got < want-0.001 || got > want+0.001 { + t.Errorf("渲染一半时总进度应为 %v，实际 %v", want, got) + } + // 结束（成功）后必须正好 100% + jobFinish("") + if got := jobSnapshot().Percent; got != 100 { + t.Errorf("完成后总进度应为 100，实际 %v", got) + } +} + +// 阶段序号只往前走，重复设置同一阶段不会把进度打回 0 +func TestJobPhaseOnlyMovesForward(t *testing.T) { + jobStart("测试", "x.mp4", 0, 100) + defer jobFinish("") + jobSetPhase(jobPhaseRender) + jobRender(60, 1, 25, 0) + before := jobSnapshot().Percent + jobSetPhase(jobPhaseAvatar) // 往回设，应该被忽略 + if got := jobSnapshot().Percent; got < before { + t.Errorf("阶段倒退把进度打回去了：%v -> %v", before, got) + } + if idx := jobSnapshot().PhaseIdx; idx != jobPhaseRender { + t.Errorf("阶段应为 %d，实际 %d", jobPhaseRender, idx) + } +} + +// 渲染阶段的 ETA = 剩余时长 ÷ 倍速 +func TestJobEtaFromSpeed(t *testing.T) { + jobStart("测试", "x.mp4", 0, 200) + defer jobFinish("") + jobSetPhase(jobPhaseRender) + jobRender(50, 2.0, 25, 0) + // 还剩 150 秒，2 倍速 → 75 秒 + if got := jobSnapshot().ETASec; got < 74.5 || got > 75.5 { + t.Errorf("ETA 应为约 75 秒，实际 %v", got) + } +} + +// 预计输出大小：先给按码率估的值，渲染过 5% 之后换成按已写出字节外推 +func TestJobEstOutSize(t *testing.T) { + jobStart("测试", "x.mp4", 0, 100) + defer jobFinish("") + jobSetPhase(jobPhaseRender) + + jobPlanSize(5000) + info := jobSnapshot() + if info.EstOutBytes != 5000 || info.EstSource != "bitrate" { + t.Fatalf("还没开始渲染时应按码率估 5000，实际 %d（%s）", info.EstOutBytes, info.EstSource) + } + + // 渲染到 50%，写出 4000 字节 → 外推整片 8000 + jobRender(50, 1, 25, 4000) + info = jobSnapshot() + if info.EstSource != "measured" { + t.Fatalf("渲染过 5%% 后应改用实测外推，实际来源 %s", info.EstSource) + } + if info.EstOutBytes != 8000 { + t.Errorf("外推应为 8000，实际 %d", info.EstOutBytes) + } + + // 比例太小的时候不急着外推，免得开头抖动 + jobStart("测试", "x.mp4", 0, 100) + jobSetPhase(jobPhaseRender) + jobPlanSize(5000) + jobRender(1, 1, 25, 100) + if got := jobSnapshot(); got.EstSource != "bitrate" { + t.Errorf("渲染不到 5%% 时不该改用外推，实际来源 %s", got.EstSource) + } +} + +// 速度折线：最多保留固定点数，且按秒采样 +func TestJobSamples(t *testing.T) { + jobStart("测试", "x.mp4", 0, 100) + defer jobFinish("") + jobSetPhase(jobPhaseRender) + // 同一秒内连着报多次只留一个点 + jobRender(1, 1.0, 25, 0) + jobRender(2, 1.1, 25, 0) + jobRender(3, 1.2, 25, 0) + if n := len(jobSnapshot().Samples); n != 1 { + t.Errorf("同一秒内的多次上报应只留 1 个采样点，实际 %d", n) + } + // 攒够 1 秒之后再报就会多一个点 + job.Lock() + job.lastSample = time.Now().Add(-2 * time.Second) + job.Unlock() + jobRender(4, 1.3, 25, 0) + samples := jobSnapshot().Samples + if len(samples) != 2 { + t.Fatalf("应有两个采样点，实际 %d", len(samples)) + } + if samples[1].Speed != 1.3 { + t.Errorf("第二个采样点的速度应为 1.3，实际 %v", samples[1].Speed) + } +} + +// 暂停 / 恢复 / 停止的状态流转；停止后闸门必须让上层拿到 errJobStopped +func TestJobPauseStopGate(t *testing.T) { + jobStart("测试", "x.mp4", 0, 100) + defer jobFinish("") + + if err := jobPause(); err != nil { + t.Fatalf("暂停失败：%v", err) + } + if !jobSnapshot().Paused { + t.Fatal("暂停后状态里应为 paused") + } + // 暂停期间闸门会一直等，放到 goroutine 里，再恢复让它出来 + done := make(chan error, 1) + go func() { done <- jobGate() }() + select { + case err := <-done: + t.Fatalf("暂停期间闸门不该放行，却返回了 %v", err) + case <-time.After(200 * time.Millisecond): + } + if err := jobResume(); err != nil { + t.Fatalf("恢复失败：%v", err) + } + select { + case err := <-done: + if err != nil { + t.Fatalf("恢复后闸门应放行，实际返回 %v", err) + } + case <-time.After(time.Second): + t.Fatal("恢复后闸门没有放行") + } + + if err := jobStop(); err != nil { + t.Fatalf("停止失败：%v", err) + } + if err := jobGate(); !errors.Is(err, errJobStopped) { + t.Errorf("停止后闸门应返回 errJobStopped，实际 %v", err) + } + // 停止不是「失败」，界面要能区分开 + jobFinish(errJobStopped.Error()) + info := jobSnapshot() + if info.Active { + t.Error("结束后不该还是 active") + } + if info.Message != "已停止" || info.Err != errJobStopped.Error() { + t.Errorf("停止后的提示应为「已停止」，实际 message=%q err=%q", info.Message, info.Err) + } + if jobErrText(nil) != "" || jobErrText(errNoDanmu) != "" { + t.Error("正常结束和「没有弹幕」都不该算失败") + } + if jobErrText(errJobStopped) != errJobStopped.Error() { + t.Error("停止的原因要原样传给界面") + } + if jobErrText(errors.New("炸了")) != "炸了" { + t.Error("其它错误要原样传给界面") + } +} + +// 没有任务时控制接口要明确报错，而不是静默成功 +func TestJobControlWithoutTask(t *testing.T) { + jobFinish("") + if err := jobPause(); err == nil { + t.Error("没有任务时暂停应该报错") + } + if err := jobPause(); err == nil { + t.Error("没有任务时恢复应该报错") + } + if err := jobStop(); err == nil { + t.Error("没有任务时停止应该报错") + } +} + +// ffmpeg 的 speed 字段形如 "1.37x"，暂停或还没开始时是 "N/A" +func TestParseFFmpegSpeed(t *testing.T) { + cases := map[string]float64{ + "1.37x": 1.37, + "2x": 2, + "N/A": 0, + "": 0, + "-1x": 0, + } + for in, want := range cases { + if got := parseFFmpegSpeed(in); got != want { + t.Errorf("parseFFmpegSpeed(%q) 应为 %v，实际 %v", in, want, got) + } + } +} + +// 解析 -progress pipe:1 的输出，只在 progress= 行回调一次 +func TestScanFFmpegProgress(t *testing.T) { + // out_time_us 和 out_time_ms 都是微秒（ffmpeg 的历史遗留） + text := strings.Join([]string{ + "frame=100", + "fps=25.0", + "total_size=1048576", + "out_time_us=4000000", + "out_time_ms=4000000", + "out_time=00:00:04.000000", + "speed=1.37x", + "progress=continue", + "frame=200", + "out_time_us=8000000", + "total_size=2097152", + "speed=N/A", + "progress=end", + }, "\n") + + type call struct { + sec, speed, fps float64 + size int64 + } + var got []call + scanFFmpegProgress(strings.NewReader(text), func(sec, speed, fps float64, size int64) { + got = append(got, call{sec, speed, fps, size}) + }) + + if len(got) != 2 { + t.Fatalf("应该回调 2 次，实际 %d 次：%+v", len(got), got) + } + if got[0].sec != 4 || got[0].speed != 1.37 || got[0].fps != 25 || got[0].size != 1048576 { + t.Errorf("第一次回调不对：%+v", got[0]) + } + if got[1].sec != 8 || got[1].size != 2097152 { + t.Errorf("第二次回调不对：%+v", got[1]) + } + // 暂停时 speed 会是 N/A，这时不该把上一次的倍速顶掉 + if got[1].speed != 0 { + t.Errorf("N/A 应该解析成 0，实际 %v", got[1].speed) + } +} + +// 混着无关行和坏数据也不能崩 +func TestScanFFmpegProgressTolerant(t *testing.T) { + text := "乱七八糟\nspeed=abc\nprogress=continue\nout_time_us=不是数字\nprogress=end\n" + n := 0 + scanFFmpegProgress(strings.NewReader(text), func(sec, speed, fps float64, size int64) { + n++ + if sec != 0 || speed != 0 { + t.Errorf("坏数据不该解析出值：sec=%v speed=%v", sec, speed) + } + }) + if n != 2 { + t.Errorf("progress= 行都应回调，期望 2 次，实际 %d", n) + } +} + +// 截图式地看一眼任务快照的字段都填上了 +func TestJobSnapshotFields(t *testing.T) { + jobStart("手动合并", "成片.mp4", 123456, 3600) + defer jobFinish("") + jobSetPhase(jobPhaseAvatar) + jobAvatars(3, 10) + jobSetTotal(3600) + info := jobSnapshot() + if !info.Active || info.Kind != "手动合并" || info.OutName != "成片.mp4" { + t.Fatalf("基本信息不对：%+v", info) + } + if info.InputBytes != 123456 || info.TotalSec != 3600 { + t.Errorf("输入大小 / 总时长不对：%+v", info) + } + if info.AvatarDone != 3 || info.AvatarTotal != 10 { + t.Errorf("头像进度不对：%+v", info) + } + if len(info.Phases) != len(jobPhaseNames) { + t.Errorf("阶段列表长度应为 %d，实际 %d", len(jobPhaseNames), len(info.Phases)) + } + if info.PhasePercent < 29.9 || info.PhasePercent > 30.1 { + t.Errorf("头像阶段进度应为 30%%，实际 %v", info.PhasePercent) + } + if !info.Pausable { + t.Error("任务在跑的时候应该可暂停") + } +} diff --git a/utils.go b/utils.go index e216540..476a8ba 100644 --- a/utils.go +++ b/utils.go @@ -282,7 +282,12 @@ func placeOutput(tmp, out string) error { // 跨盘时 os.Rename 会失败（Windows 上尤其明确），退回拷贝： // 先拷成 .part 再改名，中途失败也不会在成片目录留下像成品的文件 part := out + partSuffix - if err := copyFile(tmp, part); err != nil { + total := fileSizeOrZero(tmp) + if err := copyFileWith(tmp, part, func(done int64) { + if total > 0 { + jobSetPhasePercent(float64(done) / float64(total)) + } + }); err != nil { _ = os.Remove(part) return fmt.Errorf("拷贝到成片目录 %s 失败：%v", filepath.Dir(out), err) } @@ -296,6 +301,11 @@ func placeOutput(tmp, out string) error { // 拷贝文件，跨盘搬运成片时用 func copyFile(src, dst string) error { + return copyFileWith(src, dst, nil) +} + +// 拷贝文件。onBytes 不为空时按累计已拷贝字节回调，跨盘搬运成片时用来报进度 +func copyFileWith(src, dst string, onBytes func(int64)) error { in, err := os.Open(src) if err != nil { return err @@ -305,9 +315,27 @@ func copyFile(src, dst string) error { if err != nil { return err } - if _, err := io.Copy(out, in); err != nil { - _ = out.Close() - return err + buf := make([]byte, 1<<20) + var done int64 + for { + n, rerr := in.Read(buf) + if n > 0 { + if _, werr := out.Write(buf[:n]); werr != nil { + _ = out.Close() + return werr + } + done += int64(n) + if onBytes != nil { + onBytes(done) + } + } + if rerr == io.EOF { + break + } + if rerr != nil { + _ = out.Close() + return rerr + } } return out.Close() } diff --git a/webui.go b/webui.go index e8be0ae..20b8a5f 100644 --- a/webui.go +++ b/webui.go @@ -204,9 +204,44 @@ const webuiHTML = ` button.ghost,.mask button{padding:8px 12px;font-size:13px} #toast{bottom:14px} } + /* 转换任务进度：监控分区里的卡片 + 页面顶部的全局细进度条 */ + #jobTop{position:fixed;left:0;right:0;top:0;height:3px;z-index:12;pointer-events:none;display:none} + #jobTop.on{display:block} + #jobTop i{display:block;height:100%;width:0;background:var(--accent);transition:width .6s linear} + .jtop{display:flex;align-items:baseline;gap:8px;font-size:12px;color:var(--sub);flex-wrap:wrap} + .jtop b{font-size:15px;color:var(--text);font-variant-numeric:tabular-nums} + .jbar{height:8px;border-radius:999px;background:var(--soft);overflow:hidden;margin:8px 0 2px} + .jbar i{display:block;height:100%;width:0;border-radius:999px;background:var(--accent);transition:width .6s linear} + .jbar.paused i{background:var(--sub)} + /* 阶段步骤条 */ + .steps{display:flex;gap:6px;flex-wrap:wrap;margin-bottom:10px} + .step{display:inline-flex;align-items:center;gap:6px;font-size:11.5px;color:var(--sub);background:var(--soft2);border-radius:999px;padding:3px 10px} + .step .d{width:6px;height:6px;border-radius:50%;background:var(--line);flex:none} + .step.done .d{background:var(--ok)} + .step.on{color:var(--accent);background:var(--accent-soft);font-weight:600} + .step.on .d{background:var(--accent)} + .step.paused{color:var(--bad)} + .step.paused .d{background:var(--bad)} + /* 指标网格 */ + .jgrid{display:grid;grid-template-columns:repeat(auto-fit,minmax(108px,1fr));gap:8px 14px;margin-top:10px} + .jcell{min-width:0} + .jcell .k{font-size:11px;color:var(--sub);white-space:nowrap;overflow:hidden;text-overflow:ellipsis} + .jcell .v{font-size:13px;font-variant-numeric:tabular-nums;white-space:nowrap;overflow:hidden;text-overflow:ellipsis} + /* 输入 / 预计输出体积对比条，两段按较大值等比缩放 */ + .vbar{display:flex;height:8px;border-radius:999px;overflow:hidden;background:var(--soft);margin-top:5px} + .vbar span{height:100%} + .vbar .vin{background:var(--sub)} + .vbar .vout{background:var(--accent)} + .vlg{display:flex;gap:14px;font-size:11.5px;color:var(--sub);margin-top:4px} + .vlg i{display:inline-block;width:8px;height:8px;border-radius:2px;margin-right:5px} + /* 速度折线：颜色走 CSS 变量，属性里写 var() 在部分浏览器不生效 */ + .spark{display:block;width:100%;height:48px;margin-top:10px;border-radius:var(--radius-s);background:var(--soft2)} + .spark polyline{fill:none;stroke:var(--accent);stroke-width:1.6;stroke-linejoin:round;stroke-linecap:round;vector-effect:non-scaling-stroke} + .spark text{font-size:9px;fill:var(--sub)} +
@@ -232,6 +267,35 @@ const webuiHTML = `
+

添加主播

填 uid 后默认同时开启弹幕和录播
@@ -1424,6 +1488,191 @@ showTab(location.hash.slice(1) || localStorage.getItem("acfunlive_tab") || "mon" authed = true; refresh(); setInterval(refresh, 3000); + +// ===== 转换任务进度（进度、阶段、预计剩余、体积；暂停 / 停止）===== +// 结束之后还留一会儿，让用户看得到「已完成 / 已停止」的结果 +var JOB_LINGER_MS = 20000; +// 自动任务结束后要播报一下，同一次任务只报一次 +var jobAnnounced = ""; + +function fmtBytes(n){ + if(!n || n <= 0) return "—"; + var u = ["B","KB","MB","GB","TB"], i = 0; + while(n >= 1024 && i < u.length - 1){ n /= 1024; i++; } + return (i === 0 ? n : n.toFixed(n >= 100 ? 0 : 1)) + " " + u[i]; +} +function fmtDur(sec){ + if(!sec || sec <= 0 || !isFinite(sec)) return "—"; + sec = Math.round(sec); + var h = Math.floor(sec/3600), m = Math.floor(sec%3600/60), s = sec%60; + var p = function(x){ return x < 10 ? "0" + x : "" + x; }; + return (h ? h + ":" : "") + p(m) + ":" + p(s); +} +function fmtEta(sec){ + if(!sec || sec <= 0 || !isFinite(sec)) return "剩余时间估算中…"; + sec = Math.round(sec); + if(sec < 60) return "剩余约 " + sec + " 秒"; + var m = Math.floor(sec/60), s = sec%60; + if(m < 60) return "剩余约 " + m + " 分" + (s ? " " + s + " 秒" : ""); + var h = Math.floor(m/60); + return "剩余约 " + h + " 小时 " + (m%60) + " 分"; +} +function renderJob(j){ + var card = document.getElementById("jobCard"); + var topEl = document.getElementById("jobTop"); + var recent = j.doneAt > 0 && (Date.now() - j.doneAt*1000) < JOB_LINGER_MS; + if(!j.active && !recent){ + card.style.display = "none"; + topEl.classList.remove("on"); + return; + } + card.style.display = ""; + + var phases = j.phases || []; + var pct = Math.max(0, Math.min(100, j.percent || 0)); + // 「被手动停止」不是失败，别当成错误报出来 + var failed = !!j.err && j.err !== "任务已被手动停止"; + + document.getElementById("jobOut").textContent = + (j.kind || "") + (j.outName ? " · " + j.outName : ""); + + // 控制按钮只在跑的时候出现；暂停中也允许直接停止（后台杀挂起的进程没问题） + var pb = document.getElementById("jobPause"), sb = document.getElementById("jobStop"); + if(j.active){ + pb.style.display = ""; sb.style.display = ""; + pb.textContent = j.paused ? "继续" : "暂停"; + pb.dataset.act = j.paused ? "resume" : "pause"; + pb.disabled = !!j.stopping; + sb.disabled = !!j.stopping; + sb.title = "中断这次合并，已渲染的部分会被丢弃"; + }else{ + pb.style.display = "none"; + sb.style.display = "none"; + } + + // 阶段步骤条：走过的打勾，当前的按是否暂停上色 + document.getElementById("jobSteps").innerHTML = phases.map(function(name, i){ + var cls = "step"; + if(j.active){ + if(i < j.phaseIdx) cls += " done"; + else if(i === j.phaseIdx) cls += (j.paused ? " paused" : " on"); + }else if(!failed || i < j.phaseIdx){ + cls += " done"; + } + return '' + esc(name) + ''; + }).join(""); + + document.getElementById("jobPct").textContent = pct.toFixed(pct >= 10 ? 0 : 1) + "%"; + var phaseText = ""; + if(!j.active){ + // message 由服务端给（已完成 / 已停止 / 失败），只有真失败才补上原因 + phaseText = j.message || "已完成"; + if(j.err && j.message === "失败") phaseText += "：" + j.err; + }else if(j.paused){ + phaseText = "已暂停（ffmpeg 已挂起，CPU 不再占用）"; + }else if(j.stopping){ + phaseText = "正在停止…"; + }else if(j.phaseIdx >= 0 && j.phaseIdx < phases.length){ + phaseText = "正在" + phases[j.phaseIdx]; + if(j.phaseIdx === 2 && j.avatarTotal > 0) + phaseText += "（" + j.avatarDone + "/" + j.avatarTotal + " 个用户）"; + if(j.phaseIdx === 3 && j.chunkTotal > 1) + phaseText += "（第 " + Math.min(j.chunkDone + 1, j.chunkTotal) + "/" + j.chunkTotal + " 块）"; + } + document.getElementById("jobPhase").textContent = phaseText; + document.getElementById("jobEta").textContent = + (j.active && !j.paused) ? fmtEta(j.etaSec) : ""; + + var bar = document.getElementById("jobBarWrap"); + document.getElementById("jobBar").style.width = pct + "%"; + bar.className = "jbar" + (j.paused ? " paused" : ""); + + var cells = []; + var cell = function(k, v){ cells.push('
' + k + '
' + v + '
'); }; + cell("渲染速度", j.speed > 0 ? j.speed.toFixed(2) + "x" : "—"); + cell("帧率", j.fps > 0 ? j.fps.toFixed(1) + " fps" : "—"); + cell("已处理时长", fmtDur(j.doneSec) + " / " + fmtDur(j.totalSec)); + cell("已用时间", fmtDur(j.elapsedSec)); + cell("已写出", fmtBytes(j.outBytes)); + document.getElementById("jobNums").innerHTML = cells.join(""); + + // 源片 / 预计成片体积对比：两段按较大值等比缩放 + var inB = j.inputBytes || 0, outB = j.estOutBytes || 0; + var vol = document.getElementById("jobVol"); + if(inB > 0 || outB > 0){ + vol.style.display = ""; + var maxB = Math.max(inB, outB, 1); + document.getElementById("jobVolIn").style.width = (inB/maxB*100) + "%"; + document.getElementById("jobVolOut").style.width = (outB/maxB*100) + "%"; + document.getElementById("jobVolInText").textContent = fmtBytes(inB); + document.getElementById("jobVolOutText").textContent = fmtBytes(outB); + document.getElementById("jobVolNote").textContent = outB <= 0 ? "" + : (j.estSource === "measured" ? "（按已渲染部分推算）" : "（按目标码率估算）"); + }else{ + vol.style.display = "none"; + } + + // 渲染速度折线：横轴是任务已用时间，纵轴是倍速 + var samples = j.samples || []; + var wrap = document.getElementById("jobSparkWrap"); + if(samples.length >= 2){ + wrap.style.display = ""; + var W = 300, H = 48, PAD = 5, maxS = 0; + samples.forEach(function(s){ if(s.speed > maxS) maxS = s.speed; }); + if(maxS <= 0) maxS = 1; + var t0 = samples[0].t, span = Math.max(samples[samples.length-1].t - t0, 1); + var pts = samples.map(function(s){ + var x = PAD + (s.t - t0)/span * (W - PAD*2); + var y = H - PAD - (s.speed/maxS) * (H - PAD*2); + return x.toFixed(1) + "," + y.toFixed(1); + }).join(" "); + var last = samples[samples.length-1]; + document.getElementById("jobSpark").innerHTML = + '' + + '峰值 ' + maxS.toFixed(2) + 'x' + + '当前 ' + (last.speed || 0).toFixed(2) + 'x'; + }else{ + wrap.style.display = "none"; + } + + if(j.active){ + topEl.classList.add("on"); + document.getElementById("jobTopBar").style.width = pct + "%"; + }else{ + topEl.classList.remove("on"); + // 结束后播报一次（轮到哪个分区都看得见） + var key = String(j.doneAt); + if(jobAnnounced !== key){ + jobAnnounced = key; + if(failed) showMsg(false, "转换任务失败：" + j.err); + else showMsg(true, "转换任务" + (j.message || "已完成")); + } + } +} +async function jobControl(act){ + if(act === "stop" && !confirm("确定停止当前的合并任务吗？已经渲染的部分会被丢弃，不会生成成片。")) return; + var r = null; + try{ + r = await api("/api/job/control", {method:"POST", headers:{"Content-Type":"application/json"}, body: JSON.stringify({action: act})}); + }catch(e){} + if(!r || !r.ok){ showMsg(false, r ? await r.text() : "操作失败"); return; } + showMsg(true, act === "pause" ? "已暂停" : (act === "resume" ? "已继续" : "正在停止…")); + pollJob(); +} +// 有任务在跑就 1 秒刷一次，空闲时 4 秒 +async function pollJob(){ + var j = null; + if(authed){ + try{ + var r = await api("/api/job"); + if(r.status === 401) showLogin("管理口令已失效，请重新输入"); + else if(r.ok) j = await r.json(); + }catch(e){} + if(j) renderJob(j); + } + setTimeout(pollJob, (j && j.active) ? 1000 : 4000); +} +pollJob(); document.getElementById("tokenInput").addEventListener("keydown", function(e){ if(e.key === "Enter") login(); }); @@ -2117,6 +2366,8 @@ func startWebUI(ctx context.Context) { mux.HandleFunc("POST /api/merge/info", mergeInfoHandler) mux.HandleFunc("GET /api/merge/danmu", mergeDanmuMatchHandler) mux.HandleFunc("POST /api/merge", mergeHandler) + mux.HandleFunc("GET /api/job", jobHandler) + mux.HandleFunc("POST /api/job/control", jobControlHandler) srvCtx, srvCancel := context.WithCancel(ctx) webSrvMu.Lock()