acfunlive-remake/utils.go

// 部分数据和函数定义
package main

import (
	"context"
	"errors"
	"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               // 自动合并是否已经启动
	// 弹幕文件在写盘时出过错（磁盘满、网络盘断开等）。文件是不完整的，
	// 拿它合并会得到一个缺弹幕的成片，所以本场跳过自动合并
	danmuWriteFailed 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()
	}()
}

// 不需要重试的错误：再试一遍也只是重复同样的失败（比如头像本身就超出体积/尺寸上限），
// 硬重试只会白白把几 MB 的响应重新下一遍
type noRetryError struct{ err error }

func (e noRetryError) Error() string { return e.err.Error() }
func (e noRetryError) Unwrap() error { return e.err }

// 把错误标记成「重试也没用」
func noRetry(err error) error {
	if err == nil {
		return nil
	}
	return noRetryError{err}
}

// 尝试运行，三次出错后结束运行
func runThrice(f func() error) error {
	var err error
	for retry := 0; retry < 3; retry++ {
		if err = f(); err != nil {
			var nr noRetryError
			if errors.As(err, &nr) {
				return err
			}
			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
}

// 原地更新 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
}

// 跨盘搬运成片时的临时后缀：拷一半失败也能一眼看出不是成品
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 {
		name := e.Name()
		// 预缩放头像是个目录（chunkAvatarDir），别被整个跳过
		if e.IsDir() {
			if strings.HasPrefix(name, chunkTempPrefix) {
				path := filepath.Join(dir, name)
				if err := os.RemoveAll(path); err == nil {
					lPrintln("清理上次留下的临时文件夹：" + name)
				} else {
					lPrintWarnf("临时文件夹 %s 现在删不掉，下次启动会自动清理：%v", path, err)
				}
			}
			continue
		}
		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)
		}
	}
}