diff --git a/src/configs/config.go b/src/configs/config.go index 8da4679..dc1ea3b 100644 --- a/src/configs/config.go +++ b/src/configs/config.go @@ -593,9 +593,13 @@ type WxPusher struct { UIDs []string `yaml:"uids" json:"uids"` // 接收者 UID 列表（格式 UID_xxxx） } +// SoopLiveAuth 保存 Soop 自动登录用的账号密码。 +// Password 标记 json:"-"：它只用于 YAML 落盘，绝不通过 /api/config 等 JSON 接口下发给浏览器。 +// 写入走 PATCH /config 里对 updates map 的显式解析，读取走 /api/sooplive/auth 的状态接口与 +// 设置页的 raw-config YAML，因此屏蔽 JSON 输出不影响面板保存与显示。 type SoopLiveAuth struct { Username string `yaml:"username,omitempty" json:"username,omitempty"` - Password string `yaml:"password,omitempty" json:"password,omitempty"` + Password string `yaml:"password,omitempty" json:"-"` } // DouyuAuth 保存斗鱼 passport 域的长期凭证，用于通过 safeAuth 自动续期主站登录 cookie。 diff --git a/src/configs/config_secret_exposure_test.go b/src/configs/config_secret_exposure_test.go index 241d064..c201830 100644 --- a/src/configs/config_secret_exposure_test.go +++ b/src/configs/config_secret_exposure_test.go @@ -37,3 +37,32 @@ func TestDouyuAuthSecretsStayOutOfJSON(t *testing.T) { assert.Contains(t, string(y), "long-term-passport-ticket") assert.Contains(t, string(y), "device-id-value") } + +// Soop 账号密码也不能进 JSON：GET /api/config 直接序列化整个 Config，明文密码带 json tag +// 就会被下发到浏览器。面板写密码走 PATCH /config 对 updates map 的显式解析，读状态走 +// /api/sooplive/auth，因此屏蔽 JSON 输出不影响保存与显示；YAML 侧必须保留，否则重启后 +// 自动登录拿不到密码。 +func TestSoopLiveAuthPasswordStaysOutOfJSON(t *testing.T) { + c := &Config{ + SoopLiveAuth: SoopLiveAuth{ + Username: "soop-account", + Password: "soop-plain-password", + }, + } + + b, err := json.Marshal(c) + assert.NoError(t, err) + assert.NotContains(t, string(b), "soop-plain-password") + + var decoded map[string]interface{} + assert.NoError(t, json.Unmarshal(b, &decoded)) + soopAuth, ok := decoded["sooplive_auth"].(map[string]interface{}) + assert.True(t, ok, "sooplive_auth 节应仍存在") + assert.NotContains(t, soopAuth, "password") + // 用户名不是凭证本体，面板要显示它 + assert.Equal(t, "soop-account", soopAuth["username"]) + + y, err := yaml.Marshal(c) + assert.NoError(t, err) + assert.Contains(t, string(y), "soop-plain-password") +} diff --git a/src/live/lives.go b/src/live/lives.go index 92fcca9..4b07d77 100755 --- a/src/live/lives.go +++ b/src/live/lives.go @@ -230,6 +230,20 @@ type Live interface { // 此时不应该发起请求，也不应该按「获取失败」来对待（不是平台或网络的问题）。 var ErrPlatformToolsNotReady = errors.New("平台依赖的工具尚未就绪") +// DeferredStreamResolver 由"每一路清晰度都要单独申请一次播放凭证"的平台实现（目前只有 Soop）。 +// 每次录制实际只用一路，而面板需要的是完整候选清单（清晰度名、分辨率、码率、选择属性）， +// 并不需要每一路的可播放地址：先只读元信息列出候选，选中之后再只解析被选中的那一路， +// 取流请求量就不再随清晰度档数成倍增长。 +// 包装类型（WrappedLive、InitializingLive）必须显式转发这两个方法， +// 否则类型断言走不到平台实现，录制路径会一直退回全量解析。 +type DeferredStreamResolver interface { + // ListStreamCandidates 列出当前可录的清晰度候选，只填元信息，Url 一律为空。 + ListStreamCandidates() ([]*StreamUrlInfo, error) + // ResolveStreamCandidate 只解析传入这一路的播放地址并就地写回。 + // 平台不支持、该档已不在候选中或凭证申请失败时返回错误，调用方应换下一档或回退 GetStreamInfos。 + ResolveStreamCandidate(candidate *StreamUrlInfo) error +} + const ( // defaultInterval 没有配置轮询间隔时使用的默认值 defaultInterval = 30 * time.Second @@ -335,6 +349,25 @@ func (w *WrappedLive) GetRoomID() string { return "" } +// ListStreamCandidates 转发给底层平台实现。包装层不实现这个方法，录制路径的类型断言 +// 就只能看到本对象，永远拿不到"只解析选中的一路"这条省请求的路径。 +func (w *WrappedLive) ListStreamCandidates() ([]*StreamUrlInfo, error) { + resolver, ok := w.Live.(DeferredStreamResolver) + if !ok { + return nil, ErrNotImplemented + } + return resolver.ListStreamCandidates() +} + +// ResolveStreamCandidate 转发给底层平台实现。 +func (w *WrappedLive) ResolveStreamCandidate(candidate *StreamUrlInfo) error { + resolver, ok := w.Live.(DeferredStreamResolver) + if !ok { + return ErrNotImplemented + } + return resolver.ResolveStreamCandidate(candidate) +} + // GetCachedInfo 只读取最近一次缓存，不触发平台请求。 func (w *WrappedLive) GetCachedInfo() (*Info, bool) { if w.cache == nil { diff --git a/src/live/sooplive/auth.go b/src/live/sooplive/auth.go index 53b6708..ac7a485 100644 --- a/src/live/sooplive/auth.go +++ b/src/live/sooplive/auth.go @@ -4,6 +4,7 @@ import ( "bytes" "crypto/sha256" "encoding/hex" + "errors" "fmt" "io" "net/http" @@ -303,6 +304,64 @@ func loginAndGetCookieWithSingleflight(username, password string) (*LoginResult, return cloneLoginResult(value.(*LoginResult)), nil } +// autoLoginCooldown 是同一账号两次自动登录之间的最小间隔。 +// 平台可能持续判未登录而登录接口仍回成功，此时 30 秒一轮的轮询会变成不间断的密码撞库， +// 最容易触发风控锁号；冷却只推迟下一次请求，不改变"私有房间必须带 Cookie"的语义。 +// 成功与否都要记：只挡失败的话，"登录成功但仍判未登录"这条风暴正好绕开。 +// 面板上的手动登录走 LoginAndGetCookie，不受这里限制。 +const autoLoginCooldown = 10 * time.Minute + +var ( + autoLoginMu sync.Mutex + autoLoginAttempts = map[string]time.Time{} + autoLoginGroup singleflight.Group +) + +// remainingAutoLoginCooldown 返回该账号还需等待多久才允许再次自动登录，0 表示允许。 +// 按账号而非账号+密码计，因为要挡的是同一账号被自动反复登录，无论期间凭证串怎么变。 +func remainingAutoLoginCooldown(username string) time.Duration { + autoLoginMu.Lock() + defer autoLoginMu.Unlock() + last, ok := autoLoginAttempts[username] + if !ok { + return 0 + } + if remaining := autoLoginCooldown - nowFunc().Sub(last); remaining > 0 { + return remaining + } + delete(autoLoginAttempts, username) + return 0 +} + +// markAutoLoginAttempt 在发起自动登录前记录时刻。 +func markAutoLoginAttempt(username string) { + autoLoginMu.Lock() + defer autoLoginMu.Unlock() + autoLoginAttempts[username] = nowFunc() +} + +// errAutoLoginCoolingDown 表示"这一次根本没发登录请求，因为同账号刚试过"。 +// 调用方要据此区分"被冷却挡住"与"登录确实失败"：前者应该先去找别人已经换好的 Cookie， +// 而不是让这一路房间带着失效凭证一直等到下一个登录窗口。 +var errAutoLoginCoolingDown = errors.New("soop 自动登录处于冷却期") + +// autoLoginWithCooldown 把"查冷却、记时刻、发起登录"合成一次不可分割的动作。 +// 同账号的多个房间共用一个 singleflight 键，因此同时到达时仍只发一次登录并共享结果； +// 只有先后到达的后续轮询才会被冷却挡住。 +func autoLoginWithCooldown(username, password string) (*LoginResult, error) { + value, err, _ := autoLoginGroup.Do(username, func() (any, error) { + if remaining := remainingAutoLoginCooldown(username); remaining > 0 { + return nil, fmt.Errorf("%w，%.0f 秒后重试", errAutoLoginCoolingDown, remaining.Seconds()) + } + markAutoLoginAttempt(username) + return loginAndGetCookieWithSingleflight(username, password) + }) + if err != nil { + return nil, err + } + return cloneLoginResult(value.(*LoginResult)), nil +} + func cloneCookieVerifyResult(result *CookieVerifyResult) *CookieVerifyResult { if result == nil { return nil diff --git a/src/live/sooplive/sooplive.go b/src/live/sooplive/sooplive.go index 8b0baf3..6acc2f3 100644 --- a/src/live/sooplive/sooplive.go +++ b/src/live/sooplive/sooplive.go @@ -2,6 +2,7 @@ package sooplive import ( "context" + "errors" "fmt" "net/http" "net/url" @@ -10,6 +11,7 @@ import ( "strconv" "strings" "sync" + "time" "github.com/hr3lxphr6j/requests" "github.com/tidwall/gjson" @@ -30,12 +32,19 @@ const ( channelResultLogin = -6 channelResultEmpty = 0 channelResultBlock = -2 + + // 播放页与频道信息会被"监控判活"和紧随其后的"取流"连续读取， + // 在这个窗口内复用成功结果即可省掉重复请求。窗口必须明显小于监控轮询间隔， + // 否则下播感知会被缓存拖慢。 + pageMetaReuseTTL = 15 * time.Second + channelInfoReuseTTL = 15 * time.Second ) var ( reWindowBroadNo = regexp.MustCompile(`window\.nBroadNo\s*=\s*(\d+|null);`) setCookiesFunc = configs.SetCookies persistCookieGroup singleflight.Group + streamInfoGroup singleflight.Group ) func init() { @@ -55,6 +64,15 @@ type Live struct { runtimeCookie string ignoreStoredCookie bool stateMu sync.RWMutex + // reuseMu 保护下面两个短期复用缓存，监控线程与录制线程会同时读写它们。 + // 只缓存播放页与频道信息：aid 这类播放凭证一旦跨尝试复用， + // 平台拒绝旧凭证时本地无从得知，会把几秒的恢复拖成缓存窗口那么长的停录。 + reuseMu sync.Mutex + pageMetaCache *pageMeta + pageMetaCachedAt time.Time + channelCache *channelInfo + channelCacheKey string + channelCachedAt time.Time } func (l *Live) logRetryDetail(err error, format string, args ...interface{}) { @@ -123,7 +141,7 @@ type viewPreset struct { // 2. 再尝试调用 Soop API 纠正主播名、标题和开播状态。 func (l *Live) GetInfo() (*live.Info, error) { l.GetLogger().Debugf("Soop GetInfo 开始: url=%s", l.GetRawUrl()) - meta, err := l.fetchPageMeta() + meta, err := l.fetchPageMetaCached() if err != nil { l.GetLogger().WithError(err).Debug("Soop GetInfo 失败：页面元信息获取失败") return nil, err @@ -149,7 +167,7 @@ func (l *Live) GetInfo() (*live.Info, error) { // - 登录失效 // - Soop API 异常 // - 地区 / 风控限制 - channelInfo, err := l.resolveChannelInfo(meta.Channel, meta.BroadNo) + channelInfo, err := l.resolveChannelInfoCached(meta.Channel, meta.BroadNo) if err != nil { l.GetLogger().WithError(err).Debugf("Soop GetInfo 失败：解析频道信息失败 channel=%s broadNo=%s", meta.Channel, meta.BroadNo) return nil, err @@ -180,105 +198,33 @@ func (l *Live) GetInfo() (*live.Info, error) { // - 页面缺失 nBroadNo 字段但路径里带有 broadNo 时，仍会回退使用路径 broadNo 继续请求 API； // - 因此这里返回的“无法获取播放流”既可能表示离线，也可能表示登录态不足或页面/API 结构变化。 func (l *Live) GetStreamInfos() ([]*live.StreamUrlInfo, error) { + // 录制重试、面板 /urls 与 /probe 可能在同一房间上几乎同时解析取流地址， + // 合并为一次实际解析，避免成倍地敲播放页、播放信息和 aid/调度接口。 + result, err, _ := streamInfoGroup.Do(l.GetRawUrl(), func() (interface{}, error) { + return l.resolveStreamInfos() + }) + if err != nil { + return nil, err + } + streams, _ := result.([]*live.StreamUrlInfo) + return streams, nil +} + +func (l *Live) resolveStreamInfos() ([]*live.StreamUrlInfo, error) { l.GetLogger().Debugf("Soop GetStreamInfos 开始: url=%s", l.GetRawUrl()) - meta, err := l.fetchPageMeta() + meta, channelInfo, presets, err := l.resolveRecordingContext() if err != nil { - l.GetLogger().WithError(err).Debug("Soop GetStreamInfos 失败：页面元信息获取失败") return nil, err } - if !meta.IsLiving { - l.GetLogger().Debugf("Soop GetStreamInfos 结束：页面判定无有效 broadNo channel=%s pathBroadNo=%s pageBroadNo=%s pageBroadNoFound=%v explicitOffline=%v resolvedBroadNo=%s", - meta.Channel, meta.PathBroadNo, meta.PageBroadNo, meta.PageBroadNoFound, meta.PageExplicitlyOffline, meta.BroadNo) - if meta.PageExplicitlyOffline { - return nil, fmt.Errorf("%w: Soop 页面已明确显示下播，当前无可录制流", live.ErrLiveOffline) - } - return nil, fmt.Errorf("未从 Soop 播放页解析到有效直播场次号，可能未开播、需要登录或页面结构已变更，暂时无法获取播放流") - } - channelInfo, err := l.resolveChannelInfo(meta.Channel, meta.BroadNo) - if err != nil { - l.GetLogger().WithError(err).Debugf("Soop GetStreamInfos 失败：解析频道信息失败 channel=%s broadNo=%s", meta.Channel, meta.BroadNo) - return nil, err - } - if channelInfo.Result != channelResultOK { - return nil, explainChannelResultError("Soop 播放信息接口返回异常", channelInfo.Result) - } - - if channelInfo.BroadNo == "" || channelInfo.RMD == "" { - return nil, fmt.Errorf("soop 播放信息不完整：缺少 broadcast id 或调度节点地址") - } - - headers := l.getHeadersForDownloader() - sortViewPresetsByPriority(channelInfo.ViewPresets) - streams := make([]*live.StreamUrlInfo, 0, len(channelInfo.ViewPresets)) - for _, preset := range channelInfo.ViewPresets { - if strings.EqualFold(preset.Name, "auto") { - continue - } - l.GetLogger().Debugf("Soop 开始处理清晰度: label=%s name=%s resolution=%d bps=%d", preset.Label, preset.Name, preset.LabelResolution, preset.BPS) - - aid, result, err := l.fetchAid(meta.Channel, channelInfo.BroadNo, preset.Name) + streams := make([]*live.StreamUrlInfo, 0, len(presets)) + for _, preset := range presets { + stream, err := l.resolvePresetStream(meta.Channel, channelInfo, preset) if err != nil { - l.logRetryDetail(err, "获取 Soop AID 失败: quality=%s", preset.Name) + l.logRetryDetail(err, "解析 Soop 清晰度失败: quality=%s", preset.Name) continue } - if result == channelResultLogin { - l.logRetryDetail(nil, "Soop AID 申请提示需要登录，准备自动重登后重试: quality=%s", preset.Name) - if err = l.tryAutoLogin(); err != nil { - l.logRetryDetail(err, "Soop 自动重登失败，无法重试 AID: quality=%s", preset.Name) - continue - } - aid, result, err = l.fetchAid(meta.Channel, channelInfo.BroadNo, preset.Name) - if err != nil { - l.logRetryDetail(err, "Soop 重登后再次获取 AID 失败: quality=%s", preset.Name) - continue - } - } - if result != channelResultOK || aid == "" { - l.logRetryDetail(nil, "Soop AID 申请失败: quality=%s, reason=%s, aid_empty=%v", preset.Name, explainChannelResult(result), aid == "") - continue - } - l.GetLogger().Debugf("Soop AID 申请成功: quality=%s aid_length=%d", preset.Name, len(aid)) - - viewURL, err := l.fetchViewURL(channelInfo.RMD, channelInfo.CDN, channelInfo.BroadNo, preset.Name) - if err != nil { - l.logRetryDetail(err, "获取 Soop 播放地址失败: quality=%s", preset.Name) - continue - } - l.GetLogger().Debugf("Soop 调度成功: quality=%s view_url_host=%s", preset.Name, parseHostQuiet(viewURL)) - - streamURL, err := appendQuery(viewURL, "aid", aid) - if err != nil { - l.logRetryDetail(err, "拼接 Soop 播放地址失败: quality=%s", preset.Name) - continue - } - - u, err := url.Parse(streamURL) - if err != nil { - l.logRetryDetail(err, "解析 Soop 播放地址失败: quality=%s", preset.Name) - continue - } - - qualityLabel := preset.Label - if qualityLabel == "" { - qualityLabel = preset.Name - } - - streams = append(streams, &live.StreamUrlInfo{ - Url: u, - Name: qualityLabel, - Description: qualityLabel, - Quality: qualityLabel, - Format: "hls", - Height: preset.LabelResolution, - Bitrate: preset.BPS, - AttributesForStreamSelect: map[string]string{ - "format": "hls", - "quality_key": preset.Name, - }, - HeadersForDownloader: headers, - }) - l.GetLogger().Debugf("Soop 可用流已加入: quality=%s format=hls", qualityLabel) + streams = append(streams, stream) } if len(streams) == 0 { @@ -292,6 +238,171 @@ func (l *Live) GetStreamInfos() ([]*live.StreamUrlInfo, error) { return streams, nil } +// resolveRecordingContext 完成取流的前两步（播放页 + 播放信息）并校验是否可继续解析清晰度， +// 返回按画质优先级排好序、已剔除 auto 档的候选列表。 +// 这两步的结果在 15 秒窗口内可复用，因此"先列候选、再只解析选中的一路"不会重复敲接口。 +func (l *Live) resolveRecordingContext() (*pageMeta, *channelInfo, []viewPreset, error) { + meta, err := l.fetchPageMetaCached() + if err != nil { + l.GetLogger().WithError(err).Debug("Soop GetStreamInfos 失败：页面元信息获取失败") + return nil, nil, nil, err + } + if !meta.IsLiving { + l.GetLogger().Debugf("Soop GetStreamInfos 结束：页面判定无有效 broadNo channel=%s pathBroadNo=%s pageBroadNo=%s pageBroadNoFound=%v explicitOffline=%v resolvedBroadNo=%s", + meta.Channel, meta.PathBroadNo, meta.PageBroadNo, meta.PageBroadNoFound, meta.PageExplicitlyOffline, meta.BroadNo) + if meta.PageExplicitlyOffline { + return nil, nil, nil, fmt.Errorf("%w: Soop 页面已明确显示下播，当前无可录制流", live.ErrLiveOffline) + } + return nil, nil, nil, fmt.Errorf("未从 Soop 播放页解析到有效直播场次号，可能未开播、需要登录或页面结构已变更，暂时无法获取播放流") + } + + channelInfo, err := l.resolveChannelInfoCached(meta.Channel, meta.BroadNo) + if err != nil { + l.GetLogger().WithError(err).Debugf("Soop GetStreamInfos 失败：解析频道信息失败 channel=%s broadNo=%s", meta.Channel, meta.BroadNo) + return nil, nil, nil, err + } + if channelInfo.Result != channelResultOK { + return nil, nil, nil, explainChannelResultError("Soop 播放信息接口返回异常", channelInfo.Result) + } + if channelInfo.BroadNo == "" || channelInfo.RMD == "" { + return nil, nil, nil, fmt.Errorf("soop 播放信息不完整：缺少 broadcast id 或调度节点地址") + } + + sortViewPresetsByPriority(channelInfo.ViewPresets) + var presets []viewPreset + for _, preset := range channelInfo.ViewPresets { + // auto 档没有固定的 Name，无法据此申请 aid，只能跳过 + if strings.EqualFold(preset.Name, "auto") { + continue + } + presets = append(presets, preset) + } + return meta, channelInfo, presets, nil +} + +// resolvePresetStream 为一路清晰度申请 aid 并通过调度接口拿到播放地址。 +func (l *Live) resolvePresetStream(channel string, channelInfo *channelInfo, preset viewPreset) (*live.StreamUrlInfo, error) { + l.GetLogger().Debugf("Soop 开始处理清晰度: label=%s name=%s resolution=%d bps=%d", preset.Label, preset.Name, preset.LabelResolution, preset.BPS) + + aid, result, err := l.fetchAid(channel, channelInfo.BroadNo, preset.Name) + if err != nil { + return nil, fmt.Errorf("获取 Soop AID 失败(quality=%s): %w", preset.Name, err) + } + if result == channelResultLogin { + l.logRetryDetail(nil, "Soop AID 申请提示需要登录，准备自动重登后重试: quality=%s", preset.Name) + if err = l.tryAutoLogin(); err != nil { + return nil, fmt.Errorf("自动重登失败，无法重试 Soop AID(quality=%s): %w", preset.Name, err) + } + aid, result, err = l.fetchAid(channel, channelInfo.BroadNo, preset.Name) + if err != nil { + return nil, fmt.Errorf("重登后再次获取 Soop AID 失败(quality=%s): %w", preset.Name, err) + } + } + if result != channelResultOK || aid == "" { + return nil, fmt.Errorf("申请 Soop AID 失败: quality=%s, reason=%s", preset.Name, explainChannelResult(result)) + } + l.GetLogger().Debugf("Soop AID 申请成功: quality=%s aid_length=%d", preset.Name, len(aid)) + + viewURL, err := l.fetchViewURL(channelInfo.RMD, channelInfo.CDN, channelInfo.BroadNo, preset.Name) + if err != nil { + return nil, fmt.Errorf("获取 Soop 播放地址失败(quality=%s): %w", preset.Name, err) + } + l.GetLogger().Debugf("Soop 调度成功: quality=%s view_url_host=%s", preset.Name, parseHostQuiet(viewURL)) + + streamURL, err := appendQuery(viewURL, "aid", aid) + if err != nil { + return nil, fmt.Errorf("拼接 Soop 播放地址失败(quality=%s): %w", preset.Name, err) + } + u, err := url.Parse(streamURL) + if err != nil { + return nil, fmt.Errorf("解析 Soop 播放地址失败(quality=%s): %w", preset.Name, err) + } + + stream := l.buildStreamInfo(preset) + stream.Url = u + l.GetLogger().Debugf("Soop 可用流已加入: quality=%s format=hls", stream.Quality) + return stream, nil +} + +// buildStreamInfo 用清晰度元信息构造流条目；不带 Url 的条目只用于面板展示候选。 +func (l *Live) buildStreamInfo(preset viewPreset) *live.StreamUrlInfo { + qualityLabel := preset.Label + if qualityLabel == "" { + qualityLabel = preset.Name + } + return &live.StreamUrlInfo{ + Name: qualityLabel, + Description: qualityLabel, + Quality: qualityLabel, + Format: "hls", + Height: preset.LabelResolution, + Bitrate: preset.BPS, + AttributesForStreamSelect: map[string]string{ + "format": "hls", + "quality_key": preset.Name, + }, + HeadersForDownloader: l.getHeadersForDownloader(), + } +} + +// ListStreamCandidates 实现 live.DeferredStreamResolver：只走"播放页 + 播放信息"两步， +// 逐档不再各自申请 aid 与调度结果。录制每次只用一路，其余档位在这里以元信息形式给出， +// 面板的清晰度列表因此不受收窄影响。 +func (l *Live) ListStreamCandidates() ([]*live.StreamUrlInfo, error) { + _, _, presets, err := l.resolveRecordingContext() + if err != nil { + return nil, err + } + candidates := make([]*live.StreamUrlInfo, 0, len(presets)) + for _, preset := range presets { + candidates = append(candidates, l.buildStreamInfo(preset)) + } + if len(candidates) == 0 { + return nil, fmt.Errorf("soop 未返回任何可用清晰度") + } + l.GetLogger().Debugf("Soop 清晰度候选已列出: count=%d first=%s", len(candidates), candidates[0].Quality) + return candidates, nil +} + +// ResolveStreamCandidate 实现 live.DeferredStreamResolver：只解析传入这一路的播放地址。 +func (l *Live) ResolveStreamCandidate(candidate *live.StreamUrlInfo) error { + if candidate == nil { + return fmt.Errorf("soop 候选清晰度为空") + } + meta, channelInfo, presets, err := l.resolveRecordingContext() + if err != nil { + return err + } + qualityKey := candidate.AttributesForStreamSelect["quality_key"] + for _, preset := range presets { + if !sameSoopPreset(preset, qualityKey, candidate.Quality) { + continue + } + stream, err := l.resolvePresetStream(meta.Channel, channelInfo, preset) + if err != nil { + return err + } + candidate.Url = stream.Url + candidate.HeadersForDownloader = stream.HeadersForDownloader + return nil + } + return fmt.Errorf("soop 候选清晰度 %s(quality_key=%s) 已不在当前直播的画质列表中", candidate.Quality, qualityKey) +} + +// sameSoopPreset 按 quality_key 优先、清晰度标签兜底定位档位。 +// quality_key 是平台给的内部名（如 HD、SD），同一场内稳定；面板上的旧条目若已换场， +// 两个条件都不匹配时会返回"不在列表中"错误，由调用方回退全量解析。 +func sameSoopPreset(preset viewPreset, qualityKey, qualityLabel string) bool { + if qualityKey != "" { + return strings.EqualFold(preset.Name, qualityKey) + } + label := preset.Label + if label == "" { + label = preset.Name + } + return label == qualityLabel +} + func (l *Live) GetPlatformCNName() string { return cnName } @@ -323,8 +434,12 @@ func (l *Live) fetchPageMeta() (*pageMeta, error) { return nil, fmt.Errorf("请求 Soop 播放页失败: %w", err) } if resp.StatusCode != http.StatusOK { + // 复用的访客标识若被服务端拒绝，丢掉它让下一次退化为全新匿名访问（即持久化之前的行为）， + // 避免一个失效的 _au 长期挡住该 host 上所有房间的请求。 + soopVisitorCookieJar.clear(l.Url.Host) return nil, fmt.Errorf("soop 播放页返回异常状态码: %d", resp.StatusCode) } + soopVisitorCookieJar.remember(l.Url.Host, resp.Cookies()) body, err := resp.Text() if err != nil { @@ -363,6 +478,86 @@ func (l *Live) fetchPageMeta() (*pageMeta, error) { return meta, nil } +// fetchPageMetaCached 在 pageMetaReuseTTL 窗口内复用"已成功且页面判定开播"的元信息。 +// 同一场直播里监控判活和紧随其后的取流会连续读播放页，复用能省掉其中一次整页下载 +// （播放页实测单页 170KB 以上，且平台没有给出任何可缓存的响应头）。 +// 明确离线、字段缺失和请求失败一律不缓存，保证下播与异常能被立刻感知。 +func (l *Live) fetchPageMetaCached() (*pageMeta, error) { + l.reuseMu.Lock() + cached := l.pageMetaCache + fresh := cached != nil && time.Since(l.pageMetaCachedAt) <= pageMetaReuseTTL + l.reuseMu.Unlock() + if fresh { + l.GetLogger().Debugf("Soop 复用播放页元信息: channel=%s broadNo=%s", cached.Channel, cached.BroadNo) + return cached, nil + } + + meta, err := l.fetchPageMeta() + if err != nil { + return nil, err + } + if meta.IsLiving { + l.reuseMu.Lock() + l.pageMetaCache = meta + l.pageMetaCachedAt = time.Now() + l.reuseMu.Unlock() + } + return meta, nil +} + +// resolveChannelInfoCached 在 channelInfoReuseTTL 窗口内复用 result 正常的播放信息。 +// 需要登录（-6）、已下线（0）等地区/风控异常从不缓存，仍按原逻辑重新解析。 +func (l *Live) resolveChannelInfoCached(channel, broadNo string) (*channelInfo, error) { + key := channelInfoKey(channel, broadNo) + l.reuseMu.Lock() + cached := l.channelCache + fresh := cached != nil && l.channelCacheKey == key && time.Since(l.channelCachedAt) <= channelInfoReuseTTL + l.reuseMu.Unlock() + if fresh { + l.GetLogger().Debugf("Soop 复用播放信息: channel=%s broadNo=%s presets=%d", channel, broadNo, len(cached.ViewPresets)) + // 调用方会按画质优先级就地排序，返回副本避免污染缓存 + return copyChannelInfo(cached), nil + } + + info, err := l.resolveChannelInfo(channel, broadNo) + if err != nil { + return nil, err + } + l.reuseMu.Lock() + if info.Result == channelResultOK { + l.channelCache = info + l.channelCacheKey = key + l.channelCachedAt = time.Now() + } else { + // 状态已经转为需要登录或下线，旧的正常结果不能再复用 + l.channelCache = nil + l.channelCacheKey = "" + } + l.reuseMu.Unlock() + // 缓存里保存原始对象，返回副本给调用方，避免就地排序污染缓存 + return copyChannelInfo(info), nil +} + +func channelInfoKey(channel, broadNo string) string { + return channel + "|" + broadNo +} + +func copyChannelInfo(info *channelInfo) *channelInfo { + copied := *info + copied.ViewPresets = append([]viewPreset(nil), info.ViewPresets...) + return &copied +} + +// clearReuseCache 丢弃全部短期复用结果，供登录态变化和房间配置更新时调用。 +func (l *Live) clearReuseCache() { + l.reuseMu.Lock() + defer l.reuseMu.Unlock() + l.pageMetaCache = nil + l.pageMetaCachedAt = time.Time{} + l.channelCache = nil + l.channelCacheKey = "" +} + // fetchChannelInfo 调用 player_live_api.php(type=live) 获取房间主信息。 // 返回值同时为 GetInfo 和 GetStreamInfos 服务。 // 当前版本未实现密码房输入，因此这里固定以空密码请求。 @@ -524,10 +719,92 @@ func (l *Live) getHeadersForRequest() map[string]any { return result } +// soopVisitorCookieNames 是允许在进程内粘滞复用的访客 cookie 白名单。 +// 实测匿名请求播放页时，服务端首轮下发全新的 _au（长度 32），之后每一轮都再发一个新的； +// 把上一轮的 _au 回传，服务端就不再下发 _au（连同 _au3rd 一起停发），说明这个访客标识是可复用的。 +// 复用它就不必每次轮询都向平台申报一个"全新访客"，减少大批量房间下的异常特征。 +// 注意生效范围：配置里带了 play 域的登录 cookie 时，服务端压根不下发 _au，这里存不下来也是空转， +// 只有匿名录制形态（未登录或该 host 没有 cookie）才会真正复用。 +// 白名单外的名字一律不进存储：SESSION 等登录凭证必须继续走显式配置和登录流程， +// 不能让一次普通页面响应把它们变成进程内的隐式状态。 +var soopVisitorCookieNames = []string{"_au"} + +// visitorCookieStore 按 host 保存进程内的访客 cookie，不落盘、不进配置、重启即失效。 +type visitorCookieStore struct { + mu sync.RWMutex + byHost map[string]map[string]string +} + +func (s *visitorCookieStore) remember(host string, cookies []*http.Cookie) { + if host == "" { + return + } + kept := make(map[string]string, len(soopVisitorCookieNames)) + for _, cookie := range cookies { + if cookie == nil || !isSoopVisitorCookie(cookie.Name) { + continue + } + // 空值是服务端"清除该 cookie"的写法，留下只会一直发一个无效访客标识 + if cookie.Value == "" { + continue + } + kept[cookie.Name] = cookie.Value + } + if len(kept) == 0 { + return + } + s.mu.Lock() + defer s.mu.Unlock() + if s.byHost == nil { + s.byHost = make(map[string]map[string]string) + } + current := s.byHost[host] + if current == nil { + current = make(map[string]string, len(kept)) + s.byHost[host] = current + } + for name, value := range kept { + current[name] = value + } +} + +func (s *visitorCookieStore) get(host string) map[string]string { + s.mu.RLock() + defer s.mu.RUnlock() + stored := s.byHost[host] + if len(stored) == 0 { + return nil + } + // 返回副本，调用方合并 cookie 时不会改到共享状态 + result := make(map[string]string, len(stored)) + for name, value := range stored { + result[name] = value + } + return result +} + +func (s *visitorCookieStore) clear(host string) { + s.mu.Lock() + defer s.mu.Unlock() + delete(s.byHost, host) +} + +func isSoopVisitorCookie(name string) bool { + for _, allowed := range soopVisitorCookieNames { + if name == allowed { + return true + } + } + return false +} + +var soopVisitorCookieJar visitorCookieStore + // getCookieMap 将运行时 CookieJar 和配置文件中保存的 Cookie 合并成请求使用的 map。 // 这样可以同时兼容： // 1. 本次运行中通过登录接口注入的 Cookie； // 2. 历史配置文件里持久化的 Cookie。 +// 3. 本次运行中从平台响应里收到的访客标识（_au），仅在用户没有显式给出同名 cookie 时补上。 func (l *Live) getCookieMap() map[string]string { runtimeCookie, ignoreStoredCookie := l.getRuntimeState() if ignoreStoredCookie && strings.TrimSpace(runtimeCookie) == "" { @@ -558,6 +835,12 @@ func (l *Live) getCookieMap() map[string]string { cookieMap[name] = value } } + // 访客标识优先级最低：用户或登录流程给出的同名 cookie 永远赢过进程内缓存。 + for name, value := range soopVisitorCookieJar.get(l.Url.Host) { + if existing, exists := cookieMap[name]; !exists || existing == "" { + cookieMap[name] = value + } + } return cookieMap } @@ -589,7 +872,7 @@ func (l *Live) resolveChannelInfo(channel, broadNo string) (*channelInfo, error) } // tryVerifyAndReloginIfNeeded 在真正访问 Soop 播放接口前预检查登录态。 -// 如果已有 Cookie 失效且配置中存在账号密码，则自动重新登录。 +// 如果已有 Cookie 失效且配置中存在账号密码，则自动重新登录，失败时把错误上抛。 // 这里不会回溯重抓播放页，因此只影响后续 API 调用，不改变当前页面解析步骤的行为。 func (l *Live) tryVerifyAndReloginIfNeeded() error { cookie := l.getPrimaryCookieString() @@ -628,14 +911,11 @@ func (l *Live) tryVerifyAndReloginIfNeeded() error { l.GetLogger().Debug("Soop 普通房间将忽略已失效 Cookie，继续尝试匿名访问") return nil } + // 配了账号密码就说明这个房间需要登录态（19+ 房间匿名必定取不到流）， + // 因此这里绝不降级匿名：把失效 Cookie 留着、把错误上抛，让外层轮询退避生效并在面板上显示失败原因。 l.setIgnoreStoredCookie(false) l.GetLogger().Debug("Soop Cookie 无效，但已配置账号密码，准备自动登录") - if err := l.tryAutoLogin(); err != nil { - l.setRuntimeState("", true) - l.GetLogger().WithError(err).Warn("Soop 自动登录失败，当前运行态将忽略失效 Cookie 并降级为匿名访问") - return nil - } - return nil + return l.tryAutoLogin() } // tryAutoLogin 使用配置文件中的 Soop 账号密码重新换取 Cookie。 @@ -653,10 +933,17 @@ func (l *Live) tryAutoLogin() error { if username == "" || password == "" { return fmt.Errorf("未配置 Soop 账号密码，无法执行自动登录") } + // 同一账号的自动登录按 autoLoginCooldown 节流，失败与"登录成功但仍判未登录"都算一次尝试。 l.GetLogger().Debugf("Soop 自动登录开始: username=%s", username) - result, err := loginAndGetCookieWithSingleflight(username, password) + result, err := autoLoginWithCooldown(username, password) if err != nil { + // 被冷却挡住说明这一路根本没发登录请求，而同账号的别的房间可能刚刚已经换到新 Cookie + // 并写进了配置。先验一次配置里那份，可用就直接采用；不能因为"这个账号十分钟内登录过了" + // 就让这个房间继续举着自家的旧运行态干等下一个登录窗口。 + if errors.Is(err, errAutoLoginCoolingDown) && l.adoptPersistedCookie() { + return nil + } l.GetLogger().WithError(err).Debug("Soop 自动登录失败") return err } @@ -673,17 +960,48 @@ func (l *Live) tryAutoLogin() error { if err := persistSoopCookieWithSingleflight(result.Cookie); err != nil { l.GetLogger().WithError(err).Warn("更新 Soop Cookie 到配置失败") } - l.setRuntimeState(result.Cookie, false) + l.applySessionCookie(result.Cookie) l.GetLogger().Debugf("Soop 自动登录成功: loginID=%s cookie_length=%d", result.Verify.LoginID, len(result.Cookie)) + return nil +} + +// applySessionCookie 把一份已经可用的登录 Cookie 装到当前实例：运行态、短期复用缓存与 Options。 +func (l *Live) applySessionCookie(cookie string) { + l.setRuntimeState(cookie, false) + // 换了登录身份，旧会话抓到的播放页与播放信息一律不再复用 + l.clearReuseCache() + for _, targetURL := range []*url.URL{playSoopURL, l.Url} { if targetURL == nil { continue } - live.WithKVStringCookies(targetURL, result.Cookie)(l.Options) + live.WithKVStringCookies(targetURL, cookie)(l.Options) } +} - return nil +// adoptPersistedCookie 尝试采用配置里当前保存的 Cookie，成功返回 true。 +// 只在自动登录被冷却挡住时调用：本实例的运行态刚被验过一次不通过， +// 配置里那份与它不同才值得再验一次，相同就没必要重复请求校验接口。 +func (l *Live) adoptPersistedCookie() bool { + cfg := configs.GetCurrentConfig() + if cfg == nil || cfg.Cookies == nil { + return false + } + candidate := strings.TrimSpace(cfg.Cookies[l.Url.Host]) + if candidate == "" { + candidate = strings.TrimSpace(cfg.Cookies[domainPlaySoop]) + } + if candidate == "" || candidate == strings.TrimSpace(l.getRuntimeCookie()) { + return false + } + result, err := verifyCookieWithCache(candidate) + if err != nil || result == nil || !result.IsLogin { + return false + } + l.applySessionCookie(candidate) + l.GetLogger().Infof("冷却期内直接采用配置中已刷新的 Soop Cookie: loginID=%s", result.LoginID) + return true } func persistSoopCookieWithSingleflight(cookie string) error { @@ -751,6 +1069,8 @@ func (l *Live) setIgnoreStoredCookie(ignoreStoredCookie bool) { func (l *Live) resetRuntimeState() { l.setRuntimeState("", false) + // 登录态或房间配置已经变化，之前复用的播放页与播放信息都可能不再成立 + l.clearReuseCache() } func buildCookieStringFromCookies(cookies []*http.Cookie) string { diff --git a/src/live/sooplive/sooplive_test.go b/src/live/sooplive/sooplive_test.go index cfef69d..a8aef7b 100644 --- a/src/live/sooplive/sooplive_test.go +++ b/src/live/sooplive/sooplive_test.go @@ -2,6 +2,7 @@ package sooplive import ( "context" + "fmt" "net/http" "net/http/httptest" "net/url" @@ -446,6 +447,13 @@ func TestTryVerifyAndReloginIfNeededSkipsAutoLoginWhenVerifyEndpointFails(t *tes assert.Equal(t, "expired", l.getCookieMap()["SESS"]) } +// resetSoopAutoLoginCooldownForTest 清空按账号记的自动登录冷却，避免用例之间互相污染。 +func resetSoopAutoLoginCooldownForTest() { + autoLoginMu.Lock() + autoLoginAttempts = map[string]time.Time{} + autoLoginMu.Unlock() +} + func TestTryVerifyAndReloginIfNeededAutoLoginWhenCredentialsExist(t *testing.T) { oldVerify := verifyCookieFunc oldLogin := loginAndGetCookie @@ -453,6 +461,8 @@ func TestTryVerifyAndReloginIfNeededAutoLoginWhenCredentialsExist(t *testing.T) verifyCookieFunc = oldVerify loginAndGetCookie = oldLogin }() + resetSoopAutoLoginCooldownForTest() + defer resetSoopAutoLoginCooldownForTest() cfg := configs.NewConfig() cfg.Cookies = map[string]string{ @@ -489,7 +499,7 @@ func TestTryVerifyAndReloginIfNeededAutoLoginWhenCredentialsExist(t *testing.T) assert.Equal(t, "fresh", l.getCookieMap()["SESS"]) } -func TestTryVerifyAndReloginIfNeededFallsBackToAnonymousWhenAutoLoginFails(t *testing.T) { +func TestTryVerifyAndReloginIfNeededKeepsCookieAndErrorsWhenAutoLoginFails(t *testing.T) { oldVerify := verifyCookieFunc oldLogin := loginAndGetCookie defer func() { @@ -497,6 +507,8 @@ func TestTryVerifyAndReloginIfNeededFallsBackToAnonymousWhenAutoLoginFails(t *te loginAndGetCookie = oldLogin }() resetVerifyCookieCacheForTest() + resetSoopAutoLoginCooldownForTest() + defer resetSoopAutoLoginCooldownForTest() cfg := configs.NewConfig() cfg.Cookies = map[string]string{ @@ -524,11 +536,276 @@ func TestTryVerifyAndReloginIfNeededFallsBackToAnonymousWhenAutoLoginFails(t *te } l.Options = livepkg.MustNewOptions(livepkg.WithKVStringCookies(u, "SESS=option-expired")) + // 已配凭证说明房间需要登录态（19+ 房间匿名必定取不到流）： + // 错误必须上抛让外层轮询退避生效，且不能把请求推成完全匿名。 err = l.tryVerifyAndReloginIfNeeded() + assert.Error(t, err) + assert.False(t, l.ignoreStoredCookie) + assert.Equal(t, "SESS=runtime-expired; AUTH=stale", l.runtimeCookie) + assert.Equal(t, "runtime-expired", l.getCookieMap()["SESS"]) +} + +func TestTryAutoLoginCooldownBlocksRepeatedAttempts(t *testing.T) { + oldLogin := loginAndGetCookie + oldSetCookies := setCookiesFunc + oldNow := nowFunc + defer func() { + loginAndGetCookie = oldLogin + setCookiesFunc = oldSetCookies + nowFunc = oldNow + resetSoopAutoLoginCooldownForTest() + }() + + cfg := configs.NewConfig() + cfg.SoopLiveAuth.Username = "tester" + cfg.SoopLiveAuth.Password = "secret" + configs.SetCurrentConfig(cfg) + setCookiesFunc = func(hostCookies map[string]string) (*configs.Config, error) { + t.Fatal("登录失败时不应持久化 Cookie") + return nil, nil + } + + var loginCalls int32 + loginAndGetCookie = func(username, password string) (*LoginResult, error) { + atomic.AddInt32(&loginCalls, 1) + return nil, fmt.Errorf("soop 登录失败，平台返回业务码: -1") + } + + now := time.Unix(1700000000, 0) + nowFunc = func() time.Time { return now } + + u, err := url.Parse("https://play.sooplive.com/mbntv") assert.NoError(t, err) - assert.True(t, l.ignoreStoredCookie) - assert.Empty(t, l.runtimeCookie) - assert.Empty(t, l.getCookieMap()) + l := &Live{BaseLive: internal.NewBaseLive(u)} + l.Options = livepkg.MustNewOptions() + + assert.Error(t, l.tryAutoLogin()) + assert.Equal(t, int32(1), atomic.LoadInt32(&loginCalls)) + + // 冷却窗口内反复轮询都不得再敲登录接口，但仍要明确报错而不是静默匿名 + now = now.Add(time.Minute) + for i := 0; i < 20; i++ { + err = l.tryAutoLogin() + assert.Error(t, err) + assert.Contains(t, err.Error(), "冷却期") + } + assert.Equal(t, int32(1), atomic.LoadInt32(&loginCalls)) + + // 冷却到期后照旧重试，私有房间不会永久放弃登录 + now = now.Add(autoLoginCooldown) + assert.Error(t, l.tryAutoLogin()) + assert.Equal(t, int32(2), atomic.LoadInt32(&loginCalls)) +} + +func TestTryAutoLoginCooldownAppliesToSuccessfulAttempts(t *testing.T) { + oldLogin := loginAndGetCookie + oldSetCookies := setCookiesFunc + oldNow := nowFunc + defer func() { + loginAndGetCookie = oldLogin + setCookiesFunc = oldSetCookies + nowFunc = oldNow + resetSoopAutoLoginCooldownForTest() + }() + + cfg := configs.NewConfig() + cfg.SoopLiveAuth.Username = "tester" + cfg.SoopLiveAuth.Password = "secret" + configs.SetCurrentConfig(cfg) + + var loginCalls int32 + var persistCalls int32 + now := time.Unix(1700000000, 0) + nowFunc = func() time.Time { return now } + loginAndGetCookie = func(username, password string) (*LoginResult, error) { + atomic.AddInt32(&loginCalls, 1) + return &LoginResult{ + Cookie: "SESS=fresh; AUTH=ok", + Verify: CookieVerifyResult{IsLogin: true, LoginID: username}, + Username: username, + }, nil + } + setCookiesFunc = func(hostCookies map[string]string) (*configs.Config, error) { + atomic.AddInt32(&persistCalls, 1) + return configs.GetCurrentConfig(), nil + } + + u, err := url.Parse("https://play.sooplive.com/mbntv") + assert.NoError(t, err) + l := &Live{BaseLive: internal.NewBaseLive(u)} + l.Options = livepkg.MustNewOptions() + + assert.NoError(t, l.tryAutoLogin()) + + // "登录接口回成功、平台仍判未登录"是最省不了请求的一种形态： + // 冷却必须对成功的尝试同样生效，否则每轮都会重新登录并反复重写配置。 + // 19 次 ×30 秒 = 9.5 分钟，仍在 10 分钟窗口内。 + for i := 0; i < 19; i++ { + now = now.Add(30 * time.Second) + err = l.tryAutoLogin() + assert.Error(t, err) + assert.Contains(t, err.Error(), "冷却期") + } + assert.Equal(t, int32(1), atomic.LoadInt32(&loginCalls)) + assert.Equal(t, int32(1), persistCalls) + + // 冷却到期后仍会再试，不是一次性放弃 + now = now.Add(autoLoginCooldown) + assert.NoError(t, l.tryAutoLogin()) + assert.Equal(t, int32(2), atomic.LoadInt32(&loginCalls)) +} + +// TestTryAutoLoginAdoptsCookiePersistedByOtherRoomDuringCooldown 覆盖同账号多房间先后恢复登录态的时序： +// 房间 A 刚换到新 Cookie 并写进配置，房间 B 下一轮轮询才被 10 分钟冷却挡住。 +// B 此时不能举着自己的旧运行态干等下一个登录窗口——19+ 私有房间匿名必定取不到流， +// 白等一轮就是断录；配置里那份是别人刚换好的，验一次能用就直接采用。 +func TestTryAutoLoginAdoptsCookiePersistedByOtherRoomDuringCooldown(t *testing.T) { + oldLogin := loginAndGetCookie + oldSetCookies := setCookiesFunc + oldVerify := verifyCookieFunc + oldNow := nowFunc + defer func() { + loginAndGetCookie = oldLogin + setCookiesFunc = oldSetCookies + verifyCookieFunc = oldVerify + nowFunc = oldNow + resetSoopAutoLoginCooldownForTest() + resetVerifyCookieCacheForTest() + }() + resetSoopAutoLoginCooldownForTest() + resetVerifyCookieCacheForTest() + + cfg := configs.NewConfig() + cfg.Cookies = map[string]string{domainPlaySoop: "SESS=expired"} + cfg.SoopLiveAuth.Username = "tester" + cfg.SoopLiveAuth.Password = "secret" + configs.SetCurrentConfig(cfg) + + var loginCalls int32 + loginAndGetCookie = func(username, password string) (*LoginResult, error) { + atomic.AddInt32(&loginCalls, 1) + return &LoginResult{ + Cookie: "SESS=fresh; AUTH=ok", + Verify: CookieVerifyResult{IsLogin: true, LoginID: username}, + Username: username, + }, nil + } + // 照真实 persist 的口径把新 Cookie 落到配置里，B 才有东西可采。 + setCookiesFunc = func(hostCookies map[string]string) (*configs.Config, error) { + current := configs.GetCurrentConfig() + if current.Cookies == nil { + current.Cookies = map[string]string{} + } + for host, cookie := range hostCookies { + current.Cookies[host] = cookie + } + return current, nil + } + verifyCookieFunc = func(cookie string) (*CookieVerifyResult, error) { + if cookie == "SESS=fresh; AUTH=ok" { + return &CookieVerifyResult{IsLogin: true, LoginID: "tester"}, nil + } + return &CookieVerifyResult{IsLogin: false}, nil + } + + now := time.Unix(1700000000, 0) + nowFunc = func() time.Time { return now } + + makeLive := func() *Live { + u, err := url.Parse("https://play.sooplive.com/mbntv") + assert.NoError(t, err) + l := &Live{BaseLive: internal.NewBaseLive(u)} + l.Options = livepkg.MustNewOptions(livepkg.WithKVStringCookies(u, "SESS=expired")) + return l + } + + liveA := makeLive() + assert.NoError(t, liveA.tryAutoLogin()) + assert.Equal(t, int32(1), atomic.LoadInt32(&loginCalls)) + assert.Equal(t, "SESS=fresh; AUTH=ok", liveA.runtimeCookie) + + // B 手里还是那一份刚刚被验过不通过的旧 Cookie，此刻正处在 A 占用的冷却窗口内。 + liveB := makeLive() + liveB.setRuntimeState("SESS=stale", false) + + assert.NoError(t, liveB.tryAutoLogin()) + assert.Equal(t, int32(1), atomic.LoadInt32(&loginCalls), "被冷却挡住时不该再敲登录接口") + assert.Equal(t, "SESS=fresh; AUTH=ok", liveB.runtimeCookie) + assert.Equal(t, "fresh", liveB.getCookieMap()["SESS"]) + + // 冷却窗口过去后照旧走正常登录，采用配置 Cookie 不是一次性豁免。 + now = now.Add(autoLoginCooldown) + loginAndGetCookie = func(username, password string) (*LoginResult, error) { + atomic.AddInt32(&loginCalls, 1) + return &LoginResult{ + Cookie: "SESS=newer; AUTH=ok", + Verify: CookieVerifyResult{IsLogin: true, LoginID: username}, + Username: username, + }, nil + } + assert.NoError(t, liveB.tryAutoLogin()) + assert.Equal(t, int32(2), atomic.LoadInt32(&loginCalls)) + assert.Equal(t, "SESS=newer; AUTH=ok", liveB.runtimeCookie) +} + +// TestTryAutoLoginKeepsCooldownErrorWhenPersistedCookieAlsoInvalid 配置里那份不可用时， +// 冷却期的原始错误必须照旧上抛：不能因为尝试过采用就把它咽掉，让外层以为这一轮登录过了。 +func TestTryAutoLoginKeepsCooldownErrorWhenPersistedCookieAlsoInvalid(t *testing.T) { + oldLogin := loginAndGetCookie + oldSetCookies := setCookiesFunc + oldVerify := verifyCookieFunc + oldNow := nowFunc + defer func() { + loginAndGetCookie = oldLogin + setCookiesFunc = oldSetCookies + verifyCookieFunc = oldVerify + nowFunc = oldNow + resetSoopAutoLoginCooldownForTest() + resetVerifyCookieCacheForTest() + }() + resetSoopAutoLoginCooldownForTest() + resetVerifyCookieCacheForTest() + + cfg := configs.NewConfig() + cfg.Cookies = map[string]string{domainPlaySoop: "SESS=other-room-expired"} + cfg.SoopLiveAuth.Username = "tester" + cfg.SoopLiveAuth.Password = "secret" + configs.SetCurrentConfig(cfg) + + var loginCalls int32 + loginAndGetCookie = func(username, password string) (*LoginResult, error) { + atomic.AddInt32(&loginCalls, 1) + return &LoginResult{ + Cookie: "SESS=fresh; AUTH=ok", + Verify: CookieVerifyResult{IsLogin: true, LoginID: username}, + Username: username, + }, nil + } + setCookiesFunc = func(hostCookies map[string]string) (*configs.Config, error) { + return configs.GetCurrentConfig(), nil + } + verifyCookieFunc = func(cookie string) (*CookieVerifyResult, error) { + return &CookieVerifyResult{IsLogin: false}, nil + } + + nowFunc = func() time.Time { return time.Unix(1700000000, 0) } + + u, err := url.Parse("https://play.sooplive.com/mbntv") + assert.NoError(t, err) + l := &Live{BaseLive: internal.NewBaseLive(u)} + l.Options = livepkg.MustNewOptions() + + // 第一次照常登录并占住冷却窗口 + assert.NoError(t, l.tryAutoLogin()) + assert.Equal(t, int32(1), atomic.LoadInt32(&loginCalls)) + + // 第二次被冷却挡住，配置里那份也验不过，只能把冷却错误原样上抛 + l.setRuntimeState("SESS=stale", false) + err = l.tryAutoLogin() + assert.Error(t, err) + assert.Contains(t, err.Error(), "冷却期") + assert.Equal(t, int32(1), atomic.LoadInt32(&loginCalls)) + assert.Equal(t, "SESS=stale", l.runtimeCookie, "采用失败不能顺手改掉原有运行态") } func TestGetCookieMapPrefersLatestConfigCookieOverStaleOptions(t *testing.T) { @@ -706,6 +983,8 @@ func TestTryVerifyAndReloginIfNeededDeduplicatesAutoLogin(t *testing.T) { verifyCookieCache = map[string]cachedVerifyResult{} verifyCookieCacheMu.Unlock() }() + resetSoopAutoLoginCooldownForTest() + defer resetSoopAutoLoginCooldownForTest() verifyCookieCacheTTL = 0 @@ -773,6 +1052,8 @@ func TestTryAutoLoginDeduplicatesCookiePersistence(t *testing.T) { loginAndGetCookie = oldLogin setCookiesFunc = oldSetCookies }() + resetSoopAutoLoginCooldownForTest() + defer resetSoopAutoLoginCooldownForTest() cfg := configs.NewConfig() cfg.SoopLiveAuth.Username = "tester" @@ -833,6 +1114,8 @@ func TestTryAutoLoginRejectsUnverifiedLoginResult(t *testing.T) { loginAndGetCookie = oldLogin setCookiesFunc = oldSetCookies }() + resetSoopAutoLoginCooldownForTest() + defer resetSoopAutoLoginCooldownForTest() cfg := configs.NewConfig() cfg.SoopLiveAuth.Username = "tester" diff --git a/src/live/system/initializing_live.go b/src/live/system/initializing_live.go index 04b8cab..6d27fd1 100644 --- a/src/live/system/initializing_live.go +++ b/src/live/system/initializing_live.go @@ -174,6 +174,22 @@ func (l *InitializingLive) GetStreamInfos() ([]*live.StreamUrlInfo, error) { return l.OriginalLive.GetStreamInfos() } +// ListStreamCandidates 委托给原始 Live；平台没有实现时返回 ErrNotImplemented， +// 由调用方回退全量解析，不能谎报"候选为空"。 +func (l *InitializingLive) ListStreamCandidates() ([]*live.StreamUrlInfo, error) { + if resolver, ok := l.OriginalLive.(live.DeferredStreamResolver); ok { + return resolver.ListStreamCandidates() + } + return nil, live.ErrNotImplemented +} + +func (l *InitializingLive) ResolveStreamCandidate(candidate *live.StreamUrlInfo) error { + if resolver, ok := l.OriginalLive.(live.DeferredStreamResolver); ok { + return resolver.ResolveStreamCandidate(candidate) + } + return live.ErrNotImplemented +} + func (l *InitializingLive) GetPlatformCNName() string { return l.OriginalLive.GetPlatformCNName() } diff --git a/src/recorders/recorder.go b/src/recorders/recorder.go index 4986d39..b4db128 100644 --- a/src/recorders/recorder.go +++ b/src/recorders/recorder.go @@ -51,6 +51,18 @@ const ( const soopRetryWarnInterval = time.Minute +// fastFailThreshold 判断一次尝试是否真的拉到了流：解析持续时间短于它就按"秒退"处理。 +const fastFailThreshold = 5 * time.Second + +// 取流/录制连续失败时的重试间隔： +// 本场从未录上过说明源此刻根本不可用，慢慢试（本来也没在丢内容）； +// 已经录上过则优先保断流后的内容连续性，只稍微拉长间隔。 +const ( + minRecordRetryInterval = 5 * time.Second + maxRecordRetryIntervalNeverHeld = 60 * time.Second + maxRecordRetryIntervalAfterHold = 15 * time.Second +) + // for test var ( // newParser 根据配置的下载器类型创建 parser，并实现回退逻辑： @@ -360,6 +372,13 @@ type recorder struct { lastRetryLogKey string lastRetryLogAt time.Time suppressedRetryLogCount int + + // 下面三项用于连续"秒退"时逐步拉长重试间隔，避免录不了的房间不间断敲平台取流接口。 + // consecutiveFastFailures/lastAttemptHeldStream 只被 run() 这一个 goroutine 读写 + // （tryRecord 是它的同步调用），因此不加锁；everHeldStream 需要跨分段重启传递，用原子类型。 + consecutiveFastFailures int + everHeldStream atomic.Bool + lastAttemptHeldStream bool } func NewRecorder(ctx context.Context, live live.Live) (Recorder, error) { @@ -384,6 +403,8 @@ func (r *recorder) tryRecord(ctx context.Context) { // 每次重试前重置探测状态，避免上次录制的旧数据残留 // （例如上次探测成功但本次流分辨率已变化） r.actualStreamInfo.Store(nil) + // 早退（取流失败、FFmpeg 缺失等）同样算"这次没撑住"，由 run() 拉长下次重试间隔 + r.lastAttemptHeldStream = false cfg := configs.GetCurrentConfig() @@ -398,7 +419,11 @@ func (r *recorder) tryRecord(ctx context.Context) { var streamInfos []*live.StreamUrlInfo var err error - if streamInfos, err = r.Live.GetStreamInfos(); err == live.ErrNotImplemented { + // 本轮偏好取这一次尝试开头解析好的值，不再重读全局配置：延迟解析要跑几百毫秒的平台 + // 请求，期间用户在面板点"切换清晰度"改的就是这个偏好，重读会让下面的最终选择与 + // 上面的解析顺序按两套偏好算，选中根本没解析过 Url 的候选。 + preference := resolvedConfig.StreamPreference + if streamInfos, err = r.getStreamInfosForRecording(preference); err == live.ErrNotImplemented { var urls []*url.URL // TODO: remove deprecated method GetStreamUrls //nolint:staticcheck @@ -412,13 +437,9 @@ func (r *recorder) tryRecord(ctx context.Context) { if err != nil && r.stopRetryForExplicitOffline(err) { return } - r.logStreamURLRetry(err) - // 使用可中断的等待，确保 Ctrl+C 能立即响应 - select { - case <-ctx.Done(): - case <-r.stop: - case <-time.After(5 * time.Second): - } + // 这里不再自己 sleep：重试节奏统一由 run() 的退避决定， + // 保留固定的 5 秒只会让人误以为下限就是 5 秒。 + r.logStreamURLRetry(err, r.pendingRetryInterval()) return } r.resetRetryLogState() @@ -449,8 +470,20 @@ func (r *recorder) tryRecord(ctx context.Context) { fileName := filepath.Join(resolvedConfig.OutPutPath, buf.String()) outputPath, _ := filepath.Split(fileName) - // TODO 根据配置选择最佳流 - streamInfo := r.selectPreferredStream(streamInfos) + // 按本轮固定下来的偏好选流 + streamInfo := r.selectPreferredStream(streamInfos, preference) + if streamInfo == nil || streamInfo.Url == nil { + // 选中的这一路没有播放地址：延迟解析的列表里除被解析的那一路之外只有元信息。 + // 走到这里说明本轮的输入已经不再自洽，绝不能继续往下解引用 Url， + // 也不能持着 currentFileLock 崩掉整个录制线程（状态接口的读锁会一起挂住）。 + if streamInfo != nil { + err = fmt.Errorf("清晰度 %s 尚未解析出播放地址", streamInfo.Quality) + } else { + err = fmt.Errorf("没有可用的播放流") + } + r.logStreamURLRetry(err, r.pendingRetryInterval()) + return + } r.saveCurrentStreamInfo(streamInfo) // 更新可用流信息到 info（用于API展示） r.updateAvailableStreams(ctx, info, streamInfos) @@ -642,6 +675,7 @@ func (r *recorder) tryRecord(ctx context.Context) { } r.setAndCloseParser(p) r.startTime = time.Now() + parseStartedAt := r.startTime // 弹幕录制（支持哔哩哔哩、抖音、斗鱼平台） if resolvedConfig.DanmakuEnable { @@ -692,10 +726,14 @@ func (r *recorder) tryRecord(ctx context.Context) { dmRec.Stop() } + // 本次尝试是否"撑住过"由 run() 用来决定重试间隔：能持续拉流说明源可用， + // 连续秒退则说明源此刻录不了，应该把下一次尝试推后。 + r.lastAttemptHeldStream = time.Since(parseStartedAt) >= fastFailThreshold + if err != nil { r.getLogger().WithError(err).Error("failed to parse live stream") // 视频流快速失败时（如 404），清理没有对应视频文件的残留弹幕 - if elapsed := time.Since(r.startTime); elapsed < 5*time.Second { + if elapsed := time.Since(r.startTime); elapsed < fastFailThreshold { cleanupOrphanedDanmakuFiles(dmFile) } return @@ -928,56 +966,126 @@ func (r *recorder) stopRetryForExplicitOffline(err error) bool { return true } -func (r *recorder) selectPreferredStream(streamInfos []*live.StreamUrlInfo) (ret *live.StreamUrlInfo) { +// getStreamInfosForRecording 取得本次录制可用的流列表。 +// 平台支持"先列候选、只解析选中的那一路"时优先走那条路径：Soop 每一档清晰度都要单独申请 +// 一次播放凭证并查一次调度接口，全量解析的请求量随档位成倍增长，而每次录制实际只用一路。 +// 候选里除了被选中的一路之外只有元信息（Url 为空），面板的清晰度列表因此仍然完整。 +func (r *recorder) getStreamInfosForRecording(preference configs.StreamPreference) ([]*live.StreamUrlInfo, error) { + resolver, ok := r.Live.(live.DeferredStreamResolver) + if !ok { + return r.Live.GetStreamInfos() + } + candidates, err := resolver.ListStreamCandidates() + if err != nil { + if errors.Is(err, live.ErrNotImplemented) { + return r.Live.GetStreamInfos() + } + return nil, err + } + + ordered := orderStreamsByPreference(candidates, preference) + var firstErr error + for i, candidate := range ordered { + if err := resolver.ResolveStreamCandidate(candidate); err != nil { + wrapped := fmt.Errorf("清晰度 %s: %w", candidate.Quality, err) + if errors.Is(err, live.ErrLiveOffline) { + // 房间在列完候选之后、解析这一档的几百毫秒里下播了：换别的档位同样取不到流， + // 立即上抛让 stopRetryForExplicitOffline 收手，既不重试也不再敲剩下的档位。 + return nil, wrapped + } + if firstErr == nil { + firstErr = wrapped + } + r.getLogger().WithError(err).Debugf("候选清晰度 %s 取流失败，尝试下一档", candidate.Quality) + continue + } + // ordered[:i] 是本次尝试失败的高优先档位，与全量解析一样剔除； + // ordered[i+1:] 没有尝试过，保留其元信息供面板展示可选档位。 + streams := make([]*live.StreamUrlInfo, 0, len(ordered)) + streams = append(streams, candidate) + return append(streams, ordered[i+1:]...), nil + } + // 所有候选都解析失败：直接把第一路的错误上抛，由 run() 的退避拉长下一轮间隔。 + // 不再顺手全量解析一遍——那等于在同一次尝试里把每一档的凭证与调度请求再敲一次， + // 而"全部失败"恰恰是最需要少发请求的状态（未登录、凭证被拒、调度异常）。 + // 候选与平台档位对不上时会由下一轮重新列候选自愈，代价是一轮退避间隔。 + if firstErr == nil { + firstErr = fmt.Errorf("soop 未返回任何可解析的清晰度") + } + return nil, firstErr +} + +// orderStreamsByPreference 按偏好分值从高到低排序，分值相同保持平台返回的先后顺序， +// 使"依次尝试"选出的第一路等于 selectPreferredStream 在全量列表上的选择结果。 +func orderStreamsByPreference(candidates []*live.StreamUrlInfo, preference configs.StreamPreference) []*live.StreamUrlInfo { + if preference.Quality == nil && preference.Attributes == nil { + return candidates + } + ordered := append([]*live.StreamUrlInfo(nil), candidates...) + sort.SliceStable(ordered, func(a, b int) bool { + return streamPreferenceScore(ordered[a], preference) > streamPreferenceScore(ordered[b], preference) + }) + return ordered +} + +func (r *recorder) selectPreferredStream(streamInfos []*live.StreamUrlInfo, preference configs.StreamPreference) (ret *live.StreamUrlInfo) { // 如果没有可用流，直接返回 nil if len(streamInfos) == 0 { return nil } - streamPreference := configs.GetCurrentConfig().GetEffectiveConfigForRoom(r.Live.GetRawUrl()).StreamPreference - // 如果未配置流偏好（Quality 和 Attributes 均为 nil），直接返回第一个流 - if streamPreference.Quality == nil && streamPreference.Attributes == nil { + if preference.Quality == nil && preference.Attributes == nil { return streamInfos[0] } - // 安全获取 Quality 和 Attributes，处理 nil 情况 - var quality string - if streamPreference.Quality != nil { - quality = *streamPreference.Quality - } - var attrs map[string]string - if streamPreference.Attributes != nil { - attrs = *streamPreference.Attributes - } - - retMatchedCount := 0 + bestScore := 0 for _, info := range streamInfos { - currMatchedCount := 0 - // 仅当配置了 Quality 时才匹配 - if quality != "" && info.Quality == quality { - currMatchedCount += 100 - } - // 仅当配置了 Attributes 时才匹配 - for k, v := range attrs { - if info.AttributesForStreamSelect[k] == v { - currMatchedCount += 1 - } - } - if currMatchedCount > retMatchedCount { + if score := streamPreferenceScore(info, preference); score > bestScore { ret = info - retMatchedCount = currMatchedCount + bestScore = score } } // 如果没有任何匹配的流，回退到第一个可用流 if ret == nil { - r.getLogger().Warnf("没有流匹配配置的偏好 (quality=%s, attrs=%v)，使用第一个可用流", quality, attrs) + r.getLogger().Warnf("没有流匹配配置的偏好 (quality=%s, attrs=%v)，使用第一个可用流", preferenceQuality(preference), preferenceAttributes(preference)) return streamInfos[0] } return } +// streamPreferenceScore 给一路流按偏好打分：画质名命中 +100，属性每命中一项 +1。 +// 分值相同时由调用方保持列表原有先后顺序，与"取第一个匹配项"的旧行为一致。 +func streamPreferenceScore(info *live.StreamUrlInfo, preference configs.StreamPreference) int { + score := 0 + if quality := preferenceQuality(preference); quality != "" && info.Quality == quality { + score += 100 + } + if attrs := preferenceAttributes(preference); attrs != nil { + for k, v := range attrs { + if info.AttributesForStreamSelect[k] == v { + score += 1 + } + } + } + return score +} + +func preferenceQuality(preference configs.StreamPreference) string { + if preference.Quality == nil { + return "" + } + return *preference.Quality +} + +func preferenceAttributes(preference configs.StreamPreference) map[string]string { + if preference.Attributes == nil { + return nil + } + return *preference.Attributes +} + func (r *recorder) run(ctx context.Context) { defer close(r.done) defer func() { @@ -993,8 +1101,6 @@ func (r *recorder) run(ctx context.Context) { } }() - const minRetryInterval = 5 * time.Second - for { select { case <-r.stop: @@ -1007,10 +1113,11 @@ func (r *recorder) run(ctx context.Context) { r.tryRecord(tryCtx) tryCancel() - // 确保两次 tryRecord 之间至少间隔 minRetryInterval - // 防止快速失败（如 FFmpeg 秒退 404）导致紧密循环 - if elapsed := time.Since(start); elapsed < minRetryInterval { - delay := minRetryInterval - elapsed + // 按本次尝试是否撑住过更新退避档位： + // 防快速失败（如 FFmpeg 秒退 404）导致紧密循环，也防录不了的房间不间断敲平台取流接口 + retryInterval := r.noteAttemptResult(r.lastAttemptHeldStream) + if elapsed := time.Since(start); elapsed < retryInterval { + delay := retryInterval - elapsed select { case <-r.stop: return @@ -1023,6 +1130,41 @@ func (r *recorder) run(ctx context.Context) { } } +// noteAttemptResult 记录本次尝试有没有真正撑住过流，并返回下一次重试的间隔。 +// 只有 run() 这一个 goroutine 会调用，因此状态字段无需加锁。 +func (r *recorder) noteAttemptResult(heldStream bool) time.Duration { + if heldStream { + r.consecutiveFastFailures = 0 + r.everHeldStream.Store(true) + } else { + r.consecutiveFastFailures++ + } + return recordRetryInterval(r.consecutiveFastFailures, r.everHeldStream.Load()) +} + +// pendingRetryInterval 预测"本次尝试失败"时 run() 即将采用的间隔，仅用于日志文案。 +// 调用点在取流失败分支，此时 lastAttemptHeldStream 必为 false， +// noteAttemptResult 会把连续失败计数加一，所以这里同样按加一后的计数计算。 +func (r *recorder) pendingRetryInterval() time.Duration { + return recordRetryInterval(r.consecutiveFastFailures+1, r.everHeldStream.Load()) +} + +// recordRetryInterval 由连续"秒退"次数推出下一次重试间隔： +// 每失败一轮翻一倍，直到该档位的上限为止；一旦拉到过流就重新从最小间隔开始。 +func recordRetryInterval(consecutiveFastFailures int, everHeldStream bool) time.Duration { + maxInterval := maxRecordRetryIntervalNeverHeld + if everHeldStream { + maxInterval = maxRecordRetryIntervalAfterHold + } + // 限制位移次数，避免次数很大时溢出成负数 + shift := min(consecutiveFastFailures-1, 4) + if shift < 0 { + shift = 0 + } + interval := minRecordRetryInterval << shift + return min(interval, maxInterval) +} + // accumulateRecordedFiles 累积录制文件信息，仅记录实际存在的文件 func (r *recorder) accumulateRecordedFiles(files ...string) { r.recordedFilesMu.Lock() @@ -1289,9 +1431,10 @@ func filterOrphanedAssFromDetails(details []notify.RecordingFileDetail) []notify return result } -func (r *recorder) logStreamURLRetry(err error) { +func (r *recorder) logStreamURLRetry(err error, nextInterval time.Duration) { + intervalText := nextInterval.Round(time.Second) if configs.IsDebug() || !strings.EqualFold(r.Live.GetPlatformCNName(), "SOOP") { - r.warnStreamURLRetry(err, "failed to get stream url, will retry after 5s...", 0) + r.warnStreamURLRetry(err, fmt.Sprintf("failed to get stream url, will retry in %s...", intervalText), 0) return } @@ -1308,7 +1451,7 @@ func (r *recorder) logStreamURLRetry(err error) { r.lastRetryLogAt = now r.suppressedRetryLogCount = 0 r.retryLogMu.Unlock() - r.warnStreamURLRetry(err, "failed to get stream url, will retry after 5s...", 0) + r.warnStreamURLRetry(err, fmt.Sprintf("failed to get stream url, will retry in %s...", intervalText), 0) return } @@ -1317,7 +1460,7 @@ func (r *recorder) logStreamURLRetry(err error) { r.lastRetryLogAt = now r.suppressedRetryLogCount = 0 r.retryLogMu.Unlock() - r.warnStreamURLRetry(err, "failed to get stream url, still retrying every 5s...", suppressed) + r.warnStreamURLRetry(err, fmt.Sprintf("failed to get stream url, still retrying, current interval %s", intervalText), suppressed) return } @@ -1451,6 +1594,14 @@ func (r *recorder) SetInitialRecordedFiles(files []notify.RecordingFileDetail) { // 分段重启时由 RestartRecorder 调用，确保旧分段的 Pipeline 结果不丢失 // 使用 redirect 链替代指针覆盖，支持连续重启 A→B→C 场景 func (r *recorder) TransferPipelineState(old *recorder) { + // "本场录上过"是单调事实，只能由 false 变 true，所以这里做或而不是覆盖： + // 新 recorder 在 RestartRecorder 里先被 Start，等这里执行时它自己可能已经拉上过一次流， + // 无条件写入会把那份成功抹回旧实例的 false，让后续重试退回"从未录上过"的 60 秒长间隔。 + // old 的 run() 已在 CloseForRestart 里等待退出，此处读旧值不会有并发写。 + if old.everHeldStream.Load() { + r.everHeldStream.Store(true) + } + r.pipelineState.mu.Lock() defer r.pipelineState.mu.Unlock() diff --git a/src/recorders/recorder_stream_selection_test.go b/src/recorders/recorder_stream_selection_test.go new file mode 100644 index 0000000..2856acc --- /dev/null +++ b/src/recorders/recorder_stream_selection_test.go @@ -0,0 +1,370 @@ +package recorders + +import ( + "context" + "fmt" + "net/url" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/bluele/gcache" + "github.com/sirupsen/logrus" + gomock "go.uber.org/mock/gomock" + + "github.com/bililive-go/bililive-go/src/configs" + "github.com/bililive-go/bililive-go/src/instance" + "github.com/bililive-go/bililive-go/src/live" + livemock "github.com/bililive-go/bililive-go/src/live/mock" + eventsmock "github.com/bililive-go/bililive-go/src/pkg/events/mock" + "github.com/bililive-go/bililive-go/src/pkg/livelogger" +) + +const testRoomURL = "https://example.com/room" + +// deferredLive 模拟实现了 live.DeferredStreamResolver 的平台（当前只有 Soop）： +// 列候选只给元信息，被点名的那一路才解析出播放地址。 +type deferredLive struct { + *livemock.MockLive + + candidates []*live.StreamUrlInfo + resolveErr func(candidate *live.StreamUrlInfo) error + afterResolve func(candidate *live.StreamUrlInfo) + resolveSeen *[]string + logger *livelogger.LiveLogger +} + +func (d *deferredLive) ListStreamCandidates() ([]*live.StreamUrlInfo, error) { + return d.candidates, nil +} + +func (d *deferredLive) ResolveStreamCandidate(candidate *live.StreamUrlInfo) error { + if d.resolveSeen != nil { + *d.resolveSeen = append(*d.resolveSeen, candidate.Quality) + } + if d.resolveErr != nil { + if err := d.resolveErr(candidate); err != nil { + return err + } + } + candidate.Url = &url.URL{Scheme: "https", Host: "cdn.example.com", Path: "/" + candidate.Quality + ".m3u8"} + if d.afterResolve != nil { + d.afterResolve(candidate) + } + return nil +} + +func qualityOf(name string) *string { + value := name + return &value +} + +func testLogger() *livelogger.LiveLogger { + return livelogger.New(0, logrus.Fields{"test": "stream_selection"}) +} + +// newDeferredLive 造一个只支持延迟解析的平台，并让它把每档的解析动作记进 seen。 +func newDeferredLive(t *testing.T, candidates []*live.StreamUrlInfo, resolveErr func(*live.StreamUrlInfo) error) (*deferredLive, *[]string) { + t.Helper() + ctrl := gomock.NewController(t) + base := livemock.NewMockLive(ctrl) + logger := testLogger() + base.EXPECT().GetLogger().Return(logger).AnyTimes() + base.EXPECT().GetRawUrl().Return(testRoomURL).AnyTimes() + base.EXPECT().GetPlatformCNName().Return("SOOP").AnyTimes() + seen := make([]string, 0, len(candidates)) + return &deferredLive{ + MockLive: base, + candidates: candidates, + resolveErr: resolveErr, + resolveSeen: &seen, + logger: logger, + }, &seen +} + +func sampleCandidates() []*live.StreamUrlInfo { + return []*live.StreamUrlInfo{ + {Quality: "HD", AttributesForStreamSelect: map[string]string{"quality_key": "HD"}}, + {Quality: "SD", AttributesForStreamSelect: map[string]string{"quality_key": "SD"}}, + } +} + +func TestGetStreamInfosForRecordingResolvesOnlyPreferredCandidate(t *testing.T) { + restoreConfig := swapTestConfig(t) + defer restoreConfig() + + l, seen := newDeferredLive(t, []*live.StreamUrlInfo{ + {Quality: "SD", AttributesForStreamSelect: map[string]string{"quality_key": "SD"}}, + {Quality: "HD", AttributesForStreamSelect: map[string]string{"quality_key": "HD"}}, + {Quality: "FHD", AttributesForStreamSelect: map[string]string{"quality_key": "FHD"}}, + }, nil) + + // 偏好 FHD 时平台顺序第一档是 SD，只有按偏好排过序才会先解析 FHD。 + streams, err := (&recorder{Live: l}).getStreamInfosForRecording(configs.StreamPreference{Quality: qualityOf("FHD")}) + if err != nil { + t.Fatalf("取流失败: %v", err) + } + if len(*seen) != 1 || (*seen)[0] != "FHD" { + t.Fatalf("应只解析被选中的那一路，实际解析顺序 %v", *seen) + } + if streams[0].Quality != "FHD" || streams[0].Url == nil { + t.Fatalf("首位应为已解析的 FHD，实际 %s url=%v", streams[0].Quality, streams[0].Url) + } + // 其余档位只用于面板展示，刻意不带播放地址。 + for _, s := range streams[1:] { + if s.Url != nil { + t.Fatalf("未解析的候选 %s 不应带播放地址", s.Quality) + } + } +} + +// TestPreferredStreamIgnoresPreferenceChangedMidAttempt 覆盖"解析期间用户改了画质偏好"这一时序： +// 一次解析要跑几百毫秒的平台请求，期间面板上的"切换清晰度"就会改写房间偏好。最终选择若再读一次 +// 全局配置，就会选中根本没解析过的那一路，其 Url 为空，后面解引用直接把录制线程 panic 掉， +// 而且是在持有 currentFileLock 时崩的，面板读状态的请求会跟着一起挂住。 +func TestPreferredStreamIgnoresPreferenceChangedMidAttempt(t *testing.T) { + restoreConfig := swapTestConfig(t) + defer restoreConfig() + + l, seen := newDeferredLive(t, sampleCandidates(), nil) + r := &recorder{Live: l} + pinned := configs.StreamPreference{Quality: qualityOf("HD")} + + streams, err := r.getStreamInfosForRecording(pinned) + if err != nil { + t.Fatalf("取流失败: %v", err) + } + if len(*seen) != 1 || (*seen)[0] != "HD" { + t.Fatalf("用例前提不成立：本轮解析的不是 HD，实际 %v", *seen) + } + + // 用户在这一轮解析期间把偏好改成了 SD：全局配置已经是新值，本轮仍必须用固定下来的 HD。 + setRoomPreference(configs.GetCurrentConfig(), testRoomURL, configs.StreamPreference{Quality: qualityOf("SD")}) + + selected := r.selectPreferredStream(streams, pinned) + if selected.Quality != "HD" || selected.Url == nil { + t.Fatalf("本轮应选中已解析的 HD，实际 %s url=%v", selected.Quality, selected.Url) + } + // 未解析的 SD 仍留在列表里供面板展示档位，说明"重读全局偏好"确实会挑中空地址那一路。 + for _, s := range streams { + if s.Quality == "SD" && s.Url != nil { + t.Fatal("用例前提不成立：SD 这一路已被解析") + } + } + if got := r.selectPreferredStream(streams, configs.GetCurrentConfig().GetEffectiveConfigForRoom(testRoomURL).StreamPreference); got.Url != nil { + t.Fatal("用例前提不成立：按新全局偏好重选竟拿到了已解析的地址，本用例挡不住回归") + } +} + +// TestTryRecordKeepsResolvedCandidateThroughoutAttempt 把上一条覆盖到真实的 tryRecord 里： +// 解析要跑几百毫秒的平台请求，用户在面板上点"切换清晰度"就落在这个往返中间， +// 而本轮后续的选路必须仍然按尝试开始时那份偏好走。偏好若在读一次之后再读一次， +// 选中的就是没解析过的档位，其播放地址为空，本轮直接作废（修复前这里连错误都拿不到， +// 是解引用空指针把录制线程崩掉）。 +// 输出路径故意指向一个普通文件，让 mkdir 失败，从而在选路之后立刻收住这一轮。 +func TestTryRecordKeepsResolvedCandidateThroughoutAttempt(t *testing.T) { + restoreConfig := swapTestConfig(t) + defer restoreConfig() + + blocker := filepath.Join(t.TempDir(), "not-a-dir") + if err := os.WriteFile(blocker, []byte("x"), 0o600); err != nil { + t.Fatalf("准备输出路径占位文件失败: %v", err) + } + cfg := configs.GetCurrentConfig() + cfg.OutputTmpl = "stream.ts" + cfg.OutPutPath = blocker + setRoomPreference(cfg, testRoomURL, configs.StreamPreference{Quality: qualityOf("HD")}) + + l, seen := newDeferredLive(t, sampleCandidates(), nil) + l.afterResolve = func(*live.StreamUrlInfo) { + // 模拟解析期间用户把偏好改成了 SD：SD 这一路本轮没有被解析过，没有播放地址。 + setRoomPreference(cfg, testRoomURL, configs.StreamPreference{Quality: qualityOf("SD")}) + } + + inst := &instance.Instance{Cache: gcache.New(1).LRU().Build()} + ctx := context.WithValue(context.Background(), instance.Key, inst) + ctrl := gomock.NewController(t) + ed := eventsmock.NewMockDispatcher(ctrl) + if err := inst.Cache.Set(l, &live.Info{Live: l}); err != nil { + t.Fatalf("写入直播信息缓存失败: %v", err) + } + r := &recorder{Live: l, cache: inst.Cache, ed: ed} + + r.tryRecord(ctx) + + if len(*seen) != 1 || (*seen)[0] != "HD" { + t.Fatalf("本轮应只解析偏好指定的 HD，实际解析顺序 %v", *seen) + } + logs := l.logger.GetLogs() + if strings.Contains(logs, "尚未解析出播放地址") { + t.Fatalf("选中了未解析的候选，本轮被空地址挡下：\n%s", logs) + } + if !strings.Contains(logs, "failed to create output path") { + t.Fatalf("用例前提不成立：本轮没有走到选路之后，日志为:\n%s", logs) + } + r.currentFileLock.RLock() + streamURL := r.currentStreamURL + r.currentFileLock.RUnlock() + if !strings.Contains(streamURL, "/HD.m3u8") { + t.Fatalf("本轮应录制已解析的 HD，实际 %q", streamURL) + } +} + +// TestTryRecordStopsWhenSelectedStreamHasNoURL 选中的那一路没有播放地址时，本轮必须带着原因退出。 +// 这里曾在持有 currentFileLock 的情况下解引用空 URL：panic 被 goroutine 边界 recover 之后 +// 写锁永远不会再释放，面板查直播间状态的读请求会全部挂住，表现成整个房间无响应而不是报错。 +func TestTryRecordStopsWhenSelectedStreamHasNoURL(t *testing.T) { + restoreConfig := swapTestConfig(t) + defer restoreConfig() + + ctrl := gomock.NewController(t) + base := livemock.NewMockLive(ctrl) + logger := testLogger() + base.EXPECT().GetLogger().Return(logger).AnyTimes() + base.EXPECT().GetRawUrl().Return(testRoomURL).AnyTimes() + base.EXPECT().GetPlatformCNName().Return("SOOP").AnyTimes() + // 缓存按 key 反射取哈希，会顺带读到 Options。 + base.EXPECT().GetOptions().Return(live.MustNewOptions()).AnyTimes() + // 只有元信息、没有播放地址的一路（延迟解析的列表就是这个形状） + base.EXPECT().GetStreamInfos().Return([]*live.StreamUrlInfo{{Quality: "HD"}}, nil) + + inst := &instance.Instance{Cache: gcache.New(1).LRU().Build()} + ctx := context.WithValue(context.Background(), instance.Key, inst) + if err := inst.Cache.Set(base, &live.Info{Live: base}); err != nil { + t.Fatalf("写入直播信息缓存失败: %v", err) + } + r := &recorder{Live: base, cache: inst.Cache, ed: eventsmock.NewMockDispatcher(ctrl)} + + r.tryRecord(ctx) + + if !strings.Contains(logger.GetLogs(), "清晰度 HD 尚未解析出播放地址") { + t.Fatalf("未记录空播放地址的失败原因：\n%s", logger.GetLogs()) + } + // 抢得到写锁才说明状态接口没有被挂住的读锁拖死。 + if !r.currentFileLock.TryLock() { + t.Fatal("currentFileLock 被泄漏的写锁占住") + } + r.currentFileLock.Unlock() +} + +// TestGetStreamInfosForRecordingDoesNotReResolveAllCandidates 全部候选都失败时不能顺手再全量解析一遍： +// 那等于在同一次尝试里把每一档的凭证与调度请求重新敲一次，而"全部失败"恰恰是最需要少发请求的状态。 +func TestGetStreamInfosForRecordingDoesNotReResolveAllCandidates(t *testing.T) { + restoreConfig := swapTestConfig(t) + defer restoreConfig() + + l, seen := newDeferredLive(t, sampleCandidates(), func(*live.StreamUrlInfo) error { + return fmt.Errorf("拒绝发放播放凭证") + }) + // 不给 MockLive 设 GetStreamInfos 期望：一旦被调用 gomock 直接判失败，等于断言没有回退全量解析。 + + _, err := (&recorder{Live: l}).getStreamInfosForRecording(configs.StreamPreference{}) + if err == nil { + t.Fatal("全部候选解析失败时应返回错误") + } + if want := "清晰度 HD: 拒绝发放播放凭证"; err.Error() != want { + t.Fatalf("错误应带上第一档的失败原因，实际 %q", err.Error()) + } + if len(*seen) != 2 { + t.Fatalf("应逐档各试一次后放弃，实际 %v", *seen) + } +} + +// TestTryRecordStopsWhenCandidateResolvesToOffline 列完候选之后房间才显示下播时， +// 解析这一档拿到的 ErrLiveOffline 必须一路传到 stopRetryForExplicitOffline： +// 既不能接着敲剩下档位的凭证请求，也不能把这次失败计进重试退避让面板一直转圈。 +func TestTryRecordStopsWhenCandidateResolvesToOffline(t *testing.T) { + restoreConfig := swapTestConfig(t) + defer restoreConfig() + + candidates := []*live.StreamUrlInfo{ + {Quality: "HD", AttributesForStreamSelect: map[string]string{"quality_key": "HD"}}, + {Quality: "SD", AttributesForStreamSelect: map[string]string{"quality_key": "SD"}}, + {Quality: "FHD", AttributesForStreamSelect: map[string]string{"quality_key": "FHD"}}, + } + l, seen := newDeferredLive(t, candidates, func(candidate *live.StreamUrlInfo) error { + if candidate.Quality == "HD" { + return fmt.Errorf("拒绝发放播放凭证") + } + return fmt.Errorf("%w: Soop 页面已明确显示下播", live.ErrLiveOffline) + }) + ctrl := gomock.NewController(t) + // 不给 GetStreamInfos 设期望：判定下播后不应再有任何取流请求。 + + inst := &instance.Instance{Cache: gcache.New(1).LRU().Build()} + ctx := context.WithValue(context.Background(), instance.Key, inst) + ed := eventsmock.NewMockDispatcher(ctrl) + ed.EXPECT().DispatchEvent(gomock.Any()) + r := &recorder{Live: l, cache: inst.Cache, ed: ed} + + r.tryRecord(ctx) + + if len(*seen) != 2 || (*seen)[1] != "SD" { + t.Fatalf("确认下播后不应再解析 FHD，实际解析顺序 %v", *seen) + } + if strings.Contains(l.logger.GetLogs(), "will retry in") { + t.Fatalf("下播不该进重试退避，日志为:\n%s", l.logger.GetLogs()) + } +} + +func TestTransferPipelineStateKeepsHeldStreamFact(t *testing.T) { + ctrl := gomock.NewController(t) + base := livemock.NewMockLive(ctrl) + base.EXPECT().GetLogger().Return(testLogger()).AnyTimes() + old := &recorder{Live: base, pipelineState: &pipelineSharedState{sourceNames: make(map[string]bool)}} + fresh := &recorder{Live: base, pipelineState: &pipelineSharedState{sourceNames: make(map[string]bool)}} + + // 新 recorder 在 RestartRecorder 里先被 Start，转移到它手上时它可能已经录上过； + // 旧分段"从未录上过"不能把这个事实抹掉，否则重试间隔会退回更长的封顶档位。 + fresh.everHeldStream.Store(true) + fresh.consecutiveFastFailures = 2 + fresh.TransferPipelineState(old) + if !fresh.everHeldStream.Load() { + t.Fatal("新实例已观测到的成功不应被旧实例覆盖") + } + // 观测到成功与没观测到成功，同样的失败次数落在不同档位上，这才是用户可见的差别。 + if got := fresh.pendingRetryInterval(); got != 15*time.Second { + t.Fatalf("已录过上的分段应保持 15s 封顶档，实际 %s", got) + } + notHeld := &recorder{Live: base, pipelineState: &pipelineSharedState{sourceNames: make(map[string]bool)}} + notHeld.consecutiveFastFailures = 2 + notHeld.TransferPipelineState(old) + if got := notHeld.pendingRetryInterval(); got != 20*time.Second { + t.Fatalf("未继承成功事实时应落在升档序列的 20s，实际 %s", got) + } + + old.everHeldStream.Store(true) + inherited := &recorder{Live: base, pipelineState: &pipelineSharedState{sourceNames: make(map[string]bool)}} + inherited.TransferPipelineState(old) + if !inherited.everHeldStream.Load() { + t.Fatal("旧分段录上过，新分段应沿用该事实") + } +} + +// swapTestConfig 装一份只含测试房间的空配置，返回还原函数。 +func swapTestConfig(t *testing.T) func() { + t.Helper() + previous := configs.GetCurrentConfig() + cfg := configs.NewConfig() + configs.SetCurrentConfig(cfg) + return func() { configs.SetCurrentConfig(previous) } +} + +// setRoomPreference 把房间级画质偏好写进配置，模拟用户在面板上点"切换清晰度"。 +func setRoomPreference(cfg *configs.Config, roomURL string, preference configs.StreamPreference) { + value := preference + for i := range cfg.LiveRooms { + if cfg.LiveRooms[i].Url == roomURL { + cfg.LiveRooms[i].StreamPreference = &value + return + } + } + cfg.LiveRooms = append(cfg.LiveRooms, configs.LiveRoom{ + Url: roomURL, + OverridableConfig: configs.OverridableConfig{ + StreamPreference: &value, + }, + }) +} diff --git a/src/servers/handler.go b/src/servers/handler.go index 4b26d1f..4e3fa0d 100644 --- a/src/servers/handler.go +++ b/src/servers/handler.go @@ -1026,6 +1026,11 @@ func getRawConfig(writer http.ResponseWriter, r *http.Request) { if cfg.DouyuAuth.DyDid != "" { cfg.DouyuAuth.DyDid = douyuSecretMask } + // Soop 账号密码同理：明文页是整份 YAML 的展示与回写入口，原文输出等于把密码交给浏览器。 + // 用与调试转储同一个占位符，写回侧（applyRawConfigDoc）据此识别"这一项没被用户改过"。 + if cfg.SoopLiveAuth.Password != "" { + cfg.SoopLiveAuth.Password = configs.LogSecretMask + } b, err := yaml.Marshal(cfg) if err != nil { writeJsonWithStatusCode(writer, http.StatusInternalServerError, commonResp{ @@ -1096,6 +1101,7 @@ func putRawConfig(writer http.ResponseWriter, r *http.Request) { } topKeys := rawConfigTopKeys(rawYaml) _, authSectionEdited := topKeys["douyu_auth"] + _, soopAuthSectionEdited := topKeys["sooplive_auth"] _, cookiesSectionEdited := topKeys["cookies"] // 整体替换必须走带锁的提交路径：SetCurrentConfig 是裸写全局指针，绕开 updateMu 会把与本次 // 编辑同期完成的其它更新（保存 cookie、斗鱼自动续期）从内存里覆盖掉；锁外再 Marshal 更是 @@ -1111,7 +1117,7 @@ func putRawConfig(writer http.ResponseWriter, r *http.Request) { // c 是最新配置的私有克隆：先另存一份提交前状态，供运行态房间差异比对使用 prevConfig = configs.CloneConfigShallow(c) prevConfig.RefreshLiveRoomIndexCache() - applyRawConfigDoc(c, snapshot, newConfig, authSectionEdited, cookiesSectionEdited) + applyRawConfigDoc(c, snapshot, newConfig, authSectionEdited, soopAuthSectionEdited, cookiesSectionEdited) return nil }, 3, 10*time.Millisecond) if err != nil { @@ -1138,7 +1144,7 @@ func putRawConfig(writer http.ResponseWriter, r *http.Request) { // snapshot 是用户打开该页时的配置，仅用于判断"这一处用户到底改没改"。 // 三方规则：文档相对快照未改动的内容一律取 latest（保住解析 YAML 这几十毫秒里完成的并发写入）， // 改动过的取文档值（以用户这次编辑为准）。latest 由 UpdateWithRetry 在锁内传入，是本函数的唯一写入目标。 -func applyRawConfigDoc(latest, snapshot, doc *configs.Config, authSectionEdited, cookiesSectionEdited bool) { +func applyRawConfigDoc(latest, snapshot, doc *configs.Config, authSectionEdited, soopAuthSectionEdited, cookiesSectionEdited bool) { // 每次重试都从原始文档重新算：mutator 可能被调用多次，不能污染 doc merged := configs.CloneConfigShallow(doc) @@ -1170,6 +1176,21 @@ func applyRawConfigDoc(latest, snapshot, doc *configs.Config, authSectionEdited, } } + // Soop 密码同理：明文页输出的是占位符，直接提交会把这串字符当成新密码存下来， + // 之后自动登录拿它去换 cookie 必然失败，且面板只显示"已保存凭证"看不出异常。 + docSoop := doc.SoopLiveAuth + if docSoop.Password == configs.LogSecretMask { + docSoop.Password = snapshot.SoopLiveAuth.Password + } + if !soopAuthSectionEdited || docSoop == snapshot.SoopLiveAuth { + // 整节没提交，或整节照抄没改：以提交时刻的最新值为准（这期间后台可能刚保存过新凭证）。 + merged.SoopLiveAuth = latest.SoopLiveAuth + } else if merged.SoopLiveAuth.Password == configs.LogSecretMask { + // 用户改了这一段里的别的东西（例如只改用户名）：密码位仍还原成最新原文。 + // 真要清除时把 password 留空提交即可。 + merged.SoopLiveAuth.Password = latest.SoopLiveAuth.Password + } + // cookies 同理做逐主机三方合并：用户在编辑器里没动过的那条，若已被后台续期改写，取最新值。 // 整节缺失视为"这一节没提交"（局部客户端不认识它），不是"把所有登录 cookie 清空"。 if !cookiesSectionEdited { diff --git a/src/webapp/src/component/config-info/index.tsx b/src/webapp/src/component/config-info/index.tsx index f51b9b6..9a2313a 100644 --- a/src/webapp/src/component/config-info/index.tsx +++ b/src/webapp/src/component/config-info/index.tsx @@ -2291,6 +2291,13 @@ const ConfigInfo: React.FC = () => { // YAML 模式内容 const renderYamlMode = () => (
+ setRawConfig(code)}