From 3150971f104c96d2abfa3e38c92f0b4d9b58d5f5 Mon Sep 17 00:00:00 2001 From: truewhile <62226914+truewhile@users.noreply.github.com> Date: Sun, 6 Sep 2026 16:15:26 +0800 Subject: [PATCH] feat(sync): implement 115 adaptive hierarchical flat scan to bypass deep paging limit --- internal/service/strm_sync.go | 413 ++++++++++++++++++++++------- internal/service/strm_sync_test.go | 110 ++++++++ 2 files changed, 421 insertions(+), 102 deletions(-) diff --git a/internal/service/strm_sync.go b/internal/service/strm_sync.go index 730d10b..e06bbbc 100644 --- a/internal/service/strm_sync.go +++ b/internal/service/strm_sync.go @@ -514,28 +514,31 @@ func (st *strmSyncState) walkRemote() error { if task.rel != "" { rel = task.rel + "/" + cleanName } - if entry.IsDir { - push(dirTask{id: entry.ID, rel: rel}) - } else { - st.processRemoteFile(entry, rel) + if entry.IsDir { + st.dirCache.Store(entry.ID, rel) + st.deferDirCacheSave(entry.ID, rel) + push(dirTask{id: entry.ID, rel: rel}) + } else { + st.processRemoteFile(entry, rel) + } } + walkMu.Lock() + pending-- + if pending == 0 { + walkCond.Broadcast() + } + walkMu.Unlock() } - walkMu.Lock() - pending-- - if pending == 0 { - walkCond.Broadcast() - } - walkMu.Unlock() + }); err != nil { + cancel() } - }); err != nil { - cancel() - } - }() - } - wg.Wait() - if firstErr != nil { - return firstErr - } + }() + } + wg.Wait() + st.flushDirCacheSave() + if firstErr != nil { + return firstErr + } return ctx.Err() } @@ -674,93 +677,16 @@ func (st *strmSyncState) walk115Flat(open115 *cloud115.OpenClient) error { if pathCounts[item.Path] > 1 { continue } - st.dirCache.Store(item.DirID, cleanDirRel(item.Path)) + st.dirCache.Store(item.DirID, cleanDirRel(item.Path)) + } } } - } - // 2. 探测文件总数 - const pageSize = 1150 - firstBatch, totalCount, err := open115.GetFsListFlat(ctx, rootCID, 0, pageSize) - if err != nil { - return fmt.Errorf("115: 获取文件列表失败：%w", err) - } - - // 115 的扁平化列表（搜索底层）对 offset + limit 有 10000 的最大深度限制。 - // 当扁平模式下的总文件数 >= 9500 时，强行拒绝继续扁平拉取，而是抛出降级错误， - // 让外层回退到使用普通的按目录并发递归（walkRemote），以免截断导致后排文件被误删/重传。 - if totalCount >= 9500 { - return errFallbackToWalkRemote - } - - st.updateSyncMessage(fmt.Sprintf("正在拉取远端文件列表 (共 %d 个文件)...", totalCount)) - - allFiles := make([]cloud115.RemoteFile, 0, totalCount) - allFiles = append(allFiles, firstBatch...) - - // 3. 并发分页拉取剩余文件 - if totalCount > int64(len(firstBatch)) { - totalPages := int((totalCount + pageSize - 1) / pageSize) - type pageTask struct { - offset int + // 2. 自适应分治拉取文件列表（单目录超 9500 时自动对子目录并发分治扁平化） + allFiles, err := st.fetch115FilesAdaptive(ctx, open115, rootCID) + if err != nil { + return err } - pageTasks := make([]pageTask, 0, totalPages-1) - for page := 1; page < totalPages; page++ { - pageTasks = append(pageTasks, pageTask{offset: page * pageSize}) - } - - var ( - filesMu sync.Mutex - wg sync.WaitGroup - taskCh = make(chan pageTask, len(pageTasks)) - errMu sync.Mutex - fetchErr error - ) - - for _, t := range pageTasks { - taskCh <- t - } - close(taskCh) - - workers := 8 - if len(pageTasks) < workers { - workers = len(pageTasks) - } - - for i := 0; i < workers; i++ { - wg.Add(1) - go func() { - defer wg.Done() - // 分页拉取 panic 时取消整个同步；正常退出不取消。 - if err := helper.Recover(st.s.log, "strm.sync.walk115.page", func() error { - for t := range taskCh { - if ctx.Err() != nil { - return nil - } - files, _, err := open115.GetFsListFlat(ctx, rootCID, t.offset, pageSize) - if err != nil { - errMu.Lock() - if fetchErr == nil { - fetchErr = err - } - errMu.Unlock() - return nil - } - filesMu.Lock() - allFiles = append(allFiles, files...) - filesMu.Unlock() - } - return nil - }); err != nil { - cancel() - } - }() - } - wg.Wait() - if fetchErr != nil { - return fmt.Errorf("115: 分页拉取失败：%w", fetchErr) - } - } if ctx.Err() != nil { return ctx.Err() @@ -976,6 +902,289 @@ feed: return ctx.Err() } +const ( + flat115PageSize = 1150 + flat115Threshold = 9500 + max115AdaptiveDepth = 10 +) + +type adaptive115Task struct { + cid string + rel string + depth int +} + +// fetch115FlatSubtree 扁平拉取单个文件数在安全阈值内的子树全部文件。 +func fetch115FlatSubtree(ctx context.Context, open115 *cloud115.OpenClient, cid string, firstBatch []cloud115.RemoteFile, totalCount int64, log *zap.Logger) ([]cloud115.RemoteFile, error) { + allFiles := make([]cloud115.RemoteFile, 0, totalCount) + allFiles = append(allFiles, firstBatch...) + if totalCount <= int64(len(firstBatch)) { + return allFiles, nil + } + + totalPages := int((totalCount + flat115PageSize - 1) / flat115PageSize) + type pageTask struct { + offset int + } + pageTasks := make([]pageTask, 0, totalPages-1) + for page := 1; page < totalPages; page++ { + pageTasks = append(pageTasks, pageTask{offset: page * flat115PageSize}) + } + + var ( + filesMu sync.Mutex + wg sync.WaitGroup + taskCh = make(chan pageTask, len(pageTasks)) + errMu sync.Mutex + fetchErr error + ) + for _, t := range pageTasks { + taskCh <- t + } + close(taskCh) + + workers := 8 + if len(pageTasks) < workers { + workers = len(pageTasks) + } + + for i := 0; i < workers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + if err := helper.Recover(log, "strm.sync.walk115.page", func() error { + for t := range taskCh { + if ctx.Err() != nil { + return nil + } + files, _, err := open115.GetFsListFlat(ctx, cid, t.offset, flat115PageSize) + if err != nil { + errMu.Lock() + if fetchErr == nil { + fetchErr = err + } + errMu.Unlock() + return nil + } + filesMu.Lock() + allFiles = append(allFiles, files...) + filesMu.Unlock() + } + return nil + }); err != nil { + errMu.Lock() + if fetchErr == nil { + fetchErr = err + } + errMu.Unlock() + } + }() + } + wg.Wait() + if fetchErr != nil { + return nil, fmt.Errorf("115: 分页拉取失败：%w", fetchErr) + } + return allFiles, nil +} + +// list115DirDirect 列出指定目录下的直接子项（单层 cur=1&show_dir=1）。 +func list115DirDirect(ctx context.Context, open115 *cloud115.OpenClient, cid string) ([]cloud115.RemoteFile, error) { + var out []cloud115.RemoteFile + for offset := 0; ; offset += flat115PageSize { + files, _, err := open115.GetFsList(ctx, cid, offset, flat115PageSize) + if err != nil { + return nil, err + } + out = append(out, files...) + if len(files) < flat115PageSize { + break + } + select { + case <-ctx.Done(): + return nil, ctx.Err() + default: + } + } + return out, nil +} + +// fetch115FilesAdaptive 采用自适应分治策略抓取 115 目录树下的全部文件： +// 115 开放平台扁平搜索对 offset+limit 有 10000 的最大深度限制。 +// - 若子树文件总数 < 9500，直接使用全速扁平分页批量拉取； +// - 若子树文件总数 >= 9500（大库或超大分类目录），自动分治：仅单层列出该目录的直属子项（cur=1）， +// 直属纯文件直接收集，直属子目录则派发为独立的子树任务继续递归探测与拉取； +// - 若超大单目录下无子目录或层级过深（>10层），安全回退到 errFallbackToWalkRemote。 +func (st *strmSyncState) fetch115FilesAdaptive(ctx context.Context, open115 *cloud115.OpenClient, rootCID string) ([]cloud115.RemoteFile, error) { + var ( + allFiles []cloud115.RemoteFile + filesMu sync.Mutex + + walkMu sync.Mutex + walkCond = sync.NewCond(&walkMu) + work []adaptive115Task + pending int + + errMu sync.Mutex + firstErr error + ) + + push := func(t adaptive115Task) { + walkMu.Lock() + work = append(work, t) + pending++ + walkCond.Signal() + walkMu.Unlock() + } + + go func() { + <-ctx.Done() + walkMu.Lock() + walkCond.Broadcast() + walkMu.Unlock() + }() + + push(adaptive115Task{cid: rootCID, rel: "", depth: 0}) + + workers := 8 + var wg sync.WaitGroup + + for i := 0; i < workers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + if err := helper.Recover(st.s.log, "strm.sync.walk115.adaptive", func() error { + for { + walkMu.Lock() + for len(work) == 0 { + if ctx.Err() != nil || pending == 0 { + walkMu.Unlock() + return nil + } + walkCond.Wait() + } + task := work[0] + work = work[1:] + walkMu.Unlock() + + if ctx.Err() != nil { + walkMu.Lock() + pending-- + walkCond.Broadcast() + walkMu.Unlock() + return nil + } + + firstBatch, totalCount, err := open115.GetFsListFlat(ctx, task.cid, 0, flat115PageSize) + if err != nil { + errMu.Lock() + if firstErr == nil { + firstErr = fmt.Errorf("115: 获取文件列表失败（cid=%s）：%w", task.cid, err) + } + errMu.Unlock() + walkMu.Lock() + pending-- + walkCond.Broadcast() + walkMu.Unlock() + return nil + } + + if totalCount < flat115Threshold { + // 安全深度内：直接扁平拉取该子树全部文件 + files, err := fetch115FlatSubtree(ctx, open115, task.cid, firstBatch, totalCount, st.s.log) + if err != nil { + errMu.Lock() + if firstErr == nil { + firstErr = err + } + errMu.Unlock() + walkMu.Lock() + pending-- + walkCond.Broadcast() + walkMu.Unlock() + return nil + } + filesMu.Lock() + allFiles = append(allFiles, files...) + currentCount := len(allFiles) + filesMu.Unlock() + st.updateSyncMessage(fmt.Sprintf("正在拉取远端文件列表 (已获取 %d 个文件)...", currentCount)) + } else { + // 子树过大（>=9500）：分治展开该目录直接子项 + if task.depth >= max115AdaptiveDepth { + // 深度超限兜底：单目录嵌套超 10 层仍超 9500，回退为传统递归 + errMu.Lock() + if firstErr == nil { + firstErr = errFallbackToWalkRemote + } + errMu.Unlock() + walkMu.Lock() + pending-- + walkCond.Broadcast() + walkMu.Unlock() + return nil + } + st.s.log.Info("115: 目录文件数超限，自动分治展开子目录并发扁平拉取", + zap.String("cid", task.cid), + zap.String("rel", task.rel), + zap.Int64("total_count", totalCount), + zap.Int("depth", task.depth)) + + directEntries, err := list115DirDirect(ctx, open115, task.cid) + if err != nil { + errMu.Lock() + if firstErr == nil { + firstErr = fmt.Errorf("115: 列出单层目录失败（cid=%s）：%w", task.cid, err) + } + errMu.Unlock() + walkMu.Lock() + pending-- + walkCond.Broadcast() + walkMu.Unlock() + return nil + } + + for _, f := range directEntries { + if f.Category == cloud115.TypeDir { + cleanName := cleanEntryName(f.FileName, true) + subRel := cleanName + if task.rel != "" { + subRel = task.rel + "/" + cleanName + } + st.dirCache.Store(f.FileId, subRel) + st.deferDirCacheSave(f.FileId, subRel) + push(adaptive115Task{cid: f.FileId, rel: subRel, depth: task.depth + 1}) + } else { + filesMu.Lock() + allFiles = append(allFiles, f) + filesMu.Unlock() + } + } + } + + walkMu.Lock() + pending-- + if pending == 0 { + walkCond.Broadcast() + } + walkMu.Unlock() + } + }); err != nil { + errMu.Lock() + if firstErr == nil { + firstErr = err + } + errMu.Unlock() + } + }() + } + + wg.Wait() + if firstErr != nil { + return nil, firstErr + } + return allFiles, nil +} + // handleVideo 生成/更新 .strm 文件。 func (st *strmSyncState) handleVideo(entry cloud.FileEntry, rel, ext string) { relSansExt := rel[:len(rel)-len(ext)] diff --git a/internal/service/strm_sync_test.go b/internal/service/strm_sync_test.go index 214243f..8673234 100644 --- a/internal/service/strm_sync_test.go +++ b/internal/service/strm_sync_test.go @@ -1319,3 +1319,113 @@ func TestScanLocalMetaForUploadMultipleCopiesAllStale(t *testing.T) { t.Fatalf("expected RemoteRef to be 'old-1,old-2', got %q", tasks[0].RemoteRef) } } + +// TestWalk115AdaptiveHierarchicalFlatScan 验证自适应分治扁平化扫描： +// 当根目录探测总数 >= 9500 时，系统自动分治展开单层直接子项，对各子目录分别执行扁平拉取， +// 正确合并根目录直属文件与各子目录深层文件，突破 115 开放平台 10000 深度限制。 +func TestWalk115AdaptiveHierarchicalFlatScan(t *testing.T) { + svc := testStrmService(t) + localDir := t.TempDir() + acct := &model.StrmAccount{Name: "fake115-adaptive", 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: "adaptive-path"}, + AccountID: acct.ID, + Provider: model.StrmProvider115, + RemotePath: "0", + LocalPath: localDir, + } + + api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + q := r.URL.Query() + cid := q.Get("cid") + cur := q.Get("cur") + + switch r.URL.Path { + case "/open/ufile/files": + if cid == "0" && cur == "0" { + // 根目录扁平探测：模拟总文件数 12000 超限 (>= 9500) + w.Write([]byte(`{"state":true,"count":12000,"data":[]}`)) + return + } + if cid == "0" && cur == "1" { + // 根目录单层列举：返回 1 个直属视频和 2 个子目录 + w.Write([]byte(`{"state":true,"count":3,"data":[ + {"fid":"100","pid":"0","fc":"1","fn":"root.mkv","pc":"pcr","upt":1700000000,"fs":1024}, + {"fid":"1001","pid":"0","fc":"0","fn":"Heyzo","upt":1700000001,"fs":0}, + {"fid":"1002","pid":"0","fc":"0","fn":"S1","upt":1700000002,"fs":0}]}`)) + return + } + if cid == "1001" { + // 子目录 Heyzo 扁平拉取：文件数安全 (< 9500) + w.Write([]byte(`{"state":true,"count":2,"data":[ + {"fid":"201","pid":"1001","fc":"1","fn":"h1.mkv","pc":"pc201","upt":1700000001,"fs":1024}, + {"fid":"202","pid":"1001","fc":"1","fn":"h2.mkv","pc":"pc202","upt":1700000002,"fs":2048}]}`)) + return + } + if cid == "1002" { + // 子目录 S1 扁平拉取：文件数安全 (< 9500) + w.Write([]byte(`{"state":true,"count":1,"data":[ + {"fid":"301","pid":"1002","fc":"1","fn":"s1.mkv","pc":"pc301","upt":1700000003,"fs":4096}]}`)) + return + } + t.Errorf("unexpected files query: %s", r.URL.RawQuery) + case "/open/folder/get_info": + fileID := q.Get("file_id") + switch fileID { + case "1001": + w.Write([]byte(`{"state":true,"data":{"file_id":"1001","file_name":"Heyzo","file_category":"0", + "paths":[{"file_id":"0","file_name":"根目录"},{"file_id":"1001","file_name":"Heyzo"}]}}`)) + case "1002": + w.Write([]byte(`{"state":true,"data":{"file_id":"1002","file_name":"S1","file_category":"0", + "paths":[{"file_id":"0","file_name":"根目录"},{"file_id":"1002","file_name":"S1"}]}}`)) + default: + t.Errorf("unexpected get_info file_id: %s", fileID) + } + default: + t.Errorf("unexpected path %s", r.URL.Path) + } + })) + defer api.Close() + oldPro := cloud115.ProAPIBase + cloud115.ProAPIBase = api.URL + defer func() { cloud115.ProAPIBase = oldPro }() + + oc := cloud115.NewOpenClient("app", "at", "rt") + st := &strmSyncState{ + s: svc, + ctx: context.Background(), + p: p, + provider: cloud.NewOpenAPI115("app", "at", "rt"), + cfg: &strmPathConfig{VideoExt: []string{"mkv"}, MetaExt: []string{"nfo"}, AddPath: 1}, + rec: &model.StrmSyncRecord{}, + syncType: model.StrmSyncTypeFull, + dirCache: sync.Map{}, + seenVideo: map[string]bool{}, + seenMeta: map[string]bool{}, + remoteMeta: map[string][]remoteMetaItem{}, + seenMetaTarget: map[string]cloud.FileEntry{}, + seenVideoTarget: map[string]cloud.FileEntry{}, + } + if err := st.walk115Flat(oc); err != nil { + t.Fatalf("walk115Flat adaptive failed: %v", err) + } + + // 1个根目录视频 + 2个Heyzo视频 + 1个S1视频 = 共4个视频成功生成 .strm + if st.rec.NewStrm != 4 { + t.Fatalf("expected 4 strm created, got %d", st.rec.NewStrm) + } + expectedFiles := []string{ + filepath.Join(localDir, "root.strm"), + filepath.Join(localDir, "Heyzo", "h1.strm"), + filepath.Join(localDir, "Heyzo", "h2.strm"), + filepath.Join(localDir, "S1", "s1.strm"), + } + for _, f := range expectedFiles { + if _, err := os.Stat(f); err != nil { + t.Fatalf("expected strm %s to exist, err: %v", f, err) + } + } +}