// 直播视频录制相关 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 absExe(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 absExe(p) } } return "" } // 转成绝对路径再交给 exec:合并时 ffmpeg 的工作目录会被改成输出目录, // 相对路径到那边就找不着了;而且 Windows 上 exec 会直接拒绝执行 // 「相对当前目录」找到的可执行文件(报错信息完全看不出跟路径有关) func absExe(p string) string { if p == "" || filepath.IsAbs(p) { return p } if abs, err := filepath.Abs(p); err == nil { return abs } return p } // 检查设置里指定的 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 "" } // 连到该主播的直播间并取直播流信息(获取直播源和探测码率档位都用它) func newStreamInfo(uid int) (*acfundanmu.StreamInfo, error) { var ac *acfundanmu.AcFunLive var err error err = runThrice(func() error { ac, err = acfundanmu.NewAcFunLive(acfundanmu.SetLiverUID(int64(uid))) return err }) if err != nil { return nil, err } return ac.GetStreamInfo(), nil } // 把直播源列表转成码率档位 func streamTiers(list []acfundanmu.StreamURL) []qualityTier { tiers := make([]qualityTier, 0, len(list)) for _, st := range list { tiers = append(tiers, qualityTier{Name: st.QualityName, Bitrate: st.Bitrate}) } return tiers } // 取该主播当前的码率档位(没开播或拿不到直播源时返回错误) func fetchStreamQualities(uid int) ([]qualityTier, error) { info, err := newStreamInfo(uid) if err != nil { return nil, err } if info == nil || len(info.StreamList) == 0 { return nil, fmt.Errorf("uid 为%d的主播现在没有可用的直播源(没开播?)", uid) } return streamTiers(info.StreamList), nil } // 获取 uid 的直播源(flv)。pref/custom 是页面上选的档位偏好(最高/第二高/最低/自定义), // 按这次拿到的实际档位对应上去;拿不到档位表时退回最高档 func getStreamURL(uid int, pref string, custom int) (url string, quality string, err error) { sInfo, err := newStreamInfo(uid) if err != nil { return "", "", err } if sInfo == nil || len(sInfo.StreamList) == 0 { return "", "", fmt.Errorf("uid 为%d的主播没有可用的直播源", uid) } // 顺手把码率档位记下来:页面上要用它显示「这一档现在对应哪个档位名」,没开播时靠缓存显示 var tiers []qualityTier if tiers = streamTiers(sInfo.StreamList); len(tiers) > 0 { if setStreamerQualities(uid, tiers) { saveConfig() } } want := resolveQualityBitrate(tiers, pref, custom) // StreamList 按码率升序排列,从高往低找选中的那一档 st := sInfo.StreamList[len(sInfo.StreamList)-1] if want > 0 { st = sInfo.StreamList[0] for i := len(sInfo.StreamList) - 1; i >= 0; i-- { if sInfo.StreamList[i].Bitrate <= want { 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.QualityPref, 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() } }