package core // 下载引擎:任务编排 / 并发控制 / 指数退避重试 / 断点续传 // // 断点续传策略:以「文件系统为准」—— // // 目标文件已存在且 size > 0 → 视为已完成,直接跳过。 // // 因此即使任务状态 JSON 丢失或损坏,重跑也能正确续传。 import ( "context" "errors" "fmt" "math" "math/rand" "os" "path/filepath" "regexp" "strconv" "strings" "sync" "time" "eh-downloader-wails/internal/eh" ) // ErrCanceled 取消标记(用户主动停止不算失败) var ErrCanceled = errors.New("已取消") // retryableErrs 可重试的错误特征(对应 JS 版 RETRYABLE 集合) var retryablePatterns = []string{ "connection reset", "connection refused", "timeout", "i/o timeout", "broken pipe", "eof", "no such host", "tls handshake", "unexpected status", "server closed", "too many redirects", "429", "502", "503", "504", "socket hang up", } // EventSink 是引擎向 UI 发事件的方式(由 app 层注入) type EventSink interface { TaskUpdate(t Task) Log(level, msg string) Done() } // Engine 下载引擎 type Engine struct { store *Store client *eh.Client sink EventSink mu sync.Mutex galleryWorkers int pageConcurrency int running bool cancel context.CancelFunc queue []Task } // NewEngine 建引擎 func NewEngine(store *Store, sink EventSink, settings Settings) *Engine { e := &Engine{ store: store, sink: sink, client: eh.NewClient(time.Duration(settings.Timeout)*time.Millisecond, settings.Proxy, settings.ProxyMode, settings.Xres), galleryWorkers: clampInt(settings.GalleryConcurrency, 1, 20, 5), pageConcurrency: clampInt(settings.PageConcurrency, 1, 20, 8), } e.client.SetRequestInterval(time.Duration(settings.RequestIntervalMs) * time.Millisecond) return e } // clampInt 把值夹到 [lo, hi]。0 或负数视为「没设置」,用默认值。 func clampInt(v, lo, hi, def int) int { if v <= 0 { return def } if v < lo { return lo } if v > hi { return hi } return v } // Client 暴露给 app 层改配置用 func (e *Engine) Client() *eh.Client { return e.client } // SetConcurrency 运行时改并发 func (e *Engine) SetConcurrency(n int) { e.mu.Lock() e.galleryWorkers = clampInt(n, 1, 20, 5) e.mu.Unlock() } // SetGalleryConcurrency 改画廊并发 func (e *Engine) SetGalleryConcurrency(n int) { e.SetConcurrency(n) } // SetPageConcurrency 改单画廊页并发 func (e *Engine) SetPageConcurrency(n int) { e.mu.Lock() e.pageConcurrency = clampInt(n, 1, 20, 8) e.mu.Unlock() } // SetRequestInterval 改全局请求间隔(毫秒) func (e *Engine) SetRequestInterval(ms int) { if ms < 0 { ms = 0 } e.client.SetRequestInterval(time.Duration(ms) * time.Millisecond) } // GalleryConcurrency 当前画廊并发 func (e *Engine) GalleryConcurrency() int { e.mu.Lock() defer e.mu.Unlock() return e.galleryWorkers } // PageConcurrency 当前页并发 func (e *Engine) PageConcurrency() int { e.mu.Lock() defer e.mu.Unlock() return e.pageConcurrency } // emitTask 把任务状态推给 UI func (e *Engine) emitTask(t Task) { if e.sink != nil { e.sink.TaskUpdate(t) } } func (e *Engine) IsRunning() bool { e.mu.Lock() defer e.mu.Unlock() return e.running } func (e *Engine) QueueLength() int { e.mu.Lock() defer e.mu.Unlock() return len(e.queue) } // Stop 停止:取消 context,立刻打断在途请求 func (e *Engine) Stop() { e.mu.Lock() if e.cancel != nil { e.cancel() e.cancel = nil } e.mu.Unlock() } // Enqueue 把任务加入队列(已排队的跳过) func (e *Engine) Enqueue(tasks []Task) { e.mu.Lock() defer e.mu.Unlock() for _, t := range tasks { dup := false for _, q := range e.queue { if q.GID == t.GID && q.Token == t.Token { dup = true break } } if !dup { e.queue = append(e.queue, t) } } } // Start 开始跑队列(阻塞直到队列空或被取消) func (e *Engine) Start(ctx context.Context) { e.mu.Lock() if e.running { e.mu.Unlock() return } e.running = true e.mu.Unlock() ctx, cancel := context.WithCancel(ctx) e.mu.Lock() e.cancel = cancel n := e.galleryWorkers e.mu.Unlock() defer func() { cancel() e.mu.Lock() e.running = false e.cancel = nil e.queue = nil e.mu.Unlock() if e.sink != nil { e.sink.Done() } }() // 并发 worker 共享一个 channel 队列 work := make(chan Task) var wg sync.WaitGroup for i := 0; i < n; i++ { wg.Add(1) go func() { defer wg.Done() for { select { case <-ctx.Done(): return case t, ok := <-work: if !ok { return } e.safeRunTask(ctx, t) } } }() } // 派发 go func() { defer close(work) for { e.mu.Lock() if len(e.queue) == 0 { e.mu.Unlock() return } t := e.queue[0] e.queue = e.queue[1:] e.mu.Unlock() select { case <-ctx.Done(): return case work <- t: } } }() wg.Wait() } // worker 崩溃保护:Go 里普通 panic 可以 recover,但 runtime 的 // fatal error(并发 map 读写这类)不行 —— 那会直接杀进程。 // // 这里 recover 的目的是「万一以后又有别处的 bug 导致 panic」时, // 把这个画廊标记为失败、记一条日志,而不是让整个应用无声消失。 // 用户至少能看到「哪个画廊出的问题」,已下载的文件也不会受影响。 func (e *Engine) safeRunTask(ctx context.Context, t Task) { defer func() { if r := recover(); r != nil { // 恢复代码本身也可能 panic(比如 panic 的根因就是 store 出问题), // 所以这里再包一层,保证 recover 逻辑永远不会二次崩溃。 func() { defer func() { _ = recover() }() e.log("error", fmt.Sprintf("画廊 %s 处理时发生内部错误:%v", t.GID, r)) e.fail(t, fmt.Errorf("内部错误: %v", r)) }() } }() e.runTask(ctx, t) } // existingGallery 是磁盘上已存在的画廊信息 type existingGallery struct { dir string manifest *Manifest } // reportParseProgress 解析阶段进度反馈。 // // 解析大画廊要翻上百屏(每屏 20 张),全程可能好几分钟。 // 界面必须看得到在动,否则「解析中 0/?」会被当成卡死。 func (e *Engine) reportParseProgress(task Task, done, total, page int) { task.Status = "parsing" task.TotalPages = total task.Done = 0 // 解析期用「已解析张数」当进度,下载阶段才是真正的完成数 task.Pages = map[string]string{} if done > 0 { task.Pages["_parsing"] = strconv.Itoa(done) } e.save(task) } // findExistingGallery 在下载根目录里找这个 gid 已存在的画廊。 // // 两种匹配方式: // 1. 按 .ehdl.json 清单里的 gid 精确匹配(最可靠,拿到目录绝对路径) // 2. 没有清单时,退而按「模板算出的目录名」或「旧版标题目录名」猜 // // 找到就返回,下载时可直接复用清单里的 /s/ 路径,省掉画廊页解析。 func (e *Engine) findExistingGallery(root string, task Task) *existingGallery { // 1) 按清单里的 gid 匹配 if md := FindManifestByGID(root, task.GID); md != nil { return &existingGallery{dir: md.Dir, manifest: md.Manifest} } // 2) 没有清单:按可能的目录名猜(换个模板/旧版标题目录的情况) for _, name := range e.guessDirNames(task, nil) { p := filepath.Join(root, name) if st, err := os.Stat(p); err == nil && st.IsDir() { m, _ := ReadManifest(p) return &existingGallery{dir: p, manifest: m} } } return nil } // guessDirNames 猜这个画廊可能的目录名(当前模板算出来的 + 旧版标题目录) // // meta 优先用已找到的清单里那份(那是上次解析时拿到的真值), // 没有清单时才用任务标题试算。 func (e *Engine) guessDirNames(task Task, meta *eh.GalleryMeta) []string { tpl := e.store.Settings().DirTemplate if tpl == "" { tpl = "{title}" } probe := eh.Gallery{GID: task.GID, Title: task.Title} if meta != nil { probe.Meta = *meta } if probe.Title == "" { probe.Title = task.GID } var names []string if d := BuildDir(tpl, &probe); d != "" { names = append(names, d) } // 旧版:只按标题建目录 if task.Title != "" { if s := SanitizeSegment(task.Title, task.GID); s != "" { names = append(names, s) } } return names } func (e *Engine) runTask(ctx context.Context, task Task) { task.Status = "running" task.Error = "" task.Done = 0 task.Failed = 0 task.Pages = map[string]string{} task.FailedList = []FailedItem{} e.save(task) root := e.store.Settings().DownloadRoot if root == "" { root = DefaultDownloadRoot() } // ------------------------------------------------------------ // 先看磁盘上有没有这个画廊的目录和清单。 // 清单存在 => 之前解析过了,直接拿来下载,省掉 N 次画廊页分页请求。 // ------------------------------------------------------------ existing := e.findExistingGallery(root, task) var gallery *eh.Gallery usedManifest := false if existing != nil { if g, ok := existing.manifest.GalleryFromManifest(); ok { gallery = g usedManifest = true e.log("info", fmt.Sprintf( "画廊 %s 用本地清单(%d 页),跳过重新解析", task.GID, len(g.Pages))) } else { // 目录在、清单是旧版本(v1,没有 /s/ 路径)—— 仍需解析 e.log("info", fmt.Sprintf( "画廊 %s 有目录但清单不含页面路径,仍需重新解析", task.GID)) } } // 没有可用清单(或清单太旧)→ 正常解析画廊页 if gallery == nil { g, err := e.client.FetchGalleryPage(ctx, task.GID, task.Token, func(done, total, page int) { // 解析阶段也要有反馈,否则大画廊(上千张 = 上百屏) // 会长时间显示「解析中 0/?」,看着像卡死。 e.reportParseProgress(task, done, total, page) }) if err != nil { if ctx.Err() != nil { e.markCanceled(task) return } e.fail(task, fmt.Errorf("解析画廊页失败: %w", err)) return } gallery = g } task.Title = gallery.Title task.TotalPages = gallery.TotalPages task.Dir = BuildDir(e.store.Settings().DirTemplate, gallery) task.Status = "downloading" e.save(task) // 目录按模板生成 dir := filepath.Join(root, task.Dir) // 老版本一律用「标题」当目录名。用户中途换了模板时, // 先看老的「标题目录」在不在 —— 在就继续往那儿下, // 否则换个模板就会把几千张图全部重下一遍。 // ⚠️ 必须在 mkdir 之前判断,否则会白建出一个新目录。 legacyDir := filepath.Join(root, SanitizeSegment(gallery.Title, task.GID)) if dir != legacyDir { if _, err := os.Stat(legacyDir); err == nil { e.log("info", "沿用旧目录(模板改过,已有图在老位置):"+gallery.Title) dir = legacyDir rel, err := filepath.Rel(root, dir) if err == nil { task.Dir = rel } } } // 清单里的目录名对不上时(比如换了模板),以磁盘上实际存在的为准 if existing != nil && existing.dir != "" && existing.dir != dir { dir = existing.dir rel, err := filepath.Rel(root, dir) if err == nil { task.Dir = rel } } if err := os.MkdirAll(dir, 0o755); err != nil { e.fail(task, fmt.Errorf("创建目录失败: %w", err)) return } e.save(task) // ② ③ 页级并发下载 st := &pageState{ task: &task, dir: dir, pages: gallery.Pages, gid: task.GID, token: task.Token, title: gallery.Title, total: len(gallery.Pages), meta: gallery.Meta, engine: e, } // 先做一遍快速扫描:已存在的直接算完成,剩下的才进队列 st.scanExisting() st.saveManifest() // 页级 worker 池 runPageWorkers(ctx, e, st, st.pending) // ---- 整轮扫尾:把刚失败的页再扫一遍 ---- if st.failedCount() > 0 && ctx.Err() == nil { e.log("info", fmt.Sprintf("画廊 %s:%d 页首轮失败,开始扫尾重试", task.GID, st.failedCount())) retry := st.collectFailed() // 扫尾时清掉 failedList,让结果反映最终状态 st.clearFailures() runPageWorkers(ctx, e, st, retry) if st.failedCount() > 0 { e.log("warn", fmt.Sprintf( "画廊 %s:扫尾后仍有 %d 页失败(换个线路再点「开始下载」可续传)", task.GID, st.failedCount())) } else { e.log("ok", fmt.Sprintf("画廊 %s:扫尾重试全部成功", task.GID)) } } // ---- 清单失效自愈 ---- // 如果这次是靠清单跳过解析的,但所有待下页几乎全失败, // 说明清单里的 /s/ 路径已经过期(EH 换了 key,或作者更新了画廊)。 // 这时重新解析一次画廊页,拿到新的 /s/ 路径再试。 if usedManifest && ctx.Err() == nil && st.allFailed() { e.log("warn", fmt.Sprintf( "画廊 %s:本地清单的页面路径似乎已失效,重新解析画廊页", task.GID)) if fresh, err := e.client.FetchGalleryPage(ctx, task.GID, task.Token, func(done, total, page int) { e.reportParseProgress(task, done, total, page) }); err == nil { // 用新解析结果重跑:重置页状态 st.reset(fresh) st.scanExisting() st.saveManifest() retry := st.collectAllPending() runPageWorkers(ctx, e, st, retry) if st.failedCount() == 0 { e.log("ok", fmt.Sprintf("画廊 %s:重新解析后全部下载完成", task.GID)) } } else if ctx.Err() == nil { e.log("error", fmt.Sprintf("画廊 %s:重新解析失败:%v", task.GID, err)) } } st.saveManifest() if ctx.Err() != nil { e.markCanceled(task) return } t := st.snapshot() if t.Failed > 0 { t.Status = "partial" } else { t.Status = "done" } e.save(t) } // runPageWorkers 用页级并发下载给定页面(阻塞直到完成) func runPageWorkers(ctx context.Context, e *Engine, st *pageState, pages []eh.GalleryPageInfo) { if len(pages) == 0 { return } n := e.PageConcurrency() if n > len(pages) { n = len(pages) } work := make(chan eh.GalleryPageInfo, len(pages)) for _, p := range pages { work <- p } close(work) var wg sync.WaitGroup for i := 0; i < n; i++ { wg.Add(1) go func() { defer wg.Done() for p := range work { st.safeDownloadPage(ctx, p) } }() } wg.Wait() } // pageState 是「一个画廊的页级下载状态」,所有字段都由 mutex 保护, // 因为页级 worker 是并发的。 type pageState struct { mu sync.Mutex engine *Engine task *Task dir string gid string token string title string total int pages []eh.GalleryPageInfo pending []eh.GalleryPageInfo meta eh.GalleryMeta done map[int]bool failed map[int]string // 页码 -> 错误 failedL []FailedItem } // reset 用新的解析结果重置页状态(清单失效后重新解析时用) func (s *pageState) reset(g *eh.Gallery) { s.mu.Lock() defer s.mu.Unlock() s.pages = g.Pages s.pending = nil s.total = len(g.Pages) s.title = g.Title s.meta = g.Meta } // allFailed 判断是否「有待下页面且全部失败」—— 用来识别清单过期 func (s *pageState) allFailed() bool { s.mu.Lock() defer s.mu.Unlock() if len(s.pending) == 0 { return false // 没有待下页面,不是清单的问题 } // 只下了一页而它失败了不算(可能只是单页问题) if len(s.pending) == 1 && len(s.done) > 0 { return false } return len(s.done) == 0 && len(s.failed) > 0 } // collectAllPending 取出所有还没成功的页 func (s *pageState) collectAllPending() []eh.GalleryPageInfo { s.mu.Lock() defer s.mu.Unlock() var out []eh.GalleryPageInfo for _, p := range s.pages { if !s.done[p.Page] { out = append(out, p) } } return out } // scanExisting 扫描已有文件(断点续传的第一依据:文件在且非空 = 完成) func (s *pageState) scanExisting() { s.mu.Lock() defer s.mu.Unlock() s.done = map[int]bool{} s.failed = map[int]string{} s.failedL = nil if s.task.Pages == nil { s.task.Pages = map[string]string{} } for _, p := range s.pages { fname := targetName(p, s.gid, s.total) dest := filepath.Join(s.dir, fname) if st, err := os.Stat(dest); err == nil && st.Size() > 0 { s.done[p.Page] = true s.task.Pages[strconv.Itoa(p.Page)] = "done" } else { s.pending = append(s.pending, p) } } s.task.Done = len(s.done) } // safeDownloadPage 包一层 recover:某一页的意外 panic 不该带走整个 worker。 func (s *pageState) safeDownloadPage(ctx context.Context, p eh.GalleryPageInfo) { defer func() { if r := recover(); r != nil { fname := targetName(p, s.gid, s.total) s.mu.Lock() s.failed[p.Page] = fmt.Sprintf("内部错误: %v", r) s.failedL = append(s.failedL, FailedItem{ Page: p.Page, File: fname, Error: fmt.Sprintf("内部错误: %v", r), }) s.task.Pages[strconv.Itoa(p.Page)] = "failed" s.syncCountsLocked() snap := s.snapshotLocked() s.mu.Unlock() s.engine.emitTask(snap) } }() s.downloadPage(ctx, p) } // downloadPage 下单页(解析 URL → 下载),内部含 4 次退避重试 func (s *pageState) downloadPage(ctx context.Context, p eh.GalleryPageInfo) { if ctx.Err() != nil { return } fname := targetName(p, s.gid, s.total) dest := filepath.Join(s.dir, fname) err := s.engine.withRetry(ctx, func() error { imgURL, e := s.engine.client.ResolveImageURL(ctx, p.SPath) if e != nil { return e } _, e = s.engine.client.DownloadImage(ctx, imgURL, dest, nil) return e }) s.mu.Lock() k := strconv.Itoa(p.Page) if err != nil { if ctx.Err() != nil || errors.Is(err, ErrCanceled) { s.mu.Unlock() s.markCanceled() return } s.failed[p.Page] = err.Error() s.failedL = append(s.failedL, FailedItem{Page: p.Page, File: fname, Error: err.Error()}) s.task.Pages[k] = "failed" } else { s.done[p.Page] = true s.task.Pages[k] = "done" } s.syncCountsLocked() snap := s.snapshotLocked() s.mu.Unlock() s.engine.emitTask(snap) } func (s *pageState) syncCountsLocked() { s.task.Done = len(s.done) s.task.Failed = len(s.failed) s.task.FailedList = append([]FailedItem(nil), s.failedL...) } // snapshotLocked 生成任务快照。 // // ⚠️ 必须深拷贝 Pages / FailedList —— 这两个是 map 和 slice, // 浅拷贝会把底层内存一起交出去。而快照是要被送到 UI 做 JSON 序列化的, // 其他页 worker 此刻还在写同一份数据 —— 共享内存会导致 // fatal error: concurrent map read and map write(不可 recover,整个进程闪退)。 // 症状:图片多的画廊 + 高页并发时偶发,因为并发写窗口更长。 func (s *pageState) snapshotLocked() Task { s.syncCountsLocked() t := *s.task // 深拷贝 map pages := make(map[string]string, len(s.task.Pages)) for k, v := range s.task.Pages { pages[k] = v } t.Pages = pages // 深拷贝 slice t.FailedList = append([]FailedItem(nil), s.task.FailedList...) return t } func (s *pageState) snapshot() Task { s.mu.Lock() defer s.mu.Unlock() return s.snapshotLocked() } func (s *pageState) failedCount() int { s.mu.Lock() defer s.mu.Unlock() return len(s.failed) } func (s *pageState) clearFailures() { s.mu.Lock() defer s.mu.Unlock() s.failed = map[int]string{} s.failedL = nil s.task.Failed = 0 s.task.FailedList = []FailedItem{} } // collectFailed 收集失败页(清空失败表前先取出) func (s *pageState) collectFailed() []eh.GalleryPageInfo { s.mu.Lock() defer s.mu.Unlock() failed := intSet(keysOf(s.failed)) var out []eh.GalleryPageInfo for _, p := range s.pages { if failed[p.Page] { out = append(out, p) } } return out } // markCanceled 任一页发现取消就整体标记 func (s *pageState) markCanceled() { s.engine.markCanceled(s.snapshot()) } // saveManifest 写画廊目录清单。 // // 存的是每页的 /s/ 路径(稳定),不是 hath 直链(随机子域+随机端口+时间戳签名, // 每次都变、存了也没用)。下次重开程序时,读这份清单就能跳过画廊页解析。 func (s *pageState) saveManifest() { s.mu.Lock() done := keysOf(s.done) failed := keysOf(s.failed) t := *s.task meta := s.meta pageMap := make(map[string]ManifestPage, len(s.pages)) for _, p := range s.pages { k := strconv.Itoa(p.Page) errMsg := "" if msg, bad := s.failed[p.Page]; bad { errMsg = msg } pageMap[k] = ManifestPage{ Page: p.Page, File: targetName(p, s.gid, s.total), SPath: p.SPath, Finished: s.done[p.Page], Error: errMsg, } } s.mu.Unlock() WriteManifest(s.dir, Manifest{ GID: s.gid, Token: s.token, Title: s.title, TotalPages: s.total, DonePages: done, FailedPage: failed, PageMap: pageMap, Meta: &meta, CreatedAt: t.CreatedAt, }) } func keysOf[V any](m map[int]V) []int { out := make([]int, 0, len(m)) for k := range m { out = append(out, k) } return out } // targetName 页码补零,保证排序 func targetName(p eh.GalleryPageInfo, gid string, total int) string { width := len(strconv.Itoa(total)) n := strconv.Itoa(p.Page) for len(n) < width { n = "0" + n } return SanitizeFilename(p.Filename, gid+"-"+n+".webp") } // withRetry 指数退避重试 func (e *Engine) withRetry(ctx context.Context, fn func() error) error { const ( tries = 4 base = 800 * time.Millisecond max = 15 * time.Second ) var lastErr error for i := 0; i < tries; i++ { if ctx.Err() != nil { return ErrCanceled } err := fn() if err == nil { return nil } if ctx.Err() != nil { return ErrCanceled } lastErr = err if i == tries-1 || !isRetryable(err) { return err } wait := time.Duration(float64(base) * math.Pow(2, float64(i))) if wait > max { wait = max } wait += time.Duration(rand.Intn(400)) * time.Millisecond e.log("warn", fmt.Sprintf("第 %d 次失败,%dms 后重试:%v", i+1, wait.Milliseconds(), err)) select { case <-ctx.Done(): return ErrCanceled case <-time.After(wait): } } return lastErr } // isRetryable 判断是否值得重试 func isRetryable(err error) bool { if err == nil { return false } if errors.Is(err, ErrCanceled) || errors.Is(err, context.Canceled) { return false } s := strings.ToLower(err.Error()) for _, p := range retryablePatterns { if strings.Contains(s, p) { return true } } return false } // markCanceled 用户主动取消:退回排队,等续传 func (e *Engine) markCanceled(task Task) { task.Status = "queued" task.Error = "" e.save(task) e.log("warn", fmt.Sprintf("画廊 %s 已取消,再点开始可继续", task.GID)) } func (e *Engine) fail(task Task, err error) { msg := err.Error() task.Status = "failed" task.Error = msg e.save(task) e.log("error", fmt.Sprintf("画廊 %s 失败:%s", task.GID, msg)) } func (e *Engine) save(task Task) { task = e.store.UpsertTask(task) e.emitTask(task) } func (e *Engine) log(level, msg string) { if e.sink != nil { e.sink.Log(level, msg) } } // reSpaced 去掉路径里的多余空白(清洗用) var reSpaced = regexp.MustCompile(`\s+`) func cleanSpaces(s string) string { return reSpaced.ReplaceAllString(s, " ") }