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 concurrency 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), concurrency: settings.Concurrency, } if e.concurrency < 1 { e.concurrency = 3 } if e.concurrency > 8 { e.concurrency = 8 } return e } // Client 暴露给 app 层改配置用 func (e *Engine) Client() *eh.Client { return e.client } // SetConcurrency 运行时改并发 func (e *Engine) SetConcurrency(n int) { if n < 1 { n = 1 } if n > 8 { n = 8 } e.mu.Lock() e.concurrency = n e.mu.Unlock() } 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 ctx, cancel := context.WithCancel(ctx) e.cancel = cancel n := e.concurrency 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.runTask(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() } 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) // ① 画廊页(内部自动翻 ?p=N 分页收集全部页) gallery, err := e.client.FetchGalleryPage(ctx, task.GID, task.Token) if err != nil { if ctx.Err() != nil { e.markCanceled(task) return } e.fail(task, fmt.Errorf("解析画廊页失败: %w", err)) return } task.Title = gallery.Title task.TotalPages = gallery.TotalPages task.Dir = BuildDir(e.store.Settings().DirTemplate, gallery) task.Status = "downloading" e.save(task) // 目录按模板生成 root := e.store.Settings().DownloadRoot if root == "" { root = DefaultDownloadRoot() } 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 err := os.MkdirAll(dir, 0o755); err != nil { e.fail(task, fmt.Errorf("创建目录失败: %w", err)) return } e.save(task) // ② ③ 逐页:解析 URL → 下载 total := len(gallery.Pages) for _, p := range gallery.Pages { if ctx.Err() != nil { e.markCanceled(task) return } fname := targetName(p, task.GID, total) dest := filepath.Join(dir, fname) // 断点续传:文件在就算完成(文件系统即真相,状态文件丢了也能续) if st, err := os.Stat(dest); err == nil && st.Size() > 0 { task.Done++ task.Pages[strconv.Itoa(p.Page)] = "done" e.progress(task) continue } page := p.Page err := e.withRetry(ctx, func() error { imgURL, err := e.client.ResolveImageURL(ctx, p.SPath) if err != nil { return err } _, err = e.client.DownloadImage(ctx, imgURL, dest, nil) return err }) if err != nil { if ctx.Err() != nil || errors.Is(err, ErrCanceled) { e.markCanceled(task) return } task.Failed++ task.Pages[strconv.Itoa(page)] = "failed" task.FailedList = append(task.FailedList, FailedItem{ Page: page, File: fname, Error: err.Error(), }) e.progress(task) continue } task.Done++ task.Pages[strconv.Itoa(page)] = "done" e.progress(task) } if task.Failed > 0 { task.Status = "partial" } else { task.Status = "done" } e.save(task) } // 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) progress(task Task) { if e.sink != nil { e.sink.TaskUpdate(task) } } func (e *Engine) save(task Task) { task = e.store.UpsertTask(task) if e.sink != nil { e.sink.TaskUpdate(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, " ") }