Merge pull request #1212 from Joftal/dev
feat(live/sooplive,recorders,configs,servers): Soop 取流降请求与密码出口封堵
This commit is contained in:
11 files changed
+1454
-161
No files matched your search
@@ -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。
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
+424
-104
@@ -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 {
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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()
|
||||
}
|
||||
+201
-50
@@ -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()
|
||||
|
||||
|
||||
@@ -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,
|
||||
},
|
||||
})
|
||||
}
|
||||
+23
-2
@@ -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 {
|
||||
|
||||
@@ -2291,6 +2291,13 @@ const ConfigInfo: React.FC = () => {
|
||||
// YAML 模式内容
|
||||
const renderYamlMode = () => (
|
||||
<div className="config-content">
|
||||
<Alert
|
||||
type="warning"
|
||||
showIcon
|
||||
style={{ marginBottom: 12 }}
|
||||
message="这里是配置文件的快照,编辑久了保存前建议先重新加载一次"
|
||||
description="后台会在运行中自动续期各平台 Cookie。掩码显示的值(如 Soop 密码)保存时始终沿用服务器上的最新原文,不受这份快照影响;Cookie 则按“这条与服务器当前值是否一致”来判断你有没有改过——拿着过期快照保存,里面那条旧 Cookie 会被当成你手写的改动,把期间刚续好的新值顶回去。注意“刷新”会用服务器上的配置覆盖编辑器里尚未保存的修改。"
|
||||
/>
|
||||
<Editor
|
||||
value={rawConfig}
|
||||
onValueChange={code => setRawConfig(code)}
|
||||
|
||||
Reference in new issue
Block a user