From 60b11bc2bcf5cac5636008f8ce32ce6ccc06448a Mon Sep 17 00:00:00 2001 From: truewhile <62226914+truewhile@users.noreply.github.com> Date: Mon, 24 Aug 2026 22:38:23 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 优化 --- internal/service/cloud115/client.go | 34 ++++++++- internal/service/strm_sync.go | 106 +++++++++++++++++++++------- internal/service/strm_sync_test.go | 106 ++++++++++++++++++++++++++++ 3 files changed, 218 insertions(+), 28 deletions(-) diff --git a/internal/service/cloud115/client.go b/internal/service/cloud115/client.go index 94ecca8..bea72f4 100644 --- a/internal/service/cloud115/client.go +++ b/internal/service/cloud115/client.go @@ -11,6 +11,7 @@ import ( "net/http" "net/url" "strings" + "sync" "time" ) @@ -21,6 +22,10 @@ type OpenClient struct { AccessToken string RefreshTokenStr string executor *QueueExecutor + + // tokenMu 保护令牌刷新：业务请求中途 access_token 失效时自动刷新重试， + // 多 goroutine（同步列表 + 下载队列）并发下只允许一次刷新进行。 + tokenMu sync.Mutex } // default115HTTPClient 创建带有防 405 重定向保护的 http.Client。 @@ -175,9 +180,18 @@ func (c *OpenClient) doJSON(ctx context.Context, method, rawURL string, form map return &base, lastErr } - // Token 失效不重试 + // Token 失效（access_token 过期，如长时间同步中途过期）：自动用 + // refresh_token 刷新后重试一次。刷新失败或重试后仍失败才返回， + // 避免长时间同步因 token 过期而整体失败。 if isTokenCode(base.Code) { - return &base, nil + if access && c.tryRefreshTokenLocked() { + continue + } + if access { + // 刷新失败（或已刷新仍失败）时返回明确错误 + lastErr = NewOpenAPIResponseError(base.Code, base.Errno, base.Message, base.Error, "115: access_token 校验失败且刷新未成功") + } + return &base, lastErr } lastErr = NewOpenAPIResponseError(base.Code, base.Errno, base.Message, base.Error, "115 接口调用失败") @@ -243,6 +257,22 @@ func (c *OpenClient) doAuthJSONWithUA(ctx context.Context, method, rawURL string return c.doJSON(ctx, method, rawURL, form, true, retries, ua) } +// tryRefreshTokenLocked 并发安全地刷新 access_token；成功返回 true（调用方 +// 应使用内存中的新 token 重试原请求）。refresh_token 已失效时也会清空内存 token。 +func (c *OpenClient) tryRefreshTokenLocked() bool { + c.tokenMu.Lock() + defer c.tokenMu.Unlock() + token, err := c.RefreshToken(c.RefreshTokenStr) + if err != nil { + if IsRefreshTokenDead(err) { + c.SetAuthToken("", "") + } + return false + } + c.SetAuthToken(token.AccessToken, token.RefreshToken) + return true +} + // IsThrottleCode 判断是否为限流错误码。 func IsThrottleCode(code int) bool { return code == RequestMaxLimitCode || code == RequestRateLimitCode diff --git a/internal/service/strm_sync.go b/internal/service/strm_sync.go index 7775dc0..a708708 100644 --- a/internal/service/strm_sync.go +++ b/internal/service/strm_sync.go @@ -13,6 +13,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "time" "go.uber.org/zap" @@ -229,7 +230,14 @@ func (st *strmSyncState) run() error { return nil } -// walkRemote 广度优先遍历网盘目录树。 +// strmScanWorkers 远端目录树并发遍历的 worker 数。115 开放平台有全局 +// 令牌桶限流（QPS/QPM/QPH），并发请求自动排队，不会触发风控；并发让 +// 多个目录列表请求的网络往返彼此重叠，大幅缩短大目录树同步耗时。 +const strmScanWorkers = 8 + +// walkRemote 并发广度优先遍历网盘目录树。 +// 多个 worker 并行执行 List（受全局 115 令牌桶限流约束），子目录动态 +// 入队；任一目录失败则取消其余 worker 并返回错误（与旧串行版语义一致）。 func (st *strmSyncState) walkRemote() error { root := strings.TrimSpace(st.p.RemotePath) if root == "" { @@ -239,33 +247,78 @@ func (st *strmSyncState) walkRemote() error { id string rel string } - queue := []dirTask{{id: root, rel: ""}} - for len(queue) > 0 { - select { - case <-st.ctx.Done(): - return st.ctx.Err() - default: - } - task := queue[0] - queue = queue[1:] - entries, err := st.provider.List(st.ctx, task.id) - if err != nil { - return fmt.Errorf("列出远端目录 %s 失败：%w", task.id, err) - } - for _, entry := range entries { - cleanName := cleanEntryName(entry.Name, entry.IsDir) - rel := cleanName - if task.rel != "" { - rel = task.rel + "/" + cleanName + + ctx, cancel := context.WithCancel(st.ctx) + defer cancel() + + queue := make(chan dirTask, 512) + var pending atomic.Int64 + + // 根目录入队 + pending.Add(1) + queue <- dirTask{id: root, rel: ""} + + // 当队列中所有目录都被消费（pending 归零）或出错时关闭 channel， + // 让 worker 全部退出。 + go func() { + for { + if ctx.Err() != nil || pending.Load() == 0 { + close(queue) + return } - if entry.IsDir { - queue = append(queue, dirTask{id: entry.ID, rel: rel}) - continue - } - st.processRemoteFile(entry, rel) + time.Sleep(10 * time.Millisecond) } + }() + + var ( + wg sync.WaitGroup + errMu sync.Mutex + firstErr error + ) + for i := 0; i < strmScanWorkers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for task := range queue { + if ctx.Err() != nil { + return + } + entries, err := st.provider.List(ctx, task.id) + if err != nil { + errMu.Lock() + if firstErr == nil { + firstErr = fmt.Errorf("列出远端目录 %s 失败：%w", task.id, err) + } + errMu.Unlock() + cancel() + return + } + for _, entry := range entries { + cleanName := cleanEntryName(entry.Name, entry.IsDir) + rel := cleanName + if task.rel != "" { + rel = task.rel + "/" + cleanName + } + if entry.IsDir { + pending.Add(1) + select { + case queue <- dirTask{id: entry.ID, rel: rel}: + case <-ctx.Done(): + pending.Add(-1) + } + } else { + st.processRemoteFile(entry, rel) + } + } + pending.Add(-1) + } + }() } - return nil + wg.Wait() + if firstErr != nil { + return firstErr + } + return ctx.Err() } // processRemoteFile 分类处理远端文件：视频生成 STRM，元数据入下载队列。 @@ -340,7 +393,8 @@ func (st *strmSyncState) handleVideo(entry cloud.FileEntry, rel, ext string) { } content, err := st.strmContent(entry, rel, ext) if err != nil { - st.rec.Message = err.Error() + // 并发 worker 下 rec.Message 无锁写会有数据竞争，这里仅记录日志； + // 最终同步结果的 message 由 finishSync 统一填充。 st.s.log.Warn("build strm content failed", zap.String("file", rel), zap.Error(err)) return } diff --git a/internal/service/strm_sync_test.go b/internal/service/strm_sync_test.go index b14791c..7943bae 100644 --- a/internal/service/strm_sync_test.go +++ b/internal/service/strm_sync_test.go @@ -17,6 +17,7 @@ import ( "github.com/ShukeBta/MMTL/internal/config" "github.com/ShukeBta/MMTL/internal/model" "github.com/ShukeBta/MMTL/internal/repository" + "github.com/ShukeBta/MMTL/internal/service/cloud" ) // testStrmService 构建带内存库的 StrmService。 @@ -340,3 +341,108 @@ func TestScanLocalMetaForUpload(t *testing.T) { t.Fatalf("expected still 1 upload task after dedup, got %d", len(tasks)) } } + +// fakeRemoteProvider 是 walkRemote 并发遍历的假提供方：返回一棵固定目录树， +// 并记录每个目录被 List 的次数，用于验证并发遍历无漏目录、无重复目录。 +type fakeRemoteProvider struct { + listed map[string]int +} + +func (f *fakeRemoteProvider) Type() string { return "fake" } +func (f *fakeRemoteProvider) Ping(context.Context) error { return nil } +func (f *fakeRemoteProvider) Resolve(context.Context, string) (*cloud.DirectLink, error) { + return &cloud.DirectLink{URL: "http://cdn/x.mkv"}, nil +} + +func (f *fakeRemoteProvider) List(_ context.Context, dirID string) ([]cloud.FileEntry, error) { + if f.listed == nil { + f.listed = map[string]int{} + } + f.listed[dirID]++ + switch dirID { + case "root": + return []cloud.FileEntry{ + {ID: "a", Name: "动漫", IsDir: true}, + {ID: "b", Name: "电影", IsDir: true}, + {ID: "f1", Name: "孤儿视频.mkv", Size: 100}, + }, nil + case "a": + return []cloud.FileEntry{ + {ID: "a1", Name: "番剧", IsDir: true}, + {ID: "fa1", Name: "第01集.mkv", Size: 200}, + }, nil + case "a1": + return []cloud.FileEntry{ + {ID: "fa11", Name: "第01集.mkv", Size: 300}, + {ID: "fa12", Name: "第02集.mkv", Size: 300}, + }, nil + case "b": + return []cloud.FileEntry{ + {ID: "fb1", Name: "电影A.mkv", Size: 400}, + }, nil + default: + return nil, nil + } +} + +// TestWalkRemoteConcurrent 验证并发目录遍历：所有目录均被列出、所有文件 +// 均被处理（strm 生成 / 元数据入队），且不重复。 +func TestWalkRemoteConcurrent(t *testing.T) { + svc := testStrmService(t) + localDir := t.TempDir() + + acct := &model.StrmAccount{ + Name: "fake", + Provider: "cloud115", + Config: "{}", + Enabled: true, + } + if err := svc.repo.StrmAccount.Create(context.Background(), acct); err != nil { + t.Fatal(err) + } + p := &model.StrmSyncPath{ + Base: model.Base{ID: "walk-path"}, + AccountID: acct.ID, + Provider: model.StrmProvider115, + RemotePath: "root", + LocalPath: localDir, + } + + provider := &fakeRemoteProvider{listed: map[string]int{}} + st := &strmSyncState{ + s: svc, + ctx: context.Background(), + p: p, + provider: provider, + cfg: &strmPathConfig{VideoExt: []string{"mkv"}, MetaExt: []string{"nfo"}, AddPath: 1, DownloadMeta: false}, + rec: &model.StrmSyncRecord{}, + seenVideo: map[string]bool{}, + seenMeta: map[string]bool{}, + remoteMeta: map[string]int64{}, + } + if err := st.walkRemote(); err != nil { + t.Fatalf("walkRemote failed: %v", err) + } + + for _, dir := range []string{"root", "a", "a1", "b"} { + if provider.listed[dir] != 1 { + t.Errorf("目录 %s 被列出 %d 次，期望 1 次", dir, provider.listed[dir]) + } + } + + // 5 个视频文件应生成 5 个 .strm：孤儿视频.mkv / a目录第01集 / + // 番剧第01集+第02集 / 电影A（递归统计，含子目录） + strmCount := 0 + walkErr := filepath.WalkDir(localDir, func(path string, d os.DirEntry, err error) error { + if err == nil && !d.IsDir() && strings.HasSuffix(d.Name(), ".strm") { + strmCount++ + } + return nil + }) + if walkErr != nil { + t.Fatal(walkErr) + } + if strmCount != 5 { + t.Errorf("生成的 .strm 数量 = %d，期望 5", strmCount) + } +}