package core // 下载引擎:任务编排 / 并发控制 / 指数退避重试 / 断点续传 // // 断点续传策略:以「文件系统为准」—— // // 目标文件已存在且 size > 0 → 视为已完成,直接跳过。 // // 因此即使任务状态 JSON 丢失或损坏,重跑也能正确续传。 // // ── 与参考项目(EH)的关键差异 ───────────────────────────────── // // ① 图片 URL 是解析期一次性拿全的(4KHD 的直链稳定,不像 EH 那样每页 // 都要现去 /s/ 换 hath 直链)。所以下载阶段只有「下载」一件事, // 没有 EH 那两步(ResolveImageURL → DownloadImage)。 // ② 因此「清单失效自愈」不再是必需的:清单里的 URL 和重新解析拿到的一样。 // 但保留重解析路径,用于「站点改图后作者补了新图」的情况。 import ( "context" "errors" "fmt" "math" "math/rand" "os" "path/filepath" "regexp" "strconv" "strings" "sync" "time" "hd4k-downloader/internal/hd4k" ) // ErrCanceled 取消标记(用户主动停止不算失败) var ErrCanceled = errors.New("已取消") // retryablePatterns 可重试的错误特征 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", "500", "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 *hd4k.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: hd4k.NewClient(time.Duration(settings.Timeout)*time.Millisecond, settings.Proxy, settings.ProxyMode, settings.Sites), 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() *hd4k.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 } 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.ID == t.ID { 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() } }() 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() } // safeRunTask worker 崩溃保护: // 把这个画廊标记为失败并记一条日志,而不是让整个应用无声消失。 func (e *Engine) safeRunTask(ctx context.Context, t Task) { defer func() { if r := recover(); r != nil { // 恢复代码本身也可能 panic,再包一层保证永不二次崩溃 func() { defer func() { _ = recover() }() e.log("error", fmt.Sprintf("画廊 %s 处理时发生内部错误:%v", t.Slug, r)) e.fail(t, fmt.Errorf("内部错误: %v", r)) }() } }() e.runTask(ctx, t) } type existingGallery struct { dir string manifest *Manifest } // reportParseProgress 解析阶段进度反馈 func (e *Engine) reportParseProgress(task Task, done, total, page int) { task.Status = "parsing" task.Total = total task.Done = 0 task.Pages = map[string]string{} if done > 0 { task.Pages["_parsing"] = strconv.Itoa(done) } e.save(task) } // findExistingGallery 在下载根目录里找这个画廊已存在的目录 func (e *Engine) findExistingGallery(root string, task Task) *existingGallery { // 1) 按清单里的 ID 精确匹配(最可靠,拿到目录绝对路径) if md := FindManifestByID(root, task.ID); 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 猜这个画廊可能的目录名 func (e *Engine) guessDirNames(task Task, meta *hd4k.GalleryMeta) []string { tpl := e.store.Settings().DirTemplate if tpl == "" { tpl = "{title}" } probe := hd4k.Gallery{ ID: task.ID, URL: task.URL, Host: task.Host, Path: task.Path, Slug: task.Slug, Title: task.Title, } if meta != nil { probe.Meta = *meta } var names []string if d := BuildDir(tpl, &probe); d != "" { names = append(names, d) } if task.Title != "" { if s := SanitizeSegment(task.Title, task.Slug); s != "" { names = append(names, s) } } return names } // refFromTask 从任务还原一个 GalleryRef func refFromTask(t Task) hd4k.GalleryRef { return hd4k.GalleryRef{ ID: t.ID, URL: t.URL, Host: t.Host, Path: t.Path, Slug: t.Slug, Raw: t.Raw, Title: t.Title, } } 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() } // 先看磁盘上有没有这个画廊的目录和清单。 // 4KHD 的图片直链是稳定的,存清单后**0 次请求**就能开始下载。 existing := e.findExistingGallery(root, task) var gallery *hd4k.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.Slug, len(g.Images))) } else if existing.manifest != nil { e.log("info", fmt.Sprintf( "画廊 %s 有目录但清单不含图片直链,仍需重新解析", task.Slug)) } } if gallery == nil { g, err := e.client.FetchGallery(ctx, refFromTask(task), func(done, total, page int) { 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 } if gallery.Title != "" { task.Title = gallery.Title } task.Total = gallery.Total task.Dir = BuildDir(e.store.Settings().DirTemplate, gallery) task.Status = "downloading" e.save(task) dir := filepath.Join(root, task.Dir) // 用户中途换了模板时,先看老的「标题目录」在不在 —— 在就继续往那儿下, // 否则换个模板就会把几千张图全部重下一遍。 legacyDir := filepath.Join(root, SanitizeSegment(gallery.Title, task.Slug)) if dir != legacyDir { if _, err := os.Stat(legacyDir); err == nil { e.log("info", "沿用旧目录(模板改过,已有图在老位置):"+gallery.Title) dir = legacyDir if rel, err := filepath.Rel(root, dir); err == nil { task.Dir = rel } } } // 清单里的目录名对不上时,以磁盘上实际存在的为准 if existing != nil && existing.dir != "" && existing.dir != dir { dir = existing.dir if rel, err := filepath.Rel(root, dir); 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, id: task.ID, host: task.Host, title: gallery.Title, total: len(gallery.Images), images: gallery.Images, meta: gallery.Meta, engine: e, } // 先快速扫描:已存在的直接算完成,剩下的才进队列 st.scanExisting() st.saveManifest() runImageWorkers(ctx, e, st, st.pending) // ---- 整轮扫尾:把刚失败的图再扫一遍 ---- if st.failedCount() > 0 && ctx.Err() == nil { e.log("info", fmt.Sprintf("画廊 %s:%d 张首轮失败,开始扫尾重试", task.Slug, st.failedCount())) retry := st.collectFailed() st.clearFailures() runImageWorkers(ctx, e, st, retry) if st.failedCount() > 0 { e.log("warn", fmt.Sprintf( "画廊 %s:扫尾后仍有 %d 张失败(换个线路再点「开始下载」可续传)", task.Slug, st.failedCount())) } else { e.log("ok", fmt.Sprintf("画廊 %s:扫尾重试全部成功", task.Slug)) } } // ---- 清单自愈 ---- // 作者可能给画廊补了新图。全部待下图都失败时,重新解析一次拿最新列表。 if usedManifest && ctx.Err() == nil && st.allFailed() { e.log("warn", fmt.Sprintf( "画廊 %s:本地清单的直链似乎已失效,重新解析画廊页", task.Slug)) if fresh, err := e.client.FetchGallery(ctx, refFromTask(task), func(done, total, page int) { e.reportParseProgress(task, done, total, page) }); err == nil { st.reset(fresh) st.scanExisting() st.saveManifest() runImageWorkers(ctx, e, st, st.collectAllPending()) if st.failedCount() == 0 { e.log("ok", fmt.Sprintf("画廊 %s:重新解析后全部下载完成", task.Slug)) } } else if ctx.Err() == nil { e.log("error", fmt.Sprintf("画廊 %s:重新解析失败:%v", task.Slug, 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) } // runImageWorkers 用并发下载给定图片(阻塞直到完成) func runImageWorkers(ctx context.Context, e *Engine, st *pageState, imgs []hd4k.ImageInfo) { if len(imgs) == 0 { return } n := e.PageConcurrency() if n > len(imgs) { n = len(imgs) } work := make(chan hd4k.ImageInfo, len(imgs)) for _, im := range imgs { work <- im } close(work) var wg sync.WaitGroup for i := 0; i < n; i++ { wg.Add(1) go func() { defer wg.Done() for im := range work { st.safeDownloadImage(ctx, im) } }() } wg.Wait() } // pageState 是「一个画廊的图片级下载状态」,所有字段都由 mutex 保护, // 因为图片级 worker 是并发的。 type pageState struct { mu sync.Mutex engine *Engine task *Task dir string id string host string title string total int images []hd4k.ImageInfo pending []hd4k.ImageInfo meta hd4k.GalleryMeta done map[int]bool failed map[int]string failedL []FailedItem } func (s *pageState) reset(g *hd4k.Gallery) { s.mu.Lock() defer s.mu.Unlock() s.images = g.Images s.pending = nil s.total = len(g.Images) 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 } func (s *pageState) collectAllPending() []hd4k.ImageInfo { s.mu.Lock() defer s.mu.Unlock() var out []hd4k.ImageInfo for _, im := range s.images { if !s.done[im.Index] { out = append(out, im) } } 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 s.pending = nil if s.task.Pages == nil { s.task.Pages = map[string]string{} } for _, im := range s.images { fname := targetName(im, s.id, s.total) dest := filepath.Join(s.dir, fname) if st, err := os.Stat(dest); err == nil && st.Size() > 0 { s.done[im.Index] = true s.task.Pages[strconv.Itoa(im.Index)] = "done" } else { s.pending = append(s.pending, im) } } s.task.Done = len(s.done) } // safeDownloadImage 包一层 recover:单张图的意外 panic 不该带走整个 worker。 func (s *pageState) safeDownloadImage(ctx context.Context, im hd4k.ImageInfo) { defer func() { if r := recover(); r != nil { fname := targetName(im, s.id, s.total) s.mu.Lock() s.failed[im.Index] = fmt.Sprintf("内部错误: %v", r) s.failedL = append(s.failedL, FailedItem{ Page: im.Index, File: fname, Error: fmt.Sprintf("内部错误: %v", r), }) s.task.Pages[strconv.Itoa(im.Index)] = "failed" s.syncCountsLocked() snap := s.snapshotLocked() s.mu.Unlock() s.engine.emitTask(snap) } }() s.downloadImage(ctx, im) } // downloadImage 下载单张图,内部含 4 次退避重试 func (s *pageState) downloadImage(ctx context.Context, im hd4k.ImageInfo) { if ctx.Err() != nil { return } fname := targetName(im, s.id, s.total) dest := filepath.Join(s.dir, fname) // Referer 取画廊自身的 origin(用户可能配了 http 或带端口的站点, // 硬拼 "https://"+host 会对不上) referer := hd4k.OriginOf(s.task.URL) + "/" err := s.engine.withRetry(ctx, func() error { _, e := s.engine.client.DownloadImage(ctx, im.URL, dest, referer, nil) return e }) s.mu.Lock() k := strconv.Itoa(im.Index) if err != nil { if ctx.Err() != nil || errors.Is(err, ErrCanceled) { s.mu.Unlock() s.markCanceled() return } s.failed[im.Index] = err.Error() s.failedL = append(s.failedL, FailedItem{Page: im.Index, File: fname, Error: err.Error()}) s.task.Pages[k] = "failed" } else { s.done[im.Index] = 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(进程闪退)。 func (s *pageState) snapshotLocked() Task { s.syncCountsLocked() t := *s.task pages := make(map[string]string, len(s.task.Pages)) for k, v := range s.task.Pages { pages[k] = v } t.Pages = pages 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{} } func (s *pageState) collectFailed() []hd4k.ImageInfo { s.mu.Lock() defer s.mu.Unlock() failed := intSet(keysOf(s.failed)) var out []hd4k.ImageInfo for _, im := range s.images { if failed[im.Index] { out = append(out, im) } } return out } func (s *pageState) markCanceled() { s.engine.markCanceled(s.snapshot()) } // saveManifest 写画廊目录清单。 // // 4KHD 的图片直链是 Jetpack Photon 的哈希路径,完全稳定,存下来随时能用 —— // 下次重开程序,读这份清单就能 0 请求直接下载。 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.images)) for _, im := range s.images { k := strconv.Itoa(im.Index) errMsg := "" if msg, bad := s.failed[im.Index]; bad { errMsg = msg } pageMap[k] = ManifestPage{ Page: im.Index, File: targetName(im, s.id, s.total), URL: im.URL, Finished: s.done[im.Index], Error: errMsg, } } s.mu.Unlock() WriteManifest(s.dir, Manifest{ ID: s.id, URL: t.URL, Host: s.host, Path: t.Path, Slug: t.Slug, Title: s.title, TotalPages: 0, Total: 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 图片文件名。 // // 4KHD 的图片本身带原名和序号(tbox-vol52-4khd.com-001.webp), // 优先用它 —— 可读且天然有序。只有原名缺失时才用页码补零兜底。 func targetName(im hd4k.ImageInfo, id string, total int) string { if im.Filename != "" { return SanitizeFilename(im.Filename, fallbackName(im, id, total)) } return fallbackName(im, id, total) } func fallbackName(im hd4k.ImageInfo, id string, total int) string { width := len(strconv.Itoa(total)) n := strconv.Itoa(im.Index) for len(n) < width { n = "0" + n } short := id if i := strings.LastIndex(short, "/"); i >= 0 { short = short[i+1:] } if j := strings.Index(short, "."); j > 0 { short = short[:j] } if short == "" { short = "image" } return SanitizeFilename(short+"-"+n+".webp", n+".webp") } // withRetry 指数退避重试 // // ⚠️ 限流(429/503/403)走另一条路:触发**全局冷却**,让所有 worker // 一起停,而不是各自退避。原因见 Limiter.Cooldown 的注释 —— // 40 个 worker 错峰重试会持续往已限流的服务器打请求,永远缓不过来。 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 } // 限流:全局冷却 + 更长退避,不再各退各的 if hd4k.IsRateLimited(err) { wait := e.rateLimitBackoff(err, i) e.log("warn", fmt.Sprintf("触发限流,全部 worker 冷却 %s(第 %d 次):%v", wait.Round(time.Second), i+1, err)) e.client.Cooldown(wait) select { case <-ctx.Done(): return ErrCanceled case <-time.After(wait): } continue } 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 } // rateLimitBackoff 算限流后的等待时长。 // // 优先用服务器给的 Retry-After(它最清楚该等多久),否则按尝试次数指数退避。 // 下限 10s 是因为实测 CDN 的限流窗口通常以十几秒计,退避太短只会一直撞墙。 func (e *Engine) rateLimitBackoff(err error, attempt int) time.Duration { var se *hd4k.StatusError if errors.As(err, &se) && se.RetryAfter > 0 { // 服务器明说了,就按它的来(再加一点余量,避免卡着点又撞上) d := se.RetryAfter + time.Duration(rand.Intn(2000))*time.Millisecond if d > 10*time.Minute { d = 10 * time.Minute } return d } wait := time.Duration(10< 5*time.Minute { wait = 5 * time.Minute } return wait + time.Duration(rand.Intn(2000))*time.Millisecond } // 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.Slug)) } 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.Slug, 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) } } var reSpaced = regexp.MustCompile(`\s+`) func cleanSpaces(s string) string { return reSpaced.ReplaceAllString(s, " ") }