用户要求两件事:① 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 全干净。
933 lines
27 KiB
Go
933 lines
27 KiB
Go
// 设置相关
|
||
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)
|
||
}
|