Files
acfunlive-remake/utils.go
T
tamakyi 7dd2487cb8 合并任务的可视化进度、预计剩余时间与暂停/停止
用户反馈:合并的进度只体现在日志里(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 全过。
2026-09-26 02:05:47 +08:00

368 lines
11 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.
// 部分数据和函数定义
package main
import (
"context"
"fmt"
"io"
"log"
"os"
"path/filepath"
"strconv"
"strings"
"sync"
"time"
"github.com/orzogc/acfundanmu"
)
type control int
// 控制信息
const (
startCycle control = iota
stopCycle
quit
)
// 主播的管道信息
type controlMsg struct {
s streamer
c control
liveID string
}
// 主播信息
type streamerInfo struct {
ch chan controlMsg // 控制信息
modify bool
}
// 一段录播视频:直播流意外中断后会重新开一段
type recordSegment struct {
file string // 视频文件路径
startNs int64 // 这一段开始录制的时刻,纳秒 Unix
}
// 直播信息
type liveInfo struct {
LiveID string // 直播 ID
uid int // 主播的 uid
isDanmu bool // 是否正在录制直播弹幕
danmuCancel context.CancelFunc // 用来停止录制弹幕
assFile string // 弹幕文件路径
cfg acfundanmu.SubConfig
isRecording bool // 是否正在录制直播视频
recordCancel context.CancelFunc // 用来停止录制视频(中断当前分段)
recordStop context.CancelFunc // 用来结束整个录制循环,避免停止后又被当成意外中断而重启
ffmpegStdin io.WriteCloser // ffmpeg 的 stdin
recordFile string // 视频文件路径
firstRecordFile string // 第一个分段的视频文件路径(与弹幕文件同名配对,用于自动合并)
recordStartNs int64 // 第一个分段开始录制的时刻,纳秒 Unix(用于计算弹幕与录播的时间偏移)
recordSegments []recordSegment // 本场录播的所有分段(中断重连会产生新的分段),合并时全部参与
danmuDone bool // 弹幕录制是否已经结束
recordDone bool // 视频录制是否已经结束
danmuStarted bool // 本场是否启动过弹幕录制
recordStarted bool // 本场是否启动过视频录制
danmuStopped bool // 弹幕是否被用户手动停止(手动停止不触发自动合并)
recordStopped bool // 视频是否被用户手动停止
mergeStarted bool // 自动合并是否已经启动
doneAt int64 // 最近一次录制结束的时刻,纳秒 Unix(用于延迟清理这条记录)
resumeUntil int64 // 断流后等待直播恢复的截止时刻,纳秒 Unix;0 表示没在等待
}
// streamerInfo 的 map
var sInfoMap struct {
sync.Mutex
info map[int]*streamerInfo
}
// liveInfo 的 map
var lInfoMap struct {
sync.RWMutex
info map[string]liveInfo
}
// 设备 ID
var deviceID string
var (
exeDir string // 运行程序所在文件夹
mainCh chan controlMsg // main() 的管道
mainCtx context.Context // main() 的 ctx
configDir *string // 设置文件所在文件夹
recordDir *string // 弹幕文件保存的文件夹
logger = log.New(os.Stdout, "", log.LstdFlags) // 可以同步输出的 logger
itoa = strconv.Itoa // 将 int 转换为字符串
)
// 检查错误
func checkErr(err error) {
if err != nil {
panic(err)
}
}
// 后台任务的统一入口:兜底 recover,避免单个任务的 panic 终止整个进程
// name 用于日志里指出是哪个任务出错,f 里可以继续做自己的 recover
func safeGo(name string, f func()) {
go func() {
defer func() {
if err := recover(); err != nil {
lPrintErrf("%s 发生错误并已恢复,其它录制任务不受影响:%v", name, err)
}
}()
f()
}()
}
// 尝试运行,三次出错后结束运行
func runThrice(f func() error) error {
var err error
for retry := 0; retry < 3; retry++ {
if err = f(); err != nil {
log.Printf("%v", err)
} else {
return nil
}
time.Sleep(2 * time.Second)
}
return fmt.Errorf("运行三次都出现错误:%v", err)
}
// 获取时间
func getTime() string {
t := time.Now()
return fmt.Sprintf("%d-%02d-%02d %02d-%02d-%02d", t.Year(), t.Month(), t.Day(), t.Hour(), t.Minute(), t.Second())
}
// 打印带时间戳的 log 信息
// lPrintln 定义在 webui.go 里,同时输出到控制台和 web UI 的日志
// 获取时间,按照 log 的时间格式
func getLogTime() string {
t := time.Now()
return fmt.Sprintf("%d/%02d/%02d %02d:%02d:%02d", t.Year(), t.Month(), t.Day(), t.Hour(), t.Minute(), t.Second())
}
// 打印警告信息
func lPrintWarn(msg ...any) {
w := []any{"[WARN]"}
msg = append(w, msg...)
lPrintln(msg...)
}
// 打印错误信息
func lPrintErr(msg ...any) {
e := []any{"[ERROR]"}
msg = append(e, msg...)
lPrintln(msg...)
}
// 打印带时间戳的 log 信息(格式化字符串)
func lPrintf(format string, a ...any) {
lPrintln(fmt.Sprintf(format, a...))
}
// 打印警告信息(格式化字符串)
func lPrintWarnf(format string, a ...any) {
lPrintWarn(fmt.Sprintf(format, a...))
}
// 打印错误信息(格式化字符串)
func lPrintErrf(format string, a ...any) {
lPrintErr(fmt.Sprintf(format, a...))
}
// 返回 ID(UID)形式的字符串
func (s *streamer) longID() string {
return fmt.Sprintf("%s(%d)", s.Name, s.UID)
}
// 日志里用的主播名:只使用内存里已经有的数据,取不到就不发网络请求。
// getName 缓存未命中时会重试 3 次 HTTP(最长 30 多秒),放在日志里会把界面卡住。
func nameForLog(uid int) string {
if s, ok := getStreamer(uid); ok && s.Name != "" {
return s.Name
}
liveRooms.RLock()
defer liveRooms.RUnlock()
if room, ok := liveRooms.rooms[uid]; ok {
return room.name
}
return ""
}
// 返回 ID(UID)形式的字符串,用于日志,取不到名字时只打 uid
func longID(uid int) string {
name := nameForLog(uid)
if name == "" {
return itoa(uid)
}
return fmt.Sprintf("%s(%d)", name, uid)
}
// 根据 uid 获取 liveInfo
func getLiveInfoByUID(uid int) (infoList []liveInfo, ok bool) {
lInfoMap.RLock()
defer lInfoMap.RUnlock()
for _, info := range lInfoMap.info {
if info.uid == uid {
infoList = append(infoList, info)
ok = true
}
}
return infoList, ok
}
// 根据 liveID 获取 liveInfo
func getLiveInfo(liveID string) (liveInfo, bool) {
lInfoMap.RLock()
defer lInfoMap.RUnlock()
if info, ok := lInfoMap.info[liveID]; ok {
return info, true
}
return liveInfo{}, false
}
// 将 info 放进 lInfoMap 里
func setLiveInfo(info liveInfo) {
lInfoMap.Lock()
defer lInfoMap.Unlock()
lInfoMap.info[info.LiveID] = info
}
// 原地更新 liveInfo 的部分字段,不覆盖其他字段(弹幕和视频录制共用同一个 liveInfo)
func updateLiveInfo(liveID string, f func(*liveInfo)) {
lInfoMap.Lock()
defer lInfoMap.Unlock()
info, ok := lInfoMap.info[liveID]
if !ok {
info = liveInfo{LiveID: liveID}
}
f(&info)
lInfoMap.info[liveID] = info
}
// 根据 liveID 查询是否正在录制直播弹幕
func isDanmu(liveID string) bool {
if info, ok := getLiveInfo(liveID); ok {
return info.isDanmu
}
return false
}
// 根据 liveID 查询是否正在录制直播视频
func isRecording(liveID string) bool {
if info, ok := getLiveInfo(liveID); ok {
return info.isRecording
}
return false
}
// 跨盘搬运成片时的临时后缀:拷一半失败也能一眼看出不是成品
const partSuffix = ".acfunlive.part"
// 合成时的临时成片放在弹幕文件所在目录(本地),
// 完成后再落到成片目录 —— 成片目录可能是网络盘或外接盘,直接往里写、
// 写到一半失败会在那里留下一个看起来正常的半成品。文件名沿用临时前缀,
// 所以不会出现在文件管理的列表里,启动时也会清理残留。
func tempMergeOutput(danmuFile, outBase string) string {
return filepath.Join(filepath.Dir(danmuFile), chunkTempPrefix+"merging_"+outBase)
}
// 把合成好的本地临时文件落到成片位置:同盘直接改名(瞬时),跨盘退回拷贝
func placeOutput(tmp, out string) error {
if err := os.MkdirAll(filepath.Dir(out), 0755); err != nil {
return fmt.Errorf("创建成片文件夹失败:%v", err)
}
if err := os.Rename(tmp, out); err == nil {
return nil
}
// 跨盘时 os.Rename 会失败(Windows 上尤其明确),退回拷贝:
// 先拷成 .part 再改名,中途失败也不会在成片目录留下像成品的文件
part := out + partSuffix
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)
}
if err := os.Rename(part, out); err != nil {
_ = os.Remove(part)
return fmt.Errorf("成片改名失败:%v", err)
}
_ = os.Remove(tmp)
return nil
}
// 拷贝文件,跨盘搬运成片时用
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
}
defer in.Close()
out, err := os.Create(dst)
if err != nil {
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()
}
// 清掉上次运行残留的合成临时文件。正常结束时都会删掉,这里是兜底:
// 程序在合成中途被强杀时,临时文件会留在录播目录或成片目录里。
// 注意强杀不会连带杀掉 ffmpeg 子进程,它可能还在写这个文件(Windows 上就删不掉),
// 所以删不掉时只提示一句,下次启动会再清一次。
func cleanTempFiles(dir string) {
entries, err := os.ReadDir(dir)
if err != nil {
return
}
for _, e := range entries {
if e.IsDir() {
continue
}
name := e.Name()
if !strings.HasPrefix(name, chunkTempPrefix) && !strings.HasSuffix(name, partSuffix) {
continue
}
path := filepath.Join(dir, name)
if err := os.Remove(path); err == nil {
lPrintln("清理上次留下的临时文件:" + name)
} else {
lPrintWarnf("临时文件 %s 现在删不掉(多半是上次强杀程序后残留的 ffmpeg 还在写),下次启动会自动清理:%v", path, err)
}
}
}