Files
acfunlive-remake/config.go
T
tamakyi be8ec5f953 通知推送拆成两个渠道:QQ 机器人和 MeoW,各自带开关和测试连通
用户要求两件事:① QQ bot 推送要能测连通、并加一个开关;② 照 MeoW 的接口
文档(https://www.chuckfang.com/MeoW/api_doc.html)加一个推送渠道,同样要有
测试和开关。确认过的三个决定:QQ bot 就是现有 webhook;事件勾选两个渠道
共用、开关按渠道分开;MeoW 只填一个昵称。

1) 通知推送改成「共享事件勾选 + 每渠道一个开关」
   - notify.go: notifySetting 扩成两个渠道,新增 qqReady()/meowReady();
     sendNotify 按渠道各自起后台 goroutine 分发,事件没勾选、或两个渠道都
     没启用时什么都不做
   - config.go: 新增 notifyEnabled(QQ 机器人渠道开关)、meowEnabled、
     meowNickname、meowMsgType、meowHtmlHeight、meowImgUrl
   - 关掉开关只是暂停推送,地址、昵称这些配置都留着,随时能再打开

2) 老配置的兼容(重要)
   - notifyEnabled 是普通 bool,老 config.json 里没有这个字段会解成 false,
     升级后会把已经配好的推送静默关掉
   - 解法:loadConfig() 里先置 true 再 json.Unmarshal(Unmarshal 只覆盖文件里
     出现的字段),默认值里也写 true;文件里显式写 false 的仍然会关掉
   - 已加回归测试 TestNotifyEnabledBackwardCompat 覆盖三种写法

3) MeoW 渠道(新增 meow.go)
   - 按文档走 POST {昵称},title/msg/url/imgUrl 放 json body,
     msgType/htmlHeight 是 query 参数(文档明确说这两个不属于 JSON 参数)
   - 文档自己前后矛盾:响应格式那节写 msg,示例写 message,两个字段都解析
   - 空字段一律不放进 body:body 里的 title 优先级高于路径参数,
     传空串会把 MeoW 的默认标题顶掉
   - 业务状态码翻成中文(400 参数错、403 内容被拦、404 昵称未注册、
     429 频控、500 服务器错误),失败信息直接显示在页面上
   - msgType 为 html 时渲染成 HTML(首行加粗、链接可点、内容转义),
     高度默认 200、范围 40-2000;事件正文复用 webhook 的那份 text

4) MeoW 的错误昵称会封 IP,所以加了保护(踩出来的)
   - 实测用不存在的昵称打真实接口,返回的提示是「一天之内累计3次输错对方
     昵称您的IP将被封禁」。而开播+开始录制+下播+合并完成一天轻松触发 4 次,
     昵称配错就会把用户 IP 封了
   - meow.go: 新增 meowBadNick 记录,pushMeow 收到 404 就记下这个昵称;
     sendNotify 和测试接口命中记录时直接跳过/拒绝并说明原因,
     改掉昵称或重启程序即恢复
   - 只有 404 触发拦截:429 是频控、400 是参数问题,都应该继续尝试
   - 已加测试 TestMeowBadNickGuard(404 要拦、429/400 不能拦、不影响别的昵称)

5) 测试连通
   - /api/notify/test 合并成一个接口,用 channel 区分 qq / meow,
     每个渠道的配置下面各有一个「测试连通」按钮
   - 用页面上当前填的值直接测、不用先保存,留空的项自动用已保存的值补齐;
     测试不受渠道开关和事件勾选影响(本来就是为了打开之前先确认配置)
   - 成功返回 {ok, detail},detail 就是页面上的结论:QQ 渠道显示 webhook 的
     HTTP 状态码和耗时,MeoW 渠道显示它返回的 status 和耗时;
     webhook 返回 200 只代表地址口令没问题,能不能收到要看机器人那边,
     这点也写在结果里
   - /api/status 新增 meowBlocked,页面上红字显示「当前推不出去:…」
     (昵称没填、格式不对、或已被判过未注册)

6) 页面
   - 通知区分成「事件勾选 / 渠道一 QQ 机器人 / 渠道二 MeoW」三块,
     每块用一条带标题的细线隔开,两个渠道各带开关和测试按钮
   - 消息类型选 text 时自动收起 HTML 高度那一行

另外:README 的通知推送一节重写(两个渠道的对比表、MeoW 的频控与封禁提醒、
测试连通的说明),配置文件详解补上新字段;doc/webapi.md 补 /api/notify/test
和 meowBlocked。doc/momoko-acfunlive-notify.py 本来就处理 event="test",不用改。

已验证:新增 notify_test.go,15 个用例全部通过(URL 拼接与转义、昵称校验、
消息类型与高度归一、HTML 渲染与转义、接口成功/失败/空字段处理、6 种业务状态码
的文案、404 拦截保护、老配置兼容、测试接口各分支),其中 TestSendNotifyFanOut
是端到端分发测试:两个渠道都开→各收到 1 条、关掉任一个→只剩另一个、
两个都关→0 条、事件没勾→0 条。真机冒烟两轮:页面控件齐全、老配置默认开启、
QQ 渠道实测本地 webhook 收到完整推送(含 Authorization 和 X-Auth-Token 头与群号)、
空值回退到已保存配置、6 种非法输入全 400、开关能落盘且关开关不清地址、
MeoW 真实接口返回的 404 被正确翻成中文。全量 34 个测试通过,
Windows/Linux 双平台构建、go vet、gofmt 全干净。
2026-09-25 13:42:49 +08:00

933 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)
}
}
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)
}