perf(live/sooplive,live,live/system,recorders): Soop 只解析选中清晰度并在同轮复用播放页

- 取流改为按 DeferredStreamResolver 延迟解析:先列档位、只给要录的那一路申请 aid,
  开播房间的请求量由 8 次降到 4 次;包装层与初始化窗口一并转发,平台未实现时返回
  ErrNotImplemented 由录制器回退全量解析,其他平台选路结果与旧算法逐位等价。
- 播放页与频道信息在 15s 窗口内复用成功结果:只缓存"已成功且开播"和 result 正常的
  播放信息,返回副本供调用方就地排序,登录态变化与房间配置更新时整体作废。
- 取流请求按原始 URL 做 singleflight 合并,同一轮的重复解析只发一次网络请求。
- 录制失败重试改为升档退避(5s 起倍增,从未录上过封顶 60s、本场录上过封顶 15s),
  间隔随分段重启继承,避免下播后仍以 5s 频率反复冲击平台接口。
This commit is contained in:
Joftal committed 2026-09-29 19:59:46 +08:00
1 parent 6822de38c7
commit cdad77fa51
6 files changed
+791 -151

No files matched your search

+33
View File
@@ -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 {
+53
View File
@@ -303,6 +303,59 @@ 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()
}
// 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("soop 自动登录处于冷却期,%.0f 秒后重试", 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
+385 -101
View File
@@ -10,6 +10,7 @@ import (
"strconv"
"strings"
"sync"
"time"
"github.com/hr3lxphr6j/requests"
"github.com/tidwall/gjson"
@@ -30,12 +31,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 +63,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 +140,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 +166,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 +197,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 +237,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 +433,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 +477,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 +718,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 +834,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 +871,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 +910,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,9 +932,10 @@ 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 {
l.GetLogger().WithError(err).Debug("Soop 自动登录失败")
return err
@@ -674,6 +954,8 @@ func (l *Live) tryAutoLogin() error {
l.GetLogger().WithError(err).Warn("更新 Soop Cookie 到配置失败")
}
l.setRuntimeState(result.Cookie, false)
// 换了登录身份,旧会话抓到的播放页与播放信息一律不再复用
l.clearReuseCache()
l.GetLogger().Debugf("Soop 自动登录成功: loginID=%s cookie_length=%d", result.Verify.LoginID, len(result.Cookie))
for _, targetURL := range []*url.URL{playSoopURL, l.Url} {
@@ -751,6 +1033,8 @@ func (l *Live) setIgnoreStoredCookie(ignoreStoredCookie bool) {
func (l *Live) resetRuntimeState() {
l.setRuntimeState("", false)
// 登录态或房间配置已经变化,之前复用的播放页与播放信息都可能不再成立
l.clearReuseCache()
}
func buildCookieStringFromCookies(cookies []*http.Cookie) string {
+134 -4
View File
@@ -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,123 @@ 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))
}
func TestGetCookieMapPrefersLatestConfigCookieOverStaleOptions(t *testing.T) {
@@ -706,6 +830,8 @@ func TestTryVerifyAndReloginIfNeededDeduplicatesAutoLogin(t *testing.T) {
verifyCookieCache = map[string]cachedVerifyResult{}
verifyCookieCacheMu.Unlock()
}()
resetSoopAutoLoginCooldownForTest()
defer resetSoopAutoLoginCooldownForTest()
verifyCookieCacheTTL = 0
@@ -773,6 +899,8 @@ func TestTryAutoLoginDeduplicatesCookiePersistence(t *testing.T) {
loginAndGetCookie = oldLogin
setCookiesFunc = oldSetCookies
}()
resetSoopAutoLoginCooldownForTest()
defer resetSoopAutoLoginCooldownForTest()
cfg := configs.NewConfig()
cfg.SoopLiveAuth.Username = "tester"
@@ -833,6 +961,8 @@ func TestTryAutoLoginRejectsUnverifiedLoginResult(t *testing.T) {
loginAndGetCookie = oldLogin
setCookiesFunc = oldSetCookies
}()
resetSoopAutoLoginCooldownForTest()
defer resetSoopAutoLoginCooldownForTest()
cfg := configs.NewConfig()
cfg.SoopLiveAuth.Username = "tester"
+16
View File
@@ -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()
}
+170 -46
View File
@@ -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,7 @@ func (r *recorder) tryRecord(ctx context.Context) {
var streamInfos []*live.StreamUrlInfo
var err error
if streamInfos, err = r.Live.GetStreamInfos(); err == live.ErrNotImplemented {
if streamInfos, err = r.getStreamInfosForRecording(); err == live.ErrNotImplemented {
var urls []*url.URL
// TODO: remove deprecated method GetStreamUrls
//nolint:staticcheck
@@ -412,13 +433,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()
@@ -642,6 +659,7 @@ func (r *recorder) tryRecord(ctx context.Context) {
}
r.setAndCloseParser(p)
r.startTime = time.Now()
parseStartedAt := r.startTime
// 弹幕录制(支持哔哩哔哩、抖音、斗鱼平台)
if resolvedConfig.DanmakuEnable {
@@ -692,10 +710,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 +950,118 @@ func (r *recorder) stopRetryForExplicitOffline(err error) bool {
return true
}
// getStreamInfosForRecording 取得本次录制可用的流列表。
// 平台支持"先列候选、只解析选中的那一路"时优先走那条路径:Soop 每一档清晰度都要单独申请
// 一次播放凭证并查一次调度接口,全量解析的请求量随档位成倍增长,而每次录制实际只用一路。
// 候选里除了被选中的一路之外只有元信息(Url 为空),面板的清晰度列表因此仍然完整。
func (r *recorder) getStreamInfosForRecording() ([]*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 := r.orderStreamsByPreference(candidates)
for i, candidate := range ordered {
if err := resolver.ResolveStreamCandidate(candidate); err != nil {
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
}
// 所有候选都解析失败:回退全量解析,保留原有的错误语义
return r.Live.GetStreamInfos()
}
// orderStreamsByPreference 按偏好分值从高到低排序,分值相同保持平台返回的先后顺序,
// 使"依次尝试"选出的第一路等于 selectPreferredStream 在全量列表上的选择结果。
func (r *recorder) orderStreamsByPreference(candidates []*live.StreamUrlInfo) []*live.StreamUrlInfo {
preference := r.effectiveStreamPreference()
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) (ret *live.StreamUrlInfo) {
// 如果没有可用流,直接返回 nil
if len(streamInfos) == 0 {
return nil
}
streamPreference := configs.GetCurrentConfig().GetEffectiveConfigForRoom(r.Live.GetRawUrl()).StreamPreference
preference := r.effectiveStreamPreference()
// 如果未配置流偏好(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
}
// effectiveStreamPreference 取本房间的流偏好(已合并平台与全局层级配置)。
func (r *recorder) effectiveStreamPreference() configs.StreamPreference {
return configs.GetCurrentConfig().GetEffectiveConfigForRoom(r.Live.GetRawUrl()).StreamPreference
}
// 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 +1077,6 @@ func (r *recorder) run(ctx context.Context) {
}
}()
const minRetryInterval = 5 * time.Second
for {
select {
case <-r.stop:
@@ -1007,10 +1089,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 +1106,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 +1407,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 +1427,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 +1436,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 +1570,11 @@ func (r *recorder) SetInitialRecordedFiles(files []notify.RecordingFileDetail) {
// 分段重启时由 RestartRecorder 调用,确保旧分段的 Pipeline 结果不丢失
// 使用 redirect 链替代指针覆盖,支持连续重启 A→B→C 场景
func (r *recorder) TransferPipelineState(old *recorder) {
// 分段重启说明这条流在本会话里确实录上过,沿用旧 recorder 的事实,
// 避免新分段把重试档位退回"从未录上过"的长间隔。
// old 的 run() 已在 CloseForRestart 里等待退出,此处读取是安全的。
r.everHeldStream.Store(old.everHeldStream.Load())
r.pipelineState.mu.Lock()
defer r.pipelineState.mu.Unlock()