This commit is contained in:
tamakyi
2026-09-25 13:16:54 +08:00
commit 36fe7a978c
33 changed files with 9214 additions and 0 deletions
+312
View File
@@ -0,0 +1,312 @@
// 直播视频录制相关
package main
import (
"context"
"fmt"
"io"
"os"
"os/exec"
"path/filepath"
"runtime"
"strings"
"time"
"github.com/orzogc/acfundanmu"
)
// 查找 ffmpeg:优先设置里指定的路径,其次 PATH,最后程序所在目录及其常见子目录
func getFFmpeg() string {
name := "ffmpeg"
if runtime.GOOS == "windows" {
name = "ffmpeg.exe"
}
// 设置里显式指定的路径优先,可以填可执行文件本身或它所在的文件夹
if p := strings.TrimSpace(config.FFmpegPath); p != "" {
// 相对路径先落在程序所在目录上。合并时 ffmpeg 的工作目录会被改成输出目录,
// 留着相对路径的话它会去那里找,报一个跟真实原因完全对不上的「找不到文件」
if !filepath.IsAbs(p) {
cand := p
if joined := filepath.Join(exeDir, p); filepath.IsAbs(joined) {
if _, err := os.Stat(joined); err == nil {
cand = joined
}
}
if abs, err := filepath.Abs(cand); err == nil {
p = abs
}
}
if info, err := os.Stat(p); err == nil && info.IsDir() {
p = filepath.Join(p, name)
}
if info, err := os.Stat(p); err == nil && !info.IsDir() {
return p
}
}
if p, err := exec.LookPath("ffmpeg"); err == nil {
return p
}
for _, dir := range []string{
exeDir,
filepath.Join(exeDir, "bin"),
filepath.Join(exeDir, "ffmpeg"),
filepath.Join(exeDir, "ffmpeg", "bin"),
} {
p := filepath.Join(dir, name)
if info, err := os.Stat(p); err == nil && !info.IsDir() {
return p
}
}
return ""
}
// 检查设置里指定的 ffmpeg 路径,返回错误信息,空字符串表示可用(或没填)
func checkFFmpegPath(path string) string {
p := strings.TrimSpace(path)
if p == "" {
return ""
}
info, err := os.Stat(p)
if err != nil {
return "ffmpeg 路径不存在:" + p
}
if info.IsDir() {
name := "ffmpeg"
if runtime.GOOS == "windows" {
name = "ffmpeg.exe"
}
if _, err := os.Stat(filepath.Join(p, name)); err != nil {
return "文件夹里没有找到 " + name + ":" + p
}
return ""
}
return ""
}
// 获取 uid 的直播源(flv),bitrate 为 0 时取码率最高的源,否则取码率不超过 bitrate(Kbps)的最高档
func getStreamURL(uid, bitrate int) (url string, quality string, err error) {
var ac *acfundanmu.AcFunLive
err = runThrice(func() error {
ac, err = acfundanmu.NewAcFunLive(acfundanmu.SetLiverUID(int64(uid)))
return err
})
if err != nil {
return "", "", err
}
sInfo := ac.GetStreamInfo()
if len(sInfo.StreamList) == 0 {
return "", "", fmt.Errorf("uid 为%d的主播没有可用的直播源", uid)
}
// StreamList 按码率升序排列
st := sInfo.StreamList[len(sInfo.StreamList)-1]
if bitrate > 0 {
st = sInfo.StreamList[0]
for i := len(sInfo.StreamList) - 1; i >= 0; i-- {
if sInfo.StreamList[i].Bitrate <= bitrate {
st = sInfo.StreamList[i]
break
}
}
}
return st.URL, st.QualityName, nil
}
// 临时录制指定主播的直播视频
func startRecord(uid int) bool {
if getFFmpeg() == "" {
lPrintErr("没有找到 ffmpeg,无法录制直播视频,请安装 ffmpeg 并加入 PATH,或把 ffmpeg 放在本程序所在文件夹")
return false
}
s, ok := getStreamer(uid)
if !ok {
name := getName(uid)
if name == "" {
lPrintWarnf("不存在 uid 为%d的用户", uid)
return false
}
s = streamer{UID: uid, Name: name}
}
liveID := getLiveID(uid)
if liveID == "" {
lPrintErr(s.longID() + "不在直播,取消录制直播视频")
return false
}
if info, ok := getLiveInfo(liveID); ok && info.isRecording {
lPrintWarnf("已经在录制%s的直播视频,如要重启录制,请先运行 stoprecord %d", s.longID(), uid)
return false
}
filename := getTime() + " " + s.Name + " " + s.getTitle()
safeGo(s.longID()+"的视频录制", func() { s.initRecord(mainCtx, liveID, filename) })
return true
}
// 停止录制指定主播的直播视频
func stopRecord(uid int) bool {
infoList, ok := getLiveInfoByUID(uid)
if !ok {
lPrintWarnf("没有在录制 uid 为%d的主播的直播视频", uid)
return true
}
for _, info := range infoList {
if info.isRecording {
lPrintf("开始停止录制%s的liveID为%s的直播视频", longID(uid), info.LiveID)
updateLiveInfo(info.LiveID, func(i *liveInfo) { i.recordStopped = true })
// 先结束整个录制循环,否则单段中断会被当成意外中断而重启录制
if info.recordStop != nil {
info.recordStop()
}
if info.recordCancel != nil {
info.recordCancel()
}
}
}
return true
}
// 退出直播视频录制相关操作
func quitRecord(liveID string) {
updateLiveInfo(liveID, func(i *liveInfo) { i.isRecording = false })
}
// 初始化直播视频录制
func (s streamer) initRecord(ctx context.Context, liveID, filename string) {
defer func() {
if err := recover(); err != nil {
lPrintErr("Recovering from panic in initRecord(), the error is:", err)
lPrintErrf("录制%s的直播视频发生错误,如要重启录制,请运行 startrecord %d", s.longID(), s.UID)
}
}()
if info, ok := getLiveInfo(liveID); ok && info.isRecording {
lPrintWarnf("已经在录制%s的直播视频,如要重启录制,请先运行 stoprecord %d", s.longID(), s.UID)
return
}
dctx, dcancel := context.WithCancel(ctx)
defer dcancel()
updateLiveInfo(liveID, func(i *liveInfo) {
i.recordStop = dcancel
i.recordStarted = true
})
lPrintln("开始录制" + s.longID() + "的直播视频")
lPrintf("如果想提前结束录制%s的直播视频,运行 stoprecord %d", s.longID(), s.UID)
s.recordLive(dctx, liveID, filename)
}
// 断流重连时取不到直播源的重试次数上限,避免直播源一直不可用时无限重试
const maxResumeRetry = 3
// 录制主播的直播视频,意外中断且主播还在播时重启录制新的分段
func (s streamer) recordLive(ctx context.Context, liveID, filename string) {
defer func() {
if err := recover(); err != nil {
lPrintErr("Recovering from panic in recordLive(), the error is:", err)
lPrintErrf("下载%s的直播视频发生错误,如要重启下载,请运行 startrecord %d", s.longID(), s.UID)
}
// 录制结束(含出错),检查是否满足自动合并条件
finishRecordPart(liveID)
}()
ffmpegFile := getFFmpeg()
if ffmpegFile == "" {
lPrintErr("没有找到 ffmpeg,无法录制直播视频")
return
}
var resumeFails int // 连续取不到直播源的次数
for {
url, quality, err := getStreamURL(s.UID, s.Bitrate)
if err != nil {
// 断流重连时直播源往往要过几秒才回来,窗口内多试几次,别把一场直播切成两段
if resumeFails < maxResumeRetry && ctx.Err() == nil && s.waitResume(ctx) {
resumeFails++
lPrintWarnf("暂时取不到%s的直播源,稍后重试(%d/%d):%v", s.longID(), resumeFails, maxResumeRetry, err)
continue
}
lPrintErrf("无法获取%s的直播源,退出录制直播视频:%v", s.longID(), err)
return
}
resumeFails = 0
base := transFilename(filename)
if base == "" {
return
}
recordFile := base + recordFileExt()
// -c copy 直接封装原始流,不转码,分辨率就是直播源的实际分辨率
rctx, cancel := context.WithCancel(ctx)
cmd := exec.CommandContext(rctx, ffmpegFile,
"-y", "-rw_timeout", "20000000", "-timeout", "20000000",
"-i", url, "-c", "copy", recordFile)
stdin, err := cmd.StdinPipe()
if err != nil {
cancel()
lPrintErrf("启动 ffmpeg 失败:%v", err)
return
}
// 上下文取消时向 ffmpeg 发 q 让它正常收尾(写出索引、把缓冲刷盘),超过 WaitDelay 才强杀。
// 默认的取消行为是直接杀进程,缓冲区里的数据会丢,mp4 还会因为缺少 moov 而整个报废。
cmd.Cancel = func() error {
_, werr := io.WriteString(stdin, "q")
return werr
}
cmd.WaitDelay = 10 * time.Second
startNs := time.Now().UnixNano()
updateLiveInfo(liveID, func(i *liveInfo) {
i.uid = s.UID
i.isRecording = true
i.recordCancel = cancel
i.ffmpegStdin = stdin
i.recordFile = recordFile
// 每一段都记下来:直播流中断后重连会产生新分段,合并时要把它们拼起来
i.recordSegments = append(i.recordSegments, recordSegment{file: recordFile, startNs: startNs})
if i.firstRecordFile == "" {
i.firstRecordFile = recordFile
i.recordStartNs = startNs
}
})
lPrintln("本次录制的视频保存在" + recordFile + "(码率档位:" + quality + ")")
// 标记成正在写入,避免在 web UI 的文件管理里把它删掉
markFileActive(recordFile, true)
runErr := cmd.Run()
markFileActive(recordFile, false)
// 正常情况下上面设置的 Cancel 已经让 ffmpeg 收到 q 并正常退出,这里只做资源清理
_ = stdin.Close()
cancel()
updateLiveInfo(liveID, func(i *liveInfo) { i.isRecording = false })
s.moveFile(recordFile)
// 用户主动停止或程序退出
if ctx.Err() != nil {
lPrintln(s.longID() + "的直播视频录制已经结束")
return
}
if runErr != nil {
lPrintWarnf("下载%s的直播视频出现错误:%v", s.longID(), runErr)
}
// 断流后先观察一个窗口:网络波动或主播短暂掉线(20 秒内)恢复时接着录同一场,
// 这样视频和弹幕仍然是同一对文件,最终只会合成一个成片
markResumeWait(liveID)
resumed := s.waitResume(ctx)
clearResumeWait(liveID)
if !resumed {
lPrintln(s.longID() + "的直播视频录制已经结束")
return
}
lPrintWarn("因意外结束录制" + s.longID() + "的直播视频,尝试重启录制")
// 新分段用新文件名,避免覆盖上一段;它和第一段一起参与后面的拼接
filename = getTime() + " " + s.Name + " " + s.getTitle()
}
}