合并任务的可视化进度、预计剩余时间与暂停/停止

用户反馈:合并的进度只体现在日志里(ffmpeg 的 frame=/speed= 行),看不出
百分比、当前环节和还要多久,也没有办法中途暂停或放弃。

1. 新增全局任务状态(progress.go)
   一份状态同时服务自动合并和手动合并,记录阶段、总进度、速度、帧率、
   已处理时长、输入/预计输出/已写出体积、ETA 和速度折线采样点。
   总进度按阶段加权:准备 0.02 / 分段拼接 0.08 / 获取头像 0.10 /
   画面渲染 0.75 / 拼接与落盘 0.05,渲染阶段按「已渲染 ÷ 总时长」折算。
   ETA 在渲染阶段用「剩余时长 ÷ 倍速」,其它阶段按已用时间线性外推。

2. 用 ffmpeg 的 -progress pipe:1 取结构化进度
   runFFmpeg 拆出带回调的版本:解析 out_time_us / speed / fps / total_size,
   分块渲染时把块内时间折算到整片时间轴,并把已完成块的字节数累计进来。
   ffmpeg 自己的统计行仍在 stderr,单独用命令行时不会少东西。
   预计成片体积先按「(视频码率 + 音频 192k)× 时长 ÷ 8」估,渲染过 5%
   之后换成「已写出字节 ÷ 已完成比例」外推(estimateAudioKbps)。

3. 暂停 / 停止
   Windows 用 ntdll 的 NtSuspendProcess / NtResumeProcess(照任务管理器
   的「挂起」语义),Linux 用 SIGSTOP / SIGCONT,放在带 build tag 的
   jobctl_windows.go / jobctl_unix.go 里。
   非 ffmpeg 环节(下载头像、逐块渲染之间)用 jobGate 闸门生效。
   停止 = 杀掉当前 ffmpeg 并置位,各层返回 errJobStopped:不再回退其它
   编码器、不落成片、清掉分块临时文件,任务以「已停止」结束(而不是失败)。

4. 接口
   GET /api/job 给页面轮询任务快照;POST /api/job/control 接受
   pause / resume / stop,没有任务时 409、未知动作 400。

5. 界面
   监控分区顶部新增「转换任务」卡片:阶段步骤条、总进度条、实时指标、
   预计剩余时间、源片与预计成片的体积对比条、渲染速度折线(每秒一个点、
   最多两分钟),以及暂停/继续/停止按钮;页面最顶上再加一条全局细进度条,
   切到哪个分区都看得到。有任务时每秒轮询,空闲时 4 秒。任务结束后卡片
   再停留 20 秒显示结果。

已验证:
- 新增 progress_test.go 12 个用例:阶段权重和为 1、进度按阶段折算、阶段
  不回退、ETA、体积预估与实测外推的切换、折线采样、暂停/停止的闸门语义、
  控制接口在无任务时报错、speed 字段与 -progress 文本的解析。
  测试里把 errJobStopped 和真失败区分开(jobErrText)。
- 端到端(独立实例 51897 + 独立 config,全程没碰仓库里的实例和
  acfunlive.exe):跑通完整合并,进度到 100%、预计体积与成片实际大小
  相差 0.4%;暂停后进度冻结且 ffmpeg 的 CPU 时间 5 秒内 0 增长;
  继续后进度恢复;停止后任务以「已停止」结束、没有半成品成片、
  没有残留临时文件、没有残留 ffmpeg 进程。
- 界面(playwright 驱动系统 Edge):步骤条 5 个阶段且只有一个处于进行中、
  进度条与顶部细条同步、5 项指标(速度/帧率/已处理时长/已用时间/已写出)、
  ETA 文案、体积对比条两段宽度按比例、折线有采样点、界面按钮能暂停/继续/
  停止、结束后按钮隐藏且文案是「已停止」、深色主题与 390 宽移动端无横向溢出。
- 过程中修掉三个真缺陷:停止时残留分块临时文件(part 登记晚于渲染)、
  界面「继续」按钮点不动(onclick 写死了 pause)、停止被当成失败报红
  (现在按 message 区分 已完成/已停止/失败)。
- gofmt / go vet / go build / GOOS=linux go build / go test 全过。
This commit is contained in:
tamakyi
2026-09-26 02:05:47 +08:00
parent cbe5dfa982
commit 7dd2487cb8
11 changed files with 1472 additions and 49 deletions
+517
View File
@@ -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}`))
}