From 74694f467705dc419c463d546733d7b1e5df12bf Mon Sep 17 00:00:00 2001 From: Har01d Date: Sun, 2 Aug 2026 22:00:34 +0800 Subject: [PATCH 1/4] =?UTF-8?q?=E5=A4=B8=E5=85=8B=E5=A4=9A=E8=B4=A6?= =?UTF-8?q?=E5=8F=B7=E5=8A=A0=E9=80=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- drivers/quark_uc/extension.go | 9 + drivers/quark_uc_share/driver.go | 12 ++ drivers/quark_uc_share/driver_test.go | 79 +++++++++ drivers/quark_uc_share/util.go | 180 +++++++++++++++++-- internal/bootstrap/data/setting.go | 2 + internal/conf/const.go | 4 + internal/model/args.go | 16 +- internal/stream/util.go | 238 ++++++++++++++++++++++++++ 8 files changed, 529 insertions(+), 11 deletions(-) diff --git a/drivers/quark_uc/extension.go b/drivers/quark_uc/extension.go index fa3987539..c6cc20d74 100644 --- a/drivers/quark_uc/extension.go +++ b/drivers/quark_uc/extension.go @@ -52,6 +52,15 @@ func (d *QuarkOrUC) getTempFolder() { d.createTempFolder() } +// EnsureTempDir 确保转存用的临时目录 ID 已初始化。Init 只跑一次,若当时 GetFiles 失败 +// 会导致 TempDirId 为空(转存 to_pdir_fid 为空被夸克拒绝)。外部包转存前调用以兜底重试。 +func (d *QuarkOrUC) EnsureTempDir() { + if d.TempDirId == "" { + log.Infof("[%d] %s temp dir empty, re-init", d.ID, d.config.Name) + d.getTempFolder() + } +} + func (d *QuarkOrUC) createTempFolder() { data := base.Json{ "dir_init_lock": false, diff --git a/drivers/quark_uc_share/driver.go b/drivers/quark_uc_share/driver.go index 9ac307037..0507863ff 100755 --- a/drivers/quark_uc_share/driver.go +++ b/drivers/quark_uc_share/driver.go @@ -116,6 +116,18 @@ func (d *QuarkUCShare) Link(ctx context.Context, file model.Obj, args model.Link log.Warnf("转存取链失败,回退免转存: %v", err) link, err = resolveShareDirectLink(d, file) } + // 多账号分片并行下载:开关开启、有≥2 个网盘账号且转存主链成功时, + // 对各账号各取一条直连填 link.MultiSource,总并发按账号数放大(N×Concurrency)。 + // 任意一步失败都不影响主链,只是 MultiSource 留空 → 退回单链多线程。 + if err == nil && link != nil && multiSourceEnabled(d) { + if ms := collectMultiSource(ctx, d, file, args, link); len(ms) > 1 { + link.MultiSource = ms + if link.Concurrency > 0 { + link.Concurrency *= len(ms) + } + log.Infof("[multi-source] %s %s 用 %d 账号分片下载 file=%s", d.config.Name, file.GetName(), len(ms), file.GetID()) + } + } if err == nil && link != nil { quarkUCShareLinkCache.Set(key, link) } diff --git a/drivers/quark_uc_share/driver_test.go b/drivers/quark_uc_share/driver_test.go index 12f446eca..e927f568b 100644 --- a/drivers/quark_uc_share/driver_test.go +++ b/drivers/quark_uc_share/driver_test.go @@ -13,11 +13,28 @@ import ( "github.com/OpenListTeam/OpenList/v4/internal/model" ) +// stubMultiSourceDisabled 关闭多账号分片下载并还原,避免 Link() 读 setting 死锁(测试里 op 未初始化)。 +func stubMultiSourceDisabled(t *testing.T) { + t.Helper() + origMS := multiSourceEnabled + origCollect := collectMultiSource + multiSourceEnabled = func(d *QuarkUCShare) bool { return false } + collectMultiSource = func(ctx context.Context, d *QuarkUCShare, file model.Obj, args model.LinkArgs, primary *model.Link) []model.LinkSource { + return nil + } + t.Cleanup(func() { + multiSourceEnabled = origMS + collectMultiSource = origCollect + }) +} + func TestQuarkUCShareLink_CachesByFileID(t *testing.T) { origCache := quarkUCShareLinkCache origResolver := resolveQuarkUCShareLink origDirect := resolveShareDirectLink origSVIP := accountIsSVIP + origMS := multiSourceEnabled + origCollect := collectMultiSource quarkUCShareLinkCache = cache.NewKeyedCache[*model.Link](time.Hour) resolveCalls := 0 resolveQuarkUCShareLink = func(ctx context.Context, d *QuarkUCShare, file model.Obj, args model.LinkArgs) (*model.Link, error) { @@ -28,11 +45,17 @@ func TestQuarkUCShareLink_CachesByFileID(t *testing.T) { resolveShareDirectLink = func(d *QuarkUCShare, file model.Obj) (*model.Link, error) { return nil, errors.New("share-direct stub disabled") // 置失败,流程落到 resolveQuarkUCShareLink } + multiSourceEnabled = func(d *QuarkUCShare) bool { return false } + collectMultiSource = func(ctx context.Context, d *QuarkUCShare, file model.Obj, args model.LinkArgs, primary *model.Link) []model.LinkSource { + return nil + } t.Cleanup(func() { quarkUCShareLinkCache = origCache resolveQuarkUCShareLink = origResolver resolveShareDirectLink = origDirect accountIsSVIP = origSVIP + multiSourceEnabled = origMS + collectMultiSource = origCollect }) d := &QuarkUCShare{Addition: Addition{ShareToken: "share-token"}, config: driver.Config{Name: "QuarkShare"}} @@ -69,6 +92,10 @@ func TestQuarkUCShareLink_DoesNotCacheErrors(t *testing.T) { resolveShareDirectLink = func(d *QuarkUCShare, file model.Obj) (*model.Link, error) { return nil, errors.New("share-direct stub disabled") // 失败 → 落到同样报错的 resolveQuarkUCShareLink } + multiSourceEnabled = func(d *QuarkUCShare) bool { return false } + collectMultiSource = func(ctx context.Context, d *QuarkUCShare, file model.Obj, args model.LinkArgs, primary *model.Link) []model.LinkSource { + return nil + } t.Cleanup(func() { quarkUCShareLinkCache = origCache resolveQuarkUCShareLink = origResolver @@ -87,6 +114,7 @@ func TestQuarkUCShareLink_DoesNotCacheErrors(t *testing.T) { } func TestQuarkUCShareLink_DifferentFileIDsDoNotShareCache(t *testing.T) { + stubMultiSourceDisabled(t) origCache := quarkUCShareLinkCache origResolver := resolveQuarkUCShareLink origDirect := resolveShareDirectLink @@ -119,6 +147,7 @@ func TestQuarkUCShareLink_DifferentFileIDsDoNotShareCache(t *testing.T) { func TestQuarkUCShareLink_FallbackToShareDirectOnSaveFail(t *testing.T) { // 转存(save+speedup)为主;转存失败时回退免转存(share-direct)。 + stubMultiSourceDisabled(t) origCache := quarkUCShareLinkCache origResolver := resolveQuarkUCShareLink origDirect := resolveShareDirectLink @@ -159,6 +188,7 @@ func TestQuarkUCShareLink_FallbackToShareDirectOnSaveFail(t *testing.T) { func TestQuarkUCShareLink_PrefersSaveAndSpeedup(t *testing.T) { // 转存(save+speedup)为主:成功时不调用免转存(免转存无 speedup,被限速,仅兜底)。 + stubMultiSourceDisabled(t) origCache := quarkUCShareLinkCache origResolver := resolveQuarkUCShareLink origDirect := resolveShareDirectLink @@ -197,6 +227,55 @@ func TestQuarkUCShareLink_PrefersSaveAndSpeedup(t *testing.T) { } } +// 开关开启、collectMultiSource 返回多源时,Link() 把 MultiSource 填上并把 Concurrency 放大 N 倍。 +func TestQuarkUCShareLink_PopulatesMultiSourceWhenEnabled(t *testing.T) { + stubMultiSourceDisabled(t) + origCache := quarkUCShareLinkCache + origResolver := resolveQuarkUCShareLink + origDirect := resolveShareDirectLink + quarkUCShareLinkCache = cache.NewKeyedCache[*model.Link](time.Hour) + resolveQuarkUCShareLink = func(ctx context.Context, d *QuarkUCShare, file model.Obj, args model.LinkArgs) (*model.Link, error) { + // 主链:账号1 直链。 + return &model.Link{URL: "https://cdn.example.com/account1/fid", Concurrency: 4, PartSize: 512 << 10}, nil + } + resolveShareDirectLink = func(d *QuarkUCShare, file model.Obj) (*model.Link, error) { + return nil, errors.New("share-direct stub disabled") + } + // 开启多账号分片:collectMultiSource 返回 3 个源。 + multiSourceEnabled = func(d *QuarkUCShare) bool { return true } + collectMultiSource = func(ctx context.Context, d *QuarkUCShare, file model.Obj, args model.LinkArgs, primary *model.Link) []model.LinkSource { + primarySrc := model.LinkSource{URL: "https://cdn.example.com/account1/fid"} + return []model.LinkSource{ + primarySrc, + {URL: "https://cdn.example.com/account2/fid"}, + {URL: "https://cdn.example.com/account3/fid"}, + } + } + t.Cleanup(func() { + quarkUCShareLinkCache = origCache + resolveQuarkUCShareLink = origResolver + resolveShareDirectLink = origDirect + }) + + d := &QuarkUCShare{Addition: Addition{ShareToken: "share-token"}, config: driver.Config{Name: "QuarkShare"}} + file := &model.Object{ID: "fid-fidtoken-pid", Name: "big.mp4"} + + link, err := d.Link(context.Background(), file, model.LinkArgs{}) + if err != nil { + t.Fatalf("link: %v", err) + } + if link == nil { + t.Fatal("nil link") + } + if len(link.MultiSource) != 3 { + t.Fatalf("expected 3 MultiSource, got %d", len(link.MultiSource)) + } + // Concurrency 4 放大 3 倍 = 12。 + if link.Concurrency != 12 { + t.Fatalf("expected Concurrency=12 (4*3), got %d", link.Concurrency) + } +} + func TestBindRequestDriverUsesSelectedStorageForRequestAndTempDir(t *testing.T) { selected := &quark.QuarkOrUC{TempDirId: "temp-dir-a"} other := &quark.QuarkOrUC{TempDirId: "temp-dir-b"} diff --git a/drivers/quark_uc_share/util.go b/drivers/quark_uc_share/util.go index a19433f26..45993f264 100755 --- a/drivers/quark_uc_share/util.go +++ b/drivers/quark_uc_share/util.go @@ -89,6 +89,8 @@ type shareRequestBinding interface { getTempFile(ctx context.Context, dirID, fileID string) (model.Obj, error) deleteTempFile(ctx context.Context, fileID string) error link(ctx context.Context, file model.Obj, args model.LinkArgs) (*model.Link, error) + // cookieValue 返回该绑定账号的 Cookie,供走 drive-pc 端点的分享请求使用。 + cookieValue() string } type requestBinding struct { @@ -98,6 +100,7 @@ type requestBinding struct { } func bindRequestDriver(uc *quark.QuarkOrUC) requestBinding { + uc.EnsureTempDir() // 转存前确保临时目录 ID 已初始化(Init 时可能未成功) return requestBinding{ requestDriver: uc, cookie: uc.Cookie, @@ -167,6 +170,10 @@ func (b requestBinding) matches(uc *quark.QuarkOrUC) bool { return b.requestDriver == uc } +func (b requestBinding) cookieValue() string { + return b.cookie +} + type requestTVBinding struct { requestDriver *quark_uc_tv.QuarkUCTV cookie string @@ -257,6 +264,10 @@ func (b requestTVBinding) matches(other *requestTVBinding) bool { return b.cookie == other.cookie && b.tempDirId == other.tempDirId } +func (b requestTVBinding) cookieValue() string { + return b.cookie +} + func (d *QuarkUCShare) getDriverName() string { name := "Quark" if d.config.Name == "UCShare" { @@ -307,7 +318,33 @@ func (d *QuarkUCShare) request(pathname string, method string, callback base.Req } func (d *QuarkUCShare) directRequest(cookieStr string, pathname string, method string, callback base.ReqCallback, resp interface{}) ([]byte, error) { - u := d.conf.api + pathname + return d.requestAt(d.conf.api, cookieStr, pathname, method, callback, resp) +} + +// pcApi 返回夸克/UC 分享操作的 PC 客户端端点。夸克 sharepage/save、sharepage/token、 +// file/download 必须走 drive-pc.quark.cn:drive.quark.cn 的 token 接口宽松(能发 stoken), +// 但 save/download 严格校验 stoken,会返回 "token [st invalid,code:50052]"。 +// 对齐不夜 cloud-drive.js quarkApiUrl=drive-pc.quark.cn。UC 的 conf.api 已是 pc-api.uc.cn。 +func (d *QuarkUCShare) pcApi() string { + if d.config.Name == "UCShare" { + return d.conf.api + } + return "https://drive-pc.quark.cn/1/clouddrive" +} + +// requestSharePc 用 PC 端点 + binding(账号)Cookie(无 binding 用兜底 Cookie)发分享请求。 +func (d *QuarkUCShare) requestSharePc(binding shareRequestBinding, pathname string, method string, callback base.ReqCallback, resp interface{}) ([]byte, error) { + cookieStr := getShareCookie() + if binding != nil { + if c := binding.cookieValue(); c != "" { + cookieStr = c + } + } + return d.requestAt(d.pcApi(), cookieStr, pathname, method, callback, resp) +} + +func (d *QuarkUCShare) requestAt(api, cookieStr, pathname string, method string, callback base.ReqCallback, resp interface{}) ([]byte, error) { + u := api + pathname req := base.RestyClient.R() req.SetHeaders(map[string]string{ "Cookie": cookieStr, @@ -326,6 +363,7 @@ func (d *QuarkUCShare) directRequest(cookieStr string, pathname string, method s } var e Resp req.SetError(&e) + log.Debugf("[share-pc] %s %s cookie=%q", method, u, cookieStr) res, err := req.Execute(method, u) if err != nil { return nil, err @@ -395,7 +433,7 @@ func (d *QuarkUCShare) getShareTokenWithBinding(binding shareRequestBinding) err "share_for_transfer": true, } var resp ShareTokenResp - res, err := d.requestWithBinding(binding, "/share/sharepage/token", http.MethodPost, func(req *resty.Request) { + res, err := d.requestSharePc(binding, "/share/sharepage/token", http.MethodPost, func(req *resty.Request) { req.SetBody(data) }, &resp) log.Debugf("getShareToken: %v %v", d.ShareId, string(res)) @@ -477,7 +515,9 @@ func (d *QuarkUCShare) saveShareFile(ctx context.Context, binding shareRequestBi func (d *QuarkUCShare) doSaveShareFile(ctx context.Context, binding shareRequestBinding, parsed shareFileID) (model.Obj, error) { fidToken := parsed.FidToken - for range 2 { + stRefreshed := false + // 上限 3 次:fid_token 刷新 + stoken 刷新各占一次重试余量。 + for attempt := 0; attempt < 3; attempt++ { data := base.Json{ "fid_list": []string{parsed.FileID}, "fid_token_list": []string{fidToken}, @@ -497,19 +537,31 @@ func (d *QuarkUCShare) doSaveShareFile(ctx context.Context, binding shareRequest "__t": strconv.FormatInt(time.Now().Unix(), 10), } var resp SaveResp - res, err := d.requestWithBinding(binding, "/share/sharepage/save", http.MethodPost, func(req *resty.Request) { + res, err := d.requestSharePc(binding, "/share/sharepage/save", http.MethodPost, func(req *resty.Request) { req.SetContext(ctx).SetBody(data).SetQueryParams(query) }, &resp) log.Debugf("saveFile: %v response: %v, error: %v", parsed.FileID, string(res), err) if err != nil { + msg := err.Error() + log.Warnf("[save-debug] fid=%s fidToken=%s stoken=%q to_pdir=%s accountCookie=%q err=%s", + parsed.FileID, fidToken, d.ShareToken, binding.tempDirID(), binding.cookieValue(), msg) // fid_token 过期:按父目录重新换一个 share_fid_token 再试一次。 - if strings.Contains(err.Error(), "token校验异常") && parsed.ParentID != "" { + if strings.Contains(msg, "token校验异常") && parsed.ParentID != "" { fidToken, err = d.getFileToken(binding, parsed.ParentID, parsed.FileID) if err != nil { return nil, err } continue } + // stoken 过期(code:50052 "st invalid"):刷新分享 stoken 后重试一次。 + // stoken 是 share 级、账号无关,刷新后所有账号共享新 stoken。 + if strings.Contains(msg, "st invalid") && !stRefreshed { + if e := d.getShareTokenWithBinding(binding); e == nil { + stRefreshed = true + log.Infof("[save] stoken 失效(50052),已刷新重试 %s", parsed.FileID) + continue + } + } log.Warnf("save file failed: %v", err) return nil, err } @@ -555,7 +607,7 @@ func (d *QuarkUCShare) getSaveTaskResult(ctx context.Context, binding shareReque "__t": strconv.FormatInt(time.Now().Unix(), 10), } var resp SaveTaskResp - res, err := d.requestWithBinding(binding, "/task", http.MethodGet, func(req *resty.Request) { + res, err := d.requestSharePc(binding, "/task", http.MethodGet, func(req *resty.Request) { req.SetContext(ctx).SetQueryParams(query) }, &resp) log.Debugf("getSaveTaskResult: %v %v", taskId, string(res)) @@ -917,7 +969,7 @@ func (d *QuarkUCShare) getShareFilesWithBinding(binding shareRequestBinding, id } log.Debugf("getShareFiles query: %v", query) var resp ListResp - res, err := d.requestWithBinding(binding, "/share/sharepage/detail", http.MethodGet, func(req *resty.Request) { + res, err := d.requestSharePc(binding, "/share/sharepage/detail", http.MethodGet, func(req *resty.Request) { req.SetQueryParams(query) }, &resp) name := d.getDriverName() @@ -988,6 +1040,103 @@ var accountIsSVIP = func(d *QuarkUCShare) bool { return ok && uc.VIP } +// multiSourceEnabled 多账号分片并行下载总开关。声明为 var 便于单测替换。 +var multiSourceEnabled = func(d *QuarkUCShare) bool { + return setting.GetBool(conf.QuarkMultiAccountProxy) +} + +// multiSourceMax 取同时使用的最大账号数,0=不限。默认 3。 +func multiSourceMax() int { + m := setting.GetInt(conf.QuarkMultiAccountMax, 3) + if m < 0 { + m = 0 + } + return m +} + +// collectMultiSource 对每个网盘账号各跑一遍转存+speedup 取链,收集成功的直连成 LinkSource 列表。 +// 关键:必须用 op.GetStorages 遍历每个账号再各自 saveAndLink —— 不能用 d.link(), +// 因为 d.link() 内部 GetFirstDriver 在 DriverRoundRobin 关闭时总返回 master 账号, +// 并发调用只会取到同一条链、去重后只剩 1 个源。 +// 总数受 multiSourceMax() 限制(0=不限,默认 3)。任一账号失败跳过。 +// 发到分片代理的直连去掉 #x-referer=raw 片段标记(那是给客户端代理用的,后端代理走 link.Header)。 +// 声明为 var 便于单测替换(避免单测里 op 还原账号)。 +var collectMultiSource = func(ctx context.Context, d *QuarkUCShare, file model.Obj, args model.LinkArgs, primary *model.Link) []model.LinkSource { + name := d.getDriverName() + storages := op.GetStorages(name) + if len(storages) < 2 { + return nil + } + max := multiSourceMax() + n := len(storages) + if max > 0 && n > max { + n = max + } + if n < len(storages) { + storages = storages[:n] + } + + type result struct { + l *model.Link + } + results := make(chan result, len(storages)) + var wg sync.WaitGroup + for _, st := range storages { + uc, ok := st.(*quark.QuarkOrUC) + if !ok { + continue + } + wg.Add(1) + go func(uc *quark.QuarkOrUC) { + defer wg.Done() + if ctx.Err() != nil { + return + } + binding := bindRequestDriver(uc) + l, e := d.saveAndLink(ctx, binding, file.GetID(), args) + if e == nil && l != nil && l.URL != "" { + select { + case results <- result{l: l}: + case <-ctx.Done(): + } + } else { + log.Warnf("[multi-source] 账号 %d 取链失败: %v", uc.ID, e) + } + }(uc) + } + go func() { wg.Wait(); close(results) }() + + srcs := make([]model.LinkSource, 0, len(storages)) + seen := map[string]bool{} + if primary != nil && primary.URL != "" { + s := sourceFromLink(primary, d) + seen[s.URL] = true + srcs = append(srcs, s) + } + for r := range results { + s := sourceFromLink(r.l, d) + if seen[s.URL] { + continue + } + seen[s.URL] = true + srcs = append(srcs, s) + } + log.Infof("[multi-source] collect: driver=%s accounts=%d collected=%d", name, len(storages), len(srcs)) + return srcs +} + +// sourceFromLink 从单账号取链结果构造 LinkSource:去掉 #x-referer=raw 片段标记, +// 保留 URL + Header(后端代理直接用 link.Header,不走魔法 Referer 片段契约)。 +func sourceFromLink(l *model.Link, d *QuarkUCShare) model.LinkSource { + u := l.URL + // #x-referer=raw 是给客户端代理的片段标记(见 share-direct-referer-marker 契约), + // 后端代理直接用 link.Header,片段发往 CDN 也是无害但无意义,去掉更干净。 + if i := strings.Index(u, "#"); i >= 0 { + u = u[:i] + } + return model.LinkSource{URL: u, Header: l.Header} +} + // masterCookie 取当前驱动类型主账号(master)的 Cookie。 // 夸克 share /file/download 需账号上下文(参考脚本 quarkRequestShareDownload 用 drive.fetch 带账号), // 匿名请求会失败 → 回退转存。故夸克取链需带主账号 Cookie;UC 匿名即可。无账号返回 ""。 @@ -1041,14 +1190,25 @@ var resolveShareDirectLink = func(d *QuarkUCShare, file model.Obj) (*model.Link, "speedup_session": "", } var resp DownResp - _, err := d.directRequest(reqCookie, "/file/download", http.MethodPost, func(req *resty.Request) { + _, err := d.requestAt(d.pcApi(), reqCookie, "/file/download", http.MethodPost, func(req *resty.Request) { req.SetBody(body) }, &resp) // fid_token 失效时,按 pid 重新换取 share_fid_token 后重试一次(复用 saveFile 的回退策略)。 if err != nil && strings.Contains(err.Error(), "token校验异常") && pid != "" { if newToken, e := d.getFileToken(nil, pid, fileId); e == nil && newToken != "" { body["fids_token"] = []string{newToken} - _, err = d.directRequest(reqCookie, "/file/download", http.MethodPost, func(req *resty.Request) { + _, err = d.requestAt(d.pcApi(), reqCookie, "/file/download", http.MethodPost, func(req *resty.Request) { + req.SetBody(body) + }, &resp) + } + } + // stoken 失效(code:50052 "st invalid"):刷新分享 stoken 后重试一次。 + // 夸克夜间 stoken 易过期,刷新一次可救回,避免免转存兜底整体失败。 + if err != nil && strings.Contains(err.Error(), "st invalid") { + if e := d.getShareToken(); e == nil { + body["stoken"] = d.ShareToken + log.Infof("[share-direct] stoken 失效(50052),已刷新重试 %s", fileId) + _, err = d.requestAt(d.pcApi(), reqCookie, "/file/download", http.MethodPost, func(req *resty.Request) { req.SetBody(body) }, &resp) } @@ -1060,7 +1220,7 @@ var resolveShareDirectLink = func(d *QuarkUCShare, file model.Obj) (*model.Link, if newToken, e := d.getFileToken(nil, pid, fileId); e == nil && newToken != "" { body["fids_token"] = []string{newToken} body["stoken"] = d.ShareToken - _, err = d.directRequest(reqCookie, "/file/download", http.MethodPost, func(req *resty.Request) { + _, err = d.requestAt(d.pcApi(), reqCookie, "/file/download", http.MethodPost, func(req *resty.Request) { req.SetBody(body) }, &resp) } diff --git a/internal/bootstrap/data/setting.go b/internal/bootstrap/data/setting.go index ee0bf8507..37befdeab 100644 --- a/internal/bootstrap/data/setting.go +++ b/internal/bootstrap/data/setting.go @@ -261,6 +261,8 @@ func InitialSettings() []model.SettingItem { {Key: conf.GuangyaTo123, Value: "false", Type: conf.TypeBool, Group: model.SINGLE, Flag: model.PRIVATE}, {Key: conf.DriverRoundRobin, Value: "false", Type: conf.TypeBool, Group: model.SINGLE, Flag: model.PRIVATE}, {Key: conf.UssQuarkTv, Value: "true", Type: conf.TypeBool, Group: model.SINGLE, Flag: model.PRIVATE}, + {Key: conf.QuarkMultiAccountProxy, Value: "true", Type: conf.TypeBool, Group: model.SINGLE, Flag: model.PRIVATE}, + {Key: conf.QuarkMultiAccountMax, Value: "4", Type: conf.TypeNumber, Group: model.SINGLE, Flag: model.PRIVATE}, {Key: conf.ProxyConfig, Value: "{}", Type: conf.TypeText, Group: model.SINGLE, Flag: model.PRIVATE}, {Key: "ali_lazy_load", Value: "true", Type: conf.TypeBool, Group: model.SINGLE, Flag: model.PRIVATE}, {Key: "open_token_url", Value: "https://api.xhofe.top/alist/ali_open/token", Type: conf.TypeString, Group: model.SINGLE, Flag: model.PRIVATE}, diff --git a/internal/conf/const.go b/internal/conf/const.go index a9738691a..c0b6e1eee 100644 --- a/internal/conf/const.go +++ b/internal/conf/const.go @@ -130,6 +130,10 @@ const ( DriverRoundRobin = "driver_round_robin" UssQuarkTv = "use_quark_tv" ProxyConfig = "proxy_config" + // QuarkMultiAccountProxy 开启夸克/UC 分享多账号分片并行下载(服务端 Go 代理):把同一资源转存到多个网盘账号,各取一条直连,分片按序号轮询分发以叠加带宽。默认关(PoC,需实测是否按账号限速)。 + QuarkMultiAccountProxy = "quark_multi_account_proxy" + // QuarkMultiAccountMax 多账号分片下载同时使用的最大账号数,0=不限。默认 3。 + QuarkMultiAccountMax = "quark_multi_account_max" // SSO SSOClientId = "sso_client_id" diff --git a/internal/model/args.go b/internal/model/args.go index 393a716fa..8f546591c 100644 --- a/internal/model/args.go +++ b/internal/model/args.go @@ -37,13 +37,25 @@ type Link struct { PartSize int `json:"part_size"` ContentLength int64 `json:"content_length"` // 转码视频、缩略图 + // MultiSource 用于多账号分片并行下载。非空且长度>1 时， + // Proxy 把分片按 序号 % len(MultiSource) 轮询分发到各源直连 + // (各账号 Cookie/Referer 独立),以叠加多账号带宽。 + // 空或单元素时维持单链行为。纯 Go 内部用,不对外序列化。 + MultiSource []LinkSource `json:"-"` + utils.SyncClosers `json:"-"` // 如果SyncClosers中的资源被关闭后Link将不可用，则此值应为 true RequireReference bool `json:"-"` } +// LinkSource 是多账号分片下载的一条源直连(URL+Header 独立)。 +type LinkSource struct { + URL string + Header http.Header +} + func (l *Link) Clone() *Link { - return &Link{ + cloned := &Link{ URL: l.URL, Header: l.Header, RangeReader: l.RangeReader, @@ -51,9 +63,11 @@ func (l *Link) Clone() *Link { Concurrency: l.Concurrency, PartSize: l.PartSize, ContentLength: l.ContentLength, + MultiSource: l.MultiSource, SyncClosers: utils.NewSyncClosers(l), RequireReference: l.RequireReference, } + return cloned } type OtherArgs struct { diff --git a/internal/stream/util.go b/internal/stream/util.go index 6aa3dda5d..3d998c7a1 100644 --- a/internal/stream/util.go +++ b/internal/stream/util.go @@ -9,6 +9,7 @@ import ( "io" "net/http" "os" + "sync" "github.com/OpenListTeam/OpenList/v4/internal/conf" "github.com/OpenListTeam/OpenList/v4/internal/errs" @@ -28,6 +29,9 @@ func (f RangeReaderFunc) RangeRead(ctx context.Context, httpRange http_range.Ran } func GetRangeReaderFromLink(size int64, link *model.Link) (model.RangeReaderIF, error) { + if len(link.MultiSource) > 1 && (link.Concurrency > 0 || link.PartSize > 0) { + return newMultiSourceRangeReader(size, link), nil + } if link.RangeReader != nil { if link.Concurrency < 1 && link.PartSize < 1 { return link.RangeReader, nil @@ -122,6 +126,240 @@ func GetRangeReaderFromLink(size int64, link *model.Link) (model.RangeReaderIF, return RangeReaderFunc(rangeReader), nil } +// newMultiSourceRangeReader 构造一个把分片轮询分发到 link.MultiSource 各源直连的 RangeReader。 +// 客户端请求的整段 [start, start+length) 被切成若干 PartSize 的子分片(切片规则对齐 +// internal/net.Downloader:首片取 length%PartSize 的小片,其余 PartSize),第 i 个子分片 +// 用 sources[i % len] 的 URL+Header 发 Range 请求。总并发为 N*Concurrency(N=源数)。 +// 单片源失败时轮换到下一个源重试,所有源都失败才报错。 +func newMultiSourceRangeReader(size int64, link *model.Link) model.RangeReaderIF { + sources := link.MultiSource + partSize := link.PartSize + if partSize <= 0 { + partSize = int(net.DefaultDownloadPartSize) + } + if conf.MaxBufferLimit > 0 && partSize > conf.MaxBufferLimit { + partSize = conf.MaxBufferLimit + } + perSourceConcurrency := link.Concurrency + if perSourceConcurrency <= 0 { + perSourceConcurrency = net.DefaultDownloadConcurrency + } + // 总并发放大为 N * 单链并发。 + totalConcurrency := perSourceConcurrency * len(sources) + if totalConcurrency < 1 { + totalConcurrency = 1 + } + + type subChunk struct { + start int64 + length int64 + } + // 对齐 Downloader 切分:首片为 length%PartSize 的余片(太小则取 PartSize/2),其余 PartSize。 + computeSubchunks := func(start, length int64) []subChunk { + if length < 0 { + length = size - start + } + if length < 0 { + length = 0 + } + if start+length > size { + length = size - start + } + var chunks []subChunk + pos := start + remaining := length + nextSize := int64(partSize) + if first := length % int64(partSize); first > 0 { + minSize := int64(partSize) / 2 + if first < minSize { + nextSize = minSize + } else { + nextSize = first + } + } + for remaining > 0 { + sz := nextSize + if sz > remaining { + sz = remaining + } + chunks = append(chunks, subChunk{start: pos, length: sz}) + pos += sz + remaining -= sz + nextSize = int64(partSize) + } + return chunks + } + + rangeReader := func(ctx context.Context, httpRange http_range.Range) (io.ReadCloser, error) { + chunks := computeSubchunks(httpRange.Start, httpRange.Length) + if len(chunks) == 0 { + return io.NopCloser(bytes.NewReader(nil)), nil + } + + ctx, cancel := context.WithCancelCause(ctx) + // 每个 subChunk 一个结果槽:buf 为下载到的字节,err 为失败原因,ready 通知。 + slots := make([]multiSourceSlot, len(chunks)) + for i := range slots { + slots[i].ready = make(chan struct{}) + } + + // 信号量控制 worker 并发为 totalConcurrency;信号量在 worker 退出时释放。 + sem := make(chan struct{}, totalConcurrency) + var wg sync.WaitGroup + + launch := func(idx int) { + wg.Add(1) + go func() { + defer wg.Done() + defer func() { <-sem }() // 释放本 worker 槽位 + ch := chunks[idx] + nSrc := len(sources) + startSrc := idx % nSrc + var lastErr error + for attempt := 0; attempt < nSrc; attempt++ { + if ctx.Err() != nil { + lastErr = context.Cause(ctx) + break + } + src := sources[(startSrc+attempt)%nSrc] + data, err := downloadMultiSourceSubChunk(ctx, src, ch.start, ch.length, size) + if err == nil { + slots[idx].buf = data + close(slots[idx].ready) + return + } + lastErr = err + log.Debugf("[multi-source] subChunk_%d src_%d failed: %v", idx, (startSrc+attempt)%nSrc, err) + } + slots[idx].err = lastErr + close(slots[idx].ready) + cancel(lastErr) + }() + } + + // 调度器:满并发地启动所有分片(信号量自动限流),ctx 取消时收尾。 + go func() { + for i := range chunks { + select { + case sem <- struct{}{}: + case <-ctx.Done(): + return + } + launch(i) + } + }() + + reader := &multiSourceReader{ + ctx: ctx, + slots: slots, + cancel: cancel, + } + // reader.Close 应等 worker 退出并取消残余请求。 + reader.wg = &wg + return reader, nil + } + return RangeReaderFunc(rangeReader) +} + +// multiSourceSlot 是单个 subChunk 的下载结果槽。 +type multiSourceSlot struct { + buf []byte + err error + ready chan struct{} +} + +// multiSourceReader 按 subChunk 顺序阻塞地拼回各槽下载到的字节。 +type multiSourceReader struct { + ctx context.Context + slots []multiSourceSlot + cur int // 当前正在读的 subChunk 索引 + off int // 当前 subChunk 已读偏移 + cancel context.CancelCauseFunc + wg *sync.WaitGroup +} + +func (r *multiSourceReader) Read(p []byte) (int, error) { + for { + if r.cur >= len(r.slots) { + return 0, io.EOF + } + select { + case <-r.ctx.Done(): + return 0, context.Cause(r.ctx) + default: + } + slot := &r.slots[r.cur] + select { + case <-slot.ready: + case <-r.ctx.Done(): + return 0, context.Cause(r.ctx) + } + if slot.err != nil { + return 0, slot.err + } + // 还剩多少可从当前分片读。 + remaining := len(slot.buf) - r.off + if remaining <= 0 { + r.cur++ + r.off = 0 + continue + } + n := copy(p, slot.buf[r.off:]) + r.off += n + if r.off >= len(slot.buf) { + r.cur++ + r.off = 0 + } + return n, nil + } +} + +// Close 取消所有残余 worker 并等其退出,释放并发槽位。 +func (r *multiSourceReader) Close() error { + if r.cancel != nil { + r.cancel(context.Canceled) + } + if r.wg != nil { + r.wg.Wait() + } + return nil +} + +// downloadMultiSourceSubChunk 用单条源直连取 [start, start+length) 字节。 +// header 中预置 Range;失败由调用方据此轮换源。带 ServerDownloadLimit 限速。 +func downloadMultiSourceSubChunk(ctx context.Context, src model.LinkSource, start, length, totalSize int64) ([]byte, error) { + if length <= 0 { + if totalSize > start { + length = totalSize - start + } else { + length = 0 + } + } + header := net.ProcessHeader(nil, src.Header) + httpRange := http_range.Range{Start: start, Length: length} + header = http_range.ApplyRangeToHttpHeader(httpRange, header) + resp, err := net.RequestHttp(ctx, "GET", header, src.URL) + if err != nil { + return nil, err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusPartialContent { + return nil, fmt.Errorf("multi-source chunk status %d", resp.StatusCode) + } + var reader io.Reader = resp.Body + if ServerDownloadLimit != nil { + reader = &RateLimitReader{Ctx: ctx, Reader: resp.Body, Limiter: ServerDownloadLimit} + } + data, err := io.ReadAll(reader) + if err != nil { + return nil, err + } + if int64(len(data)) != length { + return data, fmt.Errorf("multi-source chunk size mismatch: want %d got %d", length, len(data)) + } + return data, nil +} + func GetRangeReaderFromMFile(size int64, file model.File) *model.FileRangeReader { return &model.FileRangeReader{ RangeReaderIF: RangeReaderFunc(func(ctx context.Context, httpRange http_range.Range) (io.ReadCloser, error) { From 1617ac19d7203ec28847b62d6141e210cd294773 Mon Sep 17 00:00:00 2001 From: Har01d Date: Mon, 3 Aug 2026 07:17:41 +0800 Subject: [PATCH 2/4] =?UTF-8?q?=E5=A4=B8=E5=85=8B=E5=A4=9A=E8=B4=A6?= =?UTF-8?q?=E5=8F=B7=E5=8A=A0=E9=80=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- drivers/quark_uc_share/driver.go | 6 +- drivers/quark_uc_share/driver_test.go | 6 +- drivers/quark_uc_share/util.go | 39 ++++-- drivers/thunder_share/util_test.go | 80 ++++++++++++ internal/stream/multi_source_test.go | 174 ++++++++++++++++++++++++++ internal/stream/util.go | 15 ++- server/handles/down.go | 4 +- 7 files changed, 302 insertions(+), 22 deletions(-) create mode 100644 drivers/thunder_share/util_test.go create mode 100644 internal/stream/multi_source_test.go diff --git a/drivers/quark_uc_share/driver.go b/drivers/quark_uc_share/driver.go index 0507863ff..2a684ba5d 100755 --- a/drivers/quark_uc_share/driver.go +++ b/drivers/quark_uc_share/driver.go @@ -117,14 +117,12 @@ func (d *QuarkUCShare) Link(ctx context.Context, file model.Obj, args model.Link link, err = resolveShareDirectLink(d, file) } // 多账号分片并行下载:开关开启、有≥2 个网盘账号且转存主链成功时, - // 对各账号各取一条直连填 link.MultiSource,总并发按账号数放大(N×Concurrency)。 + // 对各账号各取一条直连填 link.MultiSource。总并发在 multiSourceRangeReader 内部 + // 按 N 放大(带上限 64),这里不再改 Concurrency(applyProxyConfig 会按 Quark 覆盖为 32)。 // 任意一步失败都不影响主链,只是 MultiSource 留空 → 退回单链多线程。 if err == nil && link != nil && multiSourceEnabled(d) { if ms := collectMultiSource(ctx, d, file, args, link); len(ms) > 1 { link.MultiSource = ms - if link.Concurrency > 0 { - link.Concurrency *= len(ms) - } log.Infof("[multi-source] %s %s 用 %d 账号分片下载 file=%s", d.config.Name, file.GetName(), len(ms), file.GetID()) } } diff --git a/drivers/quark_uc_share/driver_test.go b/drivers/quark_uc_share/driver_test.go index e927f568b..4b24c87ee 100644 --- a/drivers/quark_uc_share/driver_test.go +++ b/drivers/quark_uc_share/driver_test.go @@ -270,9 +270,9 @@ func TestQuarkUCShareLink_PopulatesMultiSourceWhenEnabled(t *testing.T) { if len(link.MultiSource) != 3 { t.Fatalf("expected 3 MultiSource, got %d", len(link.MultiSource)) } - // Concurrency 4 放大 3 倍 = 12。 - if link.Concurrency != 12 { - t.Fatalf("expected Concurrency=12 (4*3), got %d", link.Concurrency) + // Concurrency 不在 Link 放大(multiSourceRangeReader 内部按 N 放大且带上限),保持原值。 + if link.Concurrency != 4 { + t.Fatalf("expected Concurrency unchanged=4, got %d", link.Concurrency) } } diff --git a/drivers/quark_uc_share/util.go b/drivers/quark_uc_share/util.go index 45993f264..bf7c9797a 100755 --- a/drivers/quark_uc_share/util.go +++ b/drivers/quark_uc_share/util.go @@ -363,7 +363,6 @@ func (d *QuarkUCShare) requestAt(api, cookieStr, pathname string, method string, } var e Resp req.SetError(&e) - log.Debugf("[share-pc] %s %s cookie=%q", method, u, cookieStr) res, err := req.Execute(method, u) if err != nil { return nil, err @@ -543,8 +542,8 @@ func (d *QuarkUCShare) doSaveShareFile(ctx context.Context, binding shareRequest log.Debugf("saveFile: %v response: %v, error: %v", parsed.FileID, string(res), err) if err != nil { msg := err.Error() - log.Warnf("[save-debug] fid=%s fidToken=%s stoken=%q to_pdir=%s accountCookie=%q err=%s", - parsed.FileID, fidToken, d.ShareToken, binding.tempDirID(), binding.cookieValue(), msg) + log.Warnf("[save-debug] fid=%s fidToken=%s stoken=%q to_pdir=%s err=%s", + parsed.FileID, fidToken, d.ShareToken, binding.tempDirID(), msg) // fid_token 过期:按父目录重新换一个 share_fid_token 再试一次。 if strings.Contains(msg, "token校验异常") && parsed.ParentID != "" { fidToken, err = d.getFileToken(binding, parsed.ParentID, parsed.FileID) @@ -645,10 +644,13 @@ func sleepCtx(ctx context.Context, d time.Duration) error { func (d *QuarkUCShare) saveAndLink(ctx context.Context, binding shareRequestBinding, id string, args model.LinkArgs) (*model.Link, error) { var lastErr error for range 2 { + saveStart := time.Now() file, reused, err := d.saveShareFile(ctx, binding, id) + saveElapsed := time.Since(saveStart) if err != nil { return nil, err } + log.Debugf("[saveAndLink] 账号 %d fid=%s 转存 耗时=%v reused=%v", binding.accountID(), file.GetID(), saveElapsed, reused) key := "" if parsed, perr := parseShareFileID(id); perr == nil { key = savedFileKey(d, binding, parsed.FileID) @@ -657,11 +659,13 @@ func (d *QuarkUCShare) saveAndLink(ctx context.Context, binding shareRequestBind // 网盘转存的文件走 speedup token 提速直链(dl-c 提速通道,实测 8x); // TV 账号或 speedup 失败时回退 binding.link 的普通直链。 var link *model.Link + linkStart := time.Now() if rb, ok := binding.(requestBinding); ok { link, err = d.speedupLink(ctx, rb, file, args) } else { link, err = binding.link(ctx, file, args) } + log.Debugf("[saveAndLink] 账号 %d fid=%s 取链 耗时=%v", binding.accountID(), file.GetID(), time.Since(linkStart)) if err == nil { return link, nil } @@ -1062,6 +1066,7 @@ func multiSourceMax() int { // 发到分片代理的直连去掉 #x-referer=raw 片段标记(那是给客户端代理用的,后端代理走 link.Header)。 // 声明为 var 便于单测替换(避免单测里 op 还原账号)。 var collectMultiSource = func(ctx context.Context, d *QuarkUCShare, file model.Obj, args model.LinkArgs, primary *model.Link) []model.LinkSource { + collectStart := time.Now() name := d.getDriverName() storages := op.GetStorages(name) if len(storages) < 2 { @@ -1076,8 +1081,14 @@ var collectMultiSource = func(ctx context.Context, d *QuarkUCShare, file model.O storages = storages[:n] } + // 收集设硬上限,避免启播被慢账号的转存/speedup 拖垮;超时的账号放弃、只用已完成的。 + collectCtx, cancel := context.WithTimeout(ctx, multiSourceCollectTimeout()) + defer cancel() + type result struct { - l *model.Link + l *model.Link + accountID uint + elapsed time.Duration } results := make(chan result, len(storages)) var wg sync.WaitGroup @@ -1089,18 +1100,21 @@ var collectMultiSource = func(ctx context.Context, d *QuarkUCShare, file model.O wg.Add(1) go func(uc *quark.QuarkOrUC) { defer wg.Done() - if ctx.Err() != nil { + if collectCtx.Err() != nil { return } + acctStart := time.Now() binding := bindRequestDriver(uc) - l, e := d.saveAndLink(ctx, binding, file.GetID(), args) + l, e := d.saveAndLink(collectCtx, binding, file.GetID(), args) + elapsed := time.Since(acctStart) if e == nil && l != nil && l.URL != "" { + log.Infof("[multi-source] 账号 %d 取链完成 耗时=%v", uc.ID, elapsed) select { - case results <- result{l: l}: - case <-ctx.Done(): + case results <- result{l: l, accountID: uc.ID, elapsed: elapsed}: + case <-collectCtx.Done(): } } else { - log.Warnf("[multi-source] 账号 %d 取链失败: %v", uc.ID, e) + log.Warnf("[multi-source] 账号 %d 取链失败 耗时=%v err=%v", uc.ID, elapsed, e) } }(uc) } @@ -1121,10 +1135,15 @@ var collectMultiSource = func(ctx context.Context, d *QuarkUCShare, file model.O seen[s.URL] = true srcs = append(srcs, s) } - log.Infof("[multi-source] collect: driver=%s accounts=%d collected=%d", name, len(storages), len(srcs)) + log.Infof("[multi-source] collect: driver=%s accounts=%d collected=%d 总耗时=%v", name, len(storages), len(srcs), time.Since(collectStart)) return srcs } +// multiSourceCollectTimeout 多账号取链的总超时,避免启播卡死。默认 8s。 +func multiSourceCollectTimeout() time.Duration { + return 8 * time.Second +} + // sourceFromLink 从单账号取链结果构造 LinkSource:去掉 #x-referer=raw 片段标记, // 保留 URL + Header(后端代理直接用 link.Header,不走魔法 Referer 片段契约)。 func sourceFromLink(l *model.Link, d *QuarkUCShare) model.LinkSource { diff --git a/drivers/thunder_share/util_test.go b/drivers/thunder_share/util_test.go new file mode 100644 index 000000000..8e181817b --- /dev/null +++ b/drivers/thunder_share/util_test.go @@ -0,0 +1,80 @@ +package thunder_share + +import ( + "context" + "testing" +) + +func TestThunderShareCleanupContextSurvivesParentCancellation(t *testing.T) { + type contextKey string + const key contextKey = "request-id" + + parent := context.WithValue(context.Background(), key, "req-1") + parent, cancel := context.WithCancel(parent) + cancel() + + cleanupCtx := thunderShareCleanupContext(parent) + if err := cleanupCtx.Err(); err != nil { + t.Fatalf("expected cleanup context to ignore parent cancellation, got %v", err) + } + if got := cleanupCtx.Value(key); got != "req-1" { + t.Fatalf("expected cleanup context to preserve values, got %v", got) + } +} + +func TestThunderShareRestoreResponseFileIDFromTask(t *testing.T) { + resp := thunderShareRestoreResponse{ + Params: thunderShareRestoreParams{ + TraceFileIDs: `{"share-file-id":"restored-file-id"}`, + }, + } + + id, ok := resp.RestoredFileID("share-file-id") + if !ok { + t.Fatal("expected restored file id from restore task") + } + if id != "restored-file-id" { + t.Fatalf("expected restored file id, got %q", id) + } +} + +func TestThunderShareRestoreResponseFileIDFromTaskIgnoresMissingAndInvalidTrace(t *testing.T) { + tests := []struct { + name string + resp thunderShareRestoreResponse + }{ + { + name: "missing source id", + resp: thunderShareRestoreResponse{ + Params: thunderShareRestoreParams{ + TraceFileIDs: `{"other-share-file-id":"restored-file-id"}`, + }, + }, + }, + { + name: "invalid trace json", + resp: thunderShareRestoreResponse{ + Params: thunderShareRestoreParams{ + TraceFileIDs: `{`, + }, + }, + }, + { + name: "empty restored id", + resp: thunderShareRestoreResponse{ + Params: thunderShareRestoreParams{ + TraceFileIDs: `{"share-file-id":""}`, + }, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + id, ok := tt.resp.RestoredFileID("share-file-id") + if ok { + t.Fatalf("did not expect restored file id, got %q", id) + } + }) + } +} diff --git a/internal/stream/multi_source_test.go b/internal/stream/multi_source_test.go new file mode 100644 index 000000000..3b7f9dbac --- /dev/null +++ b/internal/stream/multi_source_test.go @@ -0,0 +1,174 @@ +package stream + +import ( + "context" + "fmt" + "io" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + + "github.com/OpenListTeam/OpenList/v4/internal/conf" + "github.com/OpenListTeam/OpenList/v4/internal/model" + "github.com/OpenListTeam/OpenList/v4/pkg/http_range" +) + +func init() { + // NewHttpClient 读 conf.Conf.TlsInsecureSkipVerify,测试需初始化一份默认配置避免空指针。 + conf.Conf = conf.DefaultConfig("data") +} + +// parseRangeHeader 解析 "bytes=start-end" 含 end 所指偏移(开区间上界+1)。 +func parseRangeHeader(hdr string, total int) (int64, int64, error) { + if hdr == "" { + return 0, int64(total), nil + } + var start, end int64 + if _, err := fmt.Sscanf(hdr, "bytes=%d-%d", &start, &end); err != nil { + return 0, 0, err + } + return start, end + 1, nil +} + +func TestMultiSourceRangeReader_RoundRobinAcrossSources(t *testing.T) { + // 构造 4 个 PartSize 分片大小的文件,起 2 个 server,各"只"正确响应奇偶分片: + // server0 负责片 0、2;server1 负责片 1、3。验证拼回完整文件、奇偶片确实分到两 host。 + const partSize = 1 << 8 // 256 + const numParts = 4 + payload := make([]byte, partSize*numParts) + for i := range payload { + payload[i] = byte(i) + } + + // 每个 server 内部按"该片段不被任何片跨越"硬性只接自己负责的片。 + mkServer := func(responsibleParts []int) (*httptest.Server, *int32) { + allow := make(map[int]bool) + for _, p := range responsibleParts { + allow[p] = true + } + var hits int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&hits, 1) + start, end, err := parseRangeHeader(r.Header.Get("Range"), len(payload)) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + whichPart := int(start / int64(partSize)) + if !allow[whichPart] { + http.Error(w, fmt.Sprintf("not responsible for part %d", whichPart), http.StatusRequestedRangeNotSatisfiable) + return + } + if end > int64(len(payload)) { + end = int64(len(payload)) + } + w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, end-1, len(payload))) + w.Header().Set("Content-Length", fmt.Sprintf("%d", end-start)) + w.WriteHeader(http.StatusPartialContent) + _, _ = w.Write(payload[start:end]) + })) + return srv, &hits + } + s0, h0 := mkServer([]int{0, 2}) + s1, h1 := mkServer([]int{1, 3}) + defer s0.Close() + defer s1.Close() + + link := &model.Link{ + PartSize: partSize, + Concurrency: 2, + MultiSource: []model.LinkSource{ + {URL: s0.URL}, + {URL: s1.URL}, + }, + } + rr := newMultiSourceRangeReader(int64(len(payload)), link) + + // 整段读回。 + rc, err := rr.RangeRead(context.Background(), http_range.Range{Start: 0, Length: int64(len(payload))}) + if err != nil { + t.Fatalf("RangeRead: %v", err) + } + defer rc.Close() + got, err := io.ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if len(got) != len(payload) { + t.Fatalf("size mismatch: want %d got %d", len(payload), len(got)) + } + for i := range payload { + if got[i] != payload[i] { + t.Fatalf("byte %d mismatch: want %d got %d", i, payload[i], got[i]) + } + } + + // 奇偶片应分别落到两 server:server0 至少处理片0(4片首片小); + // server1 至少处理片1。 + if atomic.LoadInt32(h0) == 0 { + t.Fatalf("server0 never hit; round-robin not working") + } + if atomic.LoadInt32(h1) == 0 { + t.Fatalf("server1 never hit; round-robin not working") + } + t.Logf("hits: server0=%d server1=%d", atomic.LoadInt32(h0), atomic.LoadInt32(h1)) +} + +func TestMultiSourceRangeReader_SourceFailover(t *testing.T) { + // server0 对所有片都报错;server1 全部正确。验证任意片轮换到 server1 成功。 + const partSize = 1 << 8 + const numParts = 3 + payload := make([]byte, partSize*numParts) + for i := range payload { + payload[i] = byte(0xAB) + } + bad := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "always fail", http.StatusServiceUnavailable) + })) + defer bad.Close() + + var goodHits int32 + good := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&goodHits, 1) + start, end, err := parseRangeHeader(r.Header.Get("Range"), len(payload)) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + if end > int64(len(payload)) { + end = int64(len(payload)) + } + w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, end-1, len(payload))) + w.Header().Set("Content-Length", fmt.Sprintf("%d", end-start)) + w.WriteHeader(http.StatusPartialContent) + _, _ = w.Write(payload[start:end]) + })) + defer good.Close() + + link := &model.Link{ + PartSize: partSize, + Concurrency: 2, + MultiSource: []model.LinkSource{ + {URL: bad.URL}, + {URL: good.URL}, + }, + } + rr := newMultiSourceRangeReader(int64(len(payload)), link) + rc, err := rr.RangeRead(context.Background(), http_range.Range{Start: 0, Length: int64(len(payload))}) + if err != nil { + t.Fatalf("RangeRead: %v", err) + } + defer rc.Close() + got, err := io.ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if len(got) != len(payload) { + t.Fatalf("size mismatch: want %d got %d", len(payload), len(got)) + } + if atomic.LoadInt32(&goodHits) == 0 { + t.Fatalf("good server never hit; failover broken") + } + t.Logf("good server hits=%d", atomic.LoadInt32(&goodHits)) +} diff --git a/internal/stream/util.go b/internal/stream/util.go index 3d998c7a1..68c8b097d 100644 --- a/internal/stream/util.go +++ b/internal/stream/util.go @@ -144,8 +144,12 @@ func newMultiSourceRangeReader(size int64, link *model.Link) model.RangeReaderIF if perSourceConcurrency <= 0 { perSourceConcurrency = net.DefaultDownloadConcurrency } - // 总并发放大为 N * 单链并发。 + // 总并发按源数放大,但加上限:单 IP 出口和 DNS 解析承受不了 N*perSource 的全开并发 + // (如 5 源 × 32 = 160 并发会压垮 DNS,出现大量 lookup canceled)。 totalConcurrency := perSourceConcurrency * len(sources) + if totalConcurrency > 64 { + totalConcurrency = 64 + } if totalConcurrency < 1 { totalConcurrency = 1 } @@ -229,11 +233,16 @@ func newMultiSourceRangeReader(size int64, link *model.Link) model.RangeReaderIF return } lastErr = err - log.Debugf("[multi-source] subChunk_%d src_%d failed: %v", idx, (startSrc+attempt)%nSrc, err) + // 客户端断开/seek 导致的取消是正常的(播放器频繁 Range 取消),不打日志避免刷屏。 + if !errors.Is(err, context.Canceled) && ctx.Err() == nil { + log.Debugf("[multi-source] subChunk_%d src_%d failed: %v", idx, (startSrc+attempt)%nSrc, err) + } } slots[idx].err = lastErr close(slots[idx].ready) - cancel(lastErr) + if lastErr != nil && !errors.Is(lastErr, context.Canceled) { + cancel(lastErr) + } }() } diff --git a/server/handles/down.go b/server/handles/down.go index 624e69ca3..5cee22a7c 100644 --- a/server/handles/down.go +++ b/server/handles/down.go @@ -163,7 +163,7 @@ func applyProxyConfig(driver string, link *model.Link) { if alias, ok := proxyDriverAliases[driver]; ok { driver = alias } - log.Debugf("apply proxy config: %+v driver: %v", cfg, driver) + //log.Debugf("apply proxy config: %+v driver: %v", cfg, driver) item, ok := cfg[driver] if !ok { return @@ -174,7 +174,7 @@ func applyProxyConfig(driver string, link *model.Link) { if item.ChunkSize > 0 { link.PartSize = item.ChunkSize * utils.KB } - log.Debugf("[proxy] link: %+v", link) + //log.Debugf("[proxy] link: %+v", link) } // TODO need optimize From a01e6814d2d3b29cee7105eac3da7b91b9f06cba Mon Sep 17 00:00:00 2001 From: Har01d Date: Mon, 3 Aug 2026 07:55:39 +0800 Subject: [PATCH 3/4] =?UTF-8?q?=E5=A4=B8=E5=85=8B=E5=A4=9A=E8=B4=A6?= =?UTF-8?q?=E5=8F=B7=E5=8A=A0=E9=80=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- drivers/quark_uc_share/driver.go | 37 +++++++----- drivers/quark_uc_share/driver_test.go | 39 +++++++------ drivers/quark_uc_share/util.go | 84 ++++++++++++++++++--------- 3 files changed, 100 insertions(+), 60 deletions(-) diff --git a/drivers/quark_uc_share/driver.go b/drivers/quark_uc_share/driver.go index 2a684ba5d..4892a8a9d 100755 --- a/drivers/quark_uc_share/driver.go +++ b/drivers/quark_uc_share/driver.go @@ -106,9 +106,30 @@ func (d *QuarkUCShare) Link(ctx context.Context, file model.Obj, args model.Link return link, nil } - // 转存 + speedup token 为主(提速直链走 dl-c 通道,实测 ~14MB/s vs 免转存 ~1.7MB/s); - // speedup 必须先转存(聊天会话只认自己网盘的 fid),故转存是快速通道的前提。 - // 免转存(share-direct)降为兜底:转存失败/无网盘账号时仍可播,但被限速。 + // 多账号分片并行下载(开关开):直接对所有网盘账号 goroutine 并发取链, + // 跳过串行的 resolveQuarkUCShareLink(它失败才换下一个账号,慢;且会让主链账号被重复取)。 + // 首个账号成功后只额外等 2s 收集更多源,不等最慢账号,启播≈首源+2s。 + if multiSourceEnabled(d) { + links := collectMultiAccountLinks(ctx, d, file, args) + if len(links) > 0 { + link := links[0] + if len(links) > 1 { + link.MultiSource = sourcesFromLinks(links, d) + log.Infof("[multi-source] %s %s 用 %d 账号分片下载 file=%s", d.config.Name, file.GetName(), len(link.MultiSource), file.GetID()) + } + quarkUCShareLinkCache.Set(key, link) + return link, nil + } + // 全部账号取链失败,回退免转存兜底。 + log.Warnf("[multi-source] 全部账号取链失败,回退免转存") + link, err := resolveShareDirectLink(d, file) + if err == nil && link != nil { + quarkUCShareLinkCache.Set(key, link) + } + return link, err + } + + // 开关关:转存 + speedup 为主(串行轮询账号,失败换下一个),免转存兜底。 var link *model.Link var err error link, err = resolveQuarkUCShareLink(ctx, d, file, args) @@ -116,16 +137,6 @@ func (d *QuarkUCShare) Link(ctx context.Context, file model.Obj, args model.Link log.Warnf("转存取链失败,回退免转存: %v", err) link, err = resolveShareDirectLink(d, file) } - // 多账号分片并行下载:开关开启、有≥2 个网盘账号且转存主链成功时, - // 对各账号各取一条直连填 link.MultiSource。总并发在 multiSourceRangeReader 内部 - // 按 N 放大(带上限 64),这里不再改 Concurrency(applyProxyConfig 会按 Quark 覆盖为 32)。 - // 任意一步失败都不影响主链,只是 MultiSource 留空 → 退回单链多线程。 - if err == nil && link != nil && multiSourceEnabled(d) { - if ms := collectMultiSource(ctx, d, file, args, link); len(ms) > 1 { - link.MultiSource = ms - log.Infof("[multi-source] %s %s 用 %d 账号分片下载 file=%s", d.config.Name, file.GetName(), len(ms), file.GetID()) - } - } if err == nil && link != nil { quarkUCShareLinkCache.Set(key, link) } diff --git a/drivers/quark_uc_share/driver_test.go b/drivers/quark_uc_share/driver_test.go index 4b24c87ee..054631fc8 100644 --- a/drivers/quark_uc_share/driver_test.go +++ b/drivers/quark_uc_share/driver_test.go @@ -17,14 +17,14 @@ import ( func stubMultiSourceDisabled(t *testing.T) { t.Helper() origMS := multiSourceEnabled - origCollect := collectMultiSource + origCollect := collectMultiAccountLinks multiSourceEnabled = func(d *QuarkUCShare) bool { return false } - collectMultiSource = func(ctx context.Context, d *QuarkUCShare, file model.Obj, args model.LinkArgs, primary *model.Link) []model.LinkSource { + collectMultiAccountLinks = func(ctx context.Context, d *QuarkUCShare, file model.Obj, args model.LinkArgs) []*model.Link { return nil } t.Cleanup(func() { multiSourceEnabled = origMS - collectMultiSource = origCollect + collectMultiAccountLinks = origCollect }) } @@ -34,7 +34,7 @@ func TestQuarkUCShareLink_CachesByFileID(t *testing.T) { origDirect := resolveShareDirectLink origSVIP := accountIsSVIP origMS := multiSourceEnabled - origCollect := collectMultiSource + origCollect := collectMultiAccountLinks quarkUCShareLinkCache = cache.NewKeyedCache[*model.Link](time.Hour) resolveCalls := 0 resolveQuarkUCShareLink = func(ctx context.Context, d *QuarkUCShare, file model.Obj, args model.LinkArgs) (*model.Link, error) { @@ -46,7 +46,7 @@ func TestQuarkUCShareLink_CachesByFileID(t *testing.T) { return nil, errors.New("share-direct stub disabled") // 置失败,流程落到 resolveQuarkUCShareLink } multiSourceEnabled = func(d *QuarkUCShare) bool { return false } - collectMultiSource = func(ctx context.Context, d *QuarkUCShare, file model.Obj, args model.LinkArgs, primary *model.Link) []model.LinkSource { + collectMultiAccountLinks = func(ctx context.Context, d *QuarkUCShare, file model.Obj, args model.LinkArgs) []*model.Link { return nil } t.Cleanup(func() { @@ -55,7 +55,7 @@ func TestQuarkUCShareLink_CachesByFileID(t *testing.T) { resolveShareDirectLink = origDirect accountIsSVIP = origSVIP multiSourceEnabled = origMS - collectMultiSource = origCollect + collectMultiAccountLinks = origCollect }) d := &QuarkUCShare{Addition: Addition{ShareToken: "share-token"}, config: driver.Config{Name: "QuarkShare"}} @@ -93,7 +93,7 @@ func TestQuarkUCShareLink_DoesNotCacheErrors(t *testing.T) { return nil, errors.New("share-direct stub disabled") // 失败 → 落到同样报错的 resolveQuarkUCShareLink } multiSourceEnabled = func(d *QuarkUCShare) bool { return false } - collectMultiSource = func(ctx context.Context, d *QuarkUCShare, file model.Obj, args model.LinkArgs, primary *model.Link) []model.LinkSource { + collectMultiAccountLinks = func(ctx context.Context, d *QuarkUCShare, file model.Obj, args model.LinkArgs) []*model.Link { return nil } t.Cleanup(func() { @@ -227,28 +227,28 @@ func TestQuarkUCShareLink_PrefersSaveAndSpeedup(t *testing.T) { } } -// 开关开启、collectMultiSource 返回多源时,Link() 把 MultiSource 填上并把 Concurrency 放大 N 倍。 +// 开关开启、collectMultiAccountLinks 返回多条链时,Link() 跳过串行主链,直接用首条为 primary、全部填 MultiSource。 func TestQuarkUCShareLink_PopulatesMultiSourceWhenEnabled(t *testing.T) { stubMultiSourceDisabled(t) origCache := quarkUCShareLinkCache origResolver := resolveQuarkUCShareLink origDirect := resolveShareDirectLink quarkUCShareLinkCache = cache.NewKeyedCache[*model.Link](time.Hour) + // 开关开时 Link() 不再调 resolveQuarkUCShareLink(直接并发),这里设个哨兵确保不被调用。 resolveQuarkUCShareLink = func(ctx context.Context, d *QuarkUCShare, file model.Obj, args model.LinkArgs) (*model.Link, error) { - // 主链:账号1 直链。 - return &model.Link{URL: "https://cdn.example.com/account1/fid", Concurrency: 4, PartSize: 512 << 10}, nil + t.Fatalf("resolveQuarkUCShareLink should not be called when multi-source enabled") + return nil, nil } resolveShareDirectLink = func(d *QuarkUCShare, file model.Obj) (*model.Link, error) { return nil, errors.New("share-direct stub disabled") } - // 开启多账号分片:collectMultiSource 返回 3 个源。 + // 开启多账号分片:collectMultiAccountLinks 返回 3 条链(首条为 primary)。 multiSourceEnabled = func(d *QuarkUCShare) bool { return true } - collectMultiSource = func(ctx context.Context, d *QuarkUCShare, file model.Obj, args model.LinkArgs, primary *model.Link) []model.LinkSource { - primarySrc := model.LinkSource{URL: "https://cdn.example.com/account1/fid"} - return []model.LinkSource{ - primarySrc, - {URL: "https://cdn.example.com/account2/fid"}, - {URL: "https://cdn.example.com/account3/fid"}, + collectMultiAccountLinks = func(ctx context.Context, d *QuarkUCShare, file model.Obj, args model.LinkArgs) []*model.Link { + return []*model.Link{ + {URL: "https://cdn.example.com/account1/fid", Concurrency: 4, PartSize: 512 << 10}, + {URL: "https://cdn.example.com/account2/fid", Concurrency: 4, PartSize: 512 << 10}, + {URL: "https://cdn.example.com/account3/fid", Concurrency: 4, PartSize: 512 << 10}, } } t.Cleanup(func() { @@ -267,10 +267,13 @@ func TestQuarkUCShareLink_PopulatesMultiSourceWhenEnabled(t *testing.T) { if link == nil { t.Fatal("nil link") } + if link.URL != "https://cdn.example.com/account1/fid" { + t.Fatalf("expected primary=account1, got %s", link.URL) + } if len(link.MultiSource) != 3 { t.Fatalf("expected 3 MultiSource, got %d", len(link.MultiSource)) } - // Concurrency 不在 Link 放大(multiSourceRangeReader 内部按 N 放大且带上限),保持原值。 + // Concurrency 不在 Link 放大(multiSourceRangeReader 内部按 N 放大且带上限),保持 primary 原值。 if link.Concurrency != 4 { t.Fatalf("expected Concurrency unchanged=4, got %d", link.Concurrency) } diff --git a/drivers/quark_uc_share/util.go b/drivers/quark_uc_share/util.go index bf7c9797a..eecd0a424 100755 --- a/drivers/quark_uc_share/util.go +++ b/drivers/quark_uc_share/util.go @@ -1058,14 +1058,14 @@ func multiSourceMax() int { return m } -// collectMultiSource 对每个网盘账号各跑一遍转存+speedup 取链,收集成功的直连成 LinkSource 列表。 +// collectMultiAccountLinks 用 goroutine 对每个网盘账号并发 saveAndLink 取链。 // 关键:必须用 op.GetStorages 遍历每个账号再各自 saveAndLink —— 不能用 d.link(), // 因为 d.link() 内部 GetFirstDriver 在 DriverRoundRobin 关闭时总返回 master 账号, -// 并发调用只会取到同一条链、去重后只剩 1 个源。 -// 总数受 multiSourceMax() 限制(0=不限,默认 3)。任一账号失败跳过。 -// 发到分片代理的直连去掉 #x-referer=raw 片段标记(那是给客户端代理用的,后端代理走 link.Header)。 -// 声明为 var 便于单测替换(避免单测里 op 还原账号)。 -var collectMultiSource = func(ctx context.Context, d *QuarkUCShare, file model.Obj, args model.LinkArgs, primary *model.Link) []model.LinkSource { +// 并发调用只会取到同一条链。 +// 启播策略:所有账号同时跑,第一个成功后只额外等 firstSourceGrace(2s)收集更多源就返回, +// 不等最慢的账号 —— 这样启播≈首源时间+2s,慢账号不拖垮启播。总数受 multiSourceMax() 限制。 +// 声明为 var 便于单测替换(避免单测里 op 未初始化)。 +var collectMultiAccountLinks = func(ctx context.Context, d *QuarkUCShare, file model.Obj, args model.LinkArgs) []*model.Link { collectStart := time.Now() name := d.getDriverName() storages := op.GetStorages(name) @@ -1081,16 +1081,10 @@ var collectMultiSource = func(ctx context.Context, d *QuarkUCShare, file model.O storages = storages[:n] } - // 收集设硬上限,避免启播被慢账号的转存/speedup 拖垮;超时的账号放弃、只用已完成的。 collectCtx, cancel := context.WithTimeout(ctx, multiSourceCollectTimeout()) defer cancel() - type result struct { - l *model.Link - accountID uint - elapsed time.Duration - } - results := make(chan result, len(storages)) + results := make(chan *model.Link, len(storages)) var wg sync.WaitGroup for _, st := range storages { uc, ok := st.(*quark.QuarkOrUC) @@ -1110,7 +1104,7 @@ var collectMultiSource = func(ctx context.Context, d *QuarkUCShare, file model.O if e == nil && l != nil && l.URL != "" { log.Infof("[multi-source] 账号 %d 取链完成 耗时=%v", uc.ID, elapsed) select { - case results <- result{l: l, accountID: uc.ID, elapsed: elapsed}: + case results <- l: case <-collectCtx.Done(): } } else { @@ -1120,23 +1114,32 @@ var collectMultiSource = func(ctx context.Context, d *QuarkUCShare, file model.O } go func() { wg.Wait(); close(results) }() - srcs := make([]model.LinkSource, 0, len(storages)) - seen := map[string]bool{} - if primary != nil && primary.URL != "" { - s := sourceFromLink(primary, d) - seen[s.URL] = true - srcs = append(srcs, s) - } - for r := range results { - s := sourceFromLink(r.l, d) - if seen[s.URL] { - continue + // 首个成功后,额外等 firstSourceGrace 收集更多源;之后或全部完成或总超时,立即返回。 + var links []*model.Link + afterFirst := time.NewTimer(multiSourceCollectTimeout()) // 占位,首源前不会触发 + afterFirst.Stop() + firstDone := false + for { + select { + case l, ok := <-results: + if !ok { + log.Infof("[multi-source] collect: driver=%s accounts=%d collected=%d 总耗时=%v(全部结束)", name, len(storages), len(links), time.Since(collectStart)) + return links + } + links = append(links, l) + if !firstDone { + firstDone = true + afterFirst.Reset(multiSourceFirstGrace()) + } + case <-afterFirst.C: + log.Infof("[multi-source] collect: driver=%s accounts=%d collected=%d 总耗时=%v(首源+%v窗口)", name, len(storages), len(links), time.Since(collectStart), multiSourceFirstGrace()) + cancel() // 停止仍在跑的慢账号 + return links + case <-collectCtx.Done(): + log.Infof("[multi-source] collect: driver=%s accounts=%d collected=%d 总耗时=%v(超时)", name, len(storages), len(links), time.Since(collectStart)) + return links } - seen[s.URL] = true - srcs = append(srcs, s) } - log.Infof("[multi-source] collect: driver=%s accounts=%d collected=%d 总耗时=%v", name, len(storages), len(srcs), time.Since(collectStart)) - return srcs } // multiSourceCollectTimeout 多账号取链的总超时,避免启播卡死。默认 8s。 @@ -1144,6 +1147,11 @@ func multiSourceCollectTimeout() time.Duration { return 8 * time.Second } +// multiSourceFirstGrace 首个账号取链成功后,额外等待收集更多源的窗口。默认 2s。 +func multiSourceFirstGrace() time.Duration { + return 2 * time.Second +} + // sourceFromLink 从单账号取链结果构造 LinkSource:去掉 #x-referer=raw 片段标记, // 保留 URL + Header(后端代理直接用 link.Header,不走魔法 Referer 片段契约)。 func sourceFromLink(l *model.Link, d *QuarkUCShare) model.LinkSource { @@ -1156,6 +1164,24 @@ func sourceFromLink(l *model.Link, d *QuarkUCShare) model.LinkSource { return model.LinkSource{URL: u, Header: l.Header} } +// sourcesFromLink 把多个账号取链结果转成去重的 LinkSource 列表(供 link.MultiSource)。 +func sourcesFromLinks(links []*model.Link, d *QuarkUCShare) []model.LinkSource { + srcs := make([]model.LinkSource, 0, len(links)) + seen := map[string]bool{} + for _, l := range links { + if l == nil { + continue + } + s := sourceFromLink(l, d) + if seen[s.URL] { + continue + } + seen[s.URL] = true + srcs = append(srcs, s) + } + return srcs +} + // masterCookie 取当前驱动类型主账号(master)的 Cookie。 // 夸克 share /file/download 需账号上下文(参考脚本 quarkRequestShareDownload 用 drive.fetch 带账号), // 匿名请求会失败 → 回退转存。故夸克取链需带主账号 Cookie;UC 匿名即可。无账号返回 ""。 From 3895bdf72f8241d839f1ee195e10d40b5658f41c Mon Sep 17 00:00:00 2001 From: Har01d Date: Mon, 3 Aug 2026 19:57:21 +0800 Subject: [PATCH 4/4] =?UTF-8?q?=E5=A4=B8=E5=85=8B=E5=A4=9A=E8=B4=A6?= =?UTF-8?q?=E5=8F=B7=E5=8A=A0=E9=80=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/model/args.go | 14 +++++++------- internal/stream/util.go | 10 ++++++++++ server/handles/down.go | 1 + server/handles/fsread.go | 16 +++++++++++----- 4 files changed, 29 insertions(+), 12 deletions(-) diff --git a/internal/model/args.go b/internal/model/args.go index 8f546591c..b0994da40 100644 --- a/internal/model/args.go +++ b/internal/model/args.go @@ -37,11 +37,11 @@ type Link struct { PartSize int `json:"part_size"` ContentLength int64 `json:"content_length"` // 转码视频、缩略图 - // MultiSource 用于多账号分片并行下载。非空且长度>1 时， - // Proxy 把分片按 序号 % len(MultiSource) 轮询分发到各源直连 - // (各账号 Cookie/Referer 独立),以叠加多账号带宽。 - // 空或单元素时维持单链行为。纯 Go 内部用,不对外序列化。 - MultiSource []LinkSource `json:"-"` + // MultiSource 用于多账号分片并行下载。非空且长度>1 时, + // 服务端 Proxy 把分片按 序号 % len(MultiSource) 轮询分发到各源直连; + // 客户端代理(/api/fs/get)也把它带给客户端(spider.jar 等),由客户端自行做多账号分片。 + // 各源 URL+Header(账号 Cookie/Referer)独立。空或单元素时维持单链行为。 + MultiSource []LinkSource `json:"multi_source,omitempty"` utils.SyncClosers `json:"-"` // 如果SyncClosers中的资源被关闭后Link将不可用，则此值应为 true @@ -50,8 +50,8 @@ type Link struct { // LinkSource 是多账号分片下载的一条源直连(URL+Header 独立)。 type LinkSource struct { - URL string - Header http.Header + URL string `json:"url"` + Header http.Header `json:"header,omitempty"` } func (l *Link) Clone() *Link { diff --git a/internal/stream/util.go b/internal/stream/util.go index 68c8b097d..46f02b795 100644 --- a/internal/stream/util.go +++ b/internal/stream/util.go @@ -10,6 +10,7 @@ import ( "net/http" "os" "sync" + "time" "github.com/OpenListTeam/OpenList/v4/internal/conf" "github.com/OpenListTeam/OpenList/v4/internal/errs" @@ -195,6 +196,7 @@ func newMultiSourceRangeReader(size int64, link *model.Link) model.RangeReaderIF } rangeReader := func(ctx context.Context, httpRange http_range.Range) (io.ReadCloser, error) { + rangeStart := time.Now() chunks := computeSubchunks(httpRange.Start, httpRange.Length) if len(chunks) == 0 { return io.NopCloser(bytes.NewReader(nil)), nil @@ -262,9 +264,11 @@ func newMultiSourceRangeReader(size int64, link *model.Link) model.RangeReaderIF ctx: ctx, slots: slots, cancel: cancel, + start: rangeStart, } // reader.Close 应等 worker 退出并取消残余请求。 reader.wg = &wg + log.Debugf("[multi-source] RangeRead 启动 start=%d chunks=%d", httpRange.Start, len(chunks)) return reader, nil } return RangeReaderFunc(rangeReader) @@ -285,6 +289,8 @@ type multiSourceReader struct { off int // 当前 subChunk 已读偏移 cancel context.CancelCauseFunc wg *sync.WaitGroup + start time.Time // RangeRead 启动时刻,用于首字节耗时 + logged bool // 是否已打首字节日志 } func (r *multiSourceReader) Read(p []byte) (int, error) { @@ -319,6 +325,10 @@ func (r *multiSourceReader) Read(p []byte) (int, error) { r.cur++ r.off = 0 } + if n > 0 && !r.logged { + r.logged = true + log.Debugf("[multi-source] 首字节 耗时=%v", time.Since(r.start)) + } return n, nil } } diff --git a/server/handles/down.go b/server/handles/down.go index 5cee22a7c..b5aa09e52 100644 --- a/server/handles/down.go +++ b/server/handles/down.go @@ -46,6 +46,7 @@ func Down(c *gin.Context) { func Proxy(c *gin.Context) { rawPath := c.Request.Context().Value(conf.PathKey).(string) + log.Debugf("[proxy] 收到代理请求 path=%s range=%s", rawPath, c.Request.Header.Get("Range")) filename := stdpath.Base(rawPath) storage, err := fs.GetStorage(rawPath, &fs.GetStoragesArgs{}) if err != nil { diff --git a/server/handles/fsread.go b/server/handles/fsread.go index ff212e024..eb8e2362f 100644 --- a/server/handles/fsread.go +++ b/server/handles/fsread.go @@ -261,6 +261,9 @@ type FsGetResp struct { Header string `json:"header"` Provider string `json:"provider"` Related []ObjResp `json:"related"` + // MultiSource 多账号分片下载的各源直连(URL+Header 独立),供客户端代理(spider.jar 等) + // 自行做多账号分片;为空时客户端回落单链 raw_url。仅非 WebProxy 直链分支填充。 + MultiSource []model.LinkSource `json:"multi_source,omitempty"` } func FsGetSplit(c *gin.Context) { @@ -306,6 +309,7 @@ func FsGet(c *gin.Context, req *FsGetReq, user *model.User) { return } var rawURL string + var multiSource []model.LinkSource storage, err := fs.GetStorage(reqPath, &fs.GetStoragesArgs{}) provider, ok := model.GetProvider(obj) @@ -346,6 +350,7 @@ func FsGet(c *gin.Context, req *FsGetReq, user *model.User) { } defer link.Close() rawURL = link.URL + multiSource = link.MultiSource } } } @@ -372,11 +377,12 @@ func FsGet(c *gin.Context, req *FsGetReq, user *model.User) { Thumb: thumb, MountDetails: mountDetails, }, - RawURL: rawURL, - Readme: getReadme(meta, reqPath), - Header: getHeader(meta, reqPath), - Provider: provider, - Related: toObjsResp(related, parentPath, isEncrypt(parentMeta, parentPath)), + RawURL: rawURL, + Readme: getReadme(meta, reqPath), + Header: getHeader(meta, reqPath), + Provider: provider, + Related: toObjsResp(related, parentPath, isEncrypt(parentMeta, parentPath)), + MultiSource: multiSource, }) }