Compare commits

...

2 Commits

Author SHA1 Message Date
truewhile 3d30f6f27c bug处理 2026-09-08 21:08:47 +08:00
truewhile bd02e194ff 优化 2026-09-08 20:52:35 +08:00
9 changed files with 275 additions and 36 deletions
+13 -10
View File
@@ -150,6 +150,7 @@ func (c *OpenClient) doJSON(ctx context.Context, method, rawURL string, form map
if err != nil {
return nil, err
}
attemptedAccess := strings.TrimPrefix(req.Header.Get("Authorization"), "Bearer ")
resp, err := c.HTTP.Do(req)
if err != nil {
@@ -218,7 +219,7 @@ func (c *OpenClient) doJSON(ctx context.Context, method, rawURL string, form map
// refresh_token 刷新后重试一次。刷新失败或重试后仍失败才返回,
// 避免长时间同步因 token 过期而整体失败。
if isTokenCode(base.Code) {
if access && c.tryRefreshTokenLocked(ctx) {
if access && c.tryRefreshTokenLocked(ctx, attemptedAccess) {
continue
}
if access {
@@ -305,21 +306,23 @@ func (c *OpenClient) doAuthJSONWithUA(ctx context.Context, method, rawURL string
// tryRefreshTokenLocked 并发安全地刷新 access_token;成功返回 true(调用方
// 应使用内存中的新 token 重试原请求)。
//
// 拿到写锁后在锁内读取 oldAccess,与持锁期间的当前值对比:若已被其他
// goroutine 刷新过则直接复用新 token,避免并发请求连环轮转消耗 115 的
// 一次性 refresh_token。全程持写锁读写 token 字段,无 TOCTOU 窗口。
// failedAccess 是失败请求实际携带的 token。拿到写锁后与当前 token 对比:
// 若已被其他 goroutine 刷新过则直接复用,避免并发请求连环轮转消耗 115
// 的一次性 refresh_token。全程持写锁读写 token 字段,无 TOCTOU 窗口。
//
// 对"refresh_token 本身已失效/被吊销"(IsRefreshTokenDead,如 40140114/116/119/120)
// 这类不可恢复的错误直接放弃并清空内存 token(提示需重新授权)。
// 对其它失败(网络瞬时抖动、刷新接口可重试错误码等)做指数退避重试几次再放弃,
// 避免同步长任务中途 token 到期时恰好撞上一个短暂的刷新失败就整体失败。
func (c *OpenClient) tryRefreshTokenLocked(ctx context.Context) bool {
func (c *OpenClient) tryRefreshTokenLocked(ctx context.Context, failedAccess string) bool {
c.tokenMu.Lock()
// 在已持有写锁内读取当前 token 作为"刷新前快照",消除双重加锁窗口:
// 若在拿锁期间已有其他 goroutine 完成刷新,refreshTokenWhileLocked
// 内的 c.AccessToken != oldAccess 判断会立即命中并返回复用。
oldAccess := c.AccessToken
newToken, ok := c.refreshTokenWhileLocked(ctx, oldAccess)
// 请求发出后若其他 goroutine 已经刷新完成,直接复用新 token 重试;
// 不能再次轮换一次性的 refresh_token。
if failedAccess != "" && c.AccessToken != failedAccess {
c.tokenMu.Unlock()
return true
}
newToken, ok := c.refreshTokenWhileLocked(ctx, failedAccess)
c.tokenMu.Unlock()
// 回调必须在 tokenMu 释放后调用,避免上层在回调内访问客户端时死锁
if ok && newToken != nil && c.OnTokenRefreshed != nil {
@@ -140,6 +140,47 @@ func TestRefreshTokenDead(t *testing.T) {
}
}
func TestConcurrentTokenFailuresShareOneRefresh(t *testing.T) {
var mu sync.Mutex
refreshCalls := 0
mockAPI(t, func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/open/refreshToken" {
t.Errorf("unexpected path %s", r.URL.Path)
}
mu.Lock()
refreshCalls++
mu.Unlock()
time.Sleep(50 * time.Millisecond)
_, _ = w.Write([]byte(`{"state":true,"data":{"access_token":"at-new","refresh_token":"rt-new","expires_in":7200}}`))
})
client := NewOpenClient("100195129", "at-old", "rt-old")
var wg sync.WaitGroup
results := make(chan bool, 2)
for range 2 {
wg.Add(1)
go func() {
defer wg.Done()
results <- client.tryRefreshTokenLocked(context.Background(), "at-old")
}()
}
wg.Wait()
close(results)
for ok := range results {
if !ok {
t.Fatal("concurrent refresh should reuse the refreshed token")
}
}
mu.Lock()
defer mu.Unlock()
if refreshCalls != 1 {
t.Fatalf("refresh calls = %d, want 1", refreshCalls)
}
if client.CurrentAccessToken() != "at-new" {
t.Fatalf("access token = %q, want at-new", client.CurrentAccessToken())
}
}
func TestFsListAndDownload(t *testing.T) {
mockAPI(t, func(w http.ResponseWriter, r *http.Request) {
switch r.URL.Path {
+11 -3
View File
@@ -477,24 +477,32 @@ func (c *OpenClient) GetToken(qrCode *QrCodeDataReturn) (*TokenData, error) {
// RefreshToken 刷新访问令牌。
func (c *OpenClient) RefreshToken(refreshToken string) (*TokenData, error) {
c.tokenMu.Lock()
if refreshToken == "" {
refreshToken = c.currentRefreshToken()
refreshToken = c.RefreshTokenStr
}
if refreshToken == "" {
c.tokenMu.Unlock()
return nil, fmt.Errorf("没有可用的 refresh_token")
}
token, err := c.doRefreshToken(refreshToken)
if err != nil {
// refresh_token 已失效时清空内存令牌(提示需重新授权)
if IsRefreshTokenDead(err) {
c.SetAuthToken("", "")
c.setAuthTokenLocked("", "")
}
c.tokenMu.Unlock()
return nil, err
}
if token.AccessToken == "" || token.RefreshToken == "" {
c.tokenMu.Unlock()
return nil, fmt.Errorf("115: 刷新返回空凭证(access_token/refresh_token 缺失)")
}
c.SetAuthToken(token.AccessToken, token.RefreshToken)
c.setAuthTokenLocked(token.AccessToken, token.RefreshToken)
c.tokenMu.Unlock()
if c.OnTokenRefreshed != nil {
c.OnTokenRefreshed(token.AccessToken, token.RefreshToken)
}
return token, nil
}
+14 -3
View File
@@ -293,7 +293,11 @@ func (s *StrmService) save115OAuthToken(ctx context.Context, session *strm115Aut
acct.LastTestAt = &now
acct.LastTestResult = "授权成功"
acct.LastTestOK = true
return s.repo.StrmAccount.Update(ctx, acct)
if err := s.repo.StrmAccount.Update(ctx, acct); err != nil {
return err
}
s.invalidate115Provider(acct.ID)
return nil
}
func (s *StrmService) drop115AuthSession(sessionID string) {
@@ -382,8 +386,15 @@ func (s *StrmService) refresh115TokensOnce(ctx context.Context) {
}
}
}
client := cloud115.NewOpenClient(cfg["app_id"], cfg["access_token"], cfg["refresh_token"])
token, err := client.RefreshToken(cfg["refresh_token"])
provider, err := s.providerFor(ctx, acct)
if err != nil {
continue
}
openProvider, ok := provider.(interface{ OpenClient() *cloud115.OpenClient })
if !ok || openProvider.OpenClient() == nil {
continue
}
token, err := openProvider.OpenClient().RefreshToken("")
if err != nil {
msg := "令牌刷新失败:" + err.Error()
if cloud115.IsRefreshTokenDead(err) {
+85
View File
@@ -10,6 +10,7 @@ import (
"github.com/truewhile/MeBox/internal/config"
"github.com/truewhile/MeBox/internal/model"
"github.com/truewhile/MeBox/internal/repository"
"github.com/truewhile/MeBox/internal/service/cloud115"
)
func TestStrmAccountConfigPreviewOf(t *testing.T) {
@@ -71,6 +72,90 @@ func TestUpdateStrmAccountMergesConfigWithoutClearingSecrets(t *testing.T) {
}
}
func TestProviderForReuses115ClientUntilCredentialsChange(t *testing.T) {
svc := testStrmService(t)
ctx := context.Background()
acct, err := svc.CreateStrmAccount(ctx, "115", model.StrmProvider115, map[string]string{
"app_id": "100195129",
"access_token": "at-1",
"refresh_token": "rt-1",
})
if err != nil {
t.Fatalf("create account: %v", err)
}
first, err := svc.providerFor(ctx, acct)
if err != nil {
t.Fatalf("first provider: %v", err)
}
second, err := svc.providerFor(ctx, acct)
if err != nil {
t.Fatalf("second provider: %v", err)
}
if first != second {
t.Fatal("115 provider should be shared per account")
}
updated, err := svc.UpdateStrmAccount(ctx, acct.ID, "", nil, map[string]string{
"access_token": "at-2",
"refresh_token": "rt-2",
})
if err != nil {
t.Fatalf("update account: %v", err)
}
third, err := svc.providerFor(ctx, updated)
if err != nil {
t.Fatalf("provider after credential update: %v", err)
}
if first == third {
t.Fatal("credential update must invalidate the shared provider")
}
oldClient := first.(interface{ OpenClient() *cloud115.OpenClient }).OpenClient()
oldClient.OnTokenRefreshed("at-stale", "rt-stale")
fresh, err := svc.repo.StrmAccount.FindByID(ctx, acct.ID)
if err != nil || fresh == nil {
t.Fatalf("reload account: %v", err)
}
freshCfg, err := svc.strmAccountConfig(fresh)
if err != nil {
t.Fatalf("decode refreshed account: %v", err)
}
if freshCfg["access_token"] != "at-2" || freshCfg["refresh_token"] != "rt-2" {
t.Fatalf("stale client overwrote new credentials: %#v", freshCfg)
}
}
func TestPersist115TokensKeepsSharedClient(t *testing.T) {
svc := testStrmService(t)
ctx := context.Background()
acct, err := svc.CreateStrmAccount(ctx, "115", model.StrmProvider115, map[string]string{
"app_id": "100195129",
"access_token": "at-1",
"refresh_token": "rt-1",
})
if err != nil {
t.Fatalf("create account: %v", err)
}
first, err := svc.providerFor(ctx, acct)
if err != nil {
t.Fatalf("first provider: %v", err)
}
svc.persist115Tokens(acct.ID, "at-2", "rt-2")
fresh, err := svc.repo.StrmAccount.FindByID(ctx, acct.ID)
if err != nil || fresh == nil {
t.Fatalf("reload account: %v", err)
}
second, err := svc.providerFor(ctx, fresh)
if err != nil {
t.Fatalf("second provider: %v", err)
}
if first != second {
t.Fatal("automatic token persistence must keep the in-memory shared client")
}
}
func TestDeleteStrmAccountCascadesEmbyMounts(t *testing.T) {
ctx := context.Background()
db := newServiceTestDB(t, &model.StrmAccount{}, &model.EmbyMount{}, &model.StrmSyncPath{})
+55 -10
View File
@@ -95,6 +95,9 @@ type StrmService struct {
oauthSessions map[string]*strm115AuthSession
wafUntil time.Time // 115 风控/限流熔断截止时间(由 mu 保护)
providerMu sync.Mutex
provider115Cache map[string]cloud.Provider // account ID -> shared provider/OpenClient
downloadSem115 chan struct{} // 115 换直链+下载并发上限(风控兜底)
downloadSemDAV chan struct{} // WebDAV/OpenList/CloudDrive2 元数据下载并发上限
downloadSemOnce sync.Once
@@ -158,15 +161,16 @@ func (s *StrmService) releaseDownloadSlot(provider string) {
// NewStrmService constructs the STRM service.
func NewStrmService(cfg *config.Config, log *zap.Logger, repos *repository.Container, crypto *CryptoService) *StrmService {
return &StrmService{
log: log,
repo: repos,
cfg: cfg,
crypto: crypto,
http: &http.Client{Timeout: 90 * time.Second},
stopCh: make(chan struct{}),
baseCtx: context.Background(),
running: map[string]context.CancelFunc{},
oauthSessions: map[string]*strm115AuthSession{},
log: log,
repo: repos,
cfg: cfg,
crypto: crypto,
http: &http.Client{Timeout: 90 * time.Second},
stopCh: make(chan struct{}),
baseCtx: context.Background(),
running: map[string]context.CancelFunc{},
oauthSessions: map[string]*strm115AuthSession{},
provider115Cache: map[string]cloud.Provider{},
}
}
@@ -409,6 +413,9 @@ func (s *StrmService) UpdateStrmAccount(ctx context.Context, id, name string, en
if err := s.repo.StrmAccount.Update(ctx, acct); err != nil {
return nil, err
}
if acct.Provider == model.StrmProvider115 && len(config) > 0 {
s.invalidate115Provider(acct.ID)
}
return acct, nil
}
@@ -426,6 +433,7 @@ func (s *StrmService) DeleteStrmAccount(ctx context.Context, id string) error {
if err := s.repo.StrmAccount.Delete(ctx, id); err != nil {
return err
}
s.invalidate115Provider(id)
// 级联清理远程 Emby 挂载:否则留下孤儿挂载,挂载计数/列表仍会显示。
// 账号已删,挂载清理失败只记日志,不让删除请求报错。
if _, err := s.repo.EmbyMount.DeleteByAccountID(ctx, id); err != nil && s.log != nil {
@@ -464,6 +472,23 @@ func (s *StrmService) ListAccounts(ctx context.Context) ([]model.StrmAccount, er
// providerFor 依据账号配置构建网盘驱动。
func (s *StrmService) providerFor(ctx context.Context, acct *model.StrmAccount) (cloud.Provider, error) {
if acct != nil && acct.Provider == model.StrmProvider115 {
s.providerMu.Lock()
defer s.providerMu.Unlock()
if provider := s.provider115Cache[acct.ID]; provider != nil {
return provider, nil
}
provider, err := s.newProvider(ctx, acct)
if err != nil {
return nil, err
}
s.provider115Cache[acct.ID] = provider
return provider, nil
}
return s.newProvider(ctx, acct)
}
func (s *StrmService) newProvider(ctx context.Context, acct *model.StrmAccount) (cloud.Provider, error) {
cfg, err := s.strmAccountConfig(acct)
if err != nil {
return nil, err
@@ -482,13 +507,33 @@ func (s *StrmService) providerFor(ctx context.Context, acct *model.StrmAccount)
// refresh_token 再刷(一次性轮转),两者互相作废,最终把有效账号
// 标成“授权已失效”。
if oc, ok := provider.(interface{ OpenClient() *cloud115.OpenClient }); ok {
oc.OpenClient().OnTokenRefreshed = func(accessToken, refreshToken string) {
client := oc.OpenClient()
client.OnTokenRefreshed = func(accessToken, refreshToken string) {
// 账号重新授权/修改凭据后,旧客户端可能仍有在途请求。旧请求
// 刷新的令牌不能覆盖新授权写入的凭据。
if !s.isCurrent115Client(acct.ID, client) {
return
}
s.persist115Tokens(acct.ID, accessToken, refreshToken)
}
}
return provider, nil
}
func (s *StrmService) invalidate115Provider(accountID string) {
s.providerMu.Lock()
delete(s.provider115Cache, accountID)
s.providerMu.Unlock()
}
func (s *StrmService) isCurrent115Client(accountID string, client *cloud115.OpenClient) bool {
s.providerMu.Lock()
defer s.providerMu.Unlock()
provider := s.provider115Cache[accountID]
openProvider, ok := provider.(interface{ OpenClient() *cloud115.OpenClient })
return ok && openProvider.OpenClient() == client
}
// ─── 全局设置 ──────────────────────────────────────────────────────────────────
// GetStrmSettings 返回全局 strm 设置(含默认值)。
+20 -4
View File
@@ -27,6 +27,7 @@ func srtToVTT(body string) string {
func assToVTT(body string) string {
out := strings.Builder{}
out.WriteString("WEBVTT\n\n")
seen := make(map[string]struct{})
for i, line := range strings.Split(body, "\n") {
line = strings.TrimSpace(line)
if !strings.HasPrefix(line, "Dialogue:") {
@@ -36,11 +37,22 @@ func assToVTT(body string) string {
if len(parts) < 10 {
continue
}
start := normaliseTimecode(parts[1])
end := normaliseTimecode(parts[2])
text := stripASSTags(parts[9])
if text == "" {
continue
}
key := start + "\x00" + end + "\x00" + text
if _, duplicate := seen[key]; duplicate {
continue
}
seen[key] = struct{}{}
fmt.Fprintf(&out, "%d\n%s --> %s\n%s\n\n",
i,
normaliseTimecode(parts[1]),
normaliseTimecode(parts[2]),
stripASSTags(parts[9]),
start,
end,
text,
)
}
return out.String()
@@ -86,5 +98,9 @@ func normaliseTimecode(t string) string {
var assTag = regexp.MustCompile(`\{[^}]*\}`)
func stripASSTags(s string) string {
return assTag.ReplaceAllString(s, "")
s = assTag.ReplaceAllString(s, "")
s = strings.ReplaceAll(s, `\N`, "\n")
s = strings.ReplaceAll(s, `\n`, "\n")
s = strings.ReplaceAll(s, `\h`, "\u00a0")
return strings.TrimSpace(s)
}
+20
View File
@@ -5,6 +5,7 @@ import (
"encoding/json"
"os"
"path/filepath"
"strings"
"testing"
"github.com/glebarez/sqlite"
@@ -84,6 +85,25 @@ func TestNormaliseTimecode(t *testing.T) {
}
}
func TestAssToVTTDeduplicatesDialogueAndNormalisesLineBreaks(t *testing.T) {
body := strings.Join([]string{
`Dialogue: 0,0:00:01.00,0:00:02.00,Default,,0,0,0,,{\an2}第一行\N第二行`,
`Dialogue: 1,0:00:01.00,0:00:02.00,Copy,,0,0,0,,{\bord2}第一行\N第二行`,
`Dialogue: 0,0:00:03.00,0:00:04.00,Default,,0,0,0,,{\i1}`,
}, "\n")
got := assToVTT(body)
if strings.Count(got, "第一行\n第二行") != 1 {
t.Fatalf("duplicate ASS dialogue was not collapsed:\n%s", got)
}
if strings.Contains(got, `\N`) || strings.Contains(got, `\an2`) {
t.Fatalf("ASS control sequences leaked into WebVTT:\n%s", got)
}
if strings.Contains(got, "00:00:03.000 --> 00:00:04.000") {
t.Fatalf("empty styled dialogue should be omitted:\n%s", got)
}
}
func TestSubtitleServeRawWritesSourceBytes(t *testing.T) {
db, err := gorm.Open(sqlite.Open("file::memory:?cache=shared"), &gorm.Config{})
if err != nil {
+16 -6
View File
@@ -47,6 +47,14 @@ function parseWebVTTCues(body: string): SubtitleCue[] {
return cues
}
function uniqueSubtitleTexts(texts: string[]): string[] {
const unique = [...new Set(texts.map((text) => text.trim()).filter(Boolean))]
// ponytail: WebVTT cannot preserve ASS layers/positions. Keep at most two
// simultaneous blocks for bilingual subtitles; use libass if full ASS
// typesetting support is added later.
return unique.slice(-2)
}
type PlayerVideoStageProps = {
media: Media | null
/** 媒体元数据加载失败提示(非空时替代「加载中」展示)。 */
@@ -255,6 +263,7 @@ export function PlayerVideoStage({
const selectedTrack = subs[subtitleIndex]
if (
!video ||
!tracksArmed ||
subs.length === 0 ||
subtitleIndex < 0 ||
!selectedTrack ||
@@ -267,9 +276,11 @@ export function PlayerVideoStage({
const updateCue = () => {
const absoluteTime = video.currentTime + (streamOffset ?? 0)
if (subtitleTimeline?.path === selectedTrack.path) {
const texts = subtitleTimeline.cues
.filter((cue) => absoluteTime >= cue.startTime && absoluteTime <= cue.endTime)
.map((cue) => cue.text)
const texts = uniqueSubtitleTexts(
subtitleTimeline.cues
.filter((cue) => absoluteTime >= cue.startTime && absoluteTime <= cue.endTime)
.map((cue) => cue.text),
)
setActiveCueText(texts.join('\n'))
return
}
@@ -304,7 +315,7 @@ export function PlayerVideoStage({
}
}
}
setActiveCueText(texts.join('\n'))
setActiveCueText(uniqueSubtitleTexts(texts).join('\n'))
}
const apply = () => {
@@ -359,7 +370,7 @@ export function PlayerVideoStage({
}
}
}
}, [subtitleIndex, subs, videoRef, media, streamOffset, subtitleTimeline])
}, [subtitleIndex, subs, videoRef, media, streamOffset, subtitleTimeline, tracksArmed])
// 根据视频画面宽高比与舞台宽高比,确定视频在哪个轴向撑满 100%
const isWiderThanStage =
@@ -412,7 +423,6 @@ export function PlayerVideoStage({
src={subtitlesAPI.url(media.id, track.path)}
srcLang={track.lang}
label={track.label || track.lang}
default={subtitleIndex === index}
/>
),
)}