Files
acfunlive-remake/config.go
T
tamakyi e35a4f7cef 合并先在本地做完再落到成片目录;头像支持 gif 和静态 webp
起因是核查合并落盘和头像处理时发现的两处问题:

1) 成片落盘:ffmpeg 原来是直接写目标目录(automerge.go 里那句注释「不再事后再
   移动一次」)。成片目录可能是网络盘或外接盘,写到一半失败会在那里留下一个
   看起来正常的半成品。现在改成先本地合成、完成后再落盘:
   - 输出先写到弹幕文件所在目录(本地)的 acfunlive_chunk_merging_*
   - 写完后由 placeOutput 落盘:同盘直接 os.Rename(瞬时),跨盘退回拷贝;
     拷贝中途失败留下的是 *.acfunlive.part,一眼看得出不是成品
   - 成功或失败都会清掉本地临时文件,失败时两个目录都不留东西
   - 临时文件沿用 chunkTempPrefix,不再出现在文件管理的列表里(listFiles 过滤)
   - 启动时清理上次残留;删不掉时提示一句 —— 强杀程序不会连带杀掉 ffmpeg
     子进程,它可能还在写这个文件,这种情况下次启动会再清一次

2) 头像:downloadAvatar 只注册了 jpeg 和 png 解码器,AcFun 上的 gif 头像必然
   解码失败,静态 webp 同样。加上 _ "image/gif"(多帧取第一帧)和
   _ "golang.org/x/image/webp"(x/image 本来就是直接依赖,零新增模块)。
   动画 webp 两条路都解不开(x/image 报 invalid format,ffmpeg 5.0 也不行),
   保持只烧文字。解码失败时错误信息带上 Content-Type,方便排查是哪张图。

在隔离实例上用真 ffmpeg 跑了真合并验证:gif 头像和 webp 头像都进了缓存并烧进
画面(抽帧确认圆形头像);日志里 ffmpeg 的输出是 record 下的临时文件,完成后
才出现在 merged;坏视频合成失败后成片目录没有半成品、录播目录零残留;合并
中途强杀留下的临时文件在下次启动时被清掉。

新增 output_test.go(7 个)和 avatar_test.go(3 个);go test ./...、gofmt、
go vet 干净。
2026-09-25 17:34:31 +08:00

936 lines
27 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"
"crypto/rand"
"encoding/hex"
"encoding/json"
"fmt"
"io"
"os"
"path/filepath"
"runtime"
"strings"
"sync"
"syscall"
"time"
"github.com/fsnotify/fsnotify"
)
const configFile = "config.json" // 设置文件名字
var configFileLocation string // 设置文件位置
// 主播的设置数据
type streamer struct {
UID int `json:"uid"` // 主播 uid
Name string `json:"name"` // 主播名字
Danmu bool `json:"-"` // 是否录制直播弹幕
Record bool `json:"-"` // 是否录制直播视频
Bitrate int `json:"-"` // 录制的直播源最高码率(Kbps),0 表示码率最高的直播源
}
// 单个主播的开关设置
type uidSetting struct {
UID int `json:"uid"`
Danmu bool `json:"danmu"`
Record bool `json:"record"`
Bitrate int `json:"bitrate"` // 录制的直播源最高码率(Kbps),0 表示码率最高的直播源
}
// uidSetting 列表,兼容旧的纯数字数组格式(旧格式默认弹幕和录播都开启)
type uidList []uidSetting
func (l *uidList) UnmarshalJSON(data []byte) error {
var ints []int
if err := json.Unmarshal(data, &ints); err == nil {
*l = (*l)[:0]
for _, uid := range ints {
*l = append(*l, uidSetting{UID: uid, Danmu: true, Record: true})
}
return nil
}
type rawList []uidSetting
var arr rawList
if err := json.Unmarshal(data, &arr); err != nil {
return err
}
*l = uidList(arr)
return nil
}
// 存放主播的设置数据
var streamers struct {
sync.RWMutex // crt 的锁
crt map[int]streamer // 现在的主播的设置数据
old map[int]streamer // 旧的主播的设置数据
}
// 串行化 config.json 的写入,避免多个 goroutine 同时写坏设置文件
var configMu sync.Mutex
// 设置数据
type configData struct {
UIDs uidList `json:"uids"` // 要录制的主播 uid 列表和各自的开关
Directory string `json:"directory"` // 弹幕文件录制结束后会被移动到该文件夹
WebPort int `json:"webPort"` // web UI 的本地端口,为 0 时不启动 web UI
AutoMerge bool `json:"autoMerge"` // 直播结束后自动把弹幕层合并进录播
MergeDelVideo bool `json:"mergeDelVideo"` // 合并完成后删除源视频
MergeDelDanmu bool `json:"mergeDelDanmu"` // 合并完成后删除弹幕文件
MergeOutputDir string `json:"mergeOutputDir"` // 合并完成后成片移动到该文件夹,留空不移动
MergeBitrate int `json:"mergeBitrate"` // 合并成片的目标码率(Kbps),0 表示使用默认质量
DefaultBitrate int `json:"defaultBitrate"` // 新添加主播的默认下载码率上限(Kbps),0 表示录制码率最高的直播源
WebExternal bool `json:"webExternal"` // 是否允许外网访问 web UI,false 时只监听本机
AuthToken string `json:"authToken"` // web UI 管理接口的访问口令,首次启动自动生成
NotifyEnabled bool `json:"notifyEnabled"` // QQ 机器人(webhook)推送的渠道开关,关掉后即使填了地址也不推送
NotifyURL string `json:"notifyUrl"` // 事件推送到该地址(HTTP POST JSON),留空不推送
NotifyToken string `json:"notifyToken"` // 推送时附带的 Bearer 口令,留空不带
NotifyGroup string `json:"notifyGroup"` // 推送时带上的群号,机器人可以直接用它决定发到哪个群
NotifyLiveOn bool `json:"notifyLiveOn"` // 开播时推送
NotifyLiveOff bool `json:"notifyLiveOff"` // 下播时推送
NotifyRecordStart bool `json:"notifyRecordStart"` // 开始录制直播视频时推送
NotifyMergeDone bool `json:"notifyMergeDone"` // 自动合并完成时推送
MeowEnabled bool `json:"meowEnabled"` // MeoW 推送的渠道开关
MeowNickname string `json:"meowNickname"` // MeoW 昵称,推送地址里的路径参数
MeowMsgType string `json:"meowMsgType"` // MeoW 消息类型:text(默认,纯文本)或 html
MeowHTMLHeight int `json:"meowHtmlHeight"` // MeoW 在 msgType 为 html 时的显示高度(像素),默认 200
MeowImgURL string `json:"meowImgUrl"` // MeoW 通知图标 URL,建议 216x216 的 PNG,留空用 MeoW 默认图标
FFmpegPath string `json:"ffmpegPath"` // ffmpeg 可执行文件(或所在文件夹)的路径,留空则自动查找
RecordFormat string `json:"recordFormat"` // 录播视频的封装格式:mp4(默认,与原版一致)、flv、ts、mkv
RecordSubdir string `json:"recordSubdir"` // 录播和弹幕文件存放在保存目录下的子文件夹名,默认为 record
MergeSubdir string `json:"mergeSubdir"` // 合并成片存放在保存目录下的子文件夹名,默认为 merged
}
// 尽量转成绝对路径,失败就原样返回
func absOrSelf(path string) string {
if path == "" {
return ""
}
if abs, err := filepath.Abs(path); err == nil {
return abs
}
return path
}
// 保存录制文件的根目录:设置了 directory 就用它,否则用 -record 指定的文件夹
// 一律归一成绝对路径:录制侧的文件名由这里拼出来,而 web UI 的文件管理接口拼的是绝对路径,
// 两边形式不统一会让「这个文件正在写入」的判断失效(删掉正在录制的文件)
func saveRootDir() string {
if config.Directory != "" {
return absOrSelf(config.Directory)
}
if recordDir == nil {
return ""
}
return absOrSelf(*recordDir)
}
// 录播和弹幕文件保存的文件夹
func recordSaveDir() string {
return filepath.Join(saveRootDir(), config.RecordSubdir)
}
// 录制过程中文件直接写入的文件夹
func recordingDir() string {
if recordDir == nil {
return config.RecordSubdir
}
return filepath.Join(absOrSelf(*recordDir), config.RecordSubdir)
}
// 合并成片保存的文件夹
func mergeSaveDir() string {
if config.MergeOutputDir != "" {
return absOrSelf(config.MergeOutputDir)
}
return filepath.Join(saveRootDir(), config.MergeSubdir)
}
// 默认的存放子文件夹名
const (
defaultRecordSubdir = "record"
defaultMergeSubdir = "merged"
)
// 录播视频默认的封装格式。原版(orzogc/acfunlive)用 config.Output 决定扩展名,默认就是 mp4
const defaultRecordFormat = "mp4"
// 支持的录播封装格式。录制用的是 -c copy,只换容器不转码,所以这些都只是封装差别
var recordFormats = map[string]struct{}{
"mp4": {},
"flv": {},
"ts": {},
"mkv": {},
}
// 把设置里的录播格式规整成受支持的值,空或不认识的值回退到 mp4
func normalizeRecordFormat(format string) string {
f := strings.ToLower(strings.TrimSpace(format))
if f == "" {
return defaultRecordFormat
}
if _, ok := recordFormats[f]; !ok {
lPrintWarnf("不支持的录播格式 %q,改用 %s(可选:mp4、flv、ts、mkv)", format, defaultRecordFormat)
return defaultRecordFormat
}
return f
}
// 录播文件的扩展名(含点),供录制和合并拼文件名用
func recordFileExt() string {
f := strings.ToLower(strings.TrimSpace(config.RecordFormat))
if _, ok := recordFormats[f]; !ok {
f = defaultRecordFormat
}
return "." + f
}
// 默认设置
var config = configData{
UIDs: uidList{},
WebPort: 51880,
RecordSubdir: defaultRecordSubdir,
MergeSubdir: defaultMergeSubdir,
NotifyEnabled: true,
}
// Windows 下不能用作文件(夹)名的保留字
var reservedNames = map[string]struct{}{
"CON": {}, "PRN": {}, "AUX": {}, "NUL": {},
"COM1": {}, "COM2": {}, "COM3": {}, "COM4": {}, "COM5": {},
"COM6": {}, "COM7": {}, "COM8": {}, "COM9": {},
"LPT1": {}, "LPT2": {}, "LPT3": {}, "LPT4": {}, "LPT5": {},
"LPT6": {}, "LPT7": {}, "LPT8": {}, "LPT9": {},
}
// 子文件夹名不能包含路径分隔符和文件名非法字符,也不能是系统保留名
func isBadSubdir(name string) bool {
if name == "" || name == "." || name == ".." {
return true
}
if strings.ContainsAny(name, `\/:*?"<>|`) {
return true
}
// Windows 下结尾的空格和点会被忽略,容易和别的名字撞车
if strings.HasSuffix(name, " ") || strings.HasSuffix(name, ".") {
return true
}
if _, bad := reservedNames[strings.ToUpper(name)]; bad {
return true
}
return false
}
// 从 streamers.crt 获取 streamer
func getStreamer(uid int) (streamer, bool) {
streamers.RLock()
defer streamers.RUnlock()
s, ok := streamers.crt[uid]
return s, ok
}
// 查看设置文件是否存在
func isConfigFileExist(filename string) bool {
fileLocation := filepath.Join(*configDir, filename)
info, err := os.Stat(fileLocation)
if os.IsNotExist(err) {
return false
}
checkErr(err)
if info.IsDir() {
lPrintErr(fileLocation + " 不能是目录")
os.Exit(1)
}
return true
}
// 读取 config.json,调用方需要自己处理 streamers 的锁
func loadConfig() {
if !isConfigFileExist(configFile) {
return
}
data, err := os.ReadFile(configFileLocation)
checkErr(err)
if json.Valid(data) {
// 老设置文件里没有 notifyEnabled 这个字段,默认按「开启」处理,
// 否则升级后已经配好的推送会被静默关掉。写在 Unmarshal 前面,
// 文件里显式写了 false 时仍会被覆盖成 false
config.NotifyEnabled = true
err = json.Unmarshal(data, &config)
checkErr(err)
} else {
lPrintErr("设置文件" + configFile + "的内容不符合 json 格式,请检查其内容")
}
}
// 命令行合成模式(-danmuburn)也读取设置文件,让 ffmpeg 路径、成片目录等配置生效
func loadConfigForDanmuburn() {
if *configDir == "" {
if exePath, err := os.Executable(); err == nil {
*configDir = filepath.Dir(exePath)
}
}
if *configDir == "" {
return
}
configFileLocation = filepath.Join(*configDir, configFile)
if !isConfigFileExist(configFile) {
return
}
loadConfig()
// 设置文件里没写到的字段保持默认值
if strings.TrimSpace(config.RecordSubdir) == "" {
config.RecordSubdir = defaultRecordSubdir
}
if strings.TrimSpace(config.MergeSubdir) == "" {
config.MergeSubdir = defaultMergeSubdir
}
}
// web UI 管理接口的访问口令长度(字节),编码成十六进制字符串
const authTokenBytes = 12
// 生成 web UI 管理接口的访问口令,首次启动时自动生成并写入设置文件
func ensureAuthToken() {
if strings.TrimSpace(currentAuthToken()) != "" {
return
}
buf := make([]byte, authTokenBytes)
if _, err := rand.Read(buf); err != nil {
lPrintErrf("生成 web UI 管理口令失败:%v", err)
return
}
token := hex.EncodeToString(buf)
streamers.Lock()
config.AuthToken = token
streamers.Unlock()
saveConfig()
lPrintln("已生成 web UI 管理口令(authToken):" + token)
lPrintln("打开 web UI 后需要输入该口令才能管理录制,请妥善保存;忘记时可以在 config.json 的 authToken 里查看")
}
// 读取当前的 web UI 管理口令
func currentAuthToken() string {
streamers.RLock()
defer streamers.RUnlock()
return config.AuthToken
}
// 检查 config.json 里的配置
func checkConfig() {
for _, u := range config.UIDs {
if u.UID <= 0 {
lPrintErr(configFile + "里的 uid 必须大于 0")
os.Exit(1)
}
}
if config.WebPort != 0 && (config.WebPort < 1024 || config.WebPort > 65535) {
lPrintErr(configFile + "里的 webPort 必须为 0(不启动 web UI)或者大于 1023 且少于 65536")
os.Exit(1)
}
if config.Directory != "" {
info, err := os.Stat(config.Directory)
checkErr(err)
if !info.IsDir() {
lPrintErrf("%s里的directory必须是存在的文件夹:%s", configFile, config.Directory)
os.Exit(1)
}
}
if config.MergeOutputDir != "" {
info, err := os.Stat(config.MergeOutputDir)
checkErr(err)
if !info.IsDir() {
lPrintErrf("%s里的mergeOutputDir必须是存在的文件夹:%s", configFile, config.MergeOutputDir)
os.Exit(1)
}
}
if config.MergeBitrate < 0 {
lPrintErrf("%s里的mergeBitrate不能小于 0", configFile)
os.Exit(1)
}
if config.DefaultBitrate < 0 {
lPrintErrf("%s里的defaultBitrate不能小于 0", configFile)
os.Exit(1)
}
if strings.TrimSpace(config.RecordSubdir) == "" {
config.RecordSubdir = defaultRecordSubdir
}
if strings.TrimSpace(config.MergeSubdir) == "" {
config.MergeSubdir = defaultMergeSubdir
}
if isBadSubdir(config.RecordSubdir) {
lPrintErrf("%s里的recordSubdir不能包含路径分隔符或非法字符:%s", configFile, config.RecordSubdir)
os.Exit(1)
}
if isBadSubdir(config.MergeSubdir) {
lPrintErrf("%s里的mergeSubdir不能包含路径分隔符或非法字符:%s", configFile, config.MergeSubdir)
os.Exit(1)
}
if config.RecordSubdir == config.MergeSubdir && config.MergeOutputDir == "" {
lPrintErrf("%s里的recordSubdir和mergeSubdir不能相同", configFile)
os.Exit(1)
}
// 录播格式写错不致命,回退到 mp4,免得整个程序起不来
config.RecordFormat = normalizeRecordFormat(config.RecordFormat)
// MeoW 这几项写错同样不致命:类型和高度回退到默认值,昵称不可用只警告
config.MeowNickname = strings.TrimSpace(config.MeowNickname)
config.MeowMsgType = normalizeMeowMsgType(config.MeowMsgType)
config.MeowImgURL = strings.TrimSpace(config.MeowImgURL)
if config.MeowHTMLHeight <= 0 {
config.MeowHTMLHeight = defaultMeowHTMLHeight
}
if msg := checkMeowNickname(config.MeowNickname, true); msg != "" {
lPrintWarnf("%s里的meowNickname不可用,不会推送 MeoW:%s", configFile, msg)
}
// ffmpeg 路径填错时不致命(可能只是还没打算录视频),清空后回退到自动查找
if msg := checkFFmpegPath(config.FFmpegPath); msg != "" {
lPrintWarnf("%s里的ffmpegPath不可用,将改为自动查找 ffmpeg:%s", configFile, msg)
config.FFmpegPath = ""
}
// 确保保存文件夹存在
for _, d := range []string{recordingDir(), recordSaveDir(), mergeSaveDir()} {
if err := os.MkdirAll(d, 0755); err != nil {
lPrintErrf("创建文件夹 %s 失败:%v", d, err)
os.Exit(1)
}
}
// 上次在合成途中被关掉的话,录播目录和成片目录里会留下临时文件
cleanTempFiles(recordSaveDir())
cleanTempFiles(mergeSaveDir())
for _, u := range config.UIDs {
if u.Bitrate < 0 {
lPrintErrf("%s里的uid %d的bitrate不能小于 0", configFile, u.UID)
os.Exit(1)
}
}
}
// 根据 config.UIDs 构建 streamers.crt,调用方需要自己处理 streamers 的锁
func buildStreamers() {
news := make(map[int]streamer, len(config.UIDs))
for _, u := range config.UIDs {
if u.UID <= 0 {
continue
}
// 已有的主播保留名字,更新开关设置
if s, ok := streamers.crt[u.UID]; ok {
s.Danmu = u.Danmu
s.Record = u.Record
s.Bitrate = u.Bitrate
news[u.UID] = s
continue
}
name := getName(u.UID)
if name == "" {
name = fmt.Sprintf("uid%d", u.UID)
}
news[u.UID] = streamer{UID: u.UID, Name: name, Danmu: u.Danmu, Record: u.Record, Bitrate: u.Bitrate}
}
streamers.crt = news
}
// 保存 config.json,写入串行化并原子替换,避免并发写坏文件或读到写了一半的内容
func saveConfig() {
configMu.Lock()
defer configMu.Unlock()
// 在锁内取快照,避免与并发的设置修改(如 removeUID)竞态,顺便去掉重复的 uid
streamers.Lock()
seen := make(map[int]struct{}, len(config.UIDs))
uids := make(uidList, 0, len(config.UIDs))
for _, u := range config.UIDs {
if _, ok := seen[u.UID]; ok {
continue
}
seen[u.UID] = struct{}{}
uids = append(uids, u)
}
config.UIDs = uids
snapshot := config
streamers.Unlock()
data, err := json.MarshalIndent(snapshot, "", " ")
if err != nil {
lPrintErrf("保存%s失败:%v", configFile, err)
return
}
// 先写临时文件再改名,读取方不会看到不完整的设置文件
tmp := configFileLocation + ".tmp"
if err := os.WriteFile(tmp, data, 0644); err != nil {
lPrintErrf("保存%s失败:%v", configFile, err)
return
}
if err := os.Rename(tmp, configFileLocation); err != nil {
lPrintErrf("保存%s失败:%v", configFile, err)
_ = os.Remove(tmp)
}
}
// 添加 uid 到录制列表并保存(弹幕和录播默认都开启),启动监听由设置文件监控触发
func addUID(uid int) bool {
return setStreamerFlags(uid, true, true)
}
// 将 uid 移出录制列表并保存,同时停止其正在进行的录制,停止监听由设置文件监控触发
func removeUID(uid int) bool {
streamers.Lock()
found := false
for i, u := range config.UIDs {
if u.UID == uid {
config.UIDs = append(config.UIDs[:i], config.UIDs[i+1:]...)
found = true
break
}
}
streamers.Unlock()
if !found {
lPrintWarnf("录制列表里没有 uid 为%d的主播", uid)
return true
}
saveConfig()
stopDanmu(uid)
stopRecord(uid)
if s, ok := getStreamer(uid); ok {
lPrintln("成功将" + s.longID() + "移出录制列表")
} else {
lPrintf("成功将 uid 为%d的主播移出录制列表", uid)
}
return true
}
// 只修改指定主播的录播开关,弹幕开关保持不变,uid 不在列表里时以弹幕关闭添加
func setRecordFlag(uid int, on bool) bool {
streamers.RLock()
var danmu bool
for _, u := range config.UIDs {
if u.UID == uid {
danmu = u.Danmu
break
}
}
streamers.RUnlock()
return setStreamerFlags(uid, danmu, on)
}
// 设置指定主播的弹幕和录播开关并保存;uid 不在列表里时先添加;运行中的录制立即启停
func setStreamerFlags(uid int, danmu, record bool) bool {
if uid <= 0 {
lPrintWarn("uid 必须大于 0")
return false
}
streamers.Lock()
found := false
for i, u := range config.UIDs {
if u.UID == uid {
config.UIDs[i].Danmu = danmu
config.UIDs[i].Record = record
found = true
break
}
}
if !found {
name := getName(uid)
if name == "" {
streamers.Unlock()
lPrintWarnf("不存在 uid 为%d的用户", uid)
return false
}
config.UIDs = append(config.UIDs, uidSetting{UID: uid, Danmu: danmu, Record: record, Bitrate: config.DefaultBitrate})
}
// 同步 streamers.crt 里的开关
if s, ok := streamers.crt[uid]; ok {
s.Danmu, s.Record = danmu, record
streamers.crt[uid] = s
} else {
streamers.crt[uid] = streamer{UID: uid, Name: getName(uid), Danmu: danmu, Record: record, Bitrate: config.DefaultBitrate}
}
streamers.Unlock()
saveConfig()
// 运行中的录制按新开关立即启停
if danmu {
startDanmu(uid)
} else {
stopDanmu(uid)
}
if record {
startRecord(uid)
} else {
stopRecord(uid)
}
return true
}
// 设置指定主播的录播码率上限(Kbps,0 为码率最高的直播源)并保存,正在录制时自动重启录制让新码率生效
func setBitrate(uid, bitrate int) bool {
if uid <= 0 || bitrate < 0 {
lPrintWarn("uid 必须大于 0,bitrate 不能小于 0")
return false
}
wasRecording := false
if infoList, ok := getLiveInfoByUID(uid); ok {
for _, info := range infoList {
if info.isRecording {
wasRecording = true
}
}
}
streamers.Lock()
found := false
for i, u := range config.UIDs {
if u.UID == uid {
config.UIDs[i].Bitrate = bitrate
found = true
break
}
}
if found {
if s, ok := streamers.crt[uid]; ok {
s.Bitrate = bitrate
streamers.crt[uid] = s
}
}
streamers.Unlock()
if !found {
lPrintWarnf("录制列表里没有 uid 为%d的主播", uid)
return false
}
saveConfig()
// 码率只对新分段生效,正在录制时重启录制
if wasRecording {
stopRecord(uid)
startRecord(uid)
lPrintf("已将 uid %d 的录播码率上限改为 %d Kbps 并重启录制", uid, bitrate)
}
return true
}
// 停止或重新开始指定主播的录制,target 为 danmu、record 或 all(空字符串表示全部),action 为 stop 或 restart
func controlStreamer(uid int, target, action string) error {
if uid <= 0 {
return fmt.Errorf("uid 必须大于 0")
}
restart := action == "restart"
if !restart && action != "stop" {
return fmt.Errorf("未知的操作:%s", action)
}
streamers.Lock()
s, ok := streamers.crt[uid]
if !ok {
for _, u := range config.UIDs {
if u.UID == uid {
s = streamer{UID: uid, Name: getName(uid), Danmu: u.Danmu, Record: u.Record}
ok = true
break
}
}
}
streamers.Unlock()
if !ok {
return fmt.Errorf("录制列表里没有 uid 为%d的主播", uid)
}
danmu, record := s.Danmu, s.Record
switch target {
case "danmu":
danmu = restart
case "record":
record = restart
default:
// 重新开始只作用于原本就开启的录制,不会把用户关掉的录制也打开;停止则全部停止
if !restart {
danmu, record = false, false
}
}
if restart && !danmu && !record {
lPrintWarnf("uid 为%d的主播没有开启任何录制,跳过重新开始", uid)
return nil
}
// 重新开始:先停止正在录的部分,才会产生新的分段
if restart {
if danmu {
stopDanmu(uid)
}
if record {
stopRecord(uid)
}
time.Sleep(2 * time.Second)
resetMergeState(uid, danmu, record)
}
if !setStreamerFlags(uid, danmu, record) {
return fmt.Errorf("设置 uid 为%d的主播失败", uid)
}
return nil
}
// 重新开始录制后重置自动合并用到的状态,让新的一对文件能被合并
func resetMergeState(uid int, danmu, record bool) {
infoList, ok := getLiveInfoByUID(uid)
if !ok {
return
}
for _, info := range infoList {
updateLiveInfo(info.LiveID, func(i *liveInfo) {
if danmu {
i.assFile = ""
i.danmuDone = false
i.danmuStopped = false
}
if record {
i.firstRecordFile = ""
i.recordStartNs = 0
i.recordSegments = nil
i.recordDone = false
i.recordStopped = false
}
i.mergeStarted = false
i.resumeUntil = 0
})
}
}
// 启动指定主播的循环
func startStreamerCycle(s streamer) {
start := controlMsg{s: s, c: startCycle}
sInfoMap.Lock()
if m, ok := sInfoMap.info[s.UID]; ok {
lPrintErr("sInfoMap不应该有%s的key", s.longID())
m.modify = true
} else {
sInfoMap.info[s.UID] = &streamerInfo{modify: true}
}
sInfoMap.Unlock()
mainCh <- start
}
// 停止指定主播的循环
func stopStreamerCycle(s streamer) {
stop := controlMsg{s: s, c: stopCycle}
sInfoMap.Lock()
if m, ok := sInfoMap.info[s.UID]; ok {
m.modify = true
m.ch <- stop
} else {
lPrintErr("sInfoMap没有%s的key", s.longID())
}
sInfoMap.Unlock()
}
// 监控 config.json 是否被修改,是的话重新设置
func cycleConfig(ctx context.Context) {
defer func() {
if err := recover(); err != nil {
lPrintErr("Recovering from panic in cycleConfig(), the error is:", err)
lPrintErr("监控设置文件" + configFile + "时出错,请重启本程序")
}
}()
lPrintln("开始监控设置文件" + configFile)
watcher, err := fsnotify.NewWatcher()
checkErr(err)
defer watcher.Close()
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
for {
select {
case <-ctx.Done():
return
case event, ok := <-watcher.Events:
if !ok {
return
}
// 很多时候保存文件会分为数段写入,避免读取未完成写入的设置文件
time.Sleep(100 * time.Millisecond)
changed := isConfigFileEvent(event)
Outer:
for {
select {
case event, ok = <-watcher.Events:
if !ok {
return
}
if isConfigFileEvent(event) {
changed = true
}
time.Sleep(100 * time.Millisecond)
default:
break Outer
}
}
if changed {
lPrintln("设置文件" + configFile + "被修改,重新读取设置")
loadNewConfig()
}
case err, ok := <-watcher.Errors:
if !ok {
return
}
lPrintErr("监控设置文件"+configFile+"时出现错误:", err)
}
}
}()
err = watcher.Add(*configDir)
checkErr(err)
wg.Wait()
lPrintln("停止监控设置文件" + configFile)
}
// 判断 fsnotify 事件是否作用在设置文件上(路径比较在 Windows 下大小写不敏感)
func isConfigFileEvent(event fsnotify.Event) bool {
if !strings.EqualFold(filepath.Clean(event.Name), filepath.Clean(configFileLocation)) {
return false
}
// 监听的是目录而不是文件本身,改名替换(原子保存)只产生 Create/Rename 事件
return event.Op&(fsnotify.Write|fsnotify.Create|fsnotify.Rename) != 0
}
// 读取修改后的 config.json,对比新旧主播列表,启动和停止相应主播的循环
func loadNewConfig() {
var toStart []streamer
var toStop []streamer
streamers.Lock()
oldCrt := make(map[int]streamer, len(streamers.crt))
for uid, s := range streamers.crt {
oldCrt[uid] = s
}
loadConfig()
checkConfig()
buildStreamers()
var changedFlags []streamer
for uid, s := range streamers.crt {
old, ok := oldCrt[uid]
if !ok {
toStart = append(toStart, s)
continue
}
// 开关有变化的主播立即按新开关启停运行中的录制
if old.Danmu != s.Danmu || old.Record != s.Record {
changedFlags = append(changedFlags, s)
}
}
for uid, s := range oldCrt {
if _, ok := streamers.crt[uid]; !ok {
toStop = append(toStop, s)
}
}
streamers.Unlock()
// 必须在释放 streamers 锁之后调用:ensureAuthToken 内部要重新加锁,互斥锁不可重入会死锁
// 设置文件里的 authToken 被清空时重新生成,否则所有管理接口都会因为口令为空而拒绝访问
ensureAuthToken()
for _, s := range toStart {
lPrintln("新增" + s.longID() + "的录制设置")
startStreamerCycle(s)
}
for _, s := range toStop {
lPrintln(s.longID() + "的录制设置被删除")
stopStreamerCycle(s)
stopDanmu(s.UID)
stopRecord(s.UID)
}
for _, s := range changedFlags {
if s.Danmu {
startDanmu(s.UID)
} else {
stopDanmu(s.UID)
}
if s.Record {
startRecord(s.UID)
} else {
stopRecord(s.UID)
}
}
}
// 移动弹幕录制产出的三个文件(原格式 ASS、昵称头像版 ASS、jsonl)
func (s *streamer) moveDanmuFiles(assFile string) {
s.moveFile(assFile)
nickFile, jsonlFile := nickJsonlFile(assFile)
s.moveFile(nickFile)
s.moveFile(jsonlFile)
}
// 移动文件
func (s *streamer) moveFile(oldFile string) {
if oldFile == "" {
return
}
// 录播和弹幕文件统一放到保存目录下的子文件夹里
directory := recordSaveDir()
if err := os.MkdirAll(directory, 0755); err != nil {
lPrintErrf("创建文件夹 %s 失败:%v", directory, err)
return
}
newFile := filepath.Join(directory, filepath.Base(oldFile))
if newFile == oldFile {
return
}
_, err := os.Stat(oldFile)
if os.IsNotExist(err) {
lPrintErrf("文件 %s 不存在", oldFile)
return
}
checkErr(err)
// https://github.com/cloudfoundry/bosh-utils/blob/master/fileutil/mover.go
err = os.Rename(oldFile, newFile)
if err != nil {
le, ok := err.(*os.LinkError)
if !ok {
lPrintErrf("将文件 %s 移动到 %s 失败:%v", oldFile, newFile, err)
return
}
if le.Err == syscall.EXDEV || (runtime.GOOS == "windows" && le.Err == syscall.Errno(0x11)) {
inputFile, err := os.Open(oldFile)
checkErr(err)
defer inputFile.Close()
outputFile, err := os.Create(newFile)
checkErr(err)
defer outputFile.Close()
_, err = io.Copy(outputFile, inputFile)
checkErr(err)
_ = inputFile.Close()
err = os.Remove(oldFile)
checkErr(err)
} else {
lPrintErrf("将文件 %s 移动到 %s 失败:%+v", oldFile, newFile, le)
return
}
}
lPrintf("成功将文件 %s 移动到 %s", oldFile, newFile)
}