优化

2026-09-29 11:36:36 +08:00 · 2026-08-24 22:38:23 +08:00
parent 93c7487c96
commit 60b11bc2bc
3 changed files with 218 additions and 28 deletions
@@ -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
@@ -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
 	}
@@ -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)
+	}
+}